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

15 ModelDeclarationError, 

16 Pending, 

17 insert, 

18 mariadb, 

19 select, 

20) 

21from tests.helpers import NULL_LOGGER, provide_mariadb_server 

22 

23 

24@test() 

25def mariadb_storage_codecs_encode_and_decode_representative_values() -> None: 

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

27 

28 class Event[S = Pending](mariadb.Model[S, "Event[Fetched]"]): 

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

30 

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

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

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

34 

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

36 

37 # TODO: why do mariadb columns have both the `encode_mariadb` and `encode_sqlite` methods? 

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

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

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

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

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

43 assert_eq( 

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

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

46 ) 

47 

48 

49@test() 

50def mariadb_server_defaults_require_generated_datetime_columns() -> None: 

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

52 

53 with assert_raises(ModelDeclarationError): 

54 

55 class BadEvent[S = Pending](mariadb.Model[S, "BadEvent[Fetched]"]): 

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

57 

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

59 server_default=CurrentTimestamp(), 

60 ) 

61 

62 with assert_raises(ModelDeclarationError): 

63 

64 class BadCounter[S = Pending](mariadb.Model[S, "BadCounter[Fetched]"]): 

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

66 

67 # TODO: what if we created a type just for auto-increment ids? 

68 # Or maybe just other type hint improvements? 

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

70 

71 

72@test(mark="medium") 

73async def mariadb_value_families_round_trip_through_runtime() -> None: 

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

75 

76 server = await load_fixture(provide_mariadb_server()) 

77 

78 class Event[S = Pending](mariadb.Model[S, "Event[Fetched]"]): 

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

80 

81 __tablename__ = "issue40_event_values" 

82 

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

84 primary_key=True, 

85 auto_increment=True, 

86 default=MISSING, 

87 ) 

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

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

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

91 server_default=CurrentTimestamp(), 

92 default=MISSING, 

93 ) 

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

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

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

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

98 

99 database = await Database.initialize( 

100 server.config(), logger=NULL_LOGGER, models=[Event] 

101 ) 

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

103 try: 

104 async with database.transaction() as tx: 

105 await tx.execute( 

106 insert( 

107 Event( 

108 amount=12.5, 

109 content=b"hello", 

110 enabled=True, 

111 happened_at=happened_at, 

112 message="created", 

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

114 ) 

115 ) 

116 ) 

117 event = await tx.fetch_one(select(Event).all()) 

118 finally: 

119 await database.close() 

120 

121 assert event is not None 

122 assert_isinstance(event.id, int) 

123 assert_eq(event.amount, 12.5) 

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

125 assert_isinstance(event.created_at, datetime) 

126 assert_eq(event.enabled, True) 

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

128 assert_eq(event.message, "created") 

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