Coverage for tests/test_model_materialization.py: 100%

1"""Shared Table Model materialization tests.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6from typing import cast 

7 

8from snektest import assert_eq, assert_raises, test 

9 

10from snekql import ( 

11 Boolean, 

12 DateTime, 

13 Json, 

14 Model, 

15 ModelValidationError, 

16 Pending, 

17 mariadb, 

18) 

19from snekql._model_materialization import decode_model_row, encode_model_row 

20 

21 

22@test(mark="fast") 

23def sqlite_model_materialization_uses_one_backend_codec_path() -> None: 

24 """SQLite Pending/Fetched Model conversion is handled by the materializer.""" 

25 

26 class Event[S = Pending](Model[S, "Event[object]"]): 

27 """SQLite model used by materialization seam tests.""" 

28 

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

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

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

32 

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

34 pending_event = Event( 

35 enabled=True, 

36 happened_at=timestamp, 

37 payload={"ok": True}, 

38 ) 

39 model_class, encoded_row = encode_model_row(pending_event, backend="sqlite") 

40 fetched_event = cast( 

41 "Event[object]", 

42 decode_model_row( 

43 Event, 

44 { 

45 "enabled": 0, 

46 "happened_at": "2026-01-02T03:04:05.678Z", 

47 "payload": '{"ok":true}', 

48 }, 

49 backend="sqlite", 

50 ), 

51 ) 

52 

53 assert_eq(model_class, Event) 

54 assert_eq( 

55 encoded_row, 

56 { 

57 "enabled": 1, 

58 "happened_at": "2026-01-02T03:04:05.678Z", 

59 "payload": '{"ok":true}', 

60 }, 

61 ) 

62 assert_eq(fetched_event.enabled, False) 

63 assert_eq( 

64 fetched_event.happened_at, datetime(2026, 1, 2, 3, 4, 5, 678000, tzinfo=UTC) 

65 ) 

66 assert_eq(fetched_event.payload, {"ok": True}) 

67 

68 

69@test(mark="fast") 

70def mariadb_model_materialization_uses_one_backend_codec_path() -> None: 

71 """MariaDB Pending/Fetched Model conversion is handled by the materializer.""" 

72 

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

74 """MariaDB model used by materialization seam tests.""" 

75 

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

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

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

79 

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

81 pending_event = Event( 

82 enabled=True, 

83 happened_at=timestamp, 

84 payload={"ok": True}, 

85 ) 

86 model_class, encoded_row = encode_model_row(pending_event, backend="mariadb") 

87 fetched_event = cast( 

88 "Event[object]", 

89 decode_model_row( 

90 Event, 

91 { 

92 "enabled": 0, 

93 "happened_at": "2026-01-02 03:04:05.678", 

94 "payload": b'{"ok":true}', 

95 }, 

96 backend="mariadb", 

97 ), 

98 ) 

99 

100 assert_eq(model_class, Event) 

101 assert_eq( 

102 encoded_row, 

103 { 

104 "enabled": 1, 

105 "happened_at": "2026-01-02 03:04:05.678", 

106 "payload": '{"ok":true}', 

107 }, 

108 ) 

109 assert_eq(fetched_event.enabled, False) 

110 assert_eq( 

111 fetched_event.happened_at, datetime(2026, 1, 2, 3, 4, 5, 678000, tzinfo=UTC) 

112 ) 

113 assert_eq(fetched_event.payload, {"ok": True}) 

114 

115 

116@test(mark="fast") 

117def model_materialization_validates_database_row_shape() -> None: 

118 """Fetched Model materialization reports missing or extra database columns.""" 

119 

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

121 """MariaDB model used by row-shape checks.""" 

122 

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

124 

125 with assert_raises(ModelValidationError): 

126 _ = decode_model_row(Event, {}, backend="mariadb") 

127 

128 with assert_raises(ModelValidationError): 

129 _ = decode_model_row(Event, {"enabled": 1, "extra": 2}, backend="mariadb")