Coverage for tests/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 Index, 

17 Integer, 

18 Model, 

19 Pending, 

20 SchemaError, 

21 SchemaVerificationError, 

22 Text, 

23 sqlite, 

24) 

25from tests.logging_helpers import NULL_LOGGER 

26 

27 

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

29 connection = connect(database_path) 

30 try: 

31 _ = connection.execute(sql) 

32 connection.commit() 

33 finally: 

34 connection.close() 

35 

36 

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

38 connection = connect(database_path) 

39 try: 

40 cursor = connection.execute( 

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

42 (table_name,), 

43 ) 

44 row = cursor.fetchone() 

45 assert row is not None 

46 value = row[0] 

47 assert isinstance(value, str) 

48 return value 

49 finally: 

50 connection.close() 

51 

52 

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

54 connection = connect(database_path) 

55 try: 

56 cursor = connection.execute( 

57 """ 

58 SELECT sql FROM sqlite_master 

59 WHERE type = 'index' AND tbl_name = ? 

60 ORDER BY rowid 

61 """, 

62 (table_name,), 

63 ) 

64 values: list[str] = [] 

65 for row in cursor.fetchall(): 

66 value = row[0] 

67 assert isinstance(value, str) 

68 values.append(value) 

69 return values 

70 finally: 

71 connection.close() 

72 

73 

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

75 connection = connect(database_path) 

76 try: 

77 cursor = connection.execute( 

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

79 (table_name,), 

80 ) 

81 return cursor.fetchone() is not None 

82 finally: 

83 connection.close() 

84 

85 

86class _RecordingStructuredLogger: 

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

88 

89 def __init__(self) -> None: 

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

91 

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

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

94 

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

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

97 

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

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

100 

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

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

103 

104 

105@test(mark="medium") 

106async def initialize_creates_missing_strict_tables() -> None: 

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

108 

109 class User[S = Pending](Model[S, "User[object]"]): 

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

111 

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

113 primary_key=True, 

114 auto_increment=True, 

115 default=MISSING, 

116 ) 

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

118 

119 with TemporaryDirectory() as directory: 

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

121 database = await Database.initialize( 

122 NULL_LOGGER, database=database_path, models=[User] 

123 ) 

124 await database.close() 

125 

126 create_table = _fetch_create_table(database_path, "user") 

127 

128 expected_sql = ( 

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

130 '"email" TEXT NOT NULL) STRICT' 

131 ) 

132 assert_eq(create_table, expected_sql) 

133 

134 

135@test(mark="medium") 

136async def initialize_accepts_sqlite_config_object() -> None: 

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

138 

139 class User[S = Pending](sqlite.Model[S, "User[object]"]): 

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

141 

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

143 primary_key=True, 

144 auto_increment=True, 

145 default=MISSING, 

146 ) 

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

148 

149 with TemporaryDirectory() as directory: 

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

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

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

153 await database.close() 

154 

155 create_table = _fetch_create_table(database_path, "user") 

156 

157 expected_sql = ( 

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

159 '"email" TEXT NOT NULL) STRICT' 

160 ) 

161 assert_eq(create_table, expected_sql) 

162 

163 

164@test(mark="medium") 

165async def initialize_rejects_mixed_sqlite_config_and_legacy_database() -> None: 

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

167 

168 with TemporaryDirectory() as directory: 

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

170 config = sqlite.Config(database=database_path) 

171 

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

173 

174 with assert_raises(DatabaseRuntimeError): 

175 _ = await initialize(NULL_LOGGER, config, database=database_path) 

176 

177 

178@test(mark="medium") 

179async def initialize_creates_column_unique_indexes_after_tables() -> None: 

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

181 

182 class User[S = Pending](Model[S, "User[object]"]): 

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

184 

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

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

187 

188 with TemporaryDirectory() as directory: 

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

190 database = await Database.initialize( 

191 NULL_LOGGER, database=database_path, models=[User] 

192 ) 

193 await database.close() 

194 

195 create_table = _fetch_create_table(database_path, "user") 

196 create_indexes = _fetch_create_indexes(database_path, "user") 

197 

198 assert_eq( 

199 create_table, 

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

201 ) 

202 assert_eq( 

203 create_indexes, 

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

205 ) 

206 

207 

208@test(mark="medium") 

209async def initialize_creates_table_indexes_in_declaration_order() -> None: 

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

211 

212 class User[S = Pending](Model[S, "User[object]"]): 

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

214 

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

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

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

218 

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

220 Index(status), 

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

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

223 ] 

224 

225 with TemporaryDirectory() as directory: 

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

227 database = await Database.initialize( 

228 NULL_LOGGER, database=database_path, models=[User] 

229 ) 

230 await database.close() 

231 

232 create_indexes = _fetch_create_indexes(database_path, "user") 

233 

234 assert_eq( 

235 create_indexes, 

236 [ 

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

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

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

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

241 ], 

242 ) 

243 

244 

245@test(mark="medium") 

246async def initialize_accepts_only_path_objects_and_exact_memory_string() -> None: 

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

248 

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

250 

251 class OtherPath: 

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

253 

254 def __fspath__(self) -> str: 

255 return "app.db" 

256 

