# tests/conftest.py import pytest import tempfile import os from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from main import app from app.core.database import get_db, Base from models.database_models import User, Product, Stock, Shop from middleware.auth import AuthManager # 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 fresh database for each test""" # Create all tables Base.metadata.create_all(bind=engine) # Create session db = testing_session_local() # Override the dependency def override_get_db(): try: yield db finally: pass # Don't close here, we'll close in cleanup app.dependency_overrides[get_db] = override_get_db try: yield db finally: db.rollback() # Rollback any uncommitted changes db.close() # Clean up the dependency override if get_db in app.dependency_overrides: del app.dependency_overrides[get_db] # Drop all tables for next test Base.metadata.drop_all(bind=engine) @pytest.fixture(scope="function") def client(db): """Create a test client with database dependency override""" return TestClient(app) @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""" hashed_password = auth_manager.hash_password("testpass123") user = User( email="test@example.com", username="testuser", 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""" hashed_password = auth_manager.hash_password("adminpass123") admin = User( email="admin@example.com", username="admin", hashed_password=hashed_password, role="admin", is_active=True ) db.add(admin) db.commit() db.refresh(admin) return admin @pytest.fixture def auth_headers(client, test_user): """Get authentication headers for test user""" response = client.post("/api/v1/auth/login", json={ "username": "testuser", "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": "admin", "password": "adminpass123" }) assert response.status_code == 200, f"Admin login failed: {response.text}" token = response.json()["access_token"] return {"Authorization": f"Bearer {token}"} @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 test_shop(db, test_user): """Create a test shop""" shop = Shop( shop_code="TESTSHOP", shop_name="Test Shop", owner_id=test_user.id, is_active=True, is_verified=True ) db.add(shop) db.commit() db.refresh(shop) return shop @pytest.fixture def test_stock(db, test_product, test_shop): """Create test stock entry""" stock = Stock( product_id=test_product.product_id, shop_code=test_shop.shop_code, quantity=10, reserved_quantity=0 ) db.add(stock) db.commit() db.refresh(stock) return stock # 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()