Coverage for tests/test_mariadb_storage_codecs.py: 100%

1"""MariaDB storage codec and value family tests.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6from typing import Any 

7 

8from snektest import assert_eq, assert_isinstance, assert_raises, load_fixture, test 

9 

10from snekql import ( 

11 MISSING, 

12 CurrentTimestamp, 

13 Database, 

14 ModelDeclarationError, 

15 Pending, 

16 insert, 

17 mariadb, 

18 select, 

19) 

20from tests.logging_helpers import NULL_LOGGER 

21from tests.mariadb_server import MariaDBServer, provide_mariadb_server 

22 

23 

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

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

26 

27 return mariadb.Config( 

28 database=server.database, 

29 host=server.host, 

30 port=server.port, 

31 user=server.user, 

32 ) 

33 

34 

35@test() 

36def mariadb_storage_codecs_encode_and_decode_representative_values() -> None: 

37 """MariaDB columns expose backend-specific value codecs.""" 

38 

39 class Event[S = Pending](mariadb.Model[S, "Event[object]"]): 

40 """Model used to bind MariaDB descriptors for direct codec checks.""" 

41 

42 flag: Event.Col[bool] = mariadb.Boolean(nullable=False) 

43 payload: Event.Col[dict[str, object]] = mariadb.Json(nullable=False) 

44 happened_at: Event.Col[datetime] = mariadb.DateTime(nullable=False) 

45 

46 timestamp = datetime(2026, 1, 2, 3, 4, 5, 678901, tzinfo=UTC) 

47 

48 assert_eq(Event.flag.encode_mariadb(True), 1) 

49 assert_eq(Event.flag.decode_mariadb(0), False) 

50 assert_eq(Event.payload.encode_mariadb({"ok": True}), '{"ok":true}') 

51 assert_eq(Event.payload.decode_mariadb('{"ok":true}'), {"ok": True}) 

52 assert_eq(Event.happened_at.encode_mariadb(timestamp), "2026-01-02 03:04:05.678") 

53 assert_eq( 

54 Event.happened_at.decode_mariadb("2026-01-02 03:04:05.678"), 

55 datetime(2026, 1, 2, 3, 4, 5, 678000, tzinfo=UTC), 

56 ) 

57 

58 

59@test() 

60def mariadb_server_defaults_require_generated_datetime_columns() -> None: 

61 """CurrentTimestamp keeps the existing generated-column declaration rules.""" 

62 

63 with assert_raises(ModelDeclarationError): 

64 

65 class BadEvent[S = Pending](mariadb.Model[S, "BadEvent[object]"]): 

66 """Invalid MariaDB model using a server default on a normal column.""" 

67 

68 created_at: BadEvent.Col[datetime] = mariadb.DateTime( 

69 server_default=CurrentTimestamp(), 

70 ) 

71 

72 with assert_raises(ModelDeclarationError): 

73 

74 class BadCounter[S = Pending](mariadb.Model[S, "BadCounter[object]"]): 

75 """Invalid MariaDB model using auto increment outside a primary key.""" 

76 

77 count: BadCounter.Col[int] = mariadb.Integer(auto_increment=True) 

78 

79 

80@test(mark="medium") 

81async def mariadb_value_families_round_trip_through_runtime() -> None: 

82 """MariaDB round trips the initial value families through a live database.""" 

83 

84 class Event[S = Pending](mariadb.Model[S, "Event[object]"]): 

85 """Model covering MariaDB value family round trips.""" 

86 

87 __tablename__ = "issue40_event_values" 

88 

89 id: Event.GenCol[int] = mariadb.Integer( 

90 primary_key=True, 

91 auto_increment=True, 

92 default=MISSING, 

93 ) 

94 amount: Event.Col[float] = mariadb.Real(nullable=False) 

95 content: Event.Col[bytes] = mariadb.Blob(nullable=False) 

96 created_at: Event.GenCol[datetime] = mariadb.DateTime( 

97 server_default=CurrentTimestamp(), 

98 default=MISSING, 

99 ) 

100 enabled: Event.Col[bool] = mariadb.Boolean(nullable=False) 

101 happened_at: Event.Col[datetime] = mariadb.DateTime(nullable=False) 

102 message: Event.Col[str] = mariadb.Text(nullable=False) 

103 payload: Event.Col[dict[str, Any]] = mariadb.Json(nullable=False) 

104 

105 server = load_fixture(provide_mariadb_server()) 

106 database = await Database.initialize( 

107 NULL_LOGGER, _config_from_server(server), models=[Event] 

108 ) 

109 happened_at = datetime(2026, 1, 2, 3, 4, 5, 678901, tzinfo=UTC) 

110 try: 

111 async with database.transaction() as transaction: 

112 await transaction.execute( 

113 insert( 

114 Event( 

115 amount=12.5, 

116 content=b"hello", 

117 enabled=True, 

118 happened_at=happened_at, 

119 message="created", 

120 payload={"count": 2, "ok": True}, 

121 ) 

122 ) 

123 ) 

124 event = await transaction.fetch_one(select(Event).all()) 

125 finally: 

126 await database.close() 

127 

128 assert event is not None 

129 assert_isinstance(event.id, int) 

130 assert_eq(event.amount, 12.5) 

131 assert_eq(event.content, b"hello") 

132 assert_isinstance(event.created_at, datetime) 

133 assert_eq(event.enabled, True) 

134 assert_eq(event.happened_at, datetime(2026, 1, 2, 3, 4, 5, 678000, tzinfo=UTC)) 

135 assert_eq(event.message, "created") 

136 assert_eq(event.payload, {"count": 2, "ok": True})