Coverage for tests/test_mariadb_runtime.py: 99%

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

2 

3from __future__ import annotations 

4 

5import subprocess 

6import sys 

7 

8from snektest import assert_eq, assert_in, assert_raises, load_fixture, test 

9 

10from snekql import ( 

11 MISSING, 

12 Database, 

13 DatabaseClosedError, 

14 ExecutionError, 

15 Pending, 

16 PoolTimeoutError, 

17 delete, 

18 insert, 

19 mariadb, 

20 select, 

21 update, 

22) 

23from tests.logging_helpers import NULL_LOGGER 

24from tests.mariadb_server import MariaDBServer, provide_mariadb_server 

25 

26 

27class _RollbackSentinelError(Exception): 

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

29 

30 

31def _force_rollback() -> None: 

32 """Raise the sentinel outside the transaction test body.""" 

33 

34 raise _RollbackSentinelError 

35 

36 

37def _config_from_server(server: MariaDBServer, *, pool_size: int = 5) -> mariadb.Config: 

38 """Build a MariaDB config for the shared local test server.""" 

39 

40 return mariadb.Config( 

41 database=server.database, 

42 host=server.host, 

43 pool_size=pool_size, 

44 port=server.port, 

45 user=server.user, 

46 ) 

47 

48 

49@test(mark="medium") 

50async def mariadb_runtime_creates_schema_and_round_trips_model_rows() -> None: 

51 """A minimal MariaDB Database can create, insert, select, and close.""" 

52 

53 class User[S = Pending](mariadb.Model[S, "User[object]"]): 

54 """Table model for the first MariaDB runtime tracer bullet.""" 

55 

56 __tablename__ = "issue37_user_round_trip" 

57 

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

59 primary_key=True, 

60 auto_increment=True, 

61 default=MISSING, 

62 ) 

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

64 

65 server = load_fixture(provide_mariadb_server()) 

66 database = await Database.initialize( 

67 NULL_LOGGER, _config_from_server(server), models=[User] 

68 ) 

69 try: 

70 async with database.transaction() as transaction: 

71 await transaction.execute(insert(User(email="alice@example.com"))) 

72 fetched_user = await transaction.fetch_one( 

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

74 ) 

75 finally: 

76 await database.close() 

77 

78 assert fetched_user is not None 

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

80 assert isinstance(fetched_user.id, int) 

81 

82 

83@test(mark="medium") 

84async def mariadb_runtime_covers_rollback_pool_timeout_and_close() -> None: 

85 """The initial MariaDB adapter handles transaction and pool lifecycle.""" 

86 

87 class User[S = Pending](mariadb.Model[S, "User[object]"]): 

88 """Table model for MariaDB transaction lifecycle coverage.""" 

89 

90 __tablename__ = "issue37_user_lifecycle" 

91 

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

93 primary_key=True, 

94 auto_increment=True, 

95 default=MISSING, 

96 ) 

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

98 

99 server = load_fixture(provide_mariadb_server()) 

100 database = await Database.initialize( 

101 NULL_LOGGER, _config_from_server(server, pool_size=1), models=[User] 

102 ) 

103 try: 

104 try: 

105 async with database.transaction() as transaction: 

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

107 _force_rollback() 

108 except _RollbackSentinelError: 

109 pass 

110 

111 async with database.transaction() as transaction: 

112 rolled_back_user = await transaction.fetch_one( 

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

114 ) 

115 assert_eq(rolled_back_user, None) 

116 

117 async with database.transaction(timeout=0.5): 

118 with assert_raises(PoolTimeoutError): 

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

120 pass 

121 finally: 

122 await database.close() 

123 

124 with assert_raises(DatabaseClosedError): 

125 _ = database.transaction() 

126 

127 

128@test(mark="medium") 

129async def mariadb_runtime_executes_the_full_query_surface() -> None: 

130 """MariaDB supports result shapes, filters, ordering, updates, and deletes.""" 

131 

132 class User[S = Pending](mariadb.Model[S, "User[object]"]): 

133 """Table model for MariaDB query surface coverage.""" 

134 

135 __tablename__ = "issue38_user_query_surface" 

136 

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

138 primary_key=True, 

139 auto_increment=True, 

140 default=MISSING, 

141 ) 

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

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

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

145 

146 server = load_fixture(provide_mariadb_server()) 

147 database = await Database.initialize( 

148 NULL_LOGGER, _config_from_server(server), models=[User] 

149 ) 

