Coverage for app/tests/core/test_db.py: 100%

1from sqlalchemy import Engine, text 

2from sqlalchemy.engine.url import make_url 

3 

4from app.core.config import Settings 

5from app.tests.conftest import SQLITE_TEST_DB 

6 

7 

8def test_db_manager_pragma_fk(read_engine: Engine) -> None: 

9 with read_engine.connect() as conn: 

10 result = conn.execute(text("pragma foreign_keys;")).scalar() 

11 expected_result = 1 

12 assert result == expected_result 

13 

14 

15def test_db_manager_pragma_busy_timeout(read_engine: Engine) -> None: 

16 with read_engine.connect() as conn: 

17 result = conn.execute(text("pragma busy_timeout;")).scalar() 

18 expected_result = 5000 

19 assert result == expected_result 

20 

21 

22def test_db_manager_pragma_journal_mode(read_engine: Engine) -> None: 

23 with read_engine.connect() as conn: 

24 result = conn.execute(text("pragma journal_mode;")).scalar() 

25 expected_result = "wal" 

26 assert result == expected_result 

27 

28 

29def test_db_manager_pragma_synchronous(read_engine: Engine) -> None: 

30 with read_engine.connect() as conn: 

31 result = conn.execute(text("pragma synchronous;")).scalar() 

32 expected_result = 1 # normal 

33 assert result == expected_result 

34 

35 

36def test_db_manager_url_read_engine( 

37 read_engine: Engine, test_settings: Settings 

38) -> None: 

39 result = read_engine.url 

40 expected_result = make_url(f"{test_settings.SQLALCHEMY_DATABASE_URI}&mode=ro") 

41 assert result == expected_result 

42 assert result.database.removeprefix("file:") == str(SQLITE_TEST_DB) 

43 

44 

45def test_db_manager_url_write_engine( 

46 write_engine: Engine, test_settings: Settings 

47) -> None: 

48 result = write_engine.url 

49 expected_result = make_url(test_settings.SQLALCHEMY_DATABASE_URI) 

50 assert result == expected_result 

51 assert result.database.removeprefix("file:") == str(SQLITE_TEST_DB)