22
33import pytest
44from fastapi .testclient import TestClient
5- from sqlmodel import Session , delete
5+ from pydantic_core import MultiHostUrl
6+ from sqlalchemy import text
7+ from sqlmodel import Session , create_engine
68
9+ from app .auth .dependencies import get_db
710from app .config import settings
8- from app .database import engine , init_db
9- from app .items .models import Item
11+ from app .database import init_db
1012from app .main import app
11- from app .users .models import User
1213from tests .utils .user import authentication_token_from_email
1314from tests .utils .utils import get_superuser_token_headers
1415
16+ TEST_POSTGRES_DB = settings .POSTGRES_DB + "_test"
17+
18+ TEST_DATABASE_URI = MultiHostUrl .build (
19+ scheme = "postgresql+psycopg" ,
20+ username = settings .POSTGRES_USER ,
21+ password = settings .POSTGRES_PASSWORD ,
22+ host = settings .POSTGRES_SERVER ,
23+ port = settings .POSTGRES_PORT ,
24+ path = TEST_POSTGRES_DB ,
25+ )
26+
27+ test_engine = create_engine (str (TEST_DATABASE_URI ))
28+
29+
30+ ADMIN_DATABASE_URI = MultiHostUrl .build (
31+ scheme = "postgresql+psycopg" ,
32+ username = settings .POSTGRES_USER ,
33+ password = settings .POSTGRES_PASSWORD ,
34+ host = settings .POSTGRES_SERVER ,
35+ port = settings .POSTGRES_PORT ,
36+ path = "postgres" ,
37+ )
38+
39+ admin_engine = create_engine (str (ADMIN_DATABASE_URI ))
40+
41+
42+ def create_test_database () -> None :
43+ """Create the test database by copying the existing database."""
44+ # AUTOCOMMIT is required to drop a database
45+ with admin_engine .connect ().execution_options (isolation_level = "AUTOCOMMIT" ) as conn :
46+ conn .execute (text (f'DROP DATABASE IF EXISTS "{ TEST_POSTGRES_DB } "' ))
47+ conn .execute (
48+ text (
49+ f'CREATE DATABASE "{ TEST_POSTGRES_DB } " WITH TEMPLATE "{ settings .POSTGRES_DB } "' ,
50+ ),
51+ )
52+
1553
1654@pytest .fixture (scope = "session" , autouse = True )
17- def db () -> Generator [Session , None , None ]:
18- with Session (engine ) as session :
55+ def db () -> Generator [Session ]:
56+ create_test_database ()
57+ with Session (test_engine ) as session :
1958 init_db (session )
2059 yield session
21- statement = delete (Item )
22- session .execute (statement )
23- statement = delete (User )
24- session .execute (statement )
25- session .commit ()
60+ with admin_engine .connect ().execution_options (isolation_level = "AUTOCOMMIT" ) as conn :
61+ conn .execute (text (f'DROP DATABASE IF EXISTS "{ TEST_POSTGRES_DB } "' ))
62+
63+
64+ def get_test_db () -> Generator [Session ]:
65+ """Database dependency override for testing."""
66+ with Session (test_engine ) as session :
67+ yield session
2668
2769
2870@pytest .fixture (scope = "module" )
29- def client () -> Generator [TestClient , None , None ]:
71+ def client () -> Generator [TestClient ]:
72+ app .dependency_overrides [get_db ] = get_test_db
3073 with TestClient (app ) as c :
3174 yield c
75+ app .dependency_overrides .clear ()
3276
3377
3478@pytest .fixture (scope = "module" )
@@ -39,5 +83,7 @@ def superuser_token_headers(client: TestClient) -> dict[str, str]:
3983@pytest .fixture (scope = "module" )
4084def normal_user_token_headers (client : TestClient , db : Session ) -> dict [str , str ]:
4185 return authentication_token_from_email (
42- client = client , email = settings .EMAIL_TEST_USER , db = db
86+ client = client ,
87+ email = settings .EMAIL_TEST_USER ,
88+ db = db ,
4389 )
0 commit comments