11from collections .abc import Generator
2+ from typing import Generator as TypingGenerator
23
34import pytest
45from fastapi .testclient import TestClient
5- from sqlmodel import Session , delete
6+ from sqlmodel import Session , delete , create_engine
7+ from sqlalchemy .orm import sessionmaker
68
79from app .core .config import settings
8- from app .core .db import engine , init_db
10+ from app .core .db import engine as default_engine , init_db
911from app .main import app
12+ from app import api
1013from app .models import Item , User , Image , ImageVariant , ImageProcessingJob
1114from tests .utils .user import authentication_token_from_email
1215from tests .utils .utils import get_superuser_token_headers
1316
1417
18+ @pytest .fixture (scope = "function" )
19+ def db_engine ():
20+ """Create fresh database engine for each test to ensure complete isolation."""
21+ # Use dedicated test database URL to separate from main database
22+ test_db_url = str (settings .TEST_DATABASE_URL )
23+
24+ # Create engine with test-specific settings
25+ engine = create_engine (
26+ test_db_url ,
27+ pool_pre_ping = True ,
28+ echo = False , # Disable SQL logging for tests
29+ )
30+
31+ # Create all tables for each test
32+ from app .models import SQLModel
33+ SQLModel .metadata .create_all (engine )
34+
35+ yield engine
36+
37+ # Cleanup: Drop all tables after each test
38+ SQLModel .metadata .drop_all (engine )
39+ engine .dispose ()
40+
41+
42+ @pytest .fixture (scope = "function" )
43+ def db (db_engine ) -> Generator [Session , None , None ]:
44+ """Create transaction-isolated database session for each test."""
45+ # Bind the session to the engine
46+ with Session (bind = db_engine ) as session :
47+ # Begin a transaction
48+ transaction = session .begin ()
49+
50+ # Initialize database with superuser for tests
51+ init_db (session )
52+
53+ try :
54+ yield session
55+ finally :
56+ # Always rollback the transaction to ensure isolation
57+ try :
58+ transaction .rollback ()
59+ except Exception :
60+ pass # Ignore rollback errors
61+
62+ # Explicit cleanup of session state
63+ session .expunge_all ()
64+
65+
1566@pytest .fixture (scope = "session" )
16- def db () -> Generator [Session , None , None ]:
17- with Session (engine ) as session :
67+ def db_session_scope () -> Generator [Session , None , None ]:
68+ """Session-scoped database fixture for backward compatibility (deprecated)."""
69+ with Session (default_engine ) as session :
1870 init_db (session )
1971 yield session
72+ # Cleanup data at end of session
2073 statement = delete (ImageProcessingJob )
2174 session .execute (statement )
2275 statement = delete (ImageVariant )
@@ -30,18 +83,34 @@ def db() -> Generator[Session, None, None]:
3083 session .commit ()
3184
3285
33- @pytest .fixture (scope = "module " )
86+ @pytest .fixture (scope = "function " )
3487def client (db : Session ) -> Generator [TestClient , None , None ]:
88+ """
89+ Create a test client with database session override.
90+ This ensures each test gets an isolated database session.
91+ """
92+ def override_get_db ():
93+ try :
94+ yield db
95+ finally :
96+ pass # Transaction will be rolled back by db fixture
97+
98+ # Override the dependency
99+ app .dependency_overrides [api .deps .get_db ] = override_get_db
100+
35101 with TestClient (app ) as c :
36102 yield c
37103
104+ # Clean up overrides after test
105+ app .dependency_overrides .clear ()
106+
38107
39- @pytest .fixture (scope = "module " )
108+ @pytest .fixture (scope = "function " )
40109def superuser_token_headers (client : TestClient ) -> dict [str , str ]:
41110 return get_superuser_token_headers (client )
42111
43112
44- @pytest .fixture (scope = "module " )
113+ @pytest .fixture (scope = "function " )
45114def normal_user_token_headers (client : TestClient , db : Session ) -> dict [str , str ]:
46115 return authentication_token_from_email (
47116 client = client , email = settings .EMAIL_TEST_USER , db = db
0 commit comments