Coverage for tests/runtime/test_database_initialization.py: 99%

1"""Database initialization and schema verification tests.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from sqlite3 import connect 

7from tempfile import TemporaryDirectory 

8from typing import Any, ClassVar, cast 

9 

10from snektest import assert_eq, assert_raises, assert_true, test 

11 

12from snekql import ( 

13 MISSING, 

14 Database, 

15 DatabaseRuntimeError, 

16 Fetched, 

17 Index, 

18 Integer, 

19 Model, 

20 Pending, 

21 SchemaError, 

22 SchemaVerificationError, 

23 Text, 

24 sqlite, 

25) 

26from tests.helpers import NULL_LOGGER 

27 

28 

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

30 connection = connect(database_path) 

31 try: 

32 _ = connection.execute(sql) 

33 connection.commit() 

34 finally: 

35 connection.close() 

36 

37 

38def _fetch_create_table(database_path: Path, table_name: str) -> str: 

39 connection = connect(database_path) 

40 try: 

41 cursor = connection.execute( 

42 "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?", 

43 (table_name,), 

44 ) 

45 row = cursor.fetchone() 

46 assert row is not None 

47 value = row[0] 

48 assert isinstance(value, str) 

49 return value 

50 finally: 

51 connection.close() 

52 

53 

54def _fetch_create_indexes(database_path: Path, table_name: str) -> list[str]: 

55 connection = connect(database_path) 

56 try: 

57 cursor = connection.execute( 

58 """ 

59 SELECT sql FROM sqlite_master 

60 WHERE type = 'index' AND tbl_name = ? 

61 ORDER BY rowid 

