sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixing DQ issues

Browse Source
This commit is contained in:
Samir Boulahtit
2025-09-19 16:54:13 +02:00
parent 0ce708cf09
commit f042616fdd
45 changed files with 3625 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/api/v1/admin.py
View File

@@ -16,11 +16,8 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_admin_user
from app.core.database import get_db
from app.services.admin_service import admin_service
from models.api_models import (
MarketplaceImportJobResponse,
ShopListResponse,
UserResponse,
)
from models.api_models import (MarketplaceImportJobResponse, ShopListResponse,
UserResponse)
from models.database_models import User
router = APIRouter()

3
app/api/v1/auth.py
View File

@@ -15,7 +15,8 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.core.database import get_db
from app.services.auth_service import auth_service
from models.api_models import LoginResponse, UserLogin, UserRegister, UserResponse
from models.api_models import (LoginResponse, UserLogin, UserRegister,
UserResponse)
from models.database_models import User
router = APIRouter()

3
app/api/v1/marketplace.py
View File

@@ -18,7 +18,8 @@ from app.core.database import get_db
from app.services.marketplace_service import marketplace_service
from app.tasks.background_tasks import process_marketplace_import
from middleware.decorators import rate_limit
from models.api_models import MarketplaceImportJobResponse, MarketplaceImportRequest
from models.api_models import (MarketplaceImportJobResponse,
MarketplaceImportRequest)
from models.database_models import User
router = APIRouter()

10
app/api/v1/product.py
View File

@@ -17,13 +17,9 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.core.database import get_db
from app.services.product_service import product_service
from models.api_models import (
ProductCreate,
ProductDetailResponse,
ProductListResponse,
ProductResponse,
ProductUpdate,
)
from models.api_models import (ProductCreate, ProductDetailResponse,
ProductListResponse, ProductResponse,
ProductUpdate)
from models.database_models import User
router = APIRouter()

10
app/api/v1/shop.py
View File

@@ -15,14 +15,8 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_user, get_user_shop
from app.core.database import get_db
from app.services.shop_service import shop_service
from models.api_models import (
ShopCreate,
ShopListResponse,
ShopProductCreate,
ShopProductResponse,
ShopResponse,
)
from models.api_models import (ShopCreate, ShopListResponse, ShopProductCreate,
ShopProductResponse, ShopResponse)
from models.database_models import User
router = APIRouter()

5
app/api/v1/stats.py
View File

@@ -16,10 +16,7 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.core.database import get_db
from app.services.stats_service import stats_service
from models.api_models import (
MarketplaceStatsResponse,
StatsResponse,
)
from models.api_models import MarketplaceStatsResponse, StatsResponse
from models.database_models import User
router = APIRouter()

9
app/api/v1/stock.py
View File

@@ -16,13 +16,8 @@ from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.core.database import get_db
from app.services.stock_service import stock_service
from models.api_models import (
StockAdd,
StockCreate,
StockResponse,
StockSummaryResponse,
StockUpdate,
)
from models.api_models import (StockAdd, StockCreate, StockResponse,
StockSummaryResponse, StockUpdate)
from models.database_models import User
router = APIRouter()

2
app/services/auth_service.py
View File

