Coverage for tests/query/test_select_execution.py: 100%

1"""Select query construction, compilation, and fetch execution tests.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from pathlib import Path 

7from sqlite3 import connect 

8from tempfile import TemporaryDirectory 

9from typing import cast 

10 

11from snektest import assert_eq, assert_raises, test 

12 

13from snekql import ( 

14 MISSING, 

15 Boolean, 

16 Database, 

17 Fetched, 

18 Integer, 

19 Model, 

20 ModelValidationError, 

21 Pending, 

22 QueryCompilationError, 

23 QueryConstructionError, 

24 Text, 

25 insert, 

26 select, 

27) 

28from snekql.query import compile_select_sql 

29from tests.helpers import NULL_LOGGER 

30 

31 

32@test(mark="medium") 

33async def fetch_all_materializes_model_rows() -> None: 

34 """Model selects return fetched-state model instances decoded from rows.""" 

35 

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

37 """Table model selected through the runtime.""" 

38 

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

40 primary_key=True, 

41 auto_increment=True, 

42 default=MISSING, 

43 ) 

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

45 status: User.Col[str] = Text(nullable=False, default="active") 

46 

47 database = await Database.initialize( 

48 logger=NULL_LOGGER, database=":memory:", models=[User] 

49 ) 

50 try: 

51 async with database.transaction() as tx: 

52 await tx.execute(insert(User(email="a@example.com"))) 

53 await tx.execute( 

54 insert(User(email="b@example.com", status="disabled")), 

55 ) 

56 rows = await tx.fetch_all(select(User).all()) 

57 finally: 

58 await database.close() 

59 

60 fetched_user: User[Fetched] = rows[0] 

61 assert_eq(fetched_user.id, 1) 

62 assert_eq(fetched_user.email, "a@example.com") 

63 assert_eq(fetched_user.status, "active") 

64 assert_eq( 

65 repr(fetched_user), 

66 "User[Fetched](id=1, email='a@example.com', status='active')", 

67 ) 

68 assert_eq([row.email for row in rows], ["a@example.com", "b@example.com"]) 

69 

70 

71@test(mark="medium") 

72async def fetch_all_returns_scalar_values_for_single_column_selects() -> None: 

73 """Single-column selects return decoded scalar values instead of row tuples.""" 

74 

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

76 """Table model selected through the runtime.""" 

77 

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

79 primary_key=True, 

80 auto_increment=True, 

81 default=MISSING, 

82 ) 

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

84 status: User.Col[str] = Text(nullable=False, default="active") 

85 

86 database = await Database.initialize( 

87 logger=NULL_LOGGER, database=":memory:", models=[User] 

88 ) 

89 try: 

90 async with database.transaction() as tx: 

91 await tx.execute(insert(User(email="a@example.com"))) 

92 await tx.execute( 

93 insert(User(email="b@example.com", status="disabled")), 

94 ) 

95 emails = await tx.fetch_all( 

96 select(User.email) 

97 .where(User.status.eq("active")) 

98 .order_by(User.email.desc()), 

99 ) 

100 finally: 

101 await database.close() 

102 

103 assert_eq(emails, ["a@example.com"]) 

104 

105 

106@test(mark="medium") 

107async def fetch_all_returns_tuples_for_multi_column_selects() -> None: 

108 """Multi-column selects return value tuples in selection order.""" 

109 

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

111 """Table model selected through the runtime.""" 

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 status: User.Col[str] = Text(nullable=False, default="active") 

120 

121 database = await Database.initialize( 

122 logger=NULL_LOGGER, database=":memory:", models=[User] 

123 ) 

124 try: 

125 async with database.transaction() as tx: 

126 await tx.execute(insert(User(email="a@example.com"))) 

127 await tx.execute( 

128 insert(User(email="b@example.com", status="disabled")), 

129 ) 

130 rows = await tx.fetch_all( 

131 select(User.status, User.email).all().order_by(User.id.asc()), 

132 ) 

133 finally: 

134 await database.close() 

135 

136 assert_eq(rows, [("active", "a@example.com"), ("disabled", "b@example.com")]) 

137 

138 

139@test(mark="fast") 

140def select_rejects_mixed_model_and_field_selections() -> None: 

141 """V1 rejects mixed model+field selections before SQL compilation.""" 

142 

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

144 """Table model used by invalid select construction checks.""" 

145 

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

147 

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

149 

150 with assert_raises(QueryConstructionError): 

151 _ = select_fn(User, User.email) 

152 

153 

154@test(mark="fast") 

155def select_rejects_fields_from_multiple_models() -> None: 

156 """V1 select queries do not support joins across table models.""" 

157 

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

159 """First table model used by invalid select construction checks.""" 

160 

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

162 

163 class AuditLog[S = Pending](Model[S, "AuditLog[Fetched]"]): 

164 """Second table model used by invalid select construction checks.""" 

165 

166 message: AuditLog.Col[str] = Text(nullable=False) 

167 

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

169 

170 with assert_raises(QueryConstructionError): 

171 _ = select_fn(User.email, AuditLog.message) 

172 

173 

174@test(mark="medium") 

175async def fetch_one_returns_first_row_or_none_without_cardinality_checks() -> None: 

176 """fetch_one returns the first selected row and treats empty results as None.""" 

177 

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

179 """Table model selected through fetch_one.""" 

180 

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

182 primary_key=True, 

183 auto_increment=True, 

184 default=MISSING, 

185 ) 

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

187 

188 database = await Database.initialize( 

189 logger=NULL_LOGGER, database=":memory:", models=[User] 

190 ) 

191 try: 

192 async with database.transaction() as tx: 

193 await tx.execute(insert(User(email="a@example.com"))) 

194 await tx.execute(insert(User(email="b@example.com"))) 

195 

196 first_email = await tx.fetch_one( 

197 select(User.email).all().order_by(User.id.asc()), 

198 ) 

199 no_email = await tx.fetch_one(select(User.email).all().limit(0)) 

200 finally: 

201 await database.close() 

202 

203 assert_eq(first_email, "a@example.com") 

204 assert_eq(no_email, None) 

205 

206 

207@test(mark="medium") 

208async def fetch_all_validates_decoded_database_values() -> None: 

209 """Fetched rows are decoded and validated before model materialization.""" 

210 

211 class FeatureFlag[S = Pending](Model[S, "FeatureFlag[Fetched]"]): 

212 """Table model with a logical Boolean SQLite encoding.""" 

213 

214 id: FeatureFlag.GenCol[int] = Integer(primary_key=True, default=MISSING) 

215 enabled: FeatureFlag.Col[bool] = Boolean(nullable=False) 

216 

217 with TemporaryDirectory() as directory: 

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

219 database = await Database.initialize( 

220 logger=NULL_LOGGER, 

221 database=database_path, 

222 models=[FeatureFlag], 

223 ) 

224 await database.close() 

225 

226 connection = connect(database_path) 

227 try: 

228 _ = connection.execute( 

229 'INSERT INTO "feature_flag" ("id", "enabled") VALUES (1, 2)', 

230 ) 

231 connection.commit() 

232 finally: 

233 connection.close() 

234 

235 database = await Database.initialize( 

236 logger=NULL_LOGGER, 

237 database=database_path, 

238 models=[FeatureFlag], 

239 ) 

240 try: 

241 async with database.transaction() as tx: 

242 with assert_raises(ModelValidationError): 

243 _ = await tx.fetch_all(select(FeatureFlag).all()) 

244 finally: 

245 await database.close() 

246 

247 

248@test(mark="fast") 

249def select_compilation_requires_explicit_all_or_where() -> None: 

250 """Select queries must choose filtered or unfiltered operation to compile.""" 

251 

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

253 """Table model used by select compilation checks.""" 

254 

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

256 

257 with assert_raises(QueryCompilationError): 

258 _ = compile_select_sql(select(User)) 

259 

260 

261@test(mark="fast") 

262def select_compilation_parameterizes_filters_limits_and_offsets() -> None: 

263 """Compiled select SQL is quoted and parameterized in observable order.""" 

264 

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

266 """Table model used by select compilation checks.""" 

267 

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

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

270 

271 query = ( 

272 select(User.email, User.status) 

273 .where(User.status.in_("active", "disabled")) 

274 .order_by(User.email.asc()) 

275 .offset(2) 

276 ) 

277 

278 sql, params = compile_select_sql(query) 

279 

280 expected_sql = ( 

281 'SELECT "email", "status" FROM "user" ' 

282 'WHERE ("status" IN (?, ?)) ORDER BY "email" ASC LIMIT -1 OFFSET ?' 

283 ) 

284 assert_eq(sql, expected_sql) 

285 assert_eq(params, ("active", "disabled", 2))