62 """, 

63 (table_name,), 

64 ) 

65 values: list[str] = [] 

66 for row in cursor.fetchall(): 

67 value = row[0] 

68 assert isinstance(value, str) 

69 values.append(value) 

70 return values 

71 finally: 

72 connection.close() 

73 

74 

75def _table_exists(database_path: Path, table_name: str) -> bool: 

76 connection = connect(database_path) 

77 try: 

78 cursor = connection.execute( 

79 "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", 

80 (table_name,), 

81 ) 

82 return cursor.fetchone() is not None 

83 finally: 

84 connection.close() 

85 

86 

87class _RecordingStructuredLogger: 

88 """Structured logger fake that stores event calls for assertions.""" 

89 

90 def __init__(self) -> None: 

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

92 

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

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

95 

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

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

98 

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

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

101 

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

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

104 

105 

106@test(mark="medium") 

107async def initialize_creates_missing_strict_tables() -> None: 

108 """Initialization creates deterministic quoted STRICT tables.""" 

109 

110 class User[S = Pending](Model[S, "User[Fetched]"]): 

111 """Table model used for schema creation.""" 

112 

113 id: User.GenCol[int] = Integer( 

114 primary_key=True, 

115 auto_increment=True, 

116 default=MISSING, 

117 ) 

118 email: User.Col[str] = Text(nullable=False) 

119 

120 with TemporaryDirectory() as directory: 

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

122 database = await Database.initialize( 

123 logger=NULL_LOGGER, database=database_path, models=[User] 

124 ) 

125 await database.close() 

126 

127 create_table = _fetch_create_table(database_path, "user") 

128 

129 expected_sql = ( 

130 'CREATE TABLE "user" ("id" INTEGER PRIMARY KEY AUTOINCREMENT, ' 

131 '"email" TEXT NOT NULL) STRICT' 

132 ) 

133 assert_eq(create_table, expected_sql) 

134 

135 

136@test(mark="medium") 

137async def initialize_accepts_sqlite_config_object() -> None: 

138 """SQLite configuration objects select the SQLite runtime explicitly.""" 

139 

140 class User[S = Pending](sqlite.Model[S, "User[Fetched]"]): 

141 """Table model used for SQLite config initialization.""" 

142 

143 id: User.GenCol[int] = sqlite.Integer( 

144 primary_key=True, 

145 auto_increment=True, 

146 default=MISSING, 

147 ) 

148 email: User.Col[str] = sqlite.Text(nullable=False) 

149 

150 with TemporaryDirectory() as directory: 

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

152 config = sqlite.Config(database=database_path, pool_size=2) 

153 database = await Database.initialize(config, logger=NULL_LOGGER, models=[User]) 

154 await database.close() 

155 

156 create_table = _fetch_create_table(database_path, "user") 

157 

158 expected_sql = ( 

159 'CREATE TABLE "user" ("id" INTEGER PRIMARY KEY AUTOINCREMENT, ' 

160 '"email" TEXT NOT NULL) STRICT' 

161 ) 

162 assert_eq(create_table, expected_sql) 

163 

164 

165@test(mark="medium") 

166async def initialize_rejects_mixed_sqlite_config_and_legacy_database() -> None: 

167 """A backend config cannot be combined with legacy database arguments.""" 

168 

169 with TemporaryDirectory() as directory: 

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

171 config = sqlite.Config(database=database_path) 

172 

173 initialize = cast("Any", Database.initialize) 

174 

175 with assert_raises(DatabaseRuntimeError): 

176 _ = await initialize(config, logger=NULL_LOGGER, database=database_path) 

177 

178 

179@test(mark="medium") 

180async def initialize_creates_column_unique_indexes_after_tables() -> None: 

181 """Column unique declarations create separate deterministic unique indexes.""" 

182 

183 class User[S = Pending](Model[S, "User[Fetched]"]): 

184 """Table model with a unique public identifier.""" 

185 

186 email: User.Col[str] = Text(nullable=False, unique=True) 

187 status: User.Col[str] = Text(nullable=False) 

188 

189 with TemporaryDirectory() as directory: 

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

191 database = await Database.initialize( 

192 logger=NULL_LOGGER, database=database_path, models=[User] 

193 ) 

194 await database.close() 

195 

196 create_table = _fetch_create_table(database_path, "user") 

197 create_indexes = _fetch_create_indexes(database_path, "user") 

198 

199 assert_eq( 

200 create_table, 

201 'CREATE TABLE "user" ("email" TEXT NOT NULL, "status" TEXT NOT NULL) STRICT', 

202 ) 

203 assert_eq( 

204 create_indexes, 

205 ['CREATE UNIQUE INDEX "ux_user_email" ON "user" ("email")'], 

206 ) 

207 

208 

209@test(mark="medium") 

210async def initialize_creates_table_indexes_in_declaration_order() -> None: 

211 """Table index declarations create deterministic index SQL after uniques.""" 

212 

213 class User[S = Pending](Model[S, "User[Fetched]"]): 

214 """Table model with single and composite table indexes.""" 

215 

216 email: User.Col[str] = Text(nullable=False, unique=True) 

217 status: User.Col[str] = Text(nullable=False) 

218 tenant_id: User.Col[int] = Integer(nullable=False) 

219 

220 __indexes__: ClassVar[list[Index[Any]]] = [ 

221 Index(status), 

222 Index(tenant_id, email, unique=True), 

223 Index(tenant_id, name="ix_user_tenant_custom"), 

224 ] 

225 

226 with TemporaryDirectory() as directory: 

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

228 database = await Database.initialize( 

229 logger=NULL_LOGGER, database=database_path, models=[User] 

230 ) 

231 await database.close() 

232 

233 create_indexes = _fetch_create_indexes(database_path, "user") 

234 

235 assert_eq( 

236 create_indexes, 

237 [ 

238 'CREATE UNIQUE INDEX "ux_user_email" ON "user" ("email")', 

239 'CREATE INDEX "ix_user_status" ON "user" ("status")', 

240 'CREATE UNIQUE INDEX "ux_user_tenant_id_email" ON "user" ("tenant_id", "email")', 

241 'CREATE INDEX "ix_user_tenant_custom" ON "user" ("tenant_id")', 

242 ], 

243 ) 

244 

245 

246@test(mark="medium") 

247async def initialize_accepts_only_path_objects_and_exact_memory_string() -> None: 

248 """Only pathlib.Path and the exact in-memory database string are accepted.""" 

249 

250 initialize = cast("Any", Database.initialize) 

251 

252 class OtherPath: 

253 """Path-like object that is intentionally not pathlib.Path.""" 

254 

255 def __fspath__(self) -> str: 

256 return "app.db" 

257 

258 with assert_raises(DatabaseRuntimeError): 

259 _ = await initialize(logger=NULL_LOGGER, database="app.db") 

260 

261 with assert_raises(DatabaseRuntimeError): 

262 _ = await Database.initialize( 

263 logger=NULL_LOGGER, database=cast("Any", "app.db") 

264 ) 

265 

266 with assert_raises(DatabaseRuntimeError): 

267 _ = await Database.initialize( 

268 logger=NULL_LOGGER, database=cast("Any", "sqlite:///app.db") 

269 ) 

270 

271 with assert_raises(DatabaseRuntimeError): 

272 _ = await Database.initialize( 

273 logger=NULL_LOGGER, database=cast("Any", b"app.db") 

274 ) 

275 

276 with assert_raises(DatabaseRuntimeError): 

277 _ = await Database.initialize( 

278 logger=NULL_LOGGER, database=cast("Any", OtherPath()) 

279 ) 

280 

281 database = await Database.initialize(logger=NULL_LOGGER, database=":memory:") 

282 await database.close() 

283 

284 

285@test(mark="medium") 

286async def initialize_verifies_existing_tables_after_controlled_normalization() -> None: 

287 """Equivalent snekql DDL with formatting differences verifies cleanly.""" 

288 

289 class User[S = Pending](Model[S, "User[Fetched]"]): 

290 """Table model used for existing schema verification.""" 

291 

292 id: User.GenCol[int] = Integer( 

293 primary_key=True, 

294 auto_increment=True, 

295 default=MISSING, 

296 ) 

297 email: User.Col[str] = Text(nullable=False) 

298 

299 with TemporaryDirectory() as directory: 

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

301 existing_sql = ( 

302 'CREATE TABLE "user" (\n' 

303 ' "id" INTEGER PRIMARY KEY AUTOINCREMENT,\n' 

304 ' "email" TEXT NOT NULL\n' 

305 ") STRICT" 

306 ) 

307 _execute_sql(database_path, existing_sql) 

308 

309 database = await Database.initialize( 

310 logger=NULL_LOGGER, database=database_path, models=[User] 

311 ) 

312 await database.close() 

313 

314 create_table = _fetch_create_table(database_path, "user") 

315 

316 assert_true("STRICT" in create_table) 

317 

318 

319@test(mark="medium") 

320async def strict_schema_policy_raises_on_index_drift() -> None: 

321 """Strict schema verification rejects missing managed indexes.""" 

322 

323 class User[S = Pending](Model[S, "User[Fetched]"]): 

324 """Table model requiring an index for verification.""" 

325 

326 email: User.Col[str] = Text(nullable=False, unique=True) 

327 

328 with TemporaryDirectory() as directory: 

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

330 _execute_sql( 

331 database_path, 'CREATE TABLE "user" ("email" TEXT NOT NULL) STRICT' 

332 ) 

333 

334 with assert_raises(SchemaVerificationError): 

335 _ = await Database.initialize( 

336 logger=NULL_LOGGER, database=database_path, models=[User] 

337 ) 

338 

339 

340@test(mark="medium") 

341async def duplicate_resolved_index_names_are_rejected() -> None: 

342 """Initialization rejects duplicate index names across configured models.""" 

343 

344 class User[S = Pending](Model[S, "User[Fetched]"]): 

345 """First model using an explicit index name.""" 

346 

347 email: User.Col[str] = Text(nullable=False) 

348 __indexes__: ClassVar[list[Index[Any]]] = [ 

349 Index(email, name="ix_conflict"), 

350 ] 

351 

352 class Account[S = Pending](Model[S, "Account[Fetched]"]): 

353 """Second model reusing the explicit index name.""" 

354 

355 email: Account.Col[str] = Text(nullable=False) 

356 __indexes__: ClassVar[list[Index[Any]]] = [ 

357 Index(email, name="ix_conflict"), 

358 ] 

359 

360 with TemporaryDirectory() as directory: 

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

362 with assert_raises(SchemaError): 

363 _ = await Database.initialize( 

364 logger=NULL_LOGGER, 

365 database=database_path, 

366 models=[User, Account], 

367 ) 

368 

369 

370@test(mark="medium") 

371async def strict_schema_policy_raises_on_schema_drift() -> None: 

372 """Strict schema verification rejects existing non-STRICT tables.""" 

373 

374 class User[S = Pending](Model[S, "User[Fetched]"]): 

375 """Table model used for drift detection.""" 

376 

377 email: User.Col[str] = Text(nullable=False) 

378 

379 with TemporaryDirectory() as directory: 

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

381 _execute_sql(database_path, 'CREATE TABLE "user" ("email" TEXT NOT NULL)') 

382 

383 with assert_raises(SchemaVerificationError): 

384 _ = await Database.initialize( 

385 logger=NULL_LOGGER, database=database_path, models=[User] 

386 ) 

387 

388 

389@test(mark="medium") 

390async def warn_schema_policy_logs_drift_and_continues() -> None: 

391 """Warn schema verification reports drift without blocking startup.""" 

392 

393 class User[S = Pending](Model[S, "User[Fetched]"]): 

394 """Table model used for warn policy drift detection.""" 

395 

396 email: User.Col[str] = Text(nullable=False) 

397 

398 logger = _RecordingStructuredLogger() 

399 with TemporaryDirectory() as directory: 

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

401 _execute_sql(database_path, 'CREATE TABLE "user" ("email" TEXT NOT NULL)') 

402 

403 database = await Database.initialize( 

404 logger=logger, 

405 database=database_path, 

406 models=[User], 

407 schema_policy="warn", 

408 ) 

409 await database.close() 

410 

411 assert_true( 

412 any( 

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

414 for level, event, _fields in logger.events 

415 ) 

416 ) 

417 

418 

419@test(mark="medium") 

420async def duplicate_resolved_table_names_are_rejected() -> None: 

421 """Initialization rejects duplicate table names before schema setup.""" 

422 

423 class User[S = Pending](Model[S, "User[Fetched]"]): 

424 """First model for duplicate table detection.""" 

425 

426 __tablename__ = "account" 

427 email: User.Col[str] = Text(nullable=False) 

428 

429 class Account[S = Pending](Model[S, "Account[Fetched]"]): 

430 """Second model with the same resolved table name.""" 

431 

432 email: Account.Col[str] = Text(nullable=False) 

433 

434 with TemporaryDirectory() as directory: 

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

436 with assert_raises(SchemaError): 

437 _ = await Database.initialize( 

438 logger=NULL_LOGGER, 

439 database=database_path, 

440 models=[User, Account], 

441 ) 

442 

443 

444@test(mark="medium") 

445async def schema_setup_rolls_back_created_tables_on_strict_drift() -> None: 

446 """Schema setup is transactional for create plus verification failure.""" 

447 

448 class CreatedFirst[S = Pending](Model[S, "CreatedFirst[Fetched]"]): 

449 """Missing table that should be rolled back.""" 

450 

451 email: CreatedFirst.Col[str] = Text(nullable=False) 

452 

453 class ExistingDrift[S = Pending](Model[S, "ExistingDrift[Fetched]"]): 

454 """Existing drift table that aborts initialization.""" 

455 

456 email: ExistingDrift.Col[str] = Text(nullable=False) 

457 

458 with TemporaryDirectory() as directory: 

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

460 _execute_sql( 

461 database_path, 

462 'CREATE TABLE "existing_drift" ("email" TEXT NOT NULL)', 

463 ) 

464 

465 with assert_raises(SchemaVerificationError): 

466 _ = await Database.initialize( 

467 logger=NULL_LOGGER, 

468 database=database_path, 

469 models=[CreatedFirst, ExistingDrift], 

470 ) 

471 

472 assert_true(not _table_exists(database_path, "created_first"))