Coverage for tests/test_storage_codecs.py: 100%

1"""SQLite storage type and value codec tests.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from datetime import UTC, datetime, timedelta, timezone 

7from typing import cast 

8 

9from snektest import assert_eq, assert_false, assert_raises, assert_true, test 

10 

11import snekql 

12from snekql import ( 

13 MISSING, 

14 Blob, 

15 Boolean, 

16 CurrentTimestamp, 

17 DateTime, 

18 Fetched, 

19 Integer, 

20 Json, 

21 Model, 

22 ModelDeclarationError, 

23 ModelValidationError, 

24 Pending, 

25 Real, 

26 Text, 

27) 

28from snekql.model import decode_model_row, encode_model_row 

29 

30 

31@test() 

32def v1_exposes_only_sqlite_first_storage_classes() -> None: 

33 """Text has no length option and Varchar is not a v1 storage class.""" 

34 

35 text_constructor = cast("Callable[..., object]", Text) 

36 

37 assert_false(hasattr(snekql, "Varchar")) 

38 with assert_raises(TypeError): 

39 _ = text_constructor(length=255) 

40 

41 

42@test() 

43def storage_classes_expose_sqlite_metadata() -> None: 

44 """V1 columns record the SQLite storage class used for schema generation.""" 

45 

46 class StorageExample[S = Pending](Model[S, "StorageExample[Fetched]"]): 

47 """Table model using every v1 storage class.""" 

48 

49 integer_value: StorageExample.Col[int] = Integer(nullable=False) 

50 real_value: StorageExample.Col[float] = Real(nullable=False) 

51 text_value: StorageExample.Col[str] = Text(nullable=False) 

52 blob_value: StorageExample.Col[bytes] = Blob(nullable=False) 

53 json_value: StorageExample.Col[dict[str, object]] = Json(nullable=False) 

54 boolean_value: StorageExample.Col[bool] = Boolean(nullable=False) 

55 datetime_value: StorageExample.Col[object] = DateTime(nullable=False) 

56 

57 columns = StorageExample.__snekql_columns__ 

58 

59 assert_eq(columns["integer_value"].sqlite_storage_class, "INTEGER") 

60 assert_eq(columns["real_value"].sqlite_storage_class, "REAL") 

61 assert_eq(columns["text_value"].sqlite_storage_class, "TEXT") 

62 assert_eq(columns["blob_value"].sqlite_storage_class, "BLOB") 

63 assert_eq(columns["json_value"].sqlite_storage_class, "TEXT") 

64 assert_eq(columns["boolean_value"].sqlite_storage_class, "INTEGER") 

65 assert_eq(columns["datetime_value"].sqlite_storage_class, "TEXT") 

66 

67 

68@test() 

69def boolean_values_encode_to_integer_and_decode_before_validation() -> None: 

70 """Boolean columns use INTEGER storage while models expose bools.""" 

71 

72 class FeatureFlag[S = Pending](Model[S, "FeatureFlag[Fetched]"]): 

73 """Table model with a boolean flag.""" 

74 

75 enabled: FeatureFlag.Col[bool] = Boolean(nullable=False) 

76 

77 enabled = FeatureFlag(enabled=True) 

78 disabled = cast( 

79 "FeatureFlag[Fetched]", 

80 decode_model_row(FeatureFlag, {"enabled": 0}), 

81 ) 

82 _, encoded_enabled = encode_model_row(enabled) 

83 

84 assert_eq(encoded_enabled, {"enabled": 1}) 

85 assert_false(disabled.enabled) 

86 

87 with assert_raises(ModelValidationError): 

88 _ = decode_model_row(FeatureFlag, {"enabled": 2}) 

89 

90 

91@test() 

92def json_values_encode_to_text_and_decode_before_validation() -> None: 

93 """Json columns store JSON text and expose decoded Python values.""" 

94 

95 class Event[S = Pending](Model[S, "Event[Fetched]"]): 

96 """Table model with a JSON payload.""" 

97 

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

99 

100 event = Event(payload={"kind": "created", "count": 2}) 

101 fetched = cast( 

102 "Event[Fetched]", 

103 decode_model_row(Event, {"payload": '{"kind":"created","count":2}'}), 

104 ) 

105 _, encoded_event = encode_model_row(event) 

106 

107 assert_eq(encoded_event, {"payload": '{"kind":"created","count":2}'}) 

108 assert_eq(fetched.payload, {"kind": "created", "count": 2}) 

109 

110 with assert_raises(ModelValidationError): 

111 _ = Event(payload={"bad": {object()}}) 

112 

113 with assert_raises(ModelValidationError): 

114 _ = decode_model_row(Event, {"payload": "not json"}) 

115 

116 

117@test() 

118def datetime_values_are_utc_millisecond_text() -> None: 

119 """DateTime accepts aware values and stores UTC millisecond text.""" 

120 

121 class AuditLog[S = Pending](Model[S, "AuditLog[Fetched]"]): 

122 """Table model with a timestamp.""" 

123 

124 created_at: AuditLog.Col[datetime] = DateTime(nullable=False) 

125 

126 source_timezone = timezone(timedelta(hours=5, minutes=30)) 

127 source = datetime(2026, 5, 31, 12, 0, 1, 987654, tzinfo=source_timezone) 

128 audit_log = AuditLog(created_at=source) 

129 fetched = cast( 

130 "AuditLog[Fetched]", 

131 decode_model_row(AuditLog, {"created_at": "2026-05-31T06:30:01.987Z"}), 

132 ) 

133 _, encoded_audit_log = encode_model_row(audit_log) 

134 

135 expected = datetime(2026, 5, 31, 6, 30, 1, 987000, tzinfo=UTC) 

136 assert_eq(audit_log.created_at, expected) 

137 assert_eq(encoded_audit_log, {"created_at": "2026-05-31T06:30:01.987Z"}) 

138 assert_eq(fetched.created_at, expected) 

139 

140 with assert_raises(ModelValidationError): 

141 _ = AuditLog(created_at=datetime(2026, 5, 31, 12, 0, 1)) # noqa: DTZ001 

142 

143 

144@test() 

145def external_value_failures_are_wrapped_in_model_validation_error() -> None: 

146 """Default factories and codecs do not leak third-party exceptions.""" 

147 

148 def broken_default() -> object: 

149 msg = "outside validation failure" 

150 raise ValueError(msg) 

151 

152 class ExternalValue[S = Pending](Model[S, "ExternalValue[Fetched]"]): 

153 """Table model with an external default provider.""" 

154 

155 payload: ExternalValue.Col[object] = Json(default_factory=broken_default) 

156 

157 with assert_raises(ModelValidationError): 

158 _ = ExternalValue() 

159 

160 

161@test() 

162def current_timestamp_is_valid_only_for_datetime_generated_columns() -> None: 

163 """Server timestamp defaults are limited to generated DateTime fields.""" 

164 

165 class CreatedEvent[S = Pending](Model[S, "CreatedEvent[Fetched]"]): 

166 """Valid generated timestamp column.""" 

167 

168 created_at: CreatedEvent.GenCol[datetime] = DateTime( 

169 server_default=CurrentTimestamp(), 

170 default=MISSING, 

171 ) 

172 

173 assert_true( 

174 isinstance( 

175 CreatedEvent.__snekql_columns__["created_at"].server_default, 

176 CurrentTimestamp, 

177 ), 

178 ) 

179 

180 with assert_raises(ModelDeclarationError): 

181 

182 class NonGeneratedTimestamp[S = Pending]( 

183 Model[S, "NonGeneratedTimestamp[Fetched]"] 

184 ): 

185 """Invalid non-generated timestamp default.""" 

186 

187 created_at: NonGeneratedTimestamp.Col[datetime] = DateTime( 

188 server_default=CurrentTimestamp(), 

189 default=MISSING, 

190 ) 

191 

192 with assert_raises(ModelDeclarationError): 

193 

194 class NonDateTimeTimestamp[S = Pending]( 

195 Model[S, "NonDateTimeTimestamp[Fetched]"] 

196 ): 

197 """Invalid CurrentTimestamp use outside a DateTime server default.""" 

198 

199 created_at: NonDateTimeTimestamp.GenCol[datetime] = Text( 

200 default=CurrentTimestamp(), 

201 ) 

202 

203 with assert_raises(ModelDeclarationError): 

204 

205 class TimestampWithPythonDefault[S = Pending]( 

206 Model[S, "TimestampWithPythonDefault[Fetched]"], 

207 ): 

208 """Invalid server default paired with a Python default.""" 

209 

210 created_at: TimestampWithPythonDefault.GenCol[datetime] = DateTime( 

211 server_default=CurrentTimestamp(), 

212 default=datetime(2026, 5, 31, tzinfo=UTC), 

213 )