150 try: 

151 async with database.transaction() as transaction: 

152 await transaction.execute( 

153 insert( 

154 User(email="charlie@example.com", status="inactive", tenant_id=1) 

155 ) 

156 ) 

157 await transaction.execute( 

158 insert(User(email="alice@example.com", status="active", tenant_id=1)) 

159 ) 

160 await transaction.execute( 

161 insert(User(email="bob@example.com", status="active", tenant_id=2)) 

162 ) 

163 

164 scalar_rows = await transaction.fetch_all( 

165 select(User.email) 

166 .where(User.tenant_id.eq(1) & User.status.eq("active")) 

167 .order_by(User.email.asc()) 

168 .limit(1) 

169 .offset(0), 

170 ) 

171 tuple_rows = await transaction.fetch_all( 

172 select(User.email, User.status) 

173 .where(User.tenant_id.eq(1)) 

174 .order_by(User.email.asc()) 

175 .limit(1) 

176 .offset(1), 

177 ) 

178 

179 await transaction.execute( 

180 update(User) 

181 .set(User.status.to("disabled")) 

182 .where(User.email.eq("bob@example.com")), 

183 ) 

184 updated_status = await transaction.fetch_one( 

185 select(User.status).where(User.email.eq("bob@example.com")), 

186 ) 

187 

188 await transaction.execute(delete(User).where(User.status.eq("inactive"))) 

189 deleted_user = await transaction.fetch_one( 

190 select(User).where(User.email.eq("charlie@example.com")), 

191 ) 

192 

193 await transaction.execute(delete(User).all()) 

194 remaining_users = await transaction.fetch_all(select(User).all()) 

195 finally: 

196 await database.close() 

197 

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

199 assert_eq(tuple_rows, [("charlie@example.com", "inactive")]) 

200 assert_eq(updated_status, "disabled") 

201 assert_eq(deleted_user, None) 

202 assert_eq(remaining_users, []) 

203 

204 

205@test(mark="medium") 

206async def mariadb_execution_errors_preserve_sql_and_params() -> None: 

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

208 

209 class Account[S = Pending](mariadb.Model[S, "Account[object]"]): 

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

211 

212 __tablename__ = "issue38_account_errors" 

213 

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

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

216 

217 server = load_fixture(provide_mariadb_server()) 

218 database = await Database.initialize( 

219 NULL_LOGGER, _config_from_server(server), models=[Account] 

220 ) 

221 try: 

222 async with database.transaction() as transaction: 

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

224 with assert_raises(ExecutionError) as raised: 

225 await transaction.execute( 

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

227 ) 

228 finally: 

229 await database.close() 

230 

231 assert_in("INSERT INTO `issue38_account_errors`", raised.exception.sql) 

232 assert_in("%s", raised.exception.sql) 

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

234 

235 

236@test(mark="medium") 

237def mariadb_initialization_without_extra_reports_install_hint() -> None: 

238 """Runtime initialization explains how to install a missing MariaDB driver.""" 

239 

240 script = """ 

241from __future__ import annotations 

242 

243import asyncio 

244import importlib.abc 

245import sys 

246 

247import snekql 

248from snekql import Database, mariadb 

249from tests.logging_helpers import NULL_LOGGER 

250 

251 

252class BlockAiomysql(importlib.abc.MetaPathFinder): 

253 def find_spec(self, fullname, path=None, target=None): 

254 if fullname == "aiomysql" or fullname.startswith("aiomysql."): 

255 raise ModuleNotFoundError("No module named 'aiomysql'", name="aiomysql") 

256 return None 

257 

258 

259async def main() -> None: 

260 blocker = BlockAiomysql() 

261 sys.modules.pop("aiomysql", None) 

262 sys.meta_path.insert(0, blocker) 

263 try: 

264 _ = await Database.initialize(NULL_LOGGER, mariadb.Config(database="app", user="snekql")) 

265 except snekql.DatabaseRuntimeError as error: 

266 print(error) 

267 return 

268 finally: 

269 sys.meta_path.remove(blocker) 

270 raise AssertionError("MariaDB initialization unexpectedly succeeded") 

271 

272 

273asyncio.run(main()) 

274""" 

275 

276 result = subprocess.run( 

277 [sys.executable, "-c", script], 

278 check=False, 

279 capture_output=True, 

280 text=True, 

281 ) 

282 

283 assert_eq(result.returncode, 0) 

284 assert_in("snekql[aiomysql]", result.stdout)