257 with assert_raises(DatabaseRuntimeError): 

258 _ = await initialize(NULL_LOGGER, "app.db") 

259 

260 with assert_raises(DatabaseRuntimeError): 

261 _ = await Database.initialize(NULL_LOGGER, database=cast("Any", "app.db")) 

262 

263 with assert_raises(DatabaseRuntimeError): 

264 _ = await Database.initialize( 

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

266 ) 

267 

268 with assert_raises(DatabaseRuntimeError): 

269 _ = await Database.initialize(NULL_LOGGER, database=cast("Any", b"app.db")) 

270 

271 with assert_raises(DatabaseRuntimeError): 

272 _ = await Database.initialize(NULL_LOGGER, database=cast("Any", OtherPath())) 

273 

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

275 await database.close() 

276 

277 

278@test(mark="medium") 

279async def initialize_verifies_existing_tables_after_controlled_normalization() -> None: 

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

281 

282 class User[S = Pending](Model[S, "User[object]"]): 

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

284 

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

286 primary_key=True, 

287 auto_increment=True, 

288 default=MISSING, 

289 ) 

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

291 

292 with TemporaryDirectory() as directory: 

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

294 existing_sql = ( 

295 'CREATE TABLE "user" (\n' 

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

297 ' "email" TEXT NOT NULL\n' 

298 ") STRICT" 

299 ) 

300 _execute_sql(database_path, existing_sql) 

301 

302 database = await Database.initialize( 

303 NULL_LOGGER, database=database_path, models=[User] 

304 ) 

305 await database.close() 

306 

307 create_table = _fetch_create_table(database_path, "user") 

308 

309 assert_true("STRICT" in create_table) 

310 

311 

312@test(mark="medium") 

313async def strict_schema_policy_raises_on_index_drift() -> None: 

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

315 

316 class User[S = Pending](Model[S, "User[object]"]): 

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

318 

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

320 

321 with TemporaryDirectory() as directory: 

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

323 _execute_sql( 

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

325 ) 

326 

327 with assert_raises(SchemaVerificationError): 

328 _ = await Database.initialize( 

329 NULL_LOGGER, database=database_path, models=[User] 

330 ) 

331 

332 

333@test(mark="medium") 

334async def duplicate_resolved_index_names_are_rejected() -> None: 

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

336 

337 class User[S = Pending](Model[S, "User[object]"]): 

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

339 

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

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

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

343 ] 

344 

345 class Account[S = Pending](Model[S, "Account[object]"]): 

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

347 

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

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

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

351 ] 

352 

353 with TemporaryDirectory() as directory: 

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

355 with assert_raises(SchemaError): 

356 _ = await Database.initialize( 

357 NULL_LOGGER, 

358 database=database_path, 

359 models=[User, Account], 

360 ) 

361 

362 

363@test(mark="medium") 

364async def strict_schema_policy_raises_on_schema_drift() -> None: 

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

366 

367 class User[S = Pending](Model[S, "User[object]"]): 

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

369 

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

371 

372 with TemporaryDirectory() as directory: 

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

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

375 

376 with assert_raises(SchemaVerificationError): 

377 _ = await Database.initialize( 

378 NULL_LOGGER, database=database_path, models=[User] 

379 ) 

380 

381 

382@test(mark="medium") 

383async def warn_schema_policy_logs_drift_and_continues() -> None: 

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

385 

386 class User[S = Pending](Model[S, "User[object]"]): 

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

388 

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

390 

391 logger = _RecordingStructuredLogger() 

392 with TemporaryDirectory() as directory: 

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

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

395 

396 database = await Database.initialize( 

397 logger, 

398 database=database_path, 

399 models=[User], 

400 schema_policy="warn", 

401 ) 

402 await database.close() 

403 

404 assert_true( 

405 any( 

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

407 for level, event, _fields in logger.events 

408 ) 

409 ) 

410 

411 

412@test(mark="medium") 

413async def duplicate_resolved_table_names_are_rejected() -> None: 

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

415 

416 class User[S = Pending](Model[S, "User[object]"]): 

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

418 

419 __tablename__ = "account" 

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

421 

422 class Account[S = Pending](Model[S, "Account[object]"]): 

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

424 

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

426 

427 with TemporaryDirectory() as directory: 

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

429 with assert_raises(SchemaError): 

430 _ = await Database.initialize( 

431 NULL_LOGGER, 

432 database=database_path, 

433 models=[User, Account], 

434 ) 

435 

436 

437@test(mark="medium") 

438async def schema_setup_rolls_back_created_tables_on_strict_drift() -> None: 

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

440 

441 class CreatedFirst[S = Pending](Model[S, "CreatedFirst[object]"]): 

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

443 

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

445 

446 class ExistingDrift[S = Pending](Model[S, "ExistingDrift[object]"]): 

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

448 

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

450 

451 with TemporaryDirectory() as directory: 

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

453 _execute_sql( 

454 database_path, 

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

456 ) 

457 

458 with assert_raises(SchemaVerificationError): 

459 _ = await Database.initialize( 

460 NULL_LOGGER, 

461 database=database_path, 

462 models=[CreatedFirst, ExistingDrift], 

463 ) 

464 

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