Coverage for tests/test_v1_integration.py: 97%

1"""V1 public integration tests against real SQLite databases.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime, timedelta, timezone 

6from pathlib import Path 

7from sqlite3 import connect 

8from tempfile import TemporaryDirectory 

9from typing import Literal 

10 

11from snektest import assert_eq, assert_is_none, assert_raises, assert_true, test 

12 

13from snekql import ( 

14 MISSING, 

15 Boolean, 

16 CurrentTimestamp, 

17 Database, 

18 DatabaseClosedError, 

19 DatabaseCloseTimeoutError, 

20 DateTime, 

21 Fetched, 

22 Integer, 

23 Json, 

24 Model, 

25 Pending, 

26 PoolTimeoutError, 

27 SchemaError, 

28 SchemaVerificationError, 

29 Text, 

30 TransactionClosedError, 

31 delete, 

32 insert, 

33 select, 

34 update, 

35) 

36from tests.logging_helpers import NULL_LOGGER 

37 

38type DatabaseTarget = Path | Literal[":memory:"] 

39 

40 

41class V1Event[S = Pending](Model[S, "V1Event[Fetched]"]): 

42 """Integration model that exercises generated and encoded columns.""" 

43 

44 id: V1Event.GenCol[int] = Integer( 

45 primary_key=True, 

46 auto_increment=True, 

47 default=MISSING, 

48 ) 

49 active: V1Event.Col[bool] = Boolean(nullable=False) 

50 created_at: V1Event.GenCol[datetime] = DateTime( 

51 server_default=CurrentTimestamp(), 

52 default=MISSING, 

53 ) 

54 happened_at: V1Event.Col[datetime] = DateTime(nullable=False) 

55 name: V1Event.Col[str] = Text(nullable=False) 

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

57 

58 

59class V1Lifecycle[S = Pending](Model[S, "V1Lifecycle[Fetched]"]): 

60 """Small model for transaction, pool, and close lifecycle integration.""" 

61 

62 id: V1Lifecycle.GenCol[int] = Integer( 

63 primary_key=True, 

64 auto_increment=True, 

65 default=MISSING, 

66 ) 

67 name: V1Lifecycle.Col[str] = Text(nullable=False) 

68 

69 

70class _RecordingStructuredLogger: 

71 """Structured logger fake that stores schema warning events.""" 

72 

73 def __init__(self) -> None: 

74 self.events: list[tuple[str, str, dict[str, object]]] = [] 

75 

76 def debug(self, event: str, **fields: object) -> None: 

77 self.events.append(("debug", event, fields)) 

78 

79 def info(self, event: str, **fields: object) -> None: 

80 self.events.append(("info", event, fields)) 

81 

82 def warning(self, event: str, **fields: object) -> None: 

83 self.events.append(("warning", event, fields)) 

84 

85 def error(self, event: str, **fields: object) -> None: 

86 self.events.append(("error", event, fields)) 

87 

88 

89def _execute_sql(database_path: Path, sql: str) -> None: 

90 connection = connect(database_path) 

91 try: 

92 _ = connection.execute(sql) 

93 connection.commit() 

94 finally: 

95 connection.close() 

96 

97 

98async def _exercise_v1_runtime(database_target: DatabaseTarget) -> None: 

99 source_timezone = timezone(timedelta(hours=-4)) 

100 happened_at = datetime(2026, 5, 31, 12, 30, 1, 987654, tzinfo=source_timezone) 

101 expected_happened_at = datetime(2026, 5, 31, 16, 30, 1, 987000, tzinfo=UTC) 

102 payload: dict[str, object] = {"kind": "created", "count": 2, "tags": ["v1"]} 

103 

104 database = await Database.initialize( 

105 NULL_LOGGER, database=database_target, models=[V1Event] 

106 ) 

107 try: 

108 async with database.transaction() as transaction: 

109 insert_result = await transaction.execute( 

110 insert( 

111 V1Event( 

112 active=True, 

113 happened_at=happened_at, 

114 name="alpha", 

115 payload=payload, 

116 ) 

117 ) 

118 ) 

119 await transaction.execute( 

120 insert( 

121 V1Event( 

122 active=False, 

123 happened_at=happened_at, 

124 name="beta", 

125 payload={"kind": "ignored"}, 

126 ) 

127 ) 

128 ) 

129 

130 first_name = await transaction.fetch_one( 

131 select(V1Event.name).all().order_by(V1Event.id.asc()), 

132 ) 

133 no_name = await transaction.fetch_one(select(V1Event.name).all().limit(0)) 

134 fetched_event = await transaction.fetch_one( 

135 select(V1Event).where(V1Event.name.eq("alpha")), 

136 ) 

137 update_result = await transaction.execute( 

138 update(V1Event) 

139 .set(V1Event.name.to("alpha-renamed"), V1Event.active.to(False)) 

140 .where(V1Event.name.eq("alpha")), 

141 ) 

142 delete_result = await transaction.execute( 

143 delete(V1Event).where(V1Event.name.eq("beta")), 

144 ) 

145 remaining_names = await transaction.fetch_all( 

146 select(V1Event.name).all().order_by(V1Event.id.asc()), 

147 ) 

148 finally: 

149 await database.close() 

150 

151 assert_is_none(insert_result) 

152 assert_is_none(update_result) 

153 assert_is_none(delete_result) 

154 assert_eq(first_name, "alpha") 

155 assert_eq(no_name, None) 

156 if fetched_event is None: 

157 msg = "expected alpha event to be fetched" 

158 raise AssertionError(msg) 

159 event: V1Event[Fetched] = fetched_event 

160 assert_eq(event.id, 1) 

161 assert_true(event.active) 

162 assert_eq(event.happened_at, expected_happened_at) 

163 assert_eq(event.created_at.tzinfo, UTC) 

164 assert_eq(event.payload, payload) 

165 assert_eq(remaining_names, ["alpha-renamed"]) 

166 

167 

168@test(mark="medium") 

169async def path_database_exercises_v1_runtime_and_schema_verification() -> None: 

170 """A pathlib.Path database creates schema, runs CRUD, and verifies on reopen.""" 

171 

172 with TemporaryDirectory() as directory: 

173 database_path = Path(directory) / "app.db" 

174 await _exercise_v1_runtime(database_path) 

175 

176 database = await Database.initialize( 

177 NULL_LOGGER, database=database_path, models=[V1Event] 

178 ) 

179 await database.close() 

180 

181 

182@test(mark="medium") 

183async def memory_database_exercises_v1_runtime() -> None: 

184 """The exact :memory: database target supports the same public runtime flow.""" 

185 

186 await _exercise_v1_runtime(":memory:") 

187 

188 

189@test(mark="medium") 

190async def schema_policies_cover_duplicate_names_and_non_strict_drift() -> None: 

191 """Strict rejects drift, warn logs drift, and duplicate table names fail fast.""" 

192 

193 class DriftedEvent[S = Pending](Model[S, "DriftedEvent[object]"]): 

194 """Model whose existing table intentionally lacks SQLite STRICT.""" 

195 

196 name: DriftedEvent.Col[str] = Text(nullable=False) 

197 

198 class DuplicateOne[S = Pending](Model[S, "DuplicateOne[object]"]): 

199 """First model with a duplicate resolved table name.""" 

200 

201 __tablename__ = "duplicate_event" 

202 name: DuplicateOne.Col[str] = Text(nullable=False) 

203 

204 class DuplicateTwo[S = Pending](Model[S, "DuplicateTwo[object]"]): 

205 """Second model with a duplicate resolved table name.""" 

206 

207 __tablename__ = "duplicate_event" 

208 name: DuplicateTwo.Col[str] = Text(nullable=False) 

209 

210 with TemporaryDirectory() as directory: 

211 database_path = Path(directory) / "app.db" 

212 _execute_sql( 

213 database_path, 

214 'CREATE TABLE "drifted_event" ("name" TEXT NOT NULL)', 

215 ) 

216 

217 with assert_raises(SchemaVerificationError): 

218 _ = await Database.initialize( 

219 NULL_LOGGER, database=database_path, models=[DriftedEvent] 

220 ) 

221 

222 logger = _RecordingStructuredLogger() 

223 database = await Database.initialize( 

224 logger, 

225 database=database_path, 

226 models=[DriftedEvent], 

227 schema_policy="warn", 

228 ) 

229 await database.close() 

230 

231 with assert_raises(SchemaError): 

232 _ = await Database.initialize( 

233 NULL_LOGGER, 

234 database=database_path, 

235 models=[DuplicateOne, DuplicateTwo], 

236 ) 

237 

238 assert_true( 

239 any( 

240 level == "warning" and event == "schema drift detected" 

241 for level, event, _fields in logger.events 

242 ) 

243 ) 

244 

245 

246@test(mark="medium") 

247async def transactions_pool_and_close_lifecycle_are_integrated() -> None: 

248 """Transactions commit/roll back and close remains idempotent and retryable.""" 

249 

250 database = await Database.initialize( 

251 NULL_LOGGER, 

252 database=":memory:", 

253 models=[V1Lifecycle], 

254 pool_size=1, 

255 acquire_timeout=0.0, 

256 ) 

257 transaction = database.transaction(timeout=0.0) 

258 try: 

259 async with database.transaction() as commit_transaction: 

260 await commit_transaction.execute(insert(V1Lifecycle(name="committed"))) 

261 

262 with assert_raises(ValueError): 

263 async with database.transaction() as rollback_transaction: 

264 await rollback_transaction.execute( 

265 insert(V1Lifecycle(name="rolled-back")), 

266 ) 

267 msg = "force rollback" 

268 raise ValueError(msg) 

269 

270 async with database.transaction() as read_transaction: 

271 names = await read_transaction.fetch_all( 

272 select(V1Lifecycle.name).all().order_by(V1Lifecycle.id.asc()), 

273 ) 

274 

275 _ = await transaction.__aenter__() 

276 with assert_raises(PoolTimeoutError): 

277 async with database.transaction(timeout=0.0): 

278 pass 

279 with assert_raises(DatabaseCloseTimeoutError): 

280 await database.close() 

281 await transaction.__aexit__(None, None, None) 

282 

283 with assert_raises(TransactionClosedError): 

284 _ = await transaction.fetch_all(select(V1Lifecycle).all()) 

285 

286 async with database.transaction(timeout=0.0): 

287 pass 

288 finally: 

289 await database.close() 

290 

291 await database.close() 

292 with assert_raises(DatabaseClosedError): 

293 _ = database.transaction() 

294 assert_eq(names, ["committed"])