Upload 7 files

tests/api/test_auth.py

@@ -0,0 +1,74 @@
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.core.security import create_access_token
+
+pytestmark = pytest.mark.asyncio
+
+async def test_login(client: AsyncClient, db_session: AsyncSession):
+    # First register a test user
+    register_response = await client.post(
+        "/api/v1/auth/register",
+        data={
+            "username": "testuser@example.com",
+            "password": "testpass123"
+        }
+    )
+    assert register_response.status_code == 200
+    
+    # Test login with correct credentials
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={
+            "username": "testuser@example.com",
+            "password": "testpass123"
+        }
+    )
+    assert response.status_code == 200
+    assert "access_token" in response.json()
+    assert response.json()["token_type"] == "bearer"
+
+    # Test login with incorrect password
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={
+            "username": "testuser@example.com",
+            "password": "wrongpass"
+        }
+    )
+    assert response.status_code == 401
+
+    # Test login with non-existent user
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={
+            "username": "nonexistent@example.com",
+            "password": "testpass123"
+        }
+    )
+    assert response.status_code == 401
+
+async def test_register(client: AsyncClient, db_session: AsyncSession):
+    # Test successful registration
+    response = await client.post(
+        "/api/v1/auth/register",
+        data={
+            "username": "newuser@example.com",
+            "password": "newpass123"
+        }
+    )
+    assert response.status_code == 200
+    user_data = response.json()
+    assert user_data["email"] == "newuser@example.com"
+    assert "hashed_password" not in user_data
+
+    # Test registration with existing email
+    response = await client.post(
+        "/api/v1/auth/register",
+        data={
+            "username": "newuser@example.com",
+            "password": "anotherpass"
+        }
+    )
+    assert response.status_code == 400
+    assert "Email already registered" in response.json()["detail"]

tests/api/test_branches.py

