128 lines
No EOL
4 KiB
Python
128 lines
No EOL
4 KiB
Python
"""
|
|
Pytest configuration and shared fixtures for integration tests.
|
|
"""
|
|
import asyncio
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
from fastapi.testclient import TestClient
|
|
import database
|
|
from config import settings
|
|
from app import create_app
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_db():
|
|
"""
|
|
Create a fresh SQLite in-memory database for each test.
|
|
Returns a tuple (engine, session_factory).
|
|
"""
|
|
# Create a new in-memory SQLite engine for this test with shared cache
|
|
# Using cache=shared allows multiple connections to share the same in-memory database
|
|
test_engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:?cache=shared",
|
|
echo=False,
|
|
future=True,
|
|
poolclass=StaticPool, # Use static pool to share in-memory DB across connections
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
|
|
# Create tables
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(database.Base.metadata.create_all)
|
|
|
|
# Create session factory
|
|
test_session_factory = async_sessionmaker(
|
|
test_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
|
|
yield test_engine, test_session_factory
|
|
|
|
# Drop tables after test
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(database.Base.metadata.drop_all)
|
|
|
|
await test_engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_session(test_db):
|
|
"""
|
|
Provide an AsyncSession for database operations in tests.
|
|
"""
|
|
_, session_factory = test_db
|
|
async with session_factory() as session:
|
|
yield session
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_app(test_db):
|
|
"""
|
|
Provide a FastAPI app with a fresh in-memory database.
|
|
Overrides the database engine and session factory in the app.
|
|
"""
|
|
test_engine, test_session_factory = test_db
|
|
|
|
# Monkey-patch the database module's engine and AsyncSessionLocal
|
|
original_engine = database.engine
|
|
original_session_factory = database.AsyncSessionLocal
|
|
database.engine = test_engine
|
|
database.AsyncSessionLocal = test_session_factory
|
|
|
|
# Also patch config.settings.database_url to prevent conflicts
|
|
original_database_url = settings.database_url
|
|
settings.database_url = "sqlite+aiosqlite:///:memory:?cache=shared"
|
|
|
|
# Create app with patched database
|
|
app = create_app()
|
|
|
|
# Override get_db dependency to use our test session
|
|
from database import get_db
|
|
async def override_get_db():
|
|
async with test_session_factory() as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
# Ensure app.state.session_factory uses our test session factory
|
|
app.state.session_factory = test_session_factory
|
|
# Ensure route manager uses our test session factory
|
|
app.state.route_manager.async_session_factory = test_session_factory
|
|
|
|
yield app
|
|
|
|
# Restore original values
|
|
database.engine = original_engine
|
|
database.AsyncSessionLocal = original_session_factory
|
|
settings.database_url = original_database_url
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def test_client(test_app):
|
|
"""
|
|
Provide a TestClient with a fresh in-memory database.
|
|
"""
|
|
with TestClient(test_app) as client:
|
|
yield client
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="function")
|
|
async def admin_client(test_client):
|
|
"""
|
|
Provide a TestClient with an authenticated admin session.
|
|
Logs in via POST /admin/login and returns the client with session cookie.
|
|
"""
|
|
client = test_client
|
|
# Perform login
|
|
response = client.post(
|
|
"/admin/login",
|
|
data={"username": "admin", "password": "admin123"},
|
|
follow_redirects=False,
|
|
)
|
|
assert response.status_code == 302
|
|
# The session cookie should be set automatically
|
|
yield client |