Coverage for app/tests/conftest.py: 95%

1import os 

2import subprocess 

3from pathlib import Path 

4from typing import Generator 

5from unittest import mock 

6 

7import pytest 

8from sqlalchemy import Engine 

9 

10from app.core.config import Settings 

11from app.core.db import DBConnectionManager 

12 

13ALEMBIC_CONFIG_PATH = Path(__file__).parent.parent.parent / "alembic.ini" 

14SQLITE_TEST_DB = ( 

15 Path(__file__).parent.parent.parent.parent / "data/pypacktrends_test.db" 

16) 

17 

18 

19@pytest.fixture(scope="session") 

20def test_settings() -> Settings: 

21 with mock.patch.dict(os.environ, {"SQLITE_DB_PATH": str(SQLITE_TEST_DB)}): 

22 test_settings = Settings() 

23 return test_settings 

24 

25 

26@pytest.fixture(scope="session") 

27def run_db_migrations() -> Generator[None, None, None]: 

28 with mock.patch.dict(os.environ, {"SQLITE_DB_PATH": str(SQLITE_TEST_DB)}): 

29 if not ALEMBIC_CONFIG_PATH.exists(): 

30 raise FileNotFoundError( 

31 f"Alembic config file not found at {ALEMBIC_CONFIG_PATH}" 

32 ) 

33 

34 if SQLITE_TEST_DB.exists(): 

35 SQLITE_TEST_DB.unlink() 

36 

37 subprocess.run( 

38 ["alembic", "-c", str(ALEMBIC_CONFIG_PATH), "upgrade", "head"], check=True 

39 ) 

40 

41 yield 

42 

43 

44@pytest.fixture(scope="session") 

45def db_manager(run_db_migrations: None, test_settings: Settings) -> DBConnectionManager: 

46 return DBConnectionManager(test_settings.SQLALCHEMY_DATABASE_URI) 

47 

48 

49@pytest.fixture(scope="session") 

50def read_engine(db_manager: DBConnectionManager) -> Generator[Engine, None, None]: 

51 engine = db_manager.get_engine(read_only=True) 

52 yield engine 

53 engine.dispose() 

54 

55 

56@pytest.fixture(scope="session") 

57def write_engine(db_manager: DBConnectionManager) -> Generator[Engine, None, None]: 

58 engine = db_manager.get_engine(read_only=False) 

59 yield engine 

60 engine.dispose()