@@ -0,0 +1,154 @@
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+from ..utils import create_user_and_login, create_branch
+
+pytestmark = pytest.mark.asyncio
+
+@pytest.fixture
+async def admin_token(client: AsyncClient) -> str:
+    """Create an admin user and return their token"""
+    user_data = await create_user_and_login(
+        client,
+        email="admin@example.com",
+        password="adminpass123",
+        full_name="Admin User",
+        is_superuser=True
+    )
+    return user_data["token"]
+
+@pytest.fixture
+async def regular_token(client: AsyncClient) -> str:
+    """Create a regular user and return their token"""
+    user_data = await create_user_and_login(
+        client,
+        email="user@example.com",
+        password="userpass123",
+        full_name="Regular User",
+        is_superuser=False
+    )
+    return user_data["token"]
+
+async def test_create_branch(client: AsyncClient, admin_token: str):
+    # Test creating a branch as admin
+    response = await client.post(
+        "/api/v1/branches/",
+        json={
+            "name": "Test Branch",
+            "location": "Test Location",
+            "contact_email": "branch@example.com",
+            "phone": "1234567890"
+        },
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["name"] == "Test Branch"
+    assert data["location"] == "Test Location"
+    return data["id"]
+
+async def test_create_branch_unauthorized(client: AsyncClient, regular_token: str):
+    # Test creating a branch as regular user (should fail)
+    response = await client.post(
+        "/api/v1/branches/",
+        json={
+            "name": "Test Branch",
+            "location": "Test Location"
+        },
+        headers={"Authorization": f"Bearer {regular_token}"}
+    )
+    assert response.status_code == 403
+
+async def test_get_branch(client: AsyncClient, admin_token: str):
+    # First create a branch
+    branch_id = await test_create_branch(client, admin_token)
+    
+    # Test getting the branch
+    response = await client.get(
+        f"/api/v1/branches/{branch_id}",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["id"] == branch_id
+    assert data["name"] == "Test Branch"
+
+async def test_list_branches(client: AsyncClient, admin_token: str):
+    # Create multiple branches
+    await test_create_branch(client, admin_token)
+    await client.post(
+        "/api/v1/branches/",
+        json={
+            "name": "Another Branch",
+            "location": "Another Location",
+            "contact_email": "another@example.com",
+            "phone": "0987654321"
+        },
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    
+    # Test listing branches
+    response = await client.get(
+        "/api/v1/branches/",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) >= 2
+    assert any(b["name"] == "Test Branch" for b in data)
+    assert any(b["name"] == "Another Branch" for b in data)
+
+async def test_update_branch(client: AsyncClient, admin_token: str):
+    # First create a branch
+    branch_id = await test_create_branch(client, admin_token)
+    
+    # Test updating the branch
+    response = await client.put(
+        f"/api/v1/branches/{branch_id}",
+        json={
+            "name": "Updated Branch",
+            "location": "Updated Location",
+            "contact_email": "updated@example.com",
+            "phone": "1112223333"
+        },
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["name"] == "Updated Branch"
+    assert data["location"] == "Updated Location"
+    assert data["contact_email"] == "updated@example.com"
+
+async def test_delete_branch(client: AsyncClient, admin_token: str):
+    # First create a branch
+    branch_id = await test_create_branch(client, admin_token)
+    
+    # Test deleting the branch
+    response = await client.delete(
+        f"/api/v1/branches/{branch_id}",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    
+    # Verify branch is deleted
+    response = await client.get(
+        f"/api/v1/branches/{branch_id}",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 404
+
+async def test_branch_statistics(client: AsyncClient, admin_token: str):
+    # First create a branch
+    branch_id = await test_create_branch(client, admin_token)
+    
+    # Test getting branch statistics
+    response = await client.get(
+        f"/api/v1/branches/{branch_id}/statistics",
+        headers={"Authorization": f"Bearer {admin_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert "total_products" in data
+    assert "total_orders" in data
+    assert "total_revenue" in data
+    assert "average_order_value" in data

tests/api/test_orders.py

@@ -0,0 +1,143 @@
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+
+pytestmark = pytest.mark.asyncio
+
+@pytest.fixture
+async def user_token(client: AsyncClient) -> str:
+    """Create a test user and return their token"""
+    # Register user
+    await client.post(
+        "/api/v1/auth/register",
+        data={
+            "username": "testuser@example.com",
+            "password": "testpass123",
+            "full_name": "Test User"
+        }
+    )
+    
+    # Login to get token
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={
+            "username": "testuser@example.com",
+            "password": "testpass123"
+        }
+    )
+    return response.json()["access_token"]
+
+@pytest.fixture
+async def test_product(client: AsyncClient, user_token: str) -> dict:
+    """Create a test product and return its data"""
+    response = await client.post(
+        "/api/v1/products/",
+        json={
+            "name": "Test Product",
+            "description": "A test product",
+            "price": 99.99,
+            "category": "test",
+            "inventory_count": 10
+        },
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    return response.json()
+
+async def test_create_order(client: AsyncClient, user_token: str, test_product: dict):
+    # Test creating an order
+    response = await client.post(
+        "/api/v1/orders/",
+        json={
+            "items": [
+                {
+                    "product_id": test_product["id"],
+                    "quantity": 2
+                }
+            ]
+        },
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data["items"]) == 1
+    assert data["items"][0]["product_id"] == test_product["id"]
+    assert data["total_amount"] == test_product["price"] * 2
+    return data["id"]
+
+async def test_get_order(client: AsyncClient, user_token: str, test_product: dict):
+    # First create an order
+    order_id = await test_create_order(client, user_token, test_product)
+    
+    # Test getting the order
+    response = await client.get(
+        f"/api/v1/orders/{order_id}",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["id"] == order_id
+    assert len(data["items"]) == 1
+
+async def test_list_orders(client: AsyncClient, user_token: str, test_product: dict):
+    # Create multiple orders
+    await test_create_order(client, user_token, test_product)
+    await test_create_order(client, user_token, test_product)
+    
+    # Test listing orders
+    response = await client.get(
+        "/api/v1/orders/",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) >= 2
+
+async def test_update_order_status(client: AsyncClient, user_token: str, test_product: dict):
+    # First create an order
+    order_id = await test_create_order(client, user_token, test_product)
+    
+    # Test updating the order status
+    response = await client.put(
+        f"/api/v1/orders/{order_id}/status",
+        json={"status": "processing"},
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["status"] == "processing"
+
+    # Test invalid status
+    response = await client.put(
+        f"/api/v1/orders/{order_id}/status",
+        json={"status": "invalid_status"},
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 400
+
+async def test_cancel_order(client: AsyncClient, user_token: str, test_product: dict):
+    # First create an order
+    order_id = await test_create_order(client, user_token, test_product)
+    
+    # Get initial product inventory
+    product_response = await client.get(
+        f"/api/v1/products/{test_product['id']}",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    initial_inventory = product_response.json()["inventory_count"]
+    
+    # Cancel the order
+    response = await client.put(
+        f"/api/v1/orders/{order_id}/status",
+        json={"status": "cancelled"},
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    assert response.json()["status"] == "cancelled"
+    
+    # Verify inventory was restored
+    product_response = await client.get(
+        f"/api/v1/products/{test_product['id']}",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    final_inventory = product_response.json()["inventory_count"]
+    assert final_inventory == initial_inventory + 2  # We ordered 2 items

tests/api/test_products.py

@@ -0,0 +1,128 @@
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.core.security import create_access_token
+
+pytestmark = pytest.mark.asyncio
+
+@pytest.fixture
+async def user_token(client: AsyncClient) -> str:
+    """Create a test user and return their token"""
+    # Register user
+    await client.post(
+        "/api/v1/auth/register",
+        data={
+            "username": "testuser@example.com",
+            "password": "testpass123",
+            "full_name": "Test User"
+        }
+    )
+    
+    # Login to get token
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={
+            "username": "testuser@example.com",
+            "password": "testpass123"
+        }
+    )
+    return response.json()["access_token"]
+
+async def test_create_product(client: AsyncClient, user_token: str):
+    # Test creating a product
+    response = await client.post(
+        "/api/v1/products/",
+        json={
+            "name": "Test Product",
+            "description": "A test product",
+            "price": 99.99,
+            "category": "test",
+            "inventory_count": 10
+        },
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["name"] == "Test Product"
+    assert data["price"] == 99.99
+    return data["id"]
+
+async def test_get_product(client: AsyncClient, user_token: str):
+    # First create a product
+    product_id = await test_create_product(client, user_token)
+    
+    # Test getting the product
+    response = await client.get(
+        f"/api/v1/products/{product_id}",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["id"] == product_id
+    assert data["name"] == "Test Product"
+
+async def test_list_products(client: AsyncClient, user_token: str):
+    # Create multiple products
+    await test_create_product(client, user_token)
+    await client.post(
+        "/api/v1/products/",
+        json={
+            "name": "Another Product",
+            "description": "Another test product",
+            "price": 49.99,
+            "category": "test",
+            "inventory_count": 5
+        },
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    
+    # Test listing products
+    response = await client.get(
+        "/api/v1/products/",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) >= 2
+    assert any(p["name"] == "Test Product" for p in data)
+    assert any(p["name"] == "Another Product" for p in data)
+
+async def test_update_product(client: AsyncClient, user_token: str):
+    # First create a product
+    product_id = await test_create_product(client, user_token)
+    
+    # Test updating the product
+    response = await client.put(
+        f"/api/v1/products/{product_id}",
+        json={
+            "name": "Updated Product",
+            "description": "Updated description",
+            "price": 149.99,
+            "category": "test",
+            "inventory_count": 20
+        },
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["name"] == "Updated Product"
+    assert data["price"] == 149.99
+    assert data["inventory_count"] == 20
+
+async def test_delete_product(client: AsyncClient, user_token: str):
+    # First create a product
+    product_id = await test_create_product(client, user_token)
+    
+    # Test deleting the product
+    response = await client.delete(
+        f"/api/v1/products/{product_id}",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 200
+    
+    # Verify product is deleted
+    response = await client.get(
+        f"/api/v1/products/{product_id}",
+        headers={"Authorization": f"Bearer {user_token}"}
+    )
+    assert response.status_code == 404

tests/conftest.py

@@ -0,0 +1,84 @@
+import os
+import pytest
+import asyncio
+from typing import AsyncGenerator, Generator
+from fastapi.testclient import TestClient
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+from dotenv import load_dotenv
+from pathlib import Path
+
+# Load test environment variables
+test_env_path = Path(__file__).parent / "test.env"
+load_dotenv(test_env_path)
+
+from app.db.database import Base, get_db
+from app.main import app
+from app.core.config import settings
+
+# Use SQLite for testing
+SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
+
+engine = create_async_engine(
+    SQLALCHEMY_DATABASE_URL,
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool,
+)
+
+TestingSessionLocal = sessionmaker(
+    engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+)
+
+@pytest.fixture(scope="session")
+def event_loop() -> Generator:
+    """Create an instance of the default event loop for each test case."""
+    try:
+        loop = asyncio.get_event_loop()
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()
+
+@pytest.fixture(scope="session")
+async def test_app():
+    """Create tables for testing and clean up afterward."""
+    # Create test upload directories
+    os.makedirs("./test_uploads/documents", exist_ok=True)
+    os.makedirs("./test_uploads/images", exist_ok=True)
+    
+    # Create tables
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)  # Ensure clean state
+        await conn.run_sync(Base.metadata.create_all)
+    
+    yield app
+    
+    # Cleanup
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+@pytest.fixture
+async def db_session() -> AsyncGenerator[AsyncSession, None]:
+    """Get a test database session."""
+    async with TestingSessionLocal() as session:
+        yield session
+        # Rollback any changes made in the test
+        await session.rollback()
+
+@pytest.fixture
+async def client(test_app) -> AsyncGenerator[AsyncClient, None]:
+    """Get a test client with database session override."""
+    async def override_get_db():
+        async with TestingSessionLocal() as session:
+            yield session
+
+    app.dependency_overrides[get_db] = override_get_db
+
+    async with AsyncClient(app=app, base_url="http://test") as client:
+        yield client
+    
+    app.dependency_overrides.clear()

tests/test.env

@@ -0,0 +1,26 @@
+PROJECT_NAME=Admin Dashboard Test
+VERSION=1.0.0
+API_V1_STR=/api/v1
+SECRET_KEY=testingsecretkey123notforproduction
+ALGORITHM=HS256
+ACCESS_TOKEN_EXPIRE_MINUTES=30
+
+# Database settings
+DATABASE_URL=sqlite+aiosqlite:///:memory:
+
+# Redis settings (using dummy values for testing)
+REDIS_HOST=localhost
+REDIS_PORT=6379
+
+# Email settings (using dummy values for testing)
+SMTP_HOST=localhost
+SMTP_PORT=587
+SMTP_USER=test@example.com
+SMTP_PASSWORD=dummypassword
+EMAILS_FROM_EMAIL=test@example.com
+EMAILS_FROM_NAME=Test System
+
+# Storage settings
+MAX_UPLOAD_DIR_SIZE_MB=1024
+UPLOAD_DIR=./test_uploads
+MAX_DB_CONNECTIONS=100

tests/utils.py

@@ -0,0 +1,104 @@
+from typing import Dict, Any
+from httpx import AsyncClient
+import json
+
+async def create_user_and_login(
+    client: AsyncClient,
+    email: str = "testuser@example.com",
+    password: str = "testpass123",
+    full_name: str = "Test User",
+    is_superuser: bool = False
+) -> Dict[str, Any]:
+    """Create a test user and return their token and data"""
+    # Register user
+    register_response = await client.post(
+        "/api/v1/auth/register",
+        json={
+            "email": email,
+            "password": password,
+            "full_name": full_name,
+            "is_superuser": is_superuser
+        }
+    )
+    user_data = register_response.json()
+    
+    # Login to get token
+    login_response = await client.post(
+        "/api/v1/auth/login",
+        data={
+            "username": email,
+            "password": password
+        }
+    )
+    token_data = login_response.json()
+    
+    return {
+        "token": token_data["access_token"],
+        "user": user_data
+    }
+
+async def create_branch(
+    client: AsyncClient,
+    token: str,
+    name: str = "Test Branch",
+    location: str = "Test Location"
+) -> Dict[str, Any]:
+    """Create a test branch and return its data"""
+    response = await client.post(
+        "/api/v1/branches/",
+        json={
+            "name": name,
+            "location": location
+        },
+        headers={"Authorization": f"Bearer {token}"}
+    )
+    return response.json()
+
+async def create_product(
+    client: AsyncClient,
+    token: str,
+    branch_id: int,
+    name: str = "Test Product",
+    price: float = 99.99,
+    inventory: int = 10
+) -> Dict[str, Any]:
+    """Create a test product and return its data"""
+    response = await client.post(
+        "/api/v1/products/",
+        json={
+            "name": name,
+            "description": f"Description for {name}",
+            "price": price,
+            "category": "test",
+            "inventory_count": inventory,
+            "branch_id": branch_id
+        },
+        headers={"Authorization": f"Bearer {token}"}
+    )
+    return response.json()
+
+async def create_order(
+    client: AsyncClient,
+    token: str,
+    items: list,
+    branch_id: int
+) -> Dict[str, Any]:
+    """Create a test order and return its data"""
+    response = await client.post(
+        "/api/v1/orders/",
+        json={
+            "items": items,
+            "branch_id": branch_id
+        },
+        headers={"Authorization": f"Bearer {token}"}
+    )
+    return response.json()
+
+def assert_response_matches_schema(response_data: Dict[str, Any], schema: Dict[str, Any]):
+    """Assert that an API response matches its expected schema"""
+    assert all(key in response_data for key in schema.keys()), \
+        f"Missing required fields. Expected {schema.keys()}, got {response_data.keys()}"
+    
+    for key, value_type in schema.items():
+        assert isinstance(response_data[key], value_type), \
+            f"Field {key} should be of type {value_type}, got {type(response_data[key])}"