Coverage for tests/mariadb/test_runtime.py: 99%

1"""MariaDB runtime tracer-bullet integration tests.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Sequence 

6from typing import Any 

7 

8from snektest import ( 

9 AsyncFixture, 

10 assert_eq, 

11 assert_raises, 

12 load_fixture, 

13 test, 

14) 

15 

16from snekql import ( 

17 MISSING, 

18 Database, 

19 DatabaseClosedError, 

20 ExecutionError, 

21 Fetched, 

22 Pending, 

23 PoolTimeoutError, 

24 delete, 

25 insert, 

26 mariadb, 

27 select, 

28 update, 

29) 

30from snekql.model import Table 

31from tests.helpers import NULL_LOGGER, provide_mariadb_server 

32 

33 

34class _RollbackSentinelError(Exception): 

35 """Test-only exception used to force a transaction rollback.""" 

36 

37 

38class _UpdateUser[S = Pending](mariadb.Model[S, "_UpdateUser[Fetched]"]): 

39 """Table model for MariaDB update coverage.""" 

40 

41 __tablename__ = "issue38_user_update" 

42 

43 id: _UpdateUser.GenCol[int] = mariadb.Integer( 

44 primary_key=True, 

45 auto_increment=True, 

46 default=MISSING, 

47 ) 

48 email: _UpdateUser.Col[str] = mariadb.Text(nullable=False) 

49 status: _UpdateUser.Col[str] = mariadb.Text(nullable=False) 

50 

51 

52async def database_session( 

53 models: Sequence[type[Table[Any]]] = (), 

54 *, 

55 pool_size: int = 1, 

56) -> AsyncFixture[Database]: 

57 """Provide an initialized MariaDB Database and close it after the test.""" 

58 

59 server = await load_fixture(provide_mariadb_server()) 

60 database = await Database.initialize( 

61 server.config(pool_size=pool_size), logger=NULL_LOGGER, models=models 

62 ) 

63 try: 

64 yield database 

65 finally: 

66 await database.close() 

67 

68 

69async def database_with_update_users() -> AsyncFixture[Database]: 

70 """Provide a MariaDB Database seeded with update target rows.""" 

71 

72 database = await load_fixture(database_session([_UpdateUser])) 

73 async with database.transaction() as tx: 

74 await tx.execute( 

75 insert(_UpdateUser(email="alice@example.com", status="active")) 

76 ) 

77 await tx.execute(insert(_UpdateUser(email="bob@example.com", status="active"))) 

78 yield database 

79 

80 

81@test(mark="medium") 

82async def mariadb_runtime_creates_schema_and_round_trips_model_rows() -> None: 

83 """A MariaDB Database can insert, select, and close.""" 

84 

85 class User[S = Pending](mariadb.Model[S, "User[Fetched]"]): 

86 """Table model for the MariaDB runtime.""" 

87 

88 __tablename__ = "issue37_user_round_trip" 

89 

90 id: User.GenCol[int] = mariadb.Integer( 

91 primary_key=True, 

92 auto_increment=True, 

93 default=MISSING, 

94 ) 

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

96 

97 database = await load_fixture(database_session([User])) 

98 

99 async with database.transaction() as tx: 

100 await tx.execute(insert(User(email="alice@example.com"))) 

101 

102 async with database.transaction() as tx: 

103 fetched_user = await tx.fetch_one( 

104 select(User).where(User.email.eq("alice@example.com")), 

105 ) 

106 

107 assert fetched_user is not None 

108 assert_eq(fetched_user.email, "alice@example.com") 

109 assert isinstance(fetched_user.id, int) 

110 

111 

112@test(mark="medium") 

113async def mariadb_runtime_rolls_back_failed_transactions() -> None: 

114 """MariaDB Transactions roll back when the body raises.""" 

115 

116 class User[S = Pending](mariadb.Model[S, "User[Fetched]"]): 

117 __tablename__ = "issue37_user_lifecycle" 

118 

119 id: User.GenCol[int] = mariadb.Integer( 

120 primary_key=True, 

121 auto_increment=True, 

122 default=MISSING, 

123 ) 

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

125 

126 database = await load_fixture(database_session([User])) 

127 

128 try: 

129 async with database.transaction() as tx: 

130 await tx.execute(insert(User(email="rolled-back@example.com"))) 

131 raise _RollbackSentinelError # noqa: TRY301 

132 except _RollbackSentinelError: 

133 pass 

134 

135 async with database.transaction() as tx: 

136 rolled_back_user = await tx.fetch_one( 

137 select(User).where(User.email.eq("rolled-back@example.com")), 

138 ) 

139 

140 assert_eq(rolled_back_user, None) 

141 

142 

143@test(mark="medium") 

144async def mariadb_runtime_reports_pool_timeout() -> None: 

145 """MariaDB Database reports pool exhaustion as a timeout.""" 

146 

147 database = await load_fixture(database_session(pool_size=1)) 

148 

149 async with database.transaction(): 

150 with assert_raises(PoolTimeoutError): 

151 async with database.transaction(timeout=0.01): 

152 pass 

153 

154 

155@test(mark="medium") 

156async def mariadb_runtime_rejects_transactions_after_close() -> None: 

157 """MariaDB Database rejects new Transactions after close.""" 

158 

159 database = await load_fixture(database_session()) 

160 await database.close() 

161 

162 with assert_raises(DatabaseClosedError): 

163 _ = database.transaction() 

164 

165 

166@test(mark="medium") 

167async def mariadb_runtime_fetches_scalar_rows() -> None: 

168 """MariaDB fetch_all returns scalar rows for single-column selects.""" 

169 

170 class User[S = Pending](mariadb.Model[S, "User[Fetched]"]): 

171 """Table model for MariaDB scalar result coverage.""" 

172 

173 __tablename__ = "issue38_user_scalar_result" 

174 

175 id: User.GenCol[int] = mariadb.Integer( 

176 primary_key=True, 

177 auto_increment=True, 

178 default=MISSING, 

179 ) 

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

181 

182 database = await load_fixture(database_session([User])) 

183 

184 async with database.transaction() as tx: 

185 await tx.execute(insert(User(email="alice@example.com"))) 

186 scalar_rows = await tx.fetch_all(select(User.email).all()) 

187 

188 assert_eq(scalar_rows, ["alice@example.com"]) 

189 

190 

191@test(mark="medium") 

192async def mariadb_runtime_fetches_tuple_rows() -> None: 

193 """MariaDB fetch_all returns tuples for multi-column selects.""" 

194 

195 class User[S = Pending](mariadb.Model[S, "User[Fetched]"]): 

196 """Table model for MariaDB tuple result coverage.""" 

197 

198 __tablename__ = "issue38_user_tuple_result" 

199 

200 id: User.GenCol[int] = mariadb.Integer( 

201 primary_key=True, 

202 auto_increment=True, 

203 default=MISSING, 

204 ) 

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

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

207 

208 database = await load_fixture(database_session([User])) 

209 

210 async with database.transaction() as tx: 

211 await tx.execute(insert(User(email="alice@example.com", status="active"))) 

212 tuple_rows = await tx.fetch_all(select(User.email, User.status).all()) 

213 

214 assert_eq(tuple_rows, [("alice@example.com", "active")]) 

215 

216 

217@test(mark="medium") 

218async def mariadb_runtime_updates_matching_rows() -> None: 

219 """MariaDB update changes only rows matching the predicate.""" 

220 

221 database = await load_fixture(database_with_update_users()) 

222 

223 async with database.transaction() as tx: 

224 await tx.execute( 

225 update(_UpdateUser) 

226 .set(_UpdateUser.status.to("disabled")) 

227 .where(_UpdateUser.email.eq("bob@example.com")), 

228 ) 

229 

230 async with database.transaction() as tx: 

231 statuses = await tx.fetch_all( 

232 select(_UpdateUser.email, _UpdateUser.status).all() 

233 ) 

234 

235 assert_eq( 

236 sorted(statuses), 

237 [ 

238 ("alice@example.com", "active"), 

239 ("bob@example.com", "disabled"), 

240 ], 

241 ) 

242 

243 

244@test(mark="medium") 

245async def mariadb_runtime_deletes_filtered_rows() -> None: 

246 """MariaDB delete removes rows matching the predicate.""" 

247 

248 class User[S = Pending](mariadb.Model[S, "User[Fetched]"]): 

249 """Table model for MariaDB filtered delete coverage.""" 

250 

251 __tablename__ = "issue38_user_filtered_delete" 

252 

253 id: User.GenCol[int] = mariadb.Integer( 

254 primary_key=True, 

255 auto_increment=True, 

256 default=MISSING, 

257 ) 

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

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

260 

261 database = await load_fixture(database_session([User])) 

262 

263 async with database.transaction() as tx: 

264 await tx.execute(insert(User(email="alice@example.com", status="active"))) 

265 await tx.execute(insert(User(email="bob@example.com", status="inactive"))) 

266 

267 await tx.execute(delete(User).where(User.status.eq("inactive"))) 

268 remaining_emails = await tx.fetch_all(select(User.email).all()) 

269 

270 assert_eq(remaining_emails, ["alice@example.com"]) 

271 

272 

273@test(mark="medium") 

274async def mariadb_runtime_deletes_all_rows() -> None: 

275 """MariaDB delete all removes every row.""" 

276 

277 class User[S = Pending](mariadb.Model[S, "User[Fetched]"]): 

278 """Table model for MariaDB delete-all coverage.""" 

279 

280 __tablename__ = "issue38_user_delete_all" 

281 

282 id: User.GenCol[int] = mariadb.Integer( 

283 primary_key=True, 

284 auto_increment=True, 

285 default=MISSING, 

286 ) 

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

288 

289 database = await load_fixture(database_session([User])) 

290 

291 async with database.transaction() as tx: 

292 await tx.execute(insert(User(email="alice@example.com"))) 

293 await tx.execute(insert(User(email="bob@example.com"))) 

294 

295 await tx.execute(delete(User).all()) 

296 remaining_users = await tx.fetch_all(select(User).all()) 

297 

298 assert_eq(remaining_users, []) 

299 

300 

301@test(mark="medium") 

302async def mariadb_execution_errors_preserve_sql_and_params() -> None: 

303 """MariaDB write failures expose backend SQL and parameter context.""" 

304 

305 class Account[S = Pending](mariadb.Model[S, "Account[Fetched]"]): 

306 """Table model for MariaDB execution error coverage.""" 

307 

308 __tablename__ = "issue38_account_errors" 

309 

310 id: Account.Col[int] = mariadb.Integer(primary_key=True) 

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

312 

313 database = await load_fixture(database_session([Account])) 

314 

315 async with database.transaction() as tx: 

316 await tx.execute(insert(Account(id=1, email="first@example.com"))) 

317 with assert_raises(ExecutionError) as raised: 

318 await tx.execute( 

319 insert(Account(id=1, email="duplicate@example.com")), 

320 ) 

321 

322 assert_eq( 

323 raised.exception.sql, 

324 "INSERT INTO `issue38_account_errors` (`id`, `email`) VALUES (%s, %s)", 

325 ) 

326 assert_eq(raised.exception.params, (1, "duplicate@example.com"))