Coverage for tests/runtime/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 Fetched, 

11 Model, 

12 Pending, 

13 Text, 

14 mariadb, 

15 select, 

16 sqlite, 

17) 

18from tests.helpers import NULL_LOGGER, TemporaryMariaDBServer, provide_mariadb_server 

19 

20 

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

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[Fetched]"]): 

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

29 

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

31 

32 

33class MariadbIdentityUser[S = Pending]( 

34 mariadb.Model[S, "MariadbIdentityUser[Fetched]"] 

35): 

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

37 

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

39 

40 

41def _config_from_server(server: TemporaryMariaDBServer) -> mariadb.Config: 

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

43 

44 return server.config() 

45 

46 

47@test(mark="medium") 

48async def sqlite_initialization_rejects_mariadb_models() -> None: 

49 """SQLite Database startup rejects MariaDB Table Models.""" 

50 

51 with assert_raises(DatabaseRuntimeError) as error: 

52 _ = await Database.initialize( 

53 logger=NULL_LOGGER, database=":memory:", models=[MariadbIdentityUser] 

54 ) 

55 

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

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

58 

59 

60@test(mark="medium") 

61async def mariadb_initialization_rejects_sqlite_models() -> None: 

62 """MariaDB Database startup rejects SQLite Table Models.""" 

63 

64 with assert_raises(DatabaseRuntimeError) as error: 

65 _ = await Database.initialize( 

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

67 logger=NULL_LOGGER, 

68 models=[SqliteIdentityUser], 

69 ) 

70 

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

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

73 

74 

75@test(mark="medium") 

76async def sqlite_transaction_rejects_mariadb_queries() -> None: 

77 """SQLite Transactions reject MariaDB queries.""" 

78 

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

80 try: 

81 async with sqlite_database.transaction() as tx: 

82 with assert_raises(DatabaseRuntimeError) as error: 

83 _ = await tx.fetch_all(select(MariadbIdentityUser).all()) 

84 finally: 

85 await sqlite_database.close() 

86 

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

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

89 

90 

91@test(mark="medium") 

92async def mariadb_transaction_rejects_sqlite_queries() -> None: 

93 """MariaDB Transactions reject SQLite queries.""" 

94 

95 server = await load_fixture(provide_mariadb_server()) 

96 

97 mariadb_database = await Database.initialize( 

98 _config_from_server(server), logger=NULL_LOGGER 

99 ) 

100 try: 

101 async with mariadb_database.transaction() as tx: 

102 with assert_raises(DatabaseRuntimeError) as error: 

103 _ = await tx.fetch_all(select(SqliteIdentityUser).all()) 

104 finally: 

105 await mariadb_database.close() 

106 

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

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

109 

110 

111@test() 

112def legacy_top_level_model_is_a_sqlite_declaration() -> None: 

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

114 

115 assert_in("sqlite", LegacyIdentityUser.__snekql_backend__) 

116 assert_in("sqlite", SqliteIdentityUser.__snekql_backend__) 

117 assert_in("mariadb", MariadbIdentityUser.__snekql_backend__)