@@ -29,7 +29,7 @@ class AuthService:
def register_user(self, db: Session, user_data: UserRegister) -> User:
"""
Register a new user
Register a new user.
Args:
db: Database session

7
app/services/marketplace_service.py
View File

@@ -22,12 +22,14 @@ logger = logging.getLogger(__name__)
class MarketplaceService:
"""Service class for Marketplace operations following the application's service pattern."""
def __init__(self):
"""Class constructor."""
pass
def validate_shop_access(self, db: Session, shop_code: str, user: User) -> Shop:
"""Validate that the shop exists and user has access to it"""
"""Validate that the shop exists and user has access to it."""
# Explicit type hint to help type checker shop: Optional[Shop]
# Use case-insensitive query to handle both uppercase and lowercase codes
shop: Optional[Shop] = (
@@ -67,7 +69,8 @@ class MarketplaceService:
db.refresh(import_job)
logger.info(
f"Created marketplace import job {import_job.id}: {request.marketplace} -> {shop.shop_name} (shop_code: {shop.shop_code}) by user {user.username}"
f"Created marketplace import job {import_job.id}: "
f"{request.marketplace} -> {shop.shop_name} (shop_code: {shop.shop_code}) by user {user.username}"
)
return import_job

2
app/services/product_service.py
View File

@@ -23,6 +23,8 @@ logger = logging.getLogger(__name__)
class ProductService:
"""Service class for Product operations following the application's service pattern."""
def __init__(self):
"""Class constructor."""
self.gtin_processor = GTINProcessor()

15
app/services/shop_service.py
View File

@@ -8,7 +8,6 @@ This module provides classes and functions for:
"""
import logging
from typing import List, Optional, Tuple
from fastapi import HTTPException
@@ -28,7 +27,7 @@ class ShopService:
self, db: Session, shop_data: ShopCreate, current_user: User
) -> Shop:
"""
Create a new shop
Create a new shop.
Args:
db: Database session
@@ -256,19 +255,19 @@ class ShopService:
return shop_products, total
def get_shop_by_id(self, db: Session, shop_id: int) -> Optional[Shop]:
"""Get shop by ID"""
"""Get shop by ID."""
return db.query(Shop).filter(Shop.id == shop_id).first()
def shop_code_exists(self, db: Session, shop_code: str) -> bool:
"""Check if shop code already exists"""
"""Check if shop code already exists."""
return db.query(Shop).filter(Shop.shop_code == shop_code).first() is not None
def get_product_by_id(self, db: Session, product_id: str) -> Optional[Product]:
"""Get product by product_id"""
"""Get product by product_id."""
return db.query(Product).filter(Product.product_id == product_id).first()
def product_in_shop(self, db: Session, shop_id: int, product_id: int) -> bool:
"""Check if product is already in shop"""
"""Check if product is already in shop."""
return (
db.query(ShopProduct)
.filter(
@@ -279,11 +278,11 @@ class ShopService:
)
def is_shop_owner(self, shop: Shop, user: User) -> bool:
"""Check if user is shop owner"""
"""Check if user is shop owner."""
return shop.owner_id == user.id
def can_view_shop(self, shop: Shop, user: User) -> bool:
"""Check if user can view shop"""
"""Check if user can view shop."""
if user.role == "admin" or self.is_shop_owner(shop, user):
return True
return shop.is_active and shop.is_verified

1
app/services/stock_service.py
View File

@@ -23,6 +23,7 @@ logger = logging.getLogger(__name__)
class StockService:
"""Service class for stock operations following the application's service pattern."""
def __init__(self):
"""Class constructor."""
self.gtin_processor = GTINProcessor()

10
app/tasks/background_tasks.py
View File

@@ -1,4 +1,12 @@
# app/tasks/background_tasks.py
"""Summary description ....
This module provides classes and functions for:
- ....
- ....
- ....
"""
import logging
from datetime import datetime
@@ -12,7 +20,7 @@ logger = logging.getLogger(__name__)
async def process_marketplace_import(
job_id: int, url: str, marketplace: str, shop_name: str, batch_size: int = 1000
):
"""Background task to process marketplace CSV import"""
"""Background task to process marketplace CSV import."""
db = SessionLocal()
csv_processor = CSVProcessor()
job = None # Initialize job variable

91
auth_fixtures.py Normal file
View File

@@ -0,0 +1,91 @@
# tests/fixtures/auth_fixtures.py
import uuid
import pytest
from middleware.auth import AuthManager
from models.database_models import User
@pytest.fixture(scope="session")
def auth_manager():
"""Create auth manager instance (session scope since it's stateless)"""
return AuthManager()
@pytest.fixture
def test_user(db, auth_manager):
"""Create a test user with unique username"""
unique_id = str(uuid.uuid4())[:8] # Short unique identifier
hashed_password = auth_manager.hash_password("testpass123")
user = User(
email=f"test_{unique_id}@example.com",
username=f"testuser_{unique_id}",
hashed_password=hashed_password,
role="user",
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user
@pytest.fixture
def test_admin(db, auth_manager):
"""Create a test admin user with unique username"""
unique_id = str(uuid.uuid4())[:8] # Short unique identifier
hashed_password = auth_manager.hash_password("adminpass123")
admin = User(
email=f"admin_{unique_id}@example.com",
username=f"admin_{unique_id}",
hashed_password=hashed_password,
role="admin",
is_active=True,
)
db.add(admin)
db.commit()
db.refresh(admin)
return admin
@pytest.fixture
def other_user(db, auth_manager):
"""Create a different user for testing access controls"""
unique_id = str(uuid.uuid4())[:8]
hashed_password = auth_manager.hash_password("otherpass123")
user = User(
email=f"other_{unique_id}@example.com",
username=f"otheruser_{unique_id}",
hashed_password=hashed_password,
role="user",
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user
@pytest.fixture
def auth_headers(client, test_user):
"""Get authentication headers for test user"""
response = client.post(
"/api/v1/auth/login",
json={"username": test_user.username, "password": "testpass123"},
)
assert response.status_code == 200, f"Login failed: {response.text}"
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
@pytest.fixture
def admin_headers(client, test_admin):
"""Get authentication headers for admin user"""
response = client.post(
"/api/v1/auth/login",
json={"username": test_admin.username, "password": "adminpass123"},
)
assert response.status_code == 200, f"Admin login failed: {response.text}"
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}

80
enhanced_pytest_config.txt Normal file
View File

@@ -0,0 +1,80 @@
# pytest.ini - Enhanced configuration for your FastAPI test suite
[tool:pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
# Enhanced addopts for better development experience
addopts =
-v
--tb=short
--strict-markers
--strict-config
--color=yes
--durations=10
--showlocals
-ra
--cov=app
--cov=models
--cov=utils
--cov=middleware
--cov-report=term-missing
--cov-report=html:htmlcov
--cov-fail-under=80
# Test discovery and execution settings
minversion = 6.0
testmon = true
python_paths = .
# Markers for your specific test organization
markers =
# Test Types (for your new structure)
unit: Unit tests - fast, isolated components
integration: Integration tests - multiple components working together
system: System tests - full application behavior
e2e: End-to-end tests - complete user workflows
# Performance and Speed
slow: Slow running tests (deselect with '-m "not slow"')
performance: Performance and load tests
# Domain-specific markers (matching your application structure)
auth: Authentication and authorization tests
products: Product management functionality
stock: Stock and inventory management
shops: Shop management functionality
admin: Admin functionality and permissions
marketplace: Marketplace import functionality
stats: Statistics and reporting
# Infrastructure markers
database: Tests that require database operations
external: Tests that require external services
api: API endpoint tests
security: Security-related tests
# Test environment markers
ci: Tests that should only run in CI
dev: Development-specific tests
# Test filtering shortcuts
filterwarnings =
ignore::UserWarning
ignore::DeprecationWarning
ignore::PendingDeprecationWarning
ignore::sqlalchemy.exc.SAWarning
# Timeout settings
timeout = 300
timeout_method = thread
# Parallel execution settings (uncomment if using pytest-xdist)
# addopts = -n auto
# Additional logging configuration
log_cli = true
log_cli_level = INFO
log_cli_format = %(asctime)s [%(levelname)8s] %(name)s: %(message)s
log_cli_date_format = %Y-%m-%d %H:%M:%S

60
init_files.py Normal file
View File

@@ -0,0 +1,60 @@
# tests/fixtures/__init__.py
"""Test fixtures for the FastAPI application test suite."""
# tests/unit/__init__.py
"""Unit tests - fast, isolated component tests."""
# tests/unit/models/__init__.py
"""Database and API model unit tests."""
# tests/unit/utils/__init__.py
"""Utility function unit tests."""
# tests/unit/services/__init__.py
"""Service layer unit tests."""
# tests/integration/__init__.py
"""Integration tests - multiple components working together."""
# tests/integration/api/__init__.py
"""API integration tests."""
# tests/integration/api/v1/__init__.py
"""API v1 endpoint integration tests."""
# tests/integration/security/__init__.py
"""Security integration tests."""
# tests/performance/__init__.py
"""Performance and load tests."""
# tests/system/__init__.py
"""System-level tests - full application behavior."""
# tests/integration/conftest.py
"""Integration test specific fixtures."""
import pytest
# Add any integration-specific fixtures here if needed
# tests/unit/conftest.py
"""Unit test specific fixtures."""
import pytest
# Add any unit-specific fixtures here if needed
# tests/performance/conftest.py
"""Performance test specific fixtures."""
import pytest
@pytest.fixture
def performance_db_session(db):
"""Database session optimized for performance testing"""
# You can add performance-specific DB configurations here
return db
# tests/system/conftest.py
"""System test specific fixtures."""
import pytest
# Add any system-specific fixtures here if needed

195
integration_admin_endpoints.py Normal file
View File

@@ -0,0 +1,195 @@
# tests/integration/api/v1/test_admin_endpoints.py
import pytest
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.admin
class TestAdminAPI:
def test_get_all_users_admin(self, client, admin_headers, test_user):
"""Test admin getting all users"""
response = client.get("/api/v1/admin/users", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert len(data) >= 2 # test_user + admin user
# Check that test_user is in the response
user_ids = [user["id"] for user in data if "id" in user]
assert test_user.id in user_ids
def test_get_all_users_non_admin(self, client, auth_headers):
"""Test non-admin trying to access admin endpoint"""
response = client.get("/api/v1/admin/users", headers=auth_headers)
assert response.status_code == 403
assert (
"Access denied" in response.json()["detail"]
or "admin" in response.json()["detail"].lower()
)
def test_toggle_user_status_admin(self, client, admin_headers, test_user):
"""Test admin toggling user status"""
response = client.put(
f"/api/v1/admin/users/{test_user.id}/status", headers=admin_headers
)
assert response.status_code == 200
message = response.json()["message"]
assert "deactivated" in message or "activated" in message
# Verify the username is in the message
assert test_user.username in message
def test_toggle_user_status_user_not_found(self, client, admin_headers):
"""Test admin toggling status for non-existent user"""
response = client.put("/api/v1/admin/users/99999/status", headers=admin_headers)
assert response.status_code == 404
assert "User not found" in response.json()["detail"]
def test_toggle_user_status_cannot_deactivate_self(
self, client, admin_headers, test_admin
):
"""Test that admin cannot deactivate their own account"""
response = client.put(
f"/api/v1/admin/users/{test_admin.id}/status", headers=admin_headers
)
assert response.status_code == 400
assert "Cannot deactivate your own account" in response.json()["detail"]
def test_get_all_shops_admin(self, client, admin_headers, test_shop):
"""Test admin getting all shops"""
response = client.get("/api/v1/admin/shops", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["shops"]) >= 1
# Check that test_shop is in the response
shop_codes = [
shop["shop_code"] for shop in data["shops"] if "shop_code" in shop
]
assert test_shop.shop_code in shop_codes
def test_get_all_shops_non_admin(self, client, auth_headers):
"""Test non-admin trying to access admin shop endpoint"""
response = client.get("/api/v1/admin/shops", headers=auth_headers)
assert response.status_code == 403
assert (
"Access denied" in response.json()["detail"]
or "admin" in response.json()["detail"].lower()
)
def test_verify_shop_admin(self, client, admin_headers, test_shop):
"""Test admin verifying/unverifying shop"""
response = client.put(
f"/api/v1/admin/shops/{test_shop.id}/verify", headers=admin_headers
)
assert response.status_code == 200
message = response.json()["message"]
assert "verified" in message or "unverified" in message
assert test_shop.shop_code in message
def test_verify_shop_not_found(self, client, admin_headers):
"""Test admin verifying non-existent shop"""
response = client.put("/api/v1/admin/shops/99999/verify", headers=admin_headers)
assert response.status_code == 404
assert "Shop not found" in response.json()["detail"]
def test_toggle_shop_status_admin(self, client, admin_headers, test_shop):
"""Test admin toggling shop status"""
response = client.put(
f"/api/v1/admin/shops/{test_shop.id}/status", headers=admin_headers
)
assert response.status_code == 200
message = response.json()["message"]
assert "activated" in message or "deactivated" in message
assert test_shop.shop_code in message
def test_toggle_shop_status_not_found(self, client, admin_headers):
"""Test admin toggling status for non-existent shop"""
response = client.put("/api/v1/admin/shops/99999/status", headers=admin_headers)
assert response.status_code == 404
assert "Shop not found" in response.json()["detail"]
def test_get_marketplace_import_jobs_admin(
self, client, admin_headers, test_marketplace_job
):
"""Test admin getting marketplace import jobs"""
response = client.get(
"/api/v1/admin/marketplace-import-jobs", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
# Check that test_marketplace_job is in the response
job_ids = [job["job_id"] for job in data if "job_id" in job]
assert test_marketplace_job.id in job_ids
def test_get_marketplace_import_jobs_with_filters(
self, client, admin_headers, test_marketplace_job
):
"""Test admin getting marketplace import jobs with filters"""
response = client.get(
"/api/v1/admin/marketplace-import-jobs",
params={"marketplace": test_marketplace_job.marketplace},
headers=admin_headers,
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 1
assert all(
job["marketplace"] == test_marketplace_job.marketplace for job in data
)
def test_get_marketplace_import_jobs_non_admin(self, client, auth_headers):
"""Test non-admin trying to access marketplace import jobs"""
response = client.get(
"/api/v1/admin/marketplace-import-jobs", headers=auth_headers
)
assert response.status_code == 403
assert (
"Access denied" in response.json()["detail"]
or "admin" in response.json()["detail"].lower()
)
def test_admin_pagination_users(self, client, admin_headers, test_user, test_admin):
"""Test user pagination works correctly"""
# Test first page
response = client.get(
"/api/v1/admin/users?skip=0&limit=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) == 1
# Test second page
response = client.get(
"/api/v1/admin/users?skip=1&limit=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert len(data) >= 0 # Could be 1 or 0 depending on total users
def test_admin_pagination_shops(self, client, admin_headers, test_shop):
"""Test shop pagination works correctly"""
response = client.get(
"/api/v1/admin/shops?skip=0&limit=1", headers=admin_headers
)
assert response.status_code == 200
data = response.json()
assert data["total"] >= 1
assert len(data["shops"]) >= 0
assert "skip" in data
assert "limit" in data

75
integration_authentication.py Normal file
View File

@@ -0,0 +1,75 @@
# tests/integration/security/test_authentication.py
import pytest
@pytest.mark.integration
@pytest.mark.security
@pytest.mark.auth
class TestAuthentication:
def test_protected_endpoint_without_auth(self, client):
"""Test that protected endpoints reject unauthenticated requests"""
protected_endpoints = [
"/api/v1/admin/users",
"/api/v1/admin/shops",
"/api/v1/marketplace/import-jobs",
"/api/v1/product",
"/api/v1/shop",
"/api/v1/stats",
"/api/v1/stock",
]
for endpoint in protected_endpoints:
response = client.get(endpoint)
assert response.status_code == 401 # Authentication missing
def test_protected_endpoint_with_invalid_token(self, client):
"""Test protected endpoints with invalid token"""
headers = {"Authorization": "Bearer invalid_token_here"}
response = client.get("/api/v1/product", headers=headers)
assert response.status_code == 401 # Token is not valid
def test_debug_direct_bearer(self, client):
"""Test HTTPBearer directly"""
response = client.get("/api/v1/debug-bearer")
print(f"Direct Bearer - Status: {response.status_code}")
print(
f"Direct Bearer - Response: {response.json() if response.content else 'No content'}"
)
def test_debug_dependencies(self, client):
"""Debug the dependency chain step by step"""
# Test 1: Direct endpoint with no auth
response = client.get("/api/v1/admin/users")
print(f"Admin endpoint - Status: {response.status_code}")
try:
print(f"Admin endpoint - Response: {response.json()}")
except:
print(f"Admin endpoint - Raw: {response.content}")
# Test 2: Try a regular endpoint that uses get_current_user
response2 = client.get("/api/v1/product") # or any endpoint with get_current_user
print(f"Regular endpoint - Status: {response2.status_code}")
try:
print(f"Regular endpoint - Response: {response2.json()}")
except:
print(f"Regular endpoint - Raw: {response2.content}")
def test_debug_available_routes(self, client):
"""Debug test to see all available routes"""
print("\n=== All Available Routes ===")
for route in client.app.routes:
if hasattr(route, "path") and hasattr(route, "methods"):
print(f"{list(route.methods)} {route.path}")
print("\n=== Testing Product Endpoint Variations ===")
variations = [
"/api/v1/product", # Your current attempt
"/api/v1/product/", # With trailing slash
"/api/v1/product/list", # With list endpoint
"/api/v1/product/all", # With all endpoint
]
for path in variations:
response = client.get(path)
print(f"{path}: Status {response.status_code}")

46
integration_authorization.py Normal file
View File

@@ -0,0 +1,46 @@
# tests/integration/security/test_authorization.py
import pytest
@pytest.mark.integration
@pytest.mark.security
@pytest.mark.auth
class TestAuthorization:
def test_admin_endpoint_requires_admin_role(self, client, auth_headers):
"""Test that admin endpoints require admin role"""
response = client.get("/api/v1/admin/users", headers=auth_headers)
assert response.status_code == 403
# Regular user should be denied access
def test_admin_endpoints_with_admin_access(self, client, admin_headers):
"""Test that admin users can access admin endpoints"""
admin_endpoints = [
"/api/v1/admin/users",
"/api/v1/admin/shops",
"/api/v1/admin/marketplace-import-jobs",
]
for endpoint in admin_endpoints:
response = client.get(endpoint, headers=admin_headers)
assert response.status_code == 200 # Admin should have access
def test_regular_endpoints_with_user_access(self, client, auth_headers):
"""Test that regular users can access non-admin endpoints"""
user_endpoints = [
"/api/v1/product",
"/api/v1/stats",
"/api/v1/stock",
]
for endpoint in user_endpoints:
response = client.get(endpoint, headers=auth_headers)
assert response.status_code == 200 # Regular user should have access
def test_shop_owner_access_control(self, client, auth_headers, test_shop, other_user):
"""Test that users can only access their own shops"""
# Test accessing own shop (should work)
response = client.get(f"/api/v1/shop/{test_shop.shop_code}", headers=auth_headers)
# Response depends on your implementation - could be 200 or 404 if shop doesn't belong to user
# The exact assertion depends on your shop access control implementation
assert response.status_code in [200, 403, 404]

65
integration_input_validation.py Normal file
View File

@@ -0,0 +1,65 @@
# tests/integration/security/test_input_validation.py
import pytest
@pytest.mark.integration
@pytest.mark.security
class TestInputValidation:
def test_sql_injection_prevention(self, client, auth_headers):
"""Test SQL injection prevention in search parameters"""
# Try SQL injection in search parameter
malicious_search = "'; DROP TABLE products; --"
response = client.get(
f"/api/v1/product?search={malicious_search}", headers=auth_headers
)
# Should not crash and should return normal response
assert response.status_code == 200
# Database should still be intact (no products dropped)
# def test_input_validation(self, client, auth_headers):
# # TODO: implement sanitization
# """Test input validation and sanitization"""
# # Test XSS attempt in product creation
# xss_payload = "<script>alert('xss')</script>"
#
# product_data = {
# "product_id": "XSS_TEST",
# "title": xss_payload,
# "description": xss_payload,
# }
#
# response = client.post("/api/v1/product", headers=auth_headers, json=product_data)
#
# assert response.status_code == 200
# data = response.json()
# assert "<script>" not in data["title"]
# assert "&lt;script&gt;" in data["title"]
def test_parameter_validation(self, client, auth_headers):
"""Test parameter validation for API endpoints"""
# Test invalid pagination parameters
response = client.get("/api/v1/product?limit=-1", headers=auth_headers)
assert response.status_code == 422 # Validation error
response = client.get("/api/v1/product?skip=-1", headers=auth_headers)
assert response.status_code == 422 # Validation error
def test_json_validation(self, client, auth_headers):
"""Test JSON validation for POST requests"""
# Test invalid JSON structure
response = client.post(
"/api/v1/product",
headers=auth_headers,
content="invalid json content"
)
assert response.status_code == 422 # JSON decode error
# Test missing required fields
response = client.post(
"/api/v1/product",
headers=auth_headers,
json={"title": "Test Product"} # Missing required product_id
)
assert response.status_code == 422 # Validation error

86
integration_pagination.py Normal file
View File

@@ -0,0 +1,86 @@
# tests/integration/api/v1/test_pagination.py
import pytest
from models.database_models import Product
@pytest.mark.integration
@pytest.mark.api
@pytest.mark.database
@pytest.mark.products
class TestPagination:
def test_product_pagination(self, client, auth_headers, db):
"""Test pagination for product listing"""
# Create multiple products
products = []
for i in range(25):
product = Product(
product_id=f"PAGE{i:03d}",
title=f"Pagination Test Product {i}",
marketplace="PaginationTest",
)
products.append(product)
db.add_all(products)
db.commit()
# Test first page
response = client.get("/api/v1/product?limit=10&skip=0", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) == 10
assert data["total"] == 25
assert data["skip"] == 0
assert data["limit"] == 10
# Test second page
response = client.get("/api/v1/product?limit=10&skip=10", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) == 10
assert data["skip"] == 10
# Test last page
response = client.get("/api/v1/product?limit=10&skip=20", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert len(data["products"]) == 5 # Only 5 remaining
def test_pagination_boundaries(self, client, auth_headers):
"""Test pagination boundary conditions"""
# Test negative skip
response = client.get("/api/v1/product?skip=-1", headers=auth_headers)
assert response.status_code == 422 # Validation error
# Test zero limit
response = client.get("/api/v1/product?limit=0", headers=auth_headers)
assert response.status_code == 422 # Validation error
# Test excessive limit
response = client.get("/api/v1/product?limit=10000", headers=auth_headers)
assert response.status_code == 422 # Should be limited
def test_shop_pagination(self, client, admin_headers, db, test_user):
"""Test pagination for shop listing"""
# Create multiple shops for pagination testing
shops = []
for i in range(15):
shop = Shop(
shop_code=f"PAGESHOP{i:03d}",
shop_name=f"Pagination Shop {i}",
owner_id=test_user.id,
is_active=True,
)
shops.append(shop)
db.add_all(shops)
db.commit()
# Test first page
response = client.get("/api/v1/admin/shops?limit=5&skip=0", headers=admin_headers)
assert response.status_code == 200
data = response.json()
assert len(data["shops"]) == 5
assert data["total"] >= 15 # At least our test shops
assert data["skip"] == 0
assert data["limit"] == 5

90
main_conftest.py Normal file
View File

@@ -0,0 +1,90 @@
# tests/conftest.py - Updated main conftest with core fixtures only
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from app.core.database import Base, get_db
from main import app
# Import all models to ensure they're registered with Base metadata
from models.database_models import (MarketplaceImportJob, Product, Shop,
ShopProduct, Stock, User)
# Use in-memory SQLite database for tests
SQLALCHEMY_TEST_DATABASE_URL = "sqlite:///:memory:"
@pytest.fixture(scope="session")
def engine():
"""Create test database engine"""
return create_engine(
SQLALCHEMY_TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
echo=False, # Set to True for SQL debugging
)
@pytest.fixture(scope="session")
def testing_session_local(engine):
"""Create session factory for tests"""
return sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(scope="function")
def db(engine, testing_session_local):
"""Create a database session for direct database operations"""
# Create all tables
Base.metadata.create_all(bind=engine)
# Create session
db_session = testing_session_local()
try:
yield db_session
finally:
db_session.close()
# Clean up all data after each test
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
@pytest.fixture(scope="function")
def client(db):
"""Create a test client with database dependency override"""
# Override the dependency to use our test database
def override_get_db():
try:
yield db
finally:
pass # Don't close here, the db fixture handles it
app.dependency_overrides[get_db] = override_get_db
try:
client = TestClient(app)
yield client
finally:
# Clean up the dependency override
if get_db in app.dependency_overrides:
del app.dependency_overrides[get_db]
# Cleanup fixture to ensure clean state
@pytest.fixture(autouse=True)
def cleanup():
"""Automatically clean up after each test"""
yield
# Clear any remaining dependency overrides
app.dependency_overrides.clear()
# Import fixtures from fixture modules
pytest_plugins = [
"tests.fixtures.auth_fixtures",
"tests.fixtures.product_fixtures",
"tests.fixtures.shop_fixtures",
"tests.fixtures.marketplace_fixtures",
]

50
marketplace_fixtures.py Normal file
View File

@@ -0,0 +1,50 @@
# tests/fixtures/marketplace_fixtures.py
import pytest
from models.database_models import MarketplaceImportJob
@pytest.fixture
def test_marketplace_job(db, test_shop, test_user):
"""Create a test marketplace import job"""
job = MarketplaceImportJob(
marketplace="amazon",
shop_name="Test Import Shop",
status="completed",
source_url="https://test-marketplace.example.com/import",
shop_id=test_shop.id,
user_id=test_user.id,
imported_count=5,
updated_count=3,
total_processed=8,
error_count=0,
error_message=None,
)
db.add(job)
db.commit()
db.refresh(job)
return job
def create_test_import_job(db, shop_id, user_id, **kwargs):
"""Helper function to create MarketplaceImportJob with defaults"""
defaults = {
"marketplace": "test",
"shop_name": "Test Shop",
"status": "pending",
"source_url": "https://test.example.com/import",
"shop_id": shop_id,
"user_id": user_id,
"imported_count": 0,
"updated_count": 0,
"total_processed": 0,
"error_count": 0,
"error_message": None,
}
defaults.update(kwargs)
job = MarketplaceImportJob(**defaults)
db.add(job)
db.commit()
db.refresh(job)
return job

26
middleware/auth.py
View File

@@ -1,4 +1,12 @@
# middleware/auth.py
"""Summary description ....
This module provides classes and functions for:
- ....
- ....
- ....
"""
import logging
import os
from datetime import datetime, timedelta
@@ -30,17 +38,17 @@ class AuthManager:
self.token_expire_minutes = int(os.getenv("JWT_EXPIRE_MINUTES", "30"))
def hash_password(self, password: str) -> str:
"""Hash password using bcrypt"""
"""Hash password using bcrypt."""
return pwd_context.hash(password)
def verify_password(self, plain_password: str, hashed_password: str) -> bool:
"""Verify password against hash"""
"""Verify password against hash."""
return pwd_context.verify(plain_password, hashed_password)
def authenticate_user(
self, db: Session, username: str, password: str
) -> Optional[User]:
"""Authenticate user and return user object if valid"""
"""Authenticate user and return user object if valid."""
user = (
db.query(User)
.filter((User.username == username) | (User.email == username))
@@ -64,7 +72,7 @@ class AuthManager:
return user
def create_access_token(self, user: User) -> Dict[str, Any]:
"""Create JWT access token for user"""
"""Create JWT access token for user."""
expires_delta = timedelta(minutes=self.token_expire_minutes)
expire = datetime.utcnow() + expires_delta
@@ -86,7 +94,7 @@ class AuthManager:
}
def verify_token(self, token: str) -> Dict[str, Any]:
"""Verify JWT token and return user data"""
"""Verify JWT token and return user data."""
try:
payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
@@ -126,7 +134,7 @@ class AuthManager:
def get_current_user(
self, db: Session, credentials: HTTPAuthorizationCredentials
) -> User:
"""Get current authenticated user from database"""
"""Get current authenticated user from database."""
user_data = self.verify_token(credentials.credentials)
user = db.query(User).filter(User.id == user_data["user_id"]).first()
@@ -139,7 +147,7 @@ class AuthManager:
return user
def require_role(self, required_role: str):
"""Decorator to require specific role"""
"""Require specific role."""
def decorator(func):
def wrapper(current_user: User, *args, **kwargs):
@@ -155,13 +163,13 @@ class AuthManager:
return decorator
def require_admin(self, current_user: User):
"""Require admin role"""
"""Require admin role."""
if current_user.role != "admin":
raise HTTPException(status_code=403, detail="Admin privileges required")
return current_user
def create_default_admin_user(self, db: Session):
"""Create default admin user if it doesn't exist"""
"""Create default admin user if it doesn't exist."""
admin_user = db.query(User).filter(User.username == "admin").first()
if not admin_user:

10
middleware/decorators.py
View File

@@ -1,4 +1,12 @@
# middleware/decorators.py
"""Summary description ....
This module provides classes and functions for:
- ....
- ....
- ....
"""
from functools import wraps
from fastapi import HTTPException
@@ -10,7 +18,7 @@ rate_limiter = RateLimiter()
def rate_limit(max_requests: int = 100, window_seconds: int = 3600):
"""Rate limiting decorator for FastAPI endpoints"""
"""Rate limiting decorator for FastAPI endpoints."""
def decorator(func):
@wraps(func)

14
middleware/error_handler.py
View File

@@ -1,4 +1,12 @@
# middleware/error_handler.py
"""Summary description ....
This module provides classes and functions for:
- ....
- ....
- ....
"""
import logging
from fastapi import HTTPException, Request
@@ -9,7 +17,7 @@ logger = logging.getLogger(__name__)
async def custom_http_exception_handler(request: Request, exc: HTTPException):
"""Custom HTTP exception handler"""
"""Handle HTTP exception."""
logger.error(
f"HTTP {exc.status_code}: {exc.detail} - {request.method} {request.url}"
)
@@ -27,7 +35,7 @@ async def custom_http_exception_handler(request: Request, exc: HTTPException):
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""Handle Pydantic validation errors"""
"""Handle Pydantic validation errors."""
logger.error(f"Validation error: {exc.errors()} - {request.method} {request.url}")
return JSONResponse(
@@ -44,7 +52,7 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
async def general_exception_handler(request: Request, exc: Exception):
"""Handle unexpected exceptions"""
"""Handle unexpected exceptions."""
logger.error(
f"Unexpected error: {str(exc)} - {request.method} {request.url}", exc_info=True
)

11
middleware/logging_middleware.py
View File

@@ -1,4 +1,12 @@
# middleware/logging_middleware.py
"""Summary description ....
This module provides classes and functions for:
- ....
- ....
- ....
"""
import logging
import time
from typing import Callable
@@ -10,9 +18,10 @@ logger = logging.getLogger(__name__)
class LoggingMiddleware(BaseHTTPMiddleware):
"""Middleware for request/response logging and performance monitoring"""
"""Middleware for request/response logging and performance monitoring."""
async def dispatch(self, request: Request, call_next: Callable) -> Response:
"""Dispatch."""
# Start timing
start_time = time.time()

17
middleware/rate_limiter.py
View File

@@ -1,4 +1,12 @@
# middleware/rate_limiter.py
"""Summary description ....
This module provides classes and functions for:
- ....
- ....
- ....
"""
import logging
from collections import defaultdict, deque
from datetime import datetime, timedelta
@@ -8,7 +16,7 @@ logger = logging.getLogger(__name__)
class RateLimiter:
"""In-memory rate limiter using sliding window"""
"""In-memory rate limiter using sliding window."""
def __init__(self):
"""Class constructor."""
@@ -21,7 +29,8 @@ class RateLimiter:
self, client_id: str, max_requests: int, window_seconds: int
) -> bool:
"""
Check if client is allowed to make a request
Check if client is allowed to make a request.
Uses sliding window algorithm
"""
now = datetime.utcnow()
@@ -50,7 +59,7 @@ class RateLimiter:
return False
def _cleanup_old_entries(self):
"""Clean up old entries to prevent memory leaks"""
"""Clean up old entries to prevent memory leaks."""
cutoff_time = datetime.utcnow() - timedelta(hours=24)
clients_to_remove = []
@@ -72,7 +81,7 @@ class RateLimiter:
)
def get_client_stats(self, client_id: str) -> Dict[str, int]:
"""Get statistics for a specific client"""
"""Get statistics for a specific client."""
client_requests = self.clients.get(client_id, deque())
now = datetime.utcnow()

171
models_restructure.md Normal file
View File

@@ -0,0 +1,171 @@
# Models Restructure Plan
## New Structure
```
models/
├── database/
│ ├── __init__.py # Import all models for easy access
│ ├── base.py # Base model class and mixins
│ ├── user.py # User model
│ ├── shop.py # Shop, ShopProduct models
│ ├── product.py # Product model
│ ├── stock.py # Stock model
│ └── marketplace.py # MarketplaceImportJob model
└── api/
├── __init__.py # Common imports
├── base.py # Base response models
├── auth.py # User auth models (register, login, response)
├── shop.py # Shop management models
├── product.py # Product CRUD models
├── stock.py # Stock operation models
├── marketplace.py # Marketplace import models
└── stats.py # Statistics response models
```
## File Contents Breakdown
### Database Models
**models/database/base.py:**
```python
from datetime import datetime
from sqlalchemy import Column, DateTime
from app.core.database import Base
class TimestampMixin:
"""Mixin to add created_at and updated_at timestamps to models"""
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
)
```
**models/database/user.py:**
- `User` model (lines 12-37 from your current file)
**models/database/shop.py:**
- `Shop` model (lines 40-75)
- `ShopProduct` model (lines 155-199)
**models/database/product.py:**
- `Product` model (lines 78-152)
**models/database/stock.py:**
- `Stock` model (lines 202-232)
**models/database/marketplace.py:**
- `MarketplaceImportJob` model (lines 235-284)
### API Models
**models/api/base.py:**
```python
from typing import List, TypeVar, Generic
from pydantic import BaseModel
T = TypeVar('T')
class ListResponse(BaseModel, Generic[T]):
"""Generic list response model"""
items: List[T]
total: int
skip: int
limit: int
class StatusResponse(BaseModel):
"""Generic status response"""
success: bool
message: str
```
**models/api/auth.py:**
- `UserRegister` (lines 12-34)
- `UserLogin` (lines 37-46)
- `UserResponse` (lines 49-61)
- `LoginResponse` (lines 64-69)
**models/api/shop.py:**
- `ShopCreate` (lines 72-103)
- `ShopUpdate` (lines 106-122)
- `ShopResponse` (lines 125-145)
- `ShopListResponse` (lines 148-153)
- `ShopProductCreate` (lines 247-270)
- `ShopProductResponse` (lines 273-293)
**models/api/product.py:**
- `ProductBase` (lines 156-193)
- `ProductCreate` (lines 196-206)
- `ProductUpdate` (lines 209-211)
- `ProductResponse` (lines 214-221)
- `ProductListResponse` (lines 408-413)
- `ProductDetailResponse` (lines 416-420)
**models/api/stock.py:**
- `StockBase` (lines 296-300)
- `StockCreate` (lines 303-305)
- `StockAdd` (lines 308-310)
- `StockUpdate` (lines 313-315)
- `StockResponse` (lines 318-327)
- `StockLocationResponse` (lines 330-333)
- `StockSummaryResponse` (lines 336-342)
**models/api/marketplace.py:**
- `MarketplaceImportRequest` (lines 345-381)
- `MarketplaceImportJobResponse` (lines 384-399)
**models/api/stats.py:**
- `StatsResponse` (lines 423-431)
- `MarketplaceStatsResponse` (lines 434-439)
## Migration Steps
### Step 1: Create the new structure
1. Create the new directories
2. Create `__init__.py` files
3. Move the base classes first
### Step 2: Database models migration
1. Extract models one by one, starting with `User`
2. Update imports in each file
3. Test database connections after each model
### Step 3: API models migration
1. Extract API models by domain
2. Update imports in route files
3. Test API endpoints after each model group
### Step 4: Update imports across the application
- Update all route files to use new import paths
- Update service files
- Update test files
## Benefits of This Structure
1. **Domain separation**: Related models are grouped together
2. **Easier maintenance**: Smaller files are easier to navigate and modify
3. **Reduced conflicts**: Multiple developers can work on different domains
4. **Better testability**: Can test model groups independently
5. **Clear dependencies**: Import relationships become more explicit
## Import Examples After Restructure
```python
# In route files
from models.database import User, Product, Stock
from models.api.auth import UserLogin, UserResponse
from models.api.product import ProductCreate, ProductListResponse
# Or specific imports
from models.database.product import Product
from models.api.product import ProductCreate, ProductResponse
```
## Consideration for Relationships
When splitting database models, be careful with SQLAlchemy relationships:
- Keep related models in the same file if they have tight coupling
- Use string references for relationships across files: `relationship("User")` instead of `relationship(User)`
- Consider lazy imports in `__init__.py` files to avoid circular imports
This restructure will make your codebase much more maintainable as it grows!

476
models_restructure_script.ps1 Normal file
View File

@@ -0,0 +1,476 @@
# restructure_models.ps1 - PowerShell script to restructure models from single files to domain-organized structure
Write-Host "🔄 Starting models restructure..." -ForegroundColor Cyan
# Create new directory structure for models
Write-Host "📁 Creating models directory structure..." -ForegroundColor Yellow
$modelDirectories = @(
"models\database",
"models\api"
)
foreach ($dir in $modelDirectories) {
New-Item -Path $dir -ItemType Directory -Force | Out-Null
Write-Host " Created: $dir" -ForegroundColor Gray
}
# Backup original model files
Write-Host "💾 Backing up original model files..." -ForegroundColor Yellow
New-Item -Path "models\backup" -ItemType Directory -Force | Out-Null
$originalFiles = @("models\database_models.py", "models\api_models.py")
foreach ($file in $originalFiles) {
if (Test-Path $file) {
Copy-Item $file "models\backup\" -Force
Write-Host " Backed up: $(Split-Path $file -Leaf)" -ForegroundColor Gray
}
}
# Create database models files
Write-Host "📄 Creating database model files..." -ForegroundColor Yellow
# models/database/__init__.py
$databaseInit = @"
# models/database/__init__.py
from .user import User
from .product import Product
from .shop import Shop, ShopProduct
from .stock import Stock
from .marketplace import MarketplaceImportJob
__all__ = [
"User",
"Product",
"Shop", "ShopProduct",
"Stock",
"MarketplaceImportJob",
]
"@
$databaseInit | Out-File -FilePath "models\database\__init__.py" -Encoding UTF8
# models/database/base.py
$databaseBase = @"
# models/database/base.py
from datetime import datetime
from sqlalchemy import Column, DateTime
from app.core.database import Base
class TimestampMixin:
"""Mixin to add created_at and updated_at timestamps to models"""
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
)
"@
$databaseBase | Out-File -FilePath "models\database\base.py" -Encoding UTF8
# models/database/user.py
$userModel = @"
# models/database/user.py
from datetime import datetime
from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy.orm import relationship
from app.core.database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
username = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
role = Column(String, nullable=False, default="user") # user, admin, shop_owner
is_active = Column(Boolean, default=True, nullable=False)
last_login = Column(DateTime, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
)
# Relationships
marketplace_import_jobs = relationship(
"MarketplaceImportJob", back_populates="user"
)
owned_shops = relationship("Shop", back_populates="owner")
def __repr__(self):
return f"<User(username='{self.username}', email='{self.email}', role='{self.role}')>"
"@
$userModel | Out-File -FilePath "models\database\user.py" -Encoding UTF8
# models/database/product.py
$productModel = @"
# models/database/product.py
from datetime import datetime
from sqlalchemy import Column, DateTime, Index, Integer, String
from sqlalchemy.orm import relationship
from app.core.database import Base
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
product_id = Column(String, unique=True, index=True, nullable=False)
title = Column(String, nullable=False)
description = Column(String)
link = Column(String)
image_link = Column(String)
availability = Column(String, index=True) # Index for filtering
price = Column(String)
brand = Column(String, index=True) # Index for filtering
gtin = Column(String, index=True) # Index for stock lookups
mpn = Column(String)
condition = Column(String)
adult = Column(String)
multipack = Column(Integer)
is_bundle = Column(String)
age_group = Column(String)
color = Column(String)
gender = Column(String)
material = Column(String)
pattern = Column(String)
size = Column(String)
size_type = Column(String)
size_system = Column(String)
item_group_id = Column(String)
google_product_category = Column(String, index=True) # Index for filtering
product_type = Column(String)
custom_label_0 = Column(String)
custom_label_1 = Column(String)
custom_label_2 = Column(String)
custom_label_3 = Column(String)
custom_label_4 = Column(String)
additional_image_link = Column(String)
sale_price = Column(String)
unit_pricing_measure = Column(String)
unit_pricing_base_measure = Column(String)
identifier_exists = Column(String)
shipping = Column(String)
currency = Column(String)
# New marketplace fields
marketplace = Column(
String, index=True, nullable=True, default="Letzshop"
) # Index for marketplace filtering
shop_name = Column(String, index=True, nullable=True) # Index for shop filtering
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
)
# Relationship to stock (one-to-many via GTIN)
stock_entries = relationship(
"Stock",
foreign_keys="Stock.gtin",
primaryjoin="Product.gtin == Stock.gtin",
viewonly=True,
)
shop_products = relationship("ShopProduct", back_populates="product")
# Additional indexes for marketplace queries
__table_args__ = (
Index(
"idx_marketplace_shop", "marketplace", "shop_name"
), # Composite index for marketplace+shop queries
Index(
"idx_marketplace_brand", "marketplace", "brand"
), # Composite index for marketplace+brand queries
)
def __repr__(self):
return (
f"<Product(product_id='{self.product_id}', title='{self.title}', marketplace='{self.marketplace}', "
f"shop='{self.shop_name}')>"
)
"@
$productModel | Out-File -FilePath "models\database\product.py" -Encoding UTF8
# models/database/shop.py
$shopModel = @"
# models/database/shop.py
from datetime import datetime
from sqlalchemy import Boolean, Column, DateTime, Float, ForeignKey, Index, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import relationship
from app.core.database import Base
class Shop(Base):
__tablename__ = "shops"
id = Column(Integer, primary_key=True, index=True)
shop_code = Column(
String, unique=True, index=True, nullable=False
) # e.g., "TECHSTORE", "FASHIONHUB"
shop_name = Column(String, nullable=False) # Display name
description = Column(Text)
owner_id = Column(Integer, ForeignKey("users.id"), nullable=False)
# Contact information
contact_email = Column(String)
contact_phone = Column(String)
website = Column(String)
# Business information
business_address = Column(Text)
tax_number = Column(String)
# Status
is_active = Column(Boolean, default=True)
is_verified = Column(Boolean, default=False)
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
owner = relationship("User", back_populates="owned_shops")
shop_products = relationship("ShopProduct", back_populates="shop")
marketplace_import_jobs = relationship(
"MarketplaceImportJob", back_populates="shop"
)
class ShopProduct(Base):
__tablename__ = "shop_products"
id = Column(Integer, primary_key=True, index=True)
shop_id = Column(Integer, ForeignKey("shops.id"), nullable=False)
product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
# Shop-specific overrides (can override the main product data)
shop_product_id = Column(String) # Shop's internal product ID
shop_price = Column(Float) # Override main product price
shop_sale_price = Column(Float)
shop_currency = Column(String)
shop_availability = Column(String) # Override availability
shop_condition = Column(String)
# Shop-specific metadata
is_featured = Column(Boolean, default=False)
is_active = Column(Boolean, default=True)
display_order = Column(Integer, default=0)
# Inventory management
min_quantity = Column(Integer, default=1)
max_quantity = Column(Integer)
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
shop = relationship("Shop", back_populates="shop_products")
product = relationship("Product", back_populates="shop_products")
# Constraints
__table_args__ = (
UniqueConstraint("shop_id", "product_id", name="uq_shop_product"),
Index("idx_shop_product_active", "shop_id", "is_active"),
Index("idx_shop_product_featured", "shop_id", "is_featured"),
)
"@
$shopModel | Out-File -FilePath "models\database\shop.py" -Encoding UTF8
# models/database/stock.py
$stockModel = @"
# models/database/stock.py
from datetime import datetime
from sqlalchemy import Column, DateTime, ForeignKey, Index, Integer, String, UniqueConstraint
from sqlalchemy.orm import relationship
from app.core.database import Base
class Stock(Base):
__tablename__ = "stock"
id = Column(Integer, primary_key=True, index=True)
gtin = Column(
String, index=True, nullable=False
) # Foreign key relationship would be ideal
location = Column(String, nullable=False, index=True)
quantity = Column(Integer, nullable=False, default=0)
reserved_quantity = Column(Integer, default=0) # For orders being processed
shop_id = Column(Integer, ForeignKey("shops.id")) # Optional: shop-specific stock
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(
DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
)
# Relationships
shop = relationship("Shop")
# Composite unique constraint to prevent duplicate GTIN-location combinations
__table_args__ = (
UniqueConstraint("gtin", "location", name="uq_stock_gtin_location"),
Index(
"idx_stock_gtin_location", "gtin", "location"
), # Composite index for efficient queries
)
def __repr__(self):
return f"<Stock(gtin='{self.gtin}', location='{self.location}', quantity={self.quantity})>"
"@
$stockModel | Out-File -FilePath "models\database\stock.py" -Encoding UTF8
# models/database/marketplace.py
$marketplaceModel = @"
# models/database/marketplace.py
from datetime import datetime
from sqlalchemy import Column, DateTime, ForeignKey, Index, Integer, String
from sqlalchemy.orm import relationship
from app.core.database import Base
class MarketplaceImportJob(Base):
__tablename__ = "marketplace_import_jobs"
id = Column(Integer, primary_key=True, index=True)
status = Column(
String, nullable=False, default="pending"
) # pending, processing, completed, failed, completed_with_errors
source_url = Column(String, nullable=False)
marketplace = Column(
String, nullable=False, index=True, default="Letzshop"
) # Index for marketplace filtering
shop_name = Column(String, nullable=False, index=True) # Index for shop filtering
shop_id = Column(
Integer, ForeignKey("shops.id"), nullable=False
) # Add proper foreign key
user_id = Column(
Integer, ForeignKey("users.id"), nullable=False
) # Foreign key to users table
# Results
imported_count = Column(Integer, default=0)
updated_count = Column(Integer, default=0)
error_count = Column(Integer, default=0)
total_processed = Column(Integer, default=0)
# Error handling
error_message = Column(String)
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
started_at = Column(DateTime)
completed_at = Column(DateTime)
# Relationship to user
user = relationship("User", foreign_keys=[user_id])
shop = relationship("Shop", back_populates="marketplace_import_jobs")
# Additional indexes for marketplace import job queries
__table_args__ = (
Index(
"idx_marketplace_import_user_marketplace", "user_id", "marketplace"
), # User's marketplace imports
Index("idx_marketplace_import_shop_status", "status"), # Shop import status
Index("idx_marketplace_import_shop_id", "shop_id"),
)
def __repr__(self):
return (
f"<MarketplaceImportJob(id={self.id}, marketplace='{self.marketplace}', shop='{self.shop_name}', "
f"status='{self.status}', imported={self.imported_count})>"
)
"@
$marketplaceModel | Out-File -FilePath "models\database\marketplace.py" -Encoding UTF8
# Create API models files
Write-Host "📄 Creating API model files..." -ForegroundColor Yellow
# models/api/__init__.py
$apiInit = @"
# models/api/__init__.py
from .auth import UserRegister, UserLogin, UserResponse, LoginResponse
from .product import ProductBase, ProductCreate, ProductUpdate, ProductResponse, ProductListResponse, ProductDetailResponse
from .shop import ShopCreate, ShopUpdate, ShopResponse, ShopListResponse, ShopProductCreate, ShopProductResponse
from .stock import StockBase, StockCreate, StockAdd, StockUpdate, StockResponse, StockLocationResponse, StockSummaryResponse
from .marketplace import MarketplaceImportRequest, MarketplaceImportJobResponse
from .stats import StatsResponse, MarketplaceStatsResponse
__all__ = [
# Auth models
"UserRegister", "UserLogin", "UserResponse", "LoginResponse",
# Product models
"ProductBase", "ProductCreate", "ProductUpdate", "ProductResponse",
"ProductListResponse", "ProductDetailResponse",
# Shop models
"ShopCreate", "ShopUpdate", "ShopResponse", "ShopListResponse",
"ShopProductCreate", "ShopProductResponse",
# Stock models
"StockBase", "StockCreate", "StockAdd", "StockUpdate", "StockResponse",
"StockLocationResponse", "StockSummaryResponse",
# Marketplace models
"MarketplaceImportRequest", "MarketplaceImportJobResponse",
# Stats models
"StatsResponse", "MarketplaceStatsResponse",
]
"@
$apiInit | Out-File -FilePath "models\api\__init__.py" -Encoding UTF8
# models/api/base.py
$apiBase = @"
# models/api/base.py
from typing import List, TypeVar, Generic
from pydantic import BaseModel
T = TypeVar('T')
class ListResponse(BaseModel, Generic[T]):
"""Generic list response model"""
items: List[T]
total: int
skip: int
limit: int
class StatusResponse(BaseModel):
"""Generic status response"""
success: bool
message: str
"@
$apiBase | Out-File -FilePath "models\api\base.py" -Encoding UTF8
Write-Host "✅ Models directory structure created!" -ForegroundColor Green
Write-Host ""
Write-Host "📋 Next steps:" -ForegroundColor Cyan
Write-Host "1. You'll need to manually extract the remaining API model classes from api_models.py:" -ForegroundColor White
Write-Host " - Auth models → models\api\auth.py" -ForegroundColor Gray
Write-Host " - Product models → models\api\product.py" -ForegroundColor Gray
Write-Host " - Shop models → models\api\shop.py" -ForegroundColor Gray
Write-Host " - Stock models → models\api\stock.py" -ForegroundColor Gray
Write-Host " - Marketplace models → models\api\marketplace.py" -ForegroundColor Gray
Write-Host " - Stats models → models\api\stats.py" -ForegroundColor Gray
Write-Host ""
Write-Host "2. Update imports throughout your application:" -ForegroundColor White
Write-Host " Old: from models.database_models import User, Product" -ForegroundColor Gray
Write-Host " New: from models.database import User, Product" -ForegroundColor Gray
Write-Host ""
Write-Host " Old: from models.api_models import UserCreate, ProductResponse" -ForegroundColor Gray
Write-Host " New: from models.api import UserRegister, ProductResponse" -ForegroundColor Gray
Write-Host ""
Write-Host "3. Test imports after restructure:" -ForegroundColor White
Write-Host " python -c \"from models.database import User, Product, Shop\"" -ForegroundColor Yellow
Write-Host " python -c \"from models.api import UserRegister, ProductResponse\"" -ForegroundColor Yellow
Write-Host ""
Write-Host "4. Files to update (search and replace imports):" -ForegroundColor White
Write-Host " - All route files in api/v1/" -ForegroundColor Gray
Write-Host " - Service files in app/services/" -ForegroundColor Gray
Write-Host " - Test files" -ForegroundColor Gray
Write-Host " - Any utility files" -ForegroundColor Gray
Write-Host ""
Write-Host "✨ Database models restructure completed!" -ForegroundColor Green
Write-Host "📚 Original files backed up in models\backup\" -ForegroundColor Green

123
performance_tests.py Normal file
View File

@@ -0,0 +1,123 @@
# tests/performance/test_api_performance.py
import time
import pytest
from models.database_models import Product
@pytest.mark.performance
@pytest.mark.slow
@pytest.mark.database
class TestPerformance:
def test_product_list_performance(self, client, auth_headers, db):
"""Test performance of product listing with many products"""
# Create multiple products
products = []
for i in range(100):
product = Product(
product_id=f"PERF{i:03d}",
title=f"Performance Test Product {i}",
price=f"{i}.99",
marketplace="Performance",
)
products.append(product)
db.add_all(products)
db.commit()
# Time the request
start_time = time.time()
response = client.get("/api/v1/product?limit=100", headers=auth_headers)
end_time = time.time()
assert response.status_code == 200
assert len(response.json()["products"]) == 100
assert end_time - start_time < 2.0 # Should complete within 2 seconds
def test_search_performance(self, client, auth_headers, db):
"""Test search performance"""
# Create products with searchable content
products = []
for i in range(50):
product = Product(
product_id=f"SEARCH{i:03d}",
title=f"Searchable Product {i}",
description=f"This is a searchable product number {i}",
brand="SearchBrand",
marketplace="SearchMarket",
)
products.append(product)
db.add_all(products)
db.commit()
# Time search request
start_time = time.time()
response = client.get("/api/v1/product?search=Searchable", headers=auth_headers)
end_time = time.time()
assert response.status_code == 200
assert response.json()["total"] == 50
assert end_time - start_time < 1.0 # Search should be fast
def test_database_query_performance(self, client, auth_headers, db):
"""Test database query performance with complex filters"""
# Create products with various attributes for filtering
products = []
brands = ["Brand1", "Brand2", "Brand3"]
marketplaces = ["Market1", "Market2"]
for i in range(200):
product = Product(
product_id=f"COMPLEX{i:03d}",
title=f"Complex Product {i}",
brand=brands[i % 3],
marketplace=marketplaces[i % 2],
price=f"{10 + (i % 50)}.99",
google_product_category=f"Category{i % 5}",
)
products.append(product)
db.add_all(products)
db.commit()
# Test complex filtering performance
start_time = time.time()
response = client.get(
"/api/v1/product?brand=Brand1&marketplace=Market1&limit=50",
headers=auth_headers
)
end_time = time.time()
assert response.status_code == 200
assert end_time - start_time < 1.5 # Complex query should still be reasonably fast
def test_pagination_performance_large_dataset(self, client, auth_headers, db):
"""Test pagination performance with large dataset"""
# Create a large dataset
products = []
for i in range(500):
product = Product(
product_id=f"LARGE{i:04d}",
title=f"Large Dataset Product {i}",
marketplace="LargeTest",
)
products.append(product)
db.add_all(products)
db.commit()
# Test pagination performance at different offsets
offsets = [0, 100, 250, 400]
for offset in offsets:
start_time = time.time()
response = client.get(
f"/api/v1/product?skip={offset}&limit=20",
headers=auth_headers
)
end_time = time.time()
assert response.status_code == 200
assert len(response.json()["products"]) == 20
assert end_time - start_time < 1.0 # Pagination should be fast regardless of offset

113
powershell_migration_script.ps1 Normal file
View File

@@ -0,0 +1,113 @@
# migrate_tests.ps1 - PowerShell script to migrate your test structure
Write-Host "🚀 Starting test structure migration..." -ForegroundColor Cyan
# Create new directory structure
Write-Host "📁 Creating directory structure..." -ForegroundColor Yellow
$directories = @(
"tests\fixtures",
"tests\unit", "tests\unit\models", "tests\unit\utils", "tests\unit\services",
"tests\integration", "tests\integration\api", "tests\integration\api\v1", "tests\integration\security",
"tests\performance",
"tests\system",
"tests\test_data", "tests\test_data\csv"
)
foreach ($dir in $directories) {
New-Item -Path $dir -ItemType Directory -Force | Out-Null
Write-Host " Created: $dir" -ForegroundColor Gray
}
# Create __init__.py files
Write-Host "📄 Creating __init__.py files..." -ForegroundColor Yellow
$initFiles = @(
"tests\fixtures\__init__.py",
"tests\unit\__init__.py", "tests\unit\models\__init__.py", "tests\unit\utils\__init__.py", "tests\unit\services\__init__.py",
"tests\integration\__init__.py", "tests\integration\api\__init__.py", "tests\integration\api\v1\__init__.py", "tests\integration\security\__init__.py",
"tests\performance\__init__.py",
"tests\system\__init__.py"
)
foreach ($file in $initFiles) {
New-Item -Path $file -ItemType File -Force | Out-Null
Write-Host " Created: $file" -ForegroundColor Gray
}
# Create conftest.py files for each test category
Write-Host "⚙️ Creating conftest.py files..." -ForegroundColor Yellow
$conftestFiles = @(
"tests\unit\conftest.py",
"tests\integration\conftest.py",
"tests\performance\conftest.py",
"tests\system\conftest.py"
)
foreach ($file in $conftestFiles) {
New-Item -Path $file -ItemType File -Force | Out-Null
Write-Host " Created: $file" -ForegroundColor Gray
}
# Backup original files
Write-Host "💾 Backing up original files..." -ForegroundColor Yellow
New-Item -Path "tests\backup" -ItemType Directory -Force | Out-Null
if (Test-Path "tests\conftest.py") {
Copy-Item "tests\conftest.py" "tests\backup\" -Force
Write-Host " Backed up: conftest.py" -ForegroundColor Gray
}
Get-ChildItem "tests\test_*.py" -ErrorAction SilentlyContinue | ForEach-Object {
Copy-Item $_.FullName "tests\backup\" -Force
Write-Host " Backed up: $($_.Name)" -ForegroundColor Gray
}
# Create sample test data file
Write-Host "📊 Creating sample test data..." -ForegroundColor Yellow
$csvContent = @"
product_id,title,price,currency,brand,marketplace
TEST001,Sample Product 1,19.99,EUR,TestBrand,TestMarket
TEST002,Sample Product 2,29.99,EUR,TestBrand,TestMarket
TEST003,Sample Product 3,39.99,USD,AnotherBrand,TestMarket
"@
$csvContent | Out-File -FilePath "tests\test_data\csv\sample_products.csv" -Encoding UTF8
Write-Host " Created: sample_products.csv" -ForegroundColor Gray
Write-Host "✅ Directory structure created!" -ForegroundColor Green
Write-Host ""
Write-Host "📋 Next steps:" -ForegroundColor Cyan
Write-Host "1. Copy the fixture files I provided to tests\fixtures\" -ForegroundColor White
Write-Host "2. Update tests\conftest.py with the new version" -ForegroundColor White
Write-Host "3. Move test files to their new locations:" -ForegroundColor White
Write-Host " - test_database.py → tests\unit\models\test_database_models.py" -ForegroundColor Gray
Write-Host " - test_utils.py → tests\unit\utils\test_data_processing.py" -ForegroundColor Gray
Write-Host " - test_admin_service.py → tests\unit\services\" -ForegroundColor Gray
Write-Host " - test_admin.py → tests\integration\api\v1\test_admin_endpoints.py" -ForegroundColor Gray
Write-Host " - test_pagination.py → tests\integration\api\v1\" -ForegroundColor Gray
Write-Host " - test_performance.py → tests\performance\test_api_performance.py" -ForegroundColor Gray
Write-Host " - test_error_handling.py → tests\system\" -ForegroundColor Gray
Write-Host " - Split test_security.py into security subdirectory" -ForegroundColor Gray
Write-Host ""
Write-Host "4. Update imports in moved test files" -ForegroundColor White
Write-Host "5. Add pytest markers to test classes" -ForegroundColor White
Write-Host "6. Update pytest.ini with the enhanced configuration" -ForegroundColor White
Write-Host ""
Write-Host "🧪 Test the migration with:" -ForegroundColor Cyan
Write-Host "pytest tests\unit -v" -ForegroundColor Yellow
Write-Host "pytest tests\integration -v" -ForegroundColor Yellow
Write-Host "pytest -m unit" -ForegroundColor Yellow
Write-Host ""
Write-Host "🔧 Quick test commands after migration:" -ForegroundColor Cyan
Write-Host "pytest -m unit # Fast unit tests" -ForegroundColor White
Write-Host "pytest -m integration # Integration tests" -ForegroundColor White
Write-Host "pytest -m `"not slow`" # Skip slow tests" -ForegroundColor White
Write-Host "pytest tests\unit\models\ # Model tests only" -ForegroundColor White
Write-Host "pytest --cov=app --cov-report=html # Coverage report" -ForegroundColor White
Write-Host ""
Write-Host "✨ Migration structure ready! Follow the steps above to complete the migration." -ForegroundColor Green
Write-Host "📚 All your original files are backed up in tests\backup\" -ForegroundColor Green
# Optional: Pause to let user read the output
Write-Host ""
Write-Host "Press any key to continue..." -ForegroundColor DarkGray
$null = $Host.UI.RawUI.ReadKey("NoEcho,IncludeKeyDown")

108
product_fixtures.py Normal file
View File

@@ -0,0 +1,108 @@
# tests/fixtures/product_fixtures.py
import uuid
import pytest
from models.database_models import Product
@pytest.fixture
def test_product(db):
"""Create a test product"""
product = Product(
product_id="TEST001",
title="Test Product",
description="A test product",
price="10.99",
currency="EUR",
brand="TestBrand",
gtin="1234567890123",
availability="in stock",
marketplace="Letzshop",
shop_name="TestShop",
)
db.add(product)
db.commit()
db.refresh(product)
return product
@pytest.fixture
def unique_product(db):
"""Create a unique product for tests that need isolated product data"""
unique_id = str(uuid.uuid4())[:8]
product = Product(
product_id=f"UNIQUE_{unique_id}",
title=f"Unique Product {unique_id}",
description=f"A unique test product {unique_id}",
price="19.99",
currency="EUR",
brand=f"UniqueBrand_{unique_id}",
gtin=f"123456789{unique_id[:4]}",
availability="in stock",
marketplace="Letzshop",
shop_name=f"UniqueShop_{unique_id}",
google_product_category=f"UniqueCategory_{unique_id}",
)
db.add(product)
db.commit()
db.refresh(product)
return product
@pytest.fixture
def multiple_products(db):
"""Create multiple products for testing statistics and pagination"""
unique_id = str(uuid.uuid4())[:8]
products = []
for i in range(5):
product = Product(
product_id=f"MULTI_{unique_id}_{i}",
title=f"Multi Product {i} {unique_id}",
description=f"Multi test product {i}",
price=f"{10 + i}.99",
currency="EUR",
brand=f"MultiBrand_{i % 3}", # Create 3 different brands
marketplace=f"MultiMarket_{i % 2}", # Create 2 different marketplaces
shop_name=f"MultiShop_{i}",
google_product_category=f"MultiCategory_{i % 2}", # Create 2 different categories
gtin=f"1234567890{i}{unique_id[:2]}",
)
products.append(product)
db.add_all(products)
db.commit()
for product in products:
db.refresh(product)
return products
def create_unique_product_factory():
"""Factory function to create unique products in tests"""
def _create_product(db, **kwargs):
unique_id = str(uuid.uuid4())[:8]
defaults = {
"product_id": f"FACTORY_{unique_id}",
"title": f"Factory Product {unique_id}",
"price": "15.99",
"currency": "EUR",
"marketplace": "TestMarket",
"shop_name": "TestShop",
}
defaults.update(kwargs)
product = Product(**defaults)
db.add(product)
db.commit()
db.refresh(product)
return product
return _create_product
@pytest.fixture
def product_factory():
"""Fixture that provides a product factory function"""
return create_unique_product_factory()

1
requirements.txt
View File

@@ -1,3 +1,4 @@
starlette==0.27.0
# requirements.txt
# Core FastAPI and web framework
fastapi==0.104.1

159
restructure-model.md Normal file
View File

@@ -0,0 +1,159 @@
Yes, absolutely! As your application grows, splitting the models into separate files by domain/entity is a common and recommended practice. Here's how you could restructure your models:
## Recommended Structure:
```
models/
├── database/
│ ├── __init__.py # Import all models for easy access
│ ├── base.py # Base model class and common mixins
│ ├── user.py # User, UserProfile models
│ ├── auth.py # Authentication-related models
│ ├── product.py # Product, ProductVariant models
│ ├── stock.py # Stock, StockMovement models
│ ├── shop.py # Shop, ShopLocation models
│ ├── marketplace.py # Marketplace integration models
│ └── admin.py # Admin-specific models
└── api/
├── __init__.py # Common imports
├── base.py # Base Pydantic models
├── auth.py # Login, Token, User response models
├── product.py # Product request/response models
├── stock.py # Stock operation models
├── shop.py # Shop management models
├── marketplace.py # Marketplace import models
├── admin.py # Admin operation models
└── stats.py # Statistics response models
```
## Benefits of this approach:
1. **Maintainability**: Easier to find and modify specific model types
2. **Team collaboration**: Reduces merge conflicts when multiple developers work on different features
3. **Import clarity**: Clear imports like `from models.database.product import Product`
4. **Logical grouping**: Related models stay together
5. **Reduced file size**: Each file focuses on a specific domain
## Example implementation:
**models/database/__init__.py:**
```python
from .user import User, UserProfile
from .product import Product, ProductVariant
from .stock import Stock, StockMovement
from .shop import Shop, ShopLocation
# ... other imports
__all__ = [
"User", "UserProfile",
"Product", "ProductVariant",
"Stock", "StockMovement",
"Shop", "ShopLocation",
# ... other exports
]
```
**models/api/__init__.py:**
```python
from .auth import LoginRequest, TokenResponse, UserResponse
from .product import ProductCreate, ProductResponse, ProductUpdate
# ... other imports
__all__ = [
"LoginRequest", "TokenResponse", "UserResponse",
"ProductCreate", "ProductResponse", "ProductUpdate",
# ... other exports
]
```
This way, you can still import everything easily in your route files:
```python
from models.database import Product, Stock
from models.api import ProductCreate, ProductResponse
```
The split makes most sense when you have 5+ models per domain or when individual model files exceed ~200-300 lines.
Looking at your current `api_models.py` and `database_models.py`, they're definitely getting quite large and would benefit from being split up! Here's how I'd recommend restructuring them:Based on your current models, here are the key considerations for splitting them:
## Immediate Benefits You'll See:
1. **File Size**: Your current `api_models.py` is 439 lines and `database_models.py` is 284 lines - splitting will make them much more manageable
2. **Domain Clarity**: You have clear domains (auth, shop, product, stock, marketplace) that naturally separate
3. **Import Organization**: Instead of importing everything from two massive files, you'll have clear, purposeful imports
## Potential Challenges to Watch For:
1. **Circular Imports**: With SQLAlchemy relationships, you might need to use string references like `relationship("User")` instead of the class directly
2. **Shared Validators**: You have some common validation logic that might need to be moved to a base/utils module
3. **Import Updates**: You'll need to update all your route files, but this is a one-time cost
## Recommended Migration Order:
1. **Start with API models** - they're easier since they don't have complex relationships
2. **Begin with auth models** - they're the most independent
3. **Then do database models** - be careful with relationships
4. **Update imports gradually** - do it route by route to avoid breaking everything at once
Would you like me to create the actual split files for any specific domain (like auth or products) to get you started?
I've created a comprehensive PowerShell script that automatically generates the models restructure outlined in our discussion. This script:
## What the Script Creates:
**Database Models Structure:**
- `models/database/` with separate files for each domain
- `base.py` with TimestampMixin for common functionality
- `user.py`, `product.py`, `shop.py`, `stock.py`, `marketplace.py`
- Proper `__init__.py` with all imports configured
**API Models Structure:**
- `models/api/` directory structure
- `base.py` with generic response models
- `__init__.py` with comprehensive imports ready
**Key Features:**
- **Automatic backup** of your original files
- **Complete database models** extracted and properly organized
- **Import structure** ready with `__init__.py` files
- **Clear next steps** printed after completion
## Usage:
```powershell
# Navigate to your project root
cd your-project-directory
# Run the script
.\restructure_models.ps1
```
## What You'll Still Need to Do Manually:
The script creates the database models completely, but you'll need to extract the API model classes from your existing `api_models.py` into the new structure:
1. **Auth models** → `models/api/auth.py`
2. **Product models** → `models/api/product.py`
3. **Shop models** → `models/api/shop.py`
4. **Stock models** → `models/api/stock.py`
5. **Marketplace models** → `models/api/marketplace.py`
6. **Stats models** → `models/api/stats.py`
## Import Updates Needed:
After running the script, update imports throughout your codebase:
```python
# Old imports
from models.database_models import User, Product, Shop
from models.api_models import UserRegister, ProductResponse
# New imports
from models.database import User, Product, Shop
from models.api import UserRegister, ProductResponse
```
The script will complete the database models restructure automatically and provide you with a clear roadmap for finishing the API models migration.

9
scripts/setup_dev.py
View File

@@ -1,15 +1,14 @@
# scripts/setup_dev.py
# !/usr/bin/env python3
"""Development environment setup script"""
"""Development environment setup script."""
import os
import subprocess
import sys
from pathlib import Path
def run_command(command, description):
"""Run a shell command and handle errors"""
"""Run a shell command and handle errors."""
print(f"Running: {description}")
try:
subprocess.run(command, shell=True, check=True)
@@ -21,7 +20,7 @@ def run_command(command, description):
def setup_alembic():
"""Set up Alembic for database migrations"""
"""Set up Alembic for database migrations."""
alembic_dir = Path("alembic")
# Check if alembic directory exists and has necessary files
@@ -103,7 +102,7 @@ else:
def setup_environment():
"""Set up the development environment"""
"""Set up the development environment."""
print("🚀 Setting up ecommerce API development environment...")
# Check Python version

163
shop_fixtures.py Normal file
View File

@@ -0,0 +1,163 @@
# tests/fixtures/shop_fixtures.py
import uuid
import pytest
from models.database_models import Shop, ShopProduct, Stock
@pytest.fixture
def test_shop(db, test_user):
"""Create a test shop with unique shop code"""
unique_id = str(uuid.uuid4())[:8] # Short unique identifier
shop = Shop(
shop_code=f"TESTSHOP_{unique_id}",
shop_name=f"Test Shop {unique_id}",
owner_id=test_user.id,
is_active=True,
is_verified=True,
)
db.add(shop)
db.commit()
db.refresh(shop)
return shop
@pytest.fixture
def unique_shop(db, test_user):
"""Create a unique shop for tests that need isolated shop data"""
unique_id = str(uuid.uuid4())[:8]
shop = Shop(
shop_code=f"UNIQUESHOP_{unique_id}",
shop_name=f"Unique Test Shop {unique_id}",
description=f"A unique test shop {unique_id}",
owner_id=test_user.id,
is_active=True,
is_verified=True,
)
db.add(shop)
db.commit()
db.refresh(shop)
return shop
@pytest.fixture
def inactive_shop(db, other_user):
"""Create an inactive shop owned by other_user"""
unique_id = str(uuid.uuid4())[:8]
shop = Shop(
shop_code=f"INACTIVE_{unique_id}",
shop_name=f"Inactive Shop {unique_id}",
owner_id=other_user.id,
is_active=False,
is_verified=False,
)
db.add(shop)
db.commit()
db.refresh(shop)
return shop
@pytest.fixture
def verified_shop(db, other_user):
"""Create a verified shop owned by other_user"""
unique_id = str(uuid.uuid4())[:8]
shop = Shop(
shop_code=f"VERIFIED_{unique_id}",
shop_name=f"Verified Shop {unique_id}",
owner_id=other_user.id,
is_active=True,
is_verified=True,
)
db.add(shop)
db.commit()
db.refresh(shop)
return shop
@pytest.fixture
def shop_product(db, test_shop, unique_product):
"""Create a shop product relationship"""
shop_product = ShopProduct(
shop_id=test_shop.id, product_id=unique_product.id, is_active=True
)
# Add optional fields if they exist in your model
if hasattr(ShopProduct, "shop_price"):
shop_product.shop_price = 24.99
if hasattr(ShopProduct, "is_featured"):
shop_product.is_featured = False
if hasattr(ShopProduct, "min_quantity"):
shop_product.min_quantity = 1
db.add(shop_product)
db.commit()
db.refresh(shop_product)
return shop_product
@pytest.fixture
def test_stock(db, test_product, test_shop):
"""Create test stock entry"""
unique_id = str(uuid.uuid4())[:8].upper() # Short unique identifier
stock = Stock(
gtin=test_product.gtin, # Use gtin instead of product_id
location=f"WAREHOUSE_A_{unique_id}",
quantity=10,
reserved_quantity=0,
shop_id=test_shop.id, # Add shop_id reference
)
db.add(stock)
db.commit()
db.refresh(stock)
return stock
@pytest.fixture
def multiple_stocks(db, multiple_products, test_shop):
"""Create multiple stock entries for testing"""
stocks = []
for i, product in enumerate(multiple_products):
stock = Stock(
gtin=product.gtin,
location=f"LOC_{i}",
quantity=10 + (i * 5), # Different quantities
reserved_quantity=i,
shop_id=test_shop.id,
)
stocks.append(stock)
db.add_all(stocks)
db.commit()
for stock in stocks:
db.refresh(stock)
return stocks
def create_unique_shop_factory():
"""Factory function to create unique shops in tests"""
def _create_shop(db, owner_id, **kwargs):
unique_id = str(uuid.uuid4())[:8]
defaults = {
"shop_code": f"FACTORY_{unique_id}",
"shop_name": f"Factory Shop {unique_id}",
"owner_id": owner_id,
"is_active": True,
"is_verified": False,
}
defaults.update(kwargs)
shop = Shop(**defaults)
db.add(shop)
db.commit()
db.refresh(shop)
return shop
return _create_shop
@pytest.fixture
def shop_factory():
"""Fixture that provides a shop factory function"""
return create_unique_shop_factory()

218
specific_test_migration.md Normal file
View File

@@ -0,0 +1,218 @@
# Your Specific Test Migration Plan
## New Directory Structure
```
tests/
├── conftest.py # Your current global fixtures (keep most of it)
├── pytest.ini # New - test configuration
├── fixtures/ # Extract fixtures from conftest.py
│ ├── __init__.py
│ ├── auth_fixtures.py # User, admin fixtures
│ ├── product_fixtures.py # Product-related fixtures
│ ├── shop_fixtures.py # Shop-related fixtures
│ └── database_fixtures.py # DB setup fixtures
├── unit/ # Fast, isolated tests
│ ├── __init__.py
│ ├── conftest.py # Unit-specific fixtures
│ ├── models/
│ │ ├── __init__.py
│ │ └── test_database_models.py # From test_database.py
│ ├── utils/
│ │ ├── __init__.py
│ │ └── test_data_processing.py # From test_utils.py
│ └── services/
│ ├── __init__.py
│ └── test_admin_service.py # Keep as-is, move here
├── integration/ # Multi-component tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── conftest.py # API client fixtures
│ │ └── v1/
│ │ ├── __init__.py
│ │ └── test_admin_endpoints.py # From test_admin.py
│ └── security/
│ ├── __init__.py
│ ├── test_authentication.py # Split from test_security.py
│ ├── test_authorization.py # Split from test_security.py
│ └── test_input_validation.py # Split from test_security.py
├── system/ # Full system tests
│ ├── __init__.py
│ ├── conftest.py
│ └── test_error_handling.py # From test_error_handling.py
└── test_data/ # Test data files
└── csv/
└── sample_products.csv
```
## Specific File Mappings
### 1. Global Fixtures (Keep in main conftest.py)
**tests/conftest.py** - Keep these fixtures:
- `engine`, `testing_session_local`, `db`
- `client`
- `cleanup`
### 2. Extract to Fixture Files
**tests/fixtures/auth_fixtures.py:**
```python
# Move these from conftest.py:
- auth_manager
- test_user
- test_admin
- auth_headers
- admin_headers
- other_user
```
**tests/fixtures/product_fixtures.py:**
```python
# Move these from conftest.py:
- test_product
- unique_product
- multiple_products
- product_factory
```
**tests/fixtures/shop_fixtures.py:**
```python
# Move these from conftest.py:
- test_shop
- unique_shop
- inactive_shop
- verified_shop
- shop_product
- shop_factory
```
### 3. Unit Tests
**tests/unit/models/test_database_models.py:**
Move content from `test_database.py`:
- `TestDatabaseModels.test_user_model`
- `TestDatabaseModels.test_product_model`
- `TestDatabaseModels.test_stock_model`
- `TestDatabaseModels.test_shop_model_with_owner`
- `TestDatabaseModels.test_database_constraints`
**tests/unit/utils/test_data_processing.py:**
Move content from `test_utils.py`:
- `TestGTINProcessor` (entire class)
- `TestPriceProcessor` (entire class)
**tests/unit/services/test_admin_service.py:**
Keep `test_admin_service.py` exactly as-is, just move to this location.
### 4. Integration Tests
**tests/integration/api/v1/test_admin_endpoints.py:**
Move content from `test_admin.py`:
- `TestAdminAPI` (entire class) - all your admin endpoint tests
**tests/integration/security/test_authentication.py:**
Move from `test_security.py`:
```python
class TestAuthentication:
def test_protected_endpoint_without_auth(self, client):
# From test_security.py
def test_protected_endpoint_with_invalid_token(self, client):
# From test_security.py
```
**tests/integration/security/test_authorization.py:**
Move from `test_security.py`:
```python
class TestAuthorization:
def test_admin_endpoint_requires_admin_role(self, client, auth_headers):
# From test_security.py
```
**tests/integration/security/test_input_validation.py:**
Move from `test_security.py`:
```python
class TestInputValidation:
def test_sql_injection_prevention(self, client, auth_headers):
# From test_security.py
# def test_input_validation(self, client, auth_headers):
# # Your commented XSS test
```
### 5. System Tests
**tests/system/test_error_handling.py:**
Move `test_error_handling.py` as-is to this location.
## Migration Steps
### Step 1: Create Structure
```bash
mkdir -p tests/{fixtures,unit/{models,utils,services},integration/{api/v1,security},system,test_data}
touch tests/{fixtures,unit,integration,system}/__init__.py
touch tests/unit/{models,utils,services}/__init__.py
touch tests/integration/{api,security}/__init__.py
touch tests/integration/api/v1/__init__.py
```
### Step 2: Create pytest.ini
```ini
[tool:pytest]
testpaths = tests
python_files = test_*.py
addopts = -v --tb=short
markers =
unit: Unit tests - fast, isolated
integration: Integration tests - multiple components
system: System tests - full application
slow: Slow running tests
```
### Step 3: Extract Fixtures
Create fixture files and move relevant fixtures from your current `conftest.py`.
### Step 4: Move Test Files
Move each test file to its new location and update imports.
### Step 5: Update Imports
After moving files, update imports like:
```python
# Old import in test files
# No explicit imports needed since fixtures were in conftest.py
# New imports in test files
from tests.fixtures.auth_fixtures import test_user, auth_headers
from tests.fixtures.product_fixtures import test_product
```
## Running Tests by Category
```bash
# Fast unit tests during development
pytest tests/unit -m unit
# Integration tests before commit
pytest tests/integration -m integration
# Full test suite
pytest
# Specific domain tests
pytest tests/unit/services/ tests/integration/api/v1/test_admin_endpoints.py
# Your current debug tests (move to integration/security)
pytest tests/integration/security/ -v -s
```
## Benefits for Your Specific Tests
1. **Your admin tests** get separated into service logic (unit) vs API endpoints (integration)
2. **Your security tests** get properly categorized by concern
3. **Your database tests** become proper model unit tests
4. **Your utility tests** become isolated unit tests
5. **Your error handling** becomes system-level testing
This structure will make your test suite much more maintainable and allow for faster development cycles!

113
system_error_handling.py Normal file
View File

@@ -0,0 +1,113 @@
# tests/system/test_error_handling.py
import pytest
@pytest.mark.system
class TestErrorHandling:
def test_invalid_json(self, client, auth_headers):
"""Test handling of invalid JSON"""
response = client.post(
"/api/v1/product", headers=auth_headers, content="invalid json"
)
assert response.status_code == 422 # Validation error
def test_missing_required_fields(self, client, auth_headers):
"""Test handling of missing required fields"""
response = client.post(
"/api/v1/product", headers=auth_headers, json={"title": "Test"}
) # Missing product_id
assert response.status_code == 422
def test_invalid_authentication(self, client):
"""Test handling of invalid authentication"""
response = client.get(
"/api/v1/product", headers={"Authorization": "Bearer invalid_token"}
)
assert response.status_code == 401 # Token is not valid
def test_nonexistent_resource(self, client, auth_headers):
"""Test handling of nonexistent resource access"""
response = client.get("/api/v1/product/NONEXISTENT", headers=auth_headers)
assert response.status_code == 404
response = client.get("/api/v1/shop/NONEXISTENT", headers=auth_headers)
assert response.status_code == 404
def test_duplicate_resource_creation(self, client, auth_headers, test_product):
"""Test handling of duplicate resource creation"""
product_data = {
"product_id": test_product.product_id, # Duplicate ID
"title": "Another Product",
}
response = client.post(
"/api/v1/product", headers=auth_headers, json=product_data
)
assert response.status_code == 400
def test_server_error_handling(self, client, auth_headers):
"""Test handling of server errors"""
# This would test 500 errors if you have endpoints that can trigger them
# For now, test that the error handling middleware works
response = client.get("/api/v1/nonexistent-endpoint", headers=auth_headers)
assert response.status_code == 404
def test_rate_limiting_behavior(self, client, auth_headers):
"""Test rate limiting behavior if implemented"""
# Make multiple rapid requests to test rate limiting
responses = []
for i in range(10):
response = client.get("/api/v1/product", headers=auth_headers)
responses.append(response)
# All should succeed unless rate limiting is very aggressive
# Adjust based on your rate limiting configuration
success_count = sum(1 for r in responses if r.status_code == 200)
assert success_count >= 5 # At least half should succeed
def test_malformed_requests(self, client, auth_headers):
"""Test handling of various malformed requests"""
# Test extremely long URLs
long_search = "x" * 10000
response = client.get(f"/api/v1/product?search={long_search}", headers=auth_headers)
# Should handle gracefully, either 200 with no results or 422 for too long
assert response.status_code in [200, 422]
# Test special characters in parameters
special_chars = "!@#$%^&*(){}[]|\\:;\"'<>,.?/~`"
response = client.get(f"/api/v1/product?search={special_chars}", headers=auth_headers)
# Should handle gracefully
assert response.status_code in [200, 422]
def test_database_error_recovery(self, client, auth_headers):
"""Test application behavior during database issues"""
# This is more complex to test - you'd need to simulate DB issues
# For now, just test that basic operations work
response = client.get("/api/v1/product", headers=auth_headers)
assert response.status_code == 200
def test_content_type_errors(self, client, auth_headers):
"""Test handling of incorrect content types"""
# Send XML to JSON endpoint
response = client.post(
"/api/v1/product",
headers={**auth_headers, "Content-Type": "application/xml"},
content="<xml>not json</xml>"
)
assert response.status_code in [400, 422, 415] # Bad request or unsupported media type
def test_large_payload_handling(self, client, auth_headers):
"""Test handling of unusually large payloads"""
# Create a very large product description
large_data = {
"product_id": "LARGE_TEST",
"title": "Large Test Product",
"description": "x" * 50000 # Very long description
}
response = client.post("/api/v1/product", headers=auth_headers, json=large_data)
# Should either accept it or reject with 422 (too large)
assert response.status_code in [200, 201, 422, 413]

262
test_migration_readme.md Normal file
View File

@@ -0,0 +1,262 @@
# Test Structure Migration Guide
This document outlines the complete restructuring of the FastAPI application test suite from a single-folder approach to a comprehensive, scalable test organization.
## Overview
The test suite has been reorganized from:
```
tests/
├── conftest.py
├── test_admin.py
├── test_admin_service.py
├── test_database.py
├── test_error_handling.py
├── test_pagination.py
├── test_performance.py
├── test_security.py
├── test_utils.py
└── pytest.ini
```
To a structured, domain-organized approach that scales with application growth and provides better development workflow support.
## New Structure
```
tests/
├── conftest.py # Core test configuration and database fixtures
├── pytest.ini # Enhanced pytest configuration
├── fixtures/ # Shared test fixtures organized by domain
│ ├── __init__.py
│ ├── auth_fixtures.py # Authentication: users, tokens, headers
│ ├── product_fixtures.py # Products: test products, factories
│ ├── shop_fixtures.py # Shops: shops, stock, shop-products
│ └── marketplace_fixtures.py # Marketplace: import jobs
├── unit/ # Fast, isolated component tests
│ ├── __init__.py
│ ├── conftest.py # Unit test specific fixtures
│ ├── models/
│ │ ├── __init__.py
│ │ └── test_database_models.py # Database model tests
│ ├── utils/
│ │ ├── __init__.py
│ │ └── test_data_processing.py # Utility function tests
│ └── services/
│ ├── __init__.py
│ └── test_admin_service.py # Business logic tests
├── integration/ # Multi-component interaction tests
│ ├── __init__.py
│ ├── conftest.py # Integration test fixtures
│ ├── api/
│ │ ├── __init__.py
│ │ ├── conftest.py # API client fixtures
│ │ └── v1/
│ │ ├── __init__.py
│ │ ├── test_admin_endpoints.py # Admin API endpoint tests
│ │ └── test_pagination.py # Pagination functionality tests
│ └── security/
│ ├── __init__.py
│ ├── test_authentication.py # Authentication mechanism tests
│ ├── test_authorization.py # Permission and role tests
│ └── test_input_validation.py # Input security and validation tests
├── performance/ # Performance and load tests
│ ├── __init__.py
│ ├── conftest.py # Performance test fixtures
│ └── test_api_performance.py # API performance benchmarks
├── system/ # End-to-end system behavior tests
│ ├── __init__.py
│ ├── conftest.py # System test fixtures
│ └── test_error_handling.py # Application error handling tests
└── test_data/ # Test data files
└── csv/
└── sample_products.csv # Sample CSV data for testing
```
## Migration Changes Made
### 1. Fixture Organization
**Problem Solved:** The original `conftest.py` contained 370+ lines mixing different concerns.
**Solution:** Extracted domain-specific fixtures into separate modules:
- **auth_fixtures.py**: User authentication, tokens, and headers
- **product_fixtures.py**: Product creation, factories, and bulk data
- **shop_fixtures.py**: Shop management, stock, and shop-product relationships
- **marketplace_fixtures.py**: Import job fixtures and helpers
### 2. Test Categorization
**Problem Solved:** All tests ran together, making development cycles slow.
**Solution:** Organized tests by execution speed and scope:
- **Unit Tests**: Fast (< 100ms), isolated, no external dependencies
- **Integration Tests**: Medium speed, multiple components, database required
- **Performance Tests**: Slow, large datasets, timing-sensitive
- **System Tests**: Full application behavior, error scenarios
### 3. Enhanced pytest Configuration
**Previous:** Basic configuration with limited markers.
**Current:** Comprehensive configuration including:
```ini
[tool:pytest]
addopts =
-v --tb=short --strict-markers --strict-config
--color=yes --durations=10 --showlocals -ra
--cov=app --cov=models --cov=utils --cov=middleware
--cov-report=term-missing --cov-report=html:htmlcov
--cov-fail-under=80
markers =
unit: Unit tests - fast, isolated components
integration: Integration tests - multiple components
system: System tests - full application behavior
performance: Performance and load tests
slow: Slow running tests
# Domain-specific markers
auth: Authentication and authorization tests
products: Product management functionality
admin: Admin functionality and permissions
# ... additional markers
```
## Usage Examples
### Development Workflow
```bash
# Fast feedback during development (unit tests only)
pytest -m unit
# Before committing (unit + integration)
pytest -m "unit or integration"
# Full test suite (CI/CD)
pytest
# Skip slow tests during development
pytest -m "not slow"
```
### Domain-Specific Testing
```bash
# Test specific functionality
pytest -m auth # Authentication tests
pytest -m products # Product-related tests
pytest -m admin # Admin functionality
# Test specific layers
pytest tests/unit/ # All unit tests
pytest tests/integration/api/ # API integration tests
pytest tests/performance/ # Performance tests only
```
### Coverage and Reporting
```bash
# Generate coverage report
pytest --cov-report=html
# Find slowest tests
pytest --durations=0
# Detailed failure information
pytest -vvv --tb=long
```
## Key Improvements
### 1. Scalability
- **Modular fixture organization** allows teams to work on different domains without conflicts
- **Clear separation of concerns** makes it easy to add new test categories
- **Factory patterns** provide flexible test data generation
### 2. Development Speed
- **Fast unit tests** provide immediate feedback (typically runs in < 10 seconds)
- **Selective test execution** allows developers to run only relevant tests
- **Parallel execution support** ready (uncomment `addopts = -n auto` in pytest.ini)
### 3. Maintainability
- **Domain organization** makes it easy to locate and update tests
- **Consistent fixture patterns** reduce duplication
- **Clear naming conventions** improve code readability
### 4. CI/CD Pipeline Optimization
- **Staged test execution**: Run fast tests first, fail fast on basic issues
- **Performance monitoring**: Track test execution times and performance regressions
- **Coverage tracking**: Maintain code coverage standards
## Migration Process
### Automated Setup
1. Run the migration script:
```bash
bash migrate_tests.sh
```
### Manual Steps
1. **Copy fixture files** to `tests/fixtures/`
2. **Update main conftest.py** with the new version
3. **Move test files** to their new locations:
- `test_database.py` → `tests/unit/models/test_database_models.py`
- `test_utils.py` → `tests/unit/utils/test_data_processing.py`
- `test_admin_service.py` → `tests/unit/services/`
- `test_admin.py` → `tests/integration/api/v1/test_admin_endpoints.py`
- Split `test_security.py` into `tests/integration/security/` files
- `test_pagination.py` → `tests/integration/api/v1/`
- `test_performance.py` → `tests/performance/test_api_performance.py`
- `test_error_handling.py` → `tests/system/`
4. **Update imports** in moved files
5. **Add pytest markers** to test classes
6. **Test incrementally**:
```bash
pytest tests/unit -v
pytest tests/integration -v
pytest -m unit
```
## Benefits Realized
### For Developers
- **Faster feedback loops**: Unit tests complete in seconds
- **Focused testing**: Run only tests relevant to current work
- **Better debugging**: Clear test categorization helps identify issues quickly
### For Teams
- **Reduced merge conflicts**: Domain separation allows parallel development
- **Clear ownership**: Teams can own test domains matching their code ownership
- **Consistent patterns**: Standardized fixture and test organization
### For CI/CD
- **Optimized pipeline**: Fast tests run first, expensive tests only when needed
- **Performance monitoring**: Track performance regressions over time
- **Coverage enforcement**: Maintain quality standards automatically
## Troubleshooting
### Import Issues
If you encounter import errors after migration:
1. Ensure all `__init__.py` files are created
2. Check that `pytest_plugins` in main `conftest.py` points to correct fixture modules
3. Verify fixture dependencies (e.g., `test_shop` depends on `test_user`)
### Fixture Not Found
If pytest can't find fixtures:
1. Check that fixture modules are listed in `pytest_plugins`
2. Ensure fixture dependencies are imported properly
3. Verify fixture scope (session vs function) matches usage
### Performance Issues
If tests are running slowly:
1. Check that unit tests don't have database dependencies
2. Consider using `pytest-xdist` for parallel execution
3. Review slow tests with `pytest --durations=10`
## Future Enhancements
This structure supports future additions:
- **E2E tests**: Add `tests/e2e/` for complete user journey testing
- **Contract tests**: Add API contract testing with tools like Pact
- **Load tests**: Expand performance testing with tools like Locust
- **Visual tests**: Add screenshot testing for frontend components
- **Mutation tests**: Add mutation testing to verify test quality
The restructured test suite provides a solid foundation for maintaining high code quality as the application scales.

352
tests_restructure.md Normal file
View File

@@ -0,0 +1,352 @@
# Tests Folder Restructure Plan
## Current vs Recommended Structure
### Before (Single Folder)
```
tests/
├── test_auth.py
├── test_products.py
├── test_stock.py
├── test_shops.py
├── test_marketplace.py
├── test_admin.py
├── test_stats.py
├── test_database.py
├── test_utils.py
├── conftest.py
└── ...more files
```
### After (Organized Structure)
```
tests/
├── conftest.py # Global test configuration and fixtures
├── pytest.ini # Pytest configuration
├── __init__.py
├── fixtures/ # Shared test fixtures
│ ├── __init__.py
│ ├── auth_fixtures.py # Auth-related fixtures
│ ├── product_fixtures.py # Product fixtures
│ ├── shop_fixtures.py # Shop fixtures
│ └── database_fixtures.py # Database setup fixtures
├── unit/ # Unit tests (isolated, fast)
│ ├── __init__.py
│ ├── conftest.py # Unit test specific fixtures
│ ├── models/ # Test database and API models
│ │ ├── __init__.py
│ │ ├── test_user_model.py
│ │ ├── test_product_model.py
│ │ ├── test_shop_model.py
│ │ └── test_stock_model.py
│ ├── utils/ # Test utility functions
│ │ ├── __init__.py
│ │ ├── test_data_processing.py
│ │ ├── test_csv_processor.py
│ │ └── test_database_utils.py
│ ├── services/ # Test business logic
│ │ ├── __init__.py
│ │ ├── test_auth_service.py
│ │ ├── test_product_service.py
│ │ └── test_marketplace_service.py
│ └── middleware/ # Test middleware components
│ ├── __init__.py
│ ├── test_auth_middleware.py
│ ├── test_rate_limiter.py
│ └── test_error_handler.py
├── integration/ # Integration tests (multiple components)
│ ├── __init__.py
│ ├── conftest.py # Integration test fixtures
│ ├── api/ # API endpoint tests
│ │ ├── __init__.py
│ │ ├── conftest.py # API test fixtures (test client, etc.)
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── test_auth_endpoints.py
│ │ │ ├── test_product_endpoints.py
│ │ │ ├── test_shop_endpoints.py
│ │ │ ├── test_stock_endpoints.py
│ │ │ ├── test_marketplace_endpoints.py
│ │ │ ├── test_admin_endpoints.py
│ │ │ └── test_stats_endpoints.py
│ │ └── test_api_main.py # Test API router setup
│ ├── database/ # Database integration tests
│ │ ├── __init__.py
│ │ ├── test_crud_operations.py
│ │ ├── test_relationships.py
│ │ └── test_migrations.py
│ └── workflows/ # End-to-end workflow tests
│ ├── __init__.py
│ ├── test_product_import_workflow.py
│ ├── test_shop_setup_workflow.py
│ └── test_stock_management_workflow.py
├── e2e/ # End-to-end tests (full application)
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_user_registration_flow.py
│ ├── test_marketplace_import_flow.py
│ └── test_complete_shop_setup.py
├── performance/ # Performance and load tests
│ ├── __init__.py
│ ├── test_api_performance.py
│ ├── test_database_performance.py
│ └── test_import_performance.py
└── test_data/ # Test data files
├── csv/
│ ├── valid_products.csv
│ ├── invalid_products.csv
│ └── large_dataset.csv
├── json/
│ ├── sample_product.json
│ └── marketplace_response.json
└── fixtures/
├── test_users.json
└── test_products.json
```
## File Organization Principles
### 1. Test Types Separation
- **Unit Tests**: Fast, isolated tests for individual functions/classes
- **Integration Tests**: Tests that involve multiple components working together
- **E2E Tests**: Full application flow tests
- **Performance Tests**: Load and performance testing
### 2. Fixture Organization
```python
# tests/fixtures/auth_fixtures.py
import pytest
from models.database import User
from utils.auth import create_access_token
@pytest.fixture
def test_user_data():
return {
"email": "test@example.com",
"username": "testuser",
"password": "testpass123"
}
@pytest.fixture
def authenticated_user(db_session, test_user_data):
user = User(**test_user_data)
db_session.add(user)
db_session.commit()
return user
@pytest.fixture
def auth_headers(authenticated_user):
token = create_access_token(data={"sub": authenticated_user.username})
return {"Authorization": f"Bearer {token}"}
```
### 3. Conftest.py Structure
```python
# tests/conftest.py (Global fixtures)
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app
from app.core.database import get_db, Base
# Database fixtures
@pytest.fixture(scope="session")
def test_engine():
engine = create_engine("sqlite:///test.db")
Base.metadata.create_all(bind=engine)
yield engine
Base.metadata.drop_all(bind=engine)
@pytest.fixture
def db_session(test_engine):
TestingSessionLocal = sessionmaker(bind=test_engine)
session = TestingSessionLocal()
yield session
session.close()
@pytest.fixture
def client(db_session):
def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
```
```python
# tests/integration/api/conftest.py (API-specific fixtures)
import pytest
@pytest.fixture
def api_client(client):
"""Pre-configured API client for integration tests"""
return client
@pytest.fixture
def admin_client(client, admin_auth_headers):
"""API client with admin authentication"""
client.headers.update(admin_auth_headers)
return client
```
## Test Naming Conventions
### File Naming
- `test_*.py` for all test files
- Mirror your app structure: `test_product_endpoints.py` for `api/v1/products.py`
- Use descriptive names: `test_marketplace_import_workflow.py`
### Test Function Naming
```python
# Good test naming patterns
def test_create_product_with_valid_data_returns_201():
pass
def test_create_product_without_title_returns_422():
pass
def test_get_product_by_id_returns_product_data():
pass
def test_get_nonexistent_product_returns_404():
pass
def test_update_product_stock_updates_quantity():
pass
def test_marketplace_import_with_invalid_csv_fails_gracefully():
pass
```
## Running Tests by Category
### Pytest Configuration (pytest.ini)
```ini
[tool:pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
-v
--tb=short
--strict-markers
markers =
unit: Unit tests
integration: Integration tests
e2e: End-to-end tests
performance: Performance tests
slow: Slow running tests
database: Tests that require database
external: Tests that require external services
```
### Running Specific Test Categories
```bash
# Run only unit tests (fast)
pytest tests/unit -m unit
# Run only integration tests
pytest tests/integration -m integration
# Run all tests except slow ones
pytest -m "not slow"
# Run only database tests
pytest -m database
# Run tests for specific domain
pytest tests/unit/models/ tests/integration/api/v1/test_product_endpoints.py
# Run with coverage
pytest --cov=app --cov-report=html
```
## Benefits of This Structure
1. **Faster Development**: Developers can run relevant test subsets
2. **Clear Separation**: Easy to understand what each test covers
3. **Parallel Execution**: Can run different test types in parallel
4. **Maintenance**: Easier to maintain and update tests
5. **CI/CD Pipeline**: Can have different pipeline stages for different test types
## Migration Strategy
### Phase 1: Create Structure
1. Create new directory structure
2. Move global fixtures to `fixtures/` directory
3. Update `conftest.py` files
### Phase 2: Move Unit Tests
1. Start with utility and model tests
2. Move to `tests/unit/` with appropriate subdirectories
3. Update imports and fixtures
### Phase 3: Move Integration Tests
1. Move API endpoint tests to `tests/integration/api/`
2. Create database integration tests
3. Add workflow tests
### Phase 4: Add Missing Coverage
1. Add performance tests if needed
2. Create E2E tests for critical flows
3. Add proper test markers
## Example Test File Structure
### Unit Test Example
```python
# tests/unit/models/test_product_model.py
import pytest
from models.database import Product
class TestProductModel:
def test_product_creation_with_valid_data(self, db_session):
product = Product(
product_id="TEST123",
title="Test Product",
price="99.99"
)
db_session.add(product)
db_session.commit()
assert product.id is not None
assert product.product_id == "TEST123"
assert product.title == "Test Product"
def test_product_gtin_relationship_with_stock(self, db_session, test_product, test_stock):
# Test the GTIN-based relationship
assert test_product.gtin in [stock.gtin for stock in test_product.stock_entries]
```
### Integration Test Example
```python
# tests/integration/api/v1/test_product_endpoints.py
import pytest
class TestProductEndpoints:
def test_create_product_endpoint(self, client, auth_headers, valid_product_data):
response = client.post(
"/api/v1/products/",
json=valid_product_data,
headers=auth_headers
)
assert response.status_code == 201
assert response.json()["product_id"] == valid_product_data["product_id"]
def test_get_products_with_pagination(self, client, auth_headers, multiple_products):
response = client.get(
"/api/v1/products/?skip=0&limit=10",
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert "products" in data
assert "total" in data
assert len(data["products"]) <= 10
```
This structure will scale beautifully as your application grows and makes it much easier for your team to maintain and extend the test suite!

116
unit_admin_service.py Normal file
View File

@@ -0,0 +1,116 @@
# tests/unit/services/test_admin_service.py
import pytest
from fastapi import HTTPException
from app.services.admin_service import AdminService
from models.database_models import MarketplaceImportJob, Shop
@pytest.mark.unit
@pytest.mark.admin
class TestAdminService:
"""Test suite for AdminService following the application's testing patterns"""
def setup_method(self):
"""Setup method following the same pattern as product service tests"""
self.service = AdminService()
def test_get_all_users(self, db, test_user, test_admin):
"""Test getting all users with pagination"""
users = self.service.get_all_users(db, skip=0, limit=10)
assert len(users) >= 2 # test_user + test_admin
user_ids = [user.id for user in users]
assert test_user.id in user_ids
assert test_admin.id in user_ids
def test_get_all_users_with_pagination(self, db, test_user, test_admin):
"""Test user pagination works correctly"""
users = self.service.get_all_users(db, skip=0, limit=1)
assert len(users) == 1
users_second_page = self.service.get_all_users(db, skip=1, limit=1)
assert len(users_second_page) == 1
assert users[0].id != users_second_page[0].id
def test_toggle_user_status_deactivate(self, db, test_user, test_admin):
"""Test deactivating a user"""
assert test_user.is_active is True
user, message = self.service.toggle_user_status(db, test_user.id, test_admin.id)
assert user.id == test_user.id
assert user.is_active is False
assert f"{user.username} has been deactivated" in message
def test_toggle_user_status_activate(self, db, test_user, test_admin):
"""Test activating a user"""
# First deactivate the user
test_user.is_active = False
db.commit()
user, message = self.service.toggle_user_status(db, test_user.id, test_admin.id)
assert user.id == test_user.id
assert user.is_active is True
assert f"{user.username} has been activated" in message
def test_toggle_user_status_user_not_found(self, db, test_admin):
"""Test toggle user status when user not found"""
with pytest.raises(HTTPException) as exc_info:
self.service.toggle_user_status(db, 99999, test_admin.id)
assert exc_info.value.status_code == 404
assert "User not found" in str(exc_info.value.detail)
def test_toggle_user_status_cannot_deactivate_self(self, db, test_admin):
"""Test that admin cannot deactivate their own account"""
with pytest.raises(HTTPException) as exc_info:
self.service.toggle_user_status(db, test_admin.id, test_admin.id)
assert exc_info.value.status_code == 400
assert "Cannot deactivate your own account" in str(exc_info.value.detail)
def test_get_all_shops(self, db, test_shop):
"""Test getting all shops with total count"""
shops, total = self.service.get_all_shops(db, skip=0, limit=10)
assert total >= 1
assert len(shops) >= 1
shop_codes = [shop.shop_code for shop in shops]
assert test_shop.shop_code in shop_codes
def test_verify_shop_mark_verified(self, db, test_shop):
"""Test marking shop as verified"""
# Ensure shop starts unverified
test_shop.is_verified = False
db.commit()
shop, message = self.service.verify_shop(db, test_shop.id)
assert shop.id == test_shop.id
assert shop.is_verified is True
assert f"{shop.shop_code} has been verified" in message
def test_verify_shop_not_found(self, db):
"""Test verify shop when shop not found"""
with pytest.raises(HTTPException) as exc_info:
self.service.verify_shop(db, 99999)
assert exc_info.value.status_code == 404
assert "Shop not found" in str(exc_info.value.detail)
def test_get_marketplace_import_jobs_no_filters(self, db, test_marketplace_job):
"""Test getting marketplace import jobs without filters using fixture"""
result = self.service.get_marketplace_import_jobs(db, skip=0, limit=10)
assert len(result) >= 1
# Find our test job in the results
test_job = next(
(job for job in result if job.job_id == test_marketplace_job.id), None
)
assert test_job is not None
assert test_job.marketplace == test_marketplace_job.marketplace
assert test_job.shop_name == test_marketplace_job.shop_name
assert test_job.status == test_marketplace_job.status

121
unit_data_processing.py Normal file
View File

@@ -0,0 +1,121 @@
# tests/unit/utils/test_data_processing.py
import pytest
from utils.data_processing import GTINProcessor, PriceProcessor
@pytest.mark.unit
class TestGTINProcessor:
def setup_method(self):
self.processor = GTINProcessor()
def test_normalize_valid_gtin(self):
"""Test GTIN normalization with valid inputs"""
# Test EAN-13
assert self.processor.normalize("1234567890123") == "1234567890123"
# Test UPC-A (12 digits)
assert self.processor.normalize("123456789012") == "123456789012"
# Test with decimal point
assert self.processor.normalize("123456789012.0") == "123456789012"
# Test EAN-8
assert self.processor.normalize("12345678") == "12345678"
def test_normalize_invalid_gtin(self):
"""Test GTIN normalization with invalid inputs"""
assert self.processor.normalize("") is None
assert self.processor.normalize(None) is None
assert self.processor.normalize("abc") is None
# Test short number (gets padded)
assert self.processor.normalize("123") == "0000000000123"
def test_normalize_gtin_with_formatting(self):
"""Test GTIN normalization with various formatting"""
# Test with spaces
assert self.processor.normalize("123 456 789 012") == "123456789012"
# Test with dashes
assert self.processor.normalize("123-456-789-012") == "123456789012"
# Test with mixed formatting
assert self.processor.normalize("123 456-789 012") == "123456789012"
def test_validate_gtin(self):
"""Test GTIN validation"""
assert self.processor.validate("1234567890123") is True
assert self.processor.validate("123456789012") is True
assert self.processor.validate("12345678") is True
assert self.processor.validate("123") is False
assert self.processor.validate("") is False
assert self.processor.validate(None) is False
def test_gtin_checksum_validation(self):
"""Test GTIN checksum validation if implemented"""
# This test would verify checksum calculation if your GTINProcessor implements it
# For now, we'll test the structure validation
assert self.processor.validate("1234567890123") is True
assert self.processor.validate("12345678901234") is True # 14 digits
assert self.processor.validate("123456789012345") is False # 15 digits
@pytest.mark.unit
class TestPriceProcessor:
def setup_method(self):
self.processor = PriceProcessor()
def test_parse_price_currency_eur(self):
"""Test EUR price parsing"""
price, currency = self.processor.parse_price_currency("8.26 EUR")
assert price == "8.26"
assert currency == "EUR"
# Test with euro symbol
price, currency = self.processor.parse_price_currency("8.26 €")
assert price == "8.26"
assert currency == "EUR"
def test_parse_price_currency_usd(self):
"""Test USD price parsing"""
price, currency = self.processor.parse_price_currency("$12.50")
assert price == "12.50"
assert currency == "USD"
price, currency = self.processor.parse_price_currency("12.50 USD")
assert price == "12.50"
assert currency == "USD"
def test_parse_price_currency_comma_decimal(self):
"""Test price parsing with comma as decimal separator"""
price, currency = self.processor.parse_price_currency("8,26 EUR")
assert price == "8.26"
assert currency == "EUR"
def test_parse_invalid_price(self):
"""Test invalid price parsing"""
price, currency = self.processor.parse_price_currency("")
assert price is None
assert currency is None
price, currency = self.processor.parse_price_currency(None)
assert price is None
assert currency is None
def test_parse_price_edge_cases(self):
"""Test edge cases in price parsing"""
# Test price without currency
price, currency = self.processor.parse_price_currency("15.99")
assert price == "15.99"
# currency might be None or default value
# Test currency before price
price, currency = self.processor.parse_price_currency("EUR 25.50")
assert price == "25.50"
assert currency == "EUR"
# Test with multiple decimal places
price, currency = self.processor.parse_price_currency("12.999 USD")
assert price == "12.999"
assert currency == "USD"

96
unit_database_models.py Normal file
View File

@@ -0,0 +1,96 @@
# tests/unit/models/test_database_models.py
import pytest
from models.database_models import Product, Shop, Stock, User
@pytest.mark.unit
@pytest.mark.database
class TestDatabaseModels:
def test_user_model(self, db):
"""Test User model creation and relationships"""
user = User(
email="db_test@example.com",
username="dbtest",
hashed_password="hashed_password_123",
role="user",
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
assert user.id is not None
assert user.email == "db_test@example.com"
assert user.created_at is not None
assert user.updated_at is not None
def test_product_model(self, db):
"""Test Product model creation"""
product = Product(
product_id="DB_TEST_001",
title="Database Test Product",
description="Testing product model",
price="25.99",
currency="USD",
brand="DBTest",
gtin="1234567890123",
availability="in stock",
marketplace="TestDB",
shop_name="DBTestShop",
)
db.add(product)
db.commit()
db.refresh(product)
assert product.id is not None
assert product.product_id == "DB_TEST_001"
assert product.created_at is not None
def test_stock_model(self, db):
"""Test Stock model creation"""
stock = Stock(gtin="1234567890123", location="DB_WAREHOUSE", quantity=150)
db.add(stock)
db.commit()
db.refresh(stock)
assert stock.id is not None
assert stock.gtin == "1234567890123"
assert stock.location == "DB_WAREHOUSE"
assert stock.quantity == 150
def test_shop_model_with_owner(self, db, test_user):
"""Test Shop model with owner relationship"""
shop = Shop(
shop_code="DBTEST",
shop_name="Database Test Shop",
description="Testing shop model",
owner_id=test_user.id,
is_active=True,
is_verified=False,
)
db.add(shop)
db.commit()
db.refresh(shop)
assert shop.id is not None
assert shop.shop_code == "DBTEST"
assert shop.owner_id == test_user.id
assert shop.owner.username == test_user.username
def test_database_constraints(self, db):
"""Test database constraints and unique indexes"""
# Test unique product_id constraint
product1 = Product(product_id="UNIQUE_001", title="Product 1")
db.add(product1)
db.commit()
# This should raise an integrity error
with pytest.raises(Exception): # Could be IntegrityError or similar
product2 = Product(product_id="UNIQUE_001", title="Product 2")
db.add(product2)
db.commit()

92
windows_migration_script.txt Normal file
View File

@@ -0,0 +1,92 @@
@echo off
REM migrate_tests.bat - Windows script to migrate your test structure
echo Starting test structure migration...
REM Create new directory structure
echo Creating directory structure...
mkdir tests\fixtures 2>nul
mkdir tests\unit 2>nul
mkdir tests\unit\models 2>nul
mkdir tests\unit\utils 2>nul
mkdir tests\unit\services 2>nul
mkdir tests\integration 2>nul
mkdir tests\integration\api 2>nul
mkdir tests\integration\api\v1 2>nul
mkdir tests\integration\security 2>nul
mkdir tests\performance 2>nul
mkdir tests\system 2>nul
mkdir tests\test_data 2>nul
mkdir tests\test_data\csv 2>nul
REM Create __init__.py files
echo Creating __init__.py files...
echo. > tests\fixtures\__init__.py
echo. > tests\unit\__init__.py
echo. > tests\unit\models\__init__.py
echo. > tests\unit\utils\__init__.py
echo. > tests\unit\services\__init__.py
echo. > tests\integration\__init__.py
echo. > tests\integration\api\__init__.py
echo. > tests\integration\api\v1\__init__.py
echo. > tests\integration\security\__init__.py
echo. > tests\performance\__init__.py
echo. > tests\system\__init__.py
REM Create conftest.py files for each test category
echo Creating conftest.py files...
echo. > tests\unit\conftest.py
echo. > tests\integration\conftest.py
echo. > tests\performance\conftest.py
echo. > tests\system\conftest.py
REM Backup original files
echo Backing up original files...
mkdir tests\backup 2>nul
copy tests\conftest.py tests\backup\ >nul 2>&1
copy tests\test_*.py tests\backup\ >nul 2>&1
echo Directory structure created!
echo.
echo Next steps:
echo 1. Copy the fixture files I provided to tests\fixtures\
echo 2. Update tests\conftest.py with the new version
echo 3. Move test files to their new locations:
echo - test_database.py to tests\unit\models\test_database_models.py
echo - test_utils.py to tests\unit\utils\test_data_processing.py
echo - test_admin_service.py to tests\unit\services\
echo - test_admin.py to tests\integration\api\v1\test_admin_endpoints.py
echo - test_pagination.py to tests\integration\api\v1\
echo - test_performance.py to tests\performance\test_api_performance.py
echo - test_error_handling.py to tests\system\
echo - Split test_security.py into security subdirectory
echo.
echo 4. Update imports in moved test files
echo 5. Add pytest markers to test classes
echo 6. Update pytest.ini with the enhanced configuration
echo.
echo Test the migration with:
echo pytest tests\unit -v
echo pytest tests\integration -v
echo pytest -m unit
echo.
echo Quick test commands after migration:
echo pytest -m unit # Fast unit tests
echo pytest -m integration # Integration tests
echo pytest -m "not slow" # Skip slow tests
echo pytest tests\unit\models\ # Model tests only
echo pytest --cov=app --cov-report=html # Coverage report
REM Create a sample test data file
echo Creating sample test data...
(
echo product_id,title,price,currency,brand,marketplace
echo TEST001,Sample Product 1,19.99,EUR,TestBrand,TestMarket
echo TEST002,Sample Product 2,29.99,EUR,TestBrand,TestMarket
echo TEST003,Sample Product 3,39.99,USD,AnotherBrand,TestMarket
) > tests\test_data\csv\sample_products.csv
echo.
echo Migration structure ready! Follow the steps above to complete the migration.
echo All your original files are backed up in tests\backup\
pause