Coverage for tests/test_public_api.py: 98%

1"""Public API contract tests for snekql.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import cast 

7 

8from snektest import test 

9from snektest.assertions import ( 

10 assert_eq, 

11 assert_in, 

12 assert_is, 

13 assert_isinstance, 

14 assert_ne, 

15) 

16 

17import snekql 

18from snekql.testing import mariadb as testing_mariadb 

19 

20 

21def _assert_has_specific_docstring(value: object) -> None: 

22 docstring = getattr(value, "__doc__", None) 

23 assert_ne(docstring, None) 

24 assert_ne(docstring, "") 

25 assert_ne(docstring, object.__doc__) 

26 

27 

28def _catch_as_snekql_error(error: snekql.SnekqlError) -> None: 

29 try: 

30 raise error 

31 except snekql.SnekqlError as caught_error: 

32 assert_is(caught_error, error) 

33 

34 

35@test() 

36def public_contract_exports_canonical_names() -> None: 

37 """The package root explicitly curates canonical PRD names.""" 

38 

39 expected_names = ( 

40 "MISSING", 

41 "Assignment", 

42 "Attr", 

43 "Blob", 

44 "Boolean", 

45 "Col", 

46 "CurrentTimestamp", 

47 "Database", 

48 "DatabaseCloseTimeoutError", 

49 "DatabaseClosedError", 

50 "DatabaseClosingError", 

51 "DatabaseRuntimeError", 

52 "DateTime", 

53 "DeleteQuery", 

54 "ExecutionError", 

55 "Fetched", 

56 "FrozenModelError", 

57 "GenCol", 

58 "Index", 

59 "InsertQuery", 

60 "Integer", 

61 "Json", 

62 "Missing", 

63 "Model", 

64 "ModelDeclarationError", 

65 "ModelError", 

66 "ModelMeta", 

67 "ModelValidationError", 

68 "OrderBy", 

69 "Pending", 

70 "PoolTimeoutError", 

71 "Predicate", 

72 "QueryCompilationError", 

73 "QueryConstructionError", 

74 "QueryError", 

75 "Real", 

76 "SchemaError", 

77 "SchemaPolicy", 

78 "SchemaVerificationError", 

79 "SelectModelQuery", 

80 "SelectTupleQuery", 

81 "SelectValueQuery", 

82 "SnekqlError", 

83 "StructuredLogger", 

84 "Table", 

85 "Text", 

86 "Transaction", 

87 "TransactionClosedError", 

88 "UpdateQuery", 

89 "delete", 

90 "insert", 

91 "mariadb", 

92 "select", 

93 "sqlite", 

94 "update", 

95 ) 

96 

97 assert_eq(tuple(snekql.__all__), expected_names) 

98 for name in expected_names: 

99 assert_in(name, snekql.__all__) 

100 assert_eq(getattr(snekql, name), getattr(snekql, name)) 

101 

102 

103@test() 

104def testing_mariadb_namespace_exports_test_server_names() -> None: 

105 """The testing namespace exposes MariaDB test-server support directly.""" 

106 

107 assert_eq( 

108 tuple(testing_mariadb.__all__), 

109 ( 

110 "MariaDBAuth", 

111 "MariaDBCommandResult", 

112 "MariaDBTransport", 

113 "TemporaryMariaDBServer", 

114 "TemporaryMariaDBServerError", 

115 "temporary_mariadb_server", 

116 ), 

117 ) 

118 assert_in("mariadb", __import__("snekql.testing").testing.__all__) 

119 assert_in("temporary_mariadb_server", testing_mariadb.__all__) 

120 assert_isinstance( 

121 testing_mariadb.TemporaryMariaDBServerError("failure"), 

122 snekql.SnekqlError, 

123 ) 

124 assert "testing" not in snekql.__all__ 

125 

126 

127@test() 

128def query_factory_functions_reject_empty_selects() -> None: 

129 """Selecting no model or fields is package-originated query misuse.""" 

130 

131 select_fn = cast("Callable[..., object]", snekql.select) 

132 

133 try: 

134 _ = select_fn() 

135 except snekql.QueryConstructionError: 

136 return 

137 

138 msg = "select() should reject empty selection" 

139 raise AssertionError(msg) 

140 

141 

142@test() 

143def column_declarations_produce_query_attributes() -> None: 

144 """Column declarations leave public descriptors on table model classes.""" 

145 

146 class AttributeUser(snekql.Model[snekql.Pending, "AttributeUser[snekql.Fetched]"]): 

147 """Table model for descriptor smoke checks.""" 

148 

149 email: AttributeUser.Col[str] = snekql.Text(nullable=False) 

150 

151 assert_isinstance(AttributeUser.email, snekql.Attr) 

152 assert_isinstance(AttributeUser.email.eq("alice@example.com"), snekql.Predicate) 

153 assert_isinstance(AttributeUser.email.asc(), snekql.OrderBy) 

154 assert_isinstance(AttributeUser.email.to("new@example.com"), snekql.Assignment) 

155 

156 

157@test() 

158def sqlite_namespace_exports_backend_specific_names() -> None: 

159 """The SQLite namespace exposes the future backend-specific model shape.""" 

160 

161 assert_is(snekql.sqlite.Model, snekql.Model) 

162 assert_is(snekql.sqlite.Index, snekql.Index) 

163 assert_is(snekql.sqlite.Integer, snekql.Integer) 

164 assert_is(snekql.sqlite.Text, snekql.Text) 

165 assert_in("Config", snekql.sqlite.__all__) 

166 

167 class SqliteUser(snekql.sqlite.Model[snekql.Pending, "SqliteUser[snekql.Fetched]"]): 

168 """SQLite table model declared through the SQLite namespace.""" 

169 

170 email: SqliteUser.Col[str] = snekql.sqlite.Text(nullable=False) 

171 

172 assert_isinstance(SqliteUser.email, snekql.sqlite.Attr) 

173 assert_isinstance(SqliteUser.email.eq("alice@example.com"), snekql.Predicate) 

174 

175 

176@test() 

177def mutation_query_chain_methods_return_query_objects() -> None: 

178 """Public update/delete chain methods keep returning mutation query objects.""" 

179 

180 class MutationUser(snekql.Model[snekql.Pending, "MutationUser[snekql.Fetched]"]): 

181 """Table model for mutation chain smoke checks.""" 

182 

183 email: MutationUser.Col[str] = snekql.Text(nullable=False) 

184 status: MutationUser.Col[str] = snekql.Text(nullable=False) 

185 

186 assignment = MutationUser.status.to("disabled") 

187 predicate = MutationUser.email.eq("alice@example.com") 

188 

189 update_query = snekql.update(MutationUser) 

190 delete_query = snekql.delete(MutationUser) 

191 

192 assert_isinstance(update_query.set(assignment), snekql.UpdateQuery) 

193 assert_isinstance(update_query.where(predicate), snekql.UpdateQuery) 

194 assert_isinstance(update_query.all(), snekql.UpdateQuery) 

195 assert_isinstance(delete_query.where(predicate), snekql.DeleteQuery) 

196 assert_isinstance(delete_query.all(), snekql.DeleteQuery) 

197 

198 

199@test() 

200def select_query_chain_methods_return_query_objects() -> None: 

201 """Public select chain methods keep returning select query objects.""" 

202 

203 class ChainUser(snekql.Model[snekql.Pending, "ChainUser[snekql.Fetched]"]): 

204 """Table model for select chain smoke checks.""" 

205 

206 query = snekql.select(ChainUser) 

207 

208 assert_isinstance(query.all(), snekql.SelectModelQuery) 

209 assert_isinstance(query.limit(10), snekql.SelectModelQuery) 

210 assert_isinstance(query.offset(5), snekql.SelectModelQuery) 

211 

212 

213@test() 

214def query_factory_functions_return_public_query_objects() -> None: 

215 """Query builder entry points return stable public query classes.""" 

216 

217 class QueryUser(snekql.Model[snekql.Pending, "QueryUser[snekql.Fetched]"]): 

218 """Table model for query factory smoke checks.""" 

219 

220 row = object.__new__(QueryUser) 

221 

222 assert_isinstance(snekql.select(QueryUser), snekql.SelectModelQuery) 

223 assert_isinstance(snekql.insert(row), snekql.InsertQuery) 

224 assert_isinstance(snekql.update(QueryUser), snekql.UpdateQuery) 

225 assert_isinstance(snekql.delete(QueryUser), snekql.DeleteQuery) 

226 

227 

228@test() 

229def public_classes_have_specific_docstrings() -> None: 

230 """Public marker, error, column, query, and runtime classes explain intent.""" 

231 

232 documented_classes = ( 

233 snekql.Assignment, 

234 snekql.Attr, 

235 snekql.Blob, 

236 snekql.Boolean, 

237 snekql.CurrentTimestamp, 

238 snekql.Database, 

239 snekql.DatabaseClosedError, 

240 snekql.DatabaseCloseTimeoutError, 

241 snekql.DatabaseClosingError, 

242 snekql.DatabaseRuntimeError, 

243 snekql.DateTime, 

244 snekql.DeleteQuery, 

245 snekql.ExecutionError, 

246 snekql.Fetched, 

247 snekql.FrozenModelError, 

248 snekql.Index, 

249 snekql.InsertQuery, 

250 snekql.Integer, 

251 snekql.Json, 

252 snekql.Missing, 

253 snekql.Model, 

254 snekql.ModelDeclarationError, 

255 snekql.ModelError, 

256 snekql.ModelMeta, 

257 snekql.ModelValidationError, 

258 snekql.OrderBy, 

259 snekql.Pending, 

260 snekql.PoolTimeoutError, 

261 snekql.Predicate, 

262 snekql.QueryCompilationError, 

263 snekql.QueryConstructionError, 

264 snekql.QueryError, 

265 snekql.Real, 

266 snekql.SchemaError, 

267 snekql.SchemaVerificationError, 

268 snekql.SelectModelQuery, 

269 snekql.SelectTupleQuery, 

270 snekql.SelectValueQuery, 

271 snekql.SnekqlError, 

272 snekql.Table, 

273 snekql.Text, 

274 snekql.Transaction, 

275 snekql.TransactionClosedError, 

276 snekql.UpdateQuery, 

277 ) 

278 

279 for documented_class in documented_classes: 

280 _assert_has_specific_docstring(documented_class) 

281 

282 

283@test() 

284def missing_sentinel_has_stable_singleton_behavior() -> None: 

285 """MISSING is the only Missing value applications need to compare with.""" 

286 

287 assert_is(snekql.Missing(), snekql.MISSING) 

288 assert_eq(repr(snekql.MISSING), "MISSING") 

289 

290 

291@test() 

292def public_error_hierarchy_is_rooted_at_snekql_error() -> None: 

293 """All intentional public errors can be caught as SnekqlError.""" 

294 

295 errors = ( 

296 snekql.DatabaseClosedError("package-originated failure"), 

297 snekql.DatabaseCloseTimeoutError("package-originated failure"), 

298 snekql.DatabaseClosingError("package-originated failure"), 

299 snekql.ExecutionError( 

300 "package-originated failure", 

301 sql="SELECT ?", 

302 params=(1,), 

303 ), 

304 snekql.FrozenModelError("package-originated failure"), 

305 snekql.ModelDeclarationError("package-originated failure"), 

306 snekql.ModelValidationError("package-originated failure"), 

307 snekql.PoolTimeoutError("package-originated failure"), 

308 snekql.QueryCompilationError("package-originated failure"), 

309 snekql.QueryConstructionError("package-originated failure"), 

310 snekql.SchemaVerificationError("package-originated failure"), 

311 snekql.TransactionClosedError("package-originated failure"), 

312 ) 

313 

314 catches: tuple[Callable[[], None], ...] = tuple( 

315 lambda error=error: _catch_as_snekql_error(error) for error in errors 

316 ) 

317 

318 for catch in catches: 

319 catch() 

320 

321 

322@test() 

323def execution_error_preserves_sql_and_params() -> None: 

324 """Execution failures expose query context through the public exception.""" 

325 

326 error = snekql.ExecutionError( 

327 "insert failed", 

328 sql='INSERT INTO "user" ("email") VALUES (?)', 

329 params=("alice@example.com",), 

330 ) 

331 

332 assert_eq(error.sql, 'INSERT INTO "user" ("email") VALUES (?)') 

333 assert_eq(error.params, ("alice@example.com",)) 

334 assert_in("insert failed", str(error)) 

335 assert_in('INSERT INTO "user"', str(error)) 

336 assert_in("alice@example.com", str(error))