Coverage for tests/test_backend_identity.py: 100%

1"""Backend identity runtime safety tests.""" 

2 

3from __future__ import annotations 

4 

5from snektest import assert_in, assert_raises, load_fixture, test 

6 

7from snekql import ( 

8 Database, 

9 DatabaseRuntimeError, 

10 Model, 

11 Pending, 

12 Text, 

13 mariadb, 

14 select, 

15 sqlite, 

16) 

17from tests.logging_helpers import NULL_LOGGER 

18from tests.mariadb_server import MariaDBServer, provide_mariadb_server 

19 

20 

21class SqliteIdentityUser[S = Pending](sqlite.Model[S, "SqliteIdentityUser[object]"]): 

22 """SQLite table model for backend identity checks.""" 

23 

24 email: SqliteIdentityUser.Col[str] = sqlite.Text(nullable=False) 

25 

26 

27class LegacyIdentityUser[S = Pending](Model[S, "LegacyIdentityUser[object]"]): 

28 """Legacy top-level model remains a SQLite declaration.""" 

29 

30 email: LegacyIdentityUser.Col[str] = Text(nullable=False) 

31 

32 

33class MariadbIdentityUser[S = Pending](mariadb.Model[S, "MariadbIdentityUser[object]"]): 

34 """MariaDB table model for backend identity checks.""" 

35 

36 email: MariadbIdentityUser.Col[str] = mariadb.Text(nullable=False) 

37 

38 

39def _config_from_server(server: MariaDBServer) -> mariadb.Config: 

40 """Build a MariaDB config for the shared local test server.""" 

41 

42 return mariadb.Config( 

43 database=server.database, 

44 host=server.host, 

45 port=server.port, 

46 user=server.user, 

47 ) 

48 

49 

50@test(mark="medium") 

51async def initialization_rejects_models_from_the_wrong_backend() -> None: 

52 """Database startup rejects model/config backend mismatches before runtime work.""" 

53 

54 with assert_raises(DatabaseRuntimeError) as sqlite_error: 

55 _ = await Database.initialize( 

56 NULL_LOGGER, database=":memory:", models=[MariadbIdentityUser] 

57 ) 

58 assert_in("expected sqlite", str(sqlite_error.exception)) 

59 assert_in("received mariadb", str(sqlite_error.exception)) 

60 

61 with assert_raises(DatabaseRuntimeError) as mariadb_error: 

62 _ = await Database.initialize( 

63 NULL_LOGGER, 

64 mariadb.Config(database="app", user="snekql"), 

65 models=[SqliteIdentityUser], 

66 ) 

67 assert_in("expected mariadb", str(mariadb_error.exception)) 

68 assert_in("received sqlite", str(mariadb_error.exception)) 

69 

70 

71@test(mark="medium") 

72async def transactions_reject_queries_from_the_wrong_backend() -> None: 

73 """Transactions reject query/model backends that do not match their runtime.""" 

74 

75 sqlite_database = await Database.initialize(NULL_LOGGER, database=":memory:") 

76 try: 

77 async with sqlite_database.transaction() as transaction: 

78 with assert_raises(DatabaseRuntimeError) as error: 

79 _ = await transaction.fetch_all(select(MariadbIdentityUser).all()) 

80 assert_in("expected sqlite", str(error.exception)) 

81 assert_in("received mariadb", str(error.exception)) 

82 finally: 

83 await sqlite_database.close() 

84 

85 server = load_fixture(provide_mariadb_server()) 

86 mariadb_database = await Database.initialize( 

87 NULL_LOGGER, _config_from_server(server) 

88 ) 

89 try: 

90 async with mariadb_database.transaction() as transaction: 

91 with assert_raises(DatabaseRuntimeError) as error: 

92 _ = await transaction.fetch_all(select(SqliteIdentityUser).all()) 

93 assert_in("expected mariadb", str(error.exception)) 

94 assert_in("received sqlite", str(error.exception)) 

95 finally: 

96 await mariadb_database.close() 

97 

98 

99@test() 

100def legacy_top_level_model_is_a_sqlite_declaration() -> None: 

101 """Compatibility aliases keep behaving as SQLite declarations.""" 

102 

103 assert_in("sqlite", LegacyIdentityUser.__snekql_backend__) 

104 assert_in("sqlite", SqliteIdentityUser.__snekql_backend__) 

105 assert_in("mariadb", MariadbIdentityUser.__snekql_backend__)