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

1"""Predicate intent, immutable builders, and mutation 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_is, assert_ne, assert_raises, test 

12 

13from snekql import ( 

14 MISSING, 

15 Database, 

16 Fetched, 

17 Integer, 

18 Model, 

19 Pending, 

20 QueryCompilationError, 

21 QueryConstructionError, 

22 Text, 

23 delete, 

24 insert, 

25 select, 

26 update, 

27) 

28from snekql.query import compile_select_sql, compile_write_sql 

29from tests.helpers import NULL_LOGGER 

30 

31 

32def _fetch_rows(database_path: Path, sql: str) -> list[tuple[object, ...]]: 

33 connection = connect(database_path) 

34 try: 

35 cursor = connection.execute(sql) 

36 return [tuple(row) for row in cursor.fetchall()] 

37 finally: 

38 connection.close() 

39 

40 

41@test(mark="fast") 

42def predicates_reject_ambiguous_or_invalid_intent() -> None: 

43 """Predicate helpers reject null ambiguity, empty IN, and non-text LIKE.""" 

44 

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

46 """Table model used by predicate construction checks.""" 

47 

48 id: User.Col[int] = Integer(nullable=False) 

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

50 

51 email_eq = cast("Callable[[object], object]", User.email.eq) 

52 email_ne = cast("Callable[[object], object]", User.email.ne) 

53 email_not_in = cast("Callable[..., object]", User.email.not_in) 

54 

55 with assert_raises(QueryConstructionError): 

56 _ = email_eq(None) 

57 

58 with assert_raises(QueryConstructionError): 

59 _ = email_ne(None) 

60 

61 with assert_raises(QueryConstructionError): 

62 _ = User.email.in_() 

63 

64 with assert_raises(QueryConstructionError): 

65 _ = email_not_in("a@example.com", None) 

66 

67 with assert_raises(QueryConstructionError): 

68 _ = User.id.like("1%") 

69 

70 

71@test(mark="fast") 

72def select_builders_are_immutable_and_require_filter_intent() -> None: 

73 """Select chain methods return new queries except repeated all() no-ops.""" 

74 

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

76 """Table model used by immutable select checks.""" 

77 

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

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

80 

81 base_query = select(User.email) 

82 filtered_query = base_query.where(User.status.eq("active")) 

83 ordered_query = filtered_query.order_by(User.email.asc()) 

84 paged_query = ordered_query.limit(10).limit(2).offset(1) 

85 all_query = base_query.all() 

86 

87 assert_ne(filtered_query, base_query) 

88 assert_ne(ordered_query, filtered_query) 

89 assert_is(all_query.all(), all_query) 

90 

91 with assert_raises(QueryConstructionError): 

92 _ = base_query.where() 

93 

94 with assert_raises(QueryConstructionError): 

95 _ = base_query.order_by() 

96 

97 with assert_raises(QueryConstructionError): 

98 _ = all_query.where(User.status.eq("active")) 

99 

100 with assert_raises(QueryConstructionError): 

101 _ = filtered_query.all() 

102 

103 with assert_raises(QueryCompilationError): 

104 _ = compile_select_sql(base_query) 

105 

106 with assert_raises(QueryCompilationError): 

107 _ = compile_write_sql(all_query) 

108 

109 sql, params = compile_select_sql(paged_query) 

110 

111 expected_sql = ( 

112 'SELECT "email" FROM "user" WHERE ("status" = ?) ' 

113 'ORDER BY "email" ASC LIMIT ? OFFSET ?' 

114 ) 

115 assert_eq(sql, expected_sql) 

116 assert_eq(params, ("active", 2, 1)) 

117 

118 

119@test(mark="fast") 

120def update_compilation_requires_set_and_filter_intent() -> None: 

121 """Update SQL is parameterized and refuses implicit full-table updates.""" 

122 

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

124 """Table model used by update compilation checks.""" 

125 

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

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

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

129 

130 base_query = update(User) 

131 set_query = base_query.set(User.status.to("disabled")) 

132 filtered_query = set_query.where(User.email.eq("old@example.com")) 

133 

134 assert_ne(set_query, base_query) 

135 assert_ne(filtered_query, set_query) 

136 

137 with assert_raises(QueryConstructionError): 

138 _ = base_query.set() 

139 

140 with assert_raises(QueryConstructionError): 

141 _ = set_query.all().where(User.email.eq("old@example.com")) 

142 

143 with assert_raises(QueryConstructionError): 

144 _ = filtered_query.all() 

145 

146 with assert_raises(QueryConstructionError): 

147 _ = update(User).set(User.id.to(2)) 

148 

149 all_query = set_query.all() 

150 assert_is(all_query.all(), all_query) 

151 assert_eq( 

152 compile_write_sql(all_query), 

153 ('UPDATE "user" SET "status" = ?', ("disabled",)), 

154 ) 

155 

156 with assert_raises(QueryCompilationError): 

157 _ = compile_write_sql(base_query.all()) 

158 

159 with assert_raises(QueryCompilationError): 

160 _ = compile_write_sql(set_query) 

161 

162 sql, params = compile_write_sql(filtered_query) 

163 

164 assert_eq(sql, 'UPDATE "user" SET "status" = ? WHERE ("email" = ?)') 

165 assert_eq(params, ("disabled", "old@example.com")) 

166 

167 

168@test(mark="fast") 

169def delete_compilation_requires_filter_intent() -> None: 

170 """Delete SQL requires explicit where() or all() before compilation.""" 

171 

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

173 """Table model used by delete compilation checks.""" 

174 

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

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

177 

178 base_query = delete(User) 

179 filtered_query = base_query.where( 

180 User.status.eq("disabled"), 

181 User.email.like("%@example.com"), 

182 ) 

183 all_query = base_query.all() 

184 

185 assert_ne(filtered_query, base_query) 

186 assert_is(all_query.all(), all_query) 

187 

188 with assert_raises(QueryConstructionError): 

189 _ = base_query.where() 

190 

191 with assert_raises(QueryConstructionError): 

192 _ = all_query.where(User.status.eq("disabled")) 

193 

194 with assert_raises(QueryConstructionError): 

195 _ = filtered_query.all() 

196 

197 with assert_raises(QueryCompilationError): 

198 _ = compile_write_sql(base_query) 

199 

200 sql, params = compile_write_sql(filtered_query) 

201 

202 expected_sql = 'DELETE FROM "user" WHERE ("status" = ?) AND ("email" LIKE ?)' 

203 assert_eq(sql, expected_sql) 

204 assert_eq(params, ("disabled", "%@example.com")) 

205 assert_eq(compile_write_sql(all_query), ('DELETE FROM "user"', ())) 

206 

207 

208@test(mark="medium") 

209async def update_and_delete_execute_against_sqlite() -> None: 

210 """Mutation queries persist changes through the async transaction runtime.""" 

211 

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

213 """Table model used by mutation execution checks.""" 

214 

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

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

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

218 

219 with TemporaryDirectory() as directory: 

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

221 database = await Database.initialize( 

222 logger=NULL_LOGGER, database=database_path, models=[User] 

223 ) 

224 try: 

225 async with database.transaction() as tx: 

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

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

228 await tx.execute( 

229 update(User) 

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

231 .where(User.email.eq("a@example.com")), 

232 ) 

233 await tx.execute( 

234 delete(User).where(User.status.eq("active")), 

235 ) 

236 finally: 

237 await database.close() 

238 

239 rows = _fetch_rows( 

240 database_path, 

241 'SELECT "email", "status" FROM "user" ORDER BY "email"', 

242 ) 

243 

244 assert_eq(rows, [("a@example.com", "disabled")])