Coverage for tests/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 Integer, 

17 Model, 

18 Pending, 

19 QueryCompilationError, 

20 QueryConstructionError, 

21 Text, 

22 delete, 

23 insert, 

24 select, 

25 update, 

26) 

27from snekql.query import compile_select_sql, compile_write_sql 

28from tests.logging_helpers import NULL_LOGGER 

29 

30 

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

32 connection = connect(database_path) 

33 try: 

34 cursor = connection.execute(sql) 

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

36 finally: 

37 connection.close() 

38 

39 

40@test(mark="fast") 

41def predicates_reject_ambiguous_or_invalid_intent() -> None: 

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

43 

44 class User[S = Pending](Model[S, "User[object]"]): 

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

46 

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

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

49 

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

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

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

53 

54 with assert_raises(QueryConstructionError): 

55 _ = email_eq(None) 

56 

57 with assert_raises(QueryConstructionError): 

58 _ = email_ne(None) 

59 

60 with assert_raises(QueryConstructionError): 

61 _ = User.email.in_() 

62 

63 with assert_raises(QueryConstructionError): 

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

65 

66 with assert_raises(QueryConstructionError): 

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

68 

69 

70@test(mark="fast") 

71def select_builders_are_immutable_and_require_filter_intent() -> None: 

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

73 

74 class User[S = Pending](Model[S, "User[object]"]): 

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

76 

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

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

79 

80 base_query = select(User.email) 

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

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

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

84 all_query = base_query.all() 

85 

86 assert_ne(filtered_query, base_query) 

87 assert_ne(ordered_query, filtered_query) 

88 assert_is(all_query.all(), all_query) 

89 

90 with assert_raises(QueryConstructionError): 

91 _ = base_query.where() 

92 

93 with assert_raises(QueryConstructionError): 

94 _ = base_query.order_by() 

95 

96 with assert_raises(QueryConstructionError): 

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

98 

99 with assert_raises(QueryConstructionError): 

100 _ = filtered_query.all() 

101 

102 with assert_raises(QueryCompilationError): 

103 _ = compile_select_sql(base_query) 

104 

105 with assert_raises(QueryCompilationError): 

106 _ = compile_write_sql(all_query) 

107 

108 sql, params = compile_select_sql(paged_query) 

109 

110 expected_sql = ( 

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

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

113 ) 

114 assert_eq(sql, expected_sql) 

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

116 

117 

118@test(mark="fast") 

119def update_compilation_requires_set_and_filter_intent() -> None: 

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

121 

122 class User[S = Pending](Model[S, "User[object]"]): 

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

124 

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

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

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

128 

129 base_query = update(User) 

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

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

132 

133 assert_ne(set_query, base_query) 

134 assert_ne(filtered_query, set_query) 

135 

136 with assert_raises(QueryConstructionError): 

137 _ = base_query.set() 

138 

139 with assert_raises(QueryConstructionError): 

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

141 

142 with assert_raises(QueryConstructionError): 

143 _ = filtered_query.all() 

144 

145 with assert_raises(QueryConstructionError): 

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

147 

148 all_query = set_query.all() 

149 assert_is(all_query.all(), all_query) 

150 assert_eq( 

151 compile_write_sql(all_query), 

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

153 ) 

154 

155 with assert_raises(QueryCompilationError): 

156 _ = compile_write_sql(base_query.all()) 

157 

158 with assert_raises(QueryCompilationError): 

159 _ = compile_write_sql(set_query) 

160 

161 sql, params = compile_write_sql(filtered_query) 

162 

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

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

165 

166 

167@test(mark="fast") 

168def delete_compilation_requires_filter_intent() -> None: 

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

170 

171 class User[S = Pending](Model[S, "User[object]"]): 

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

173 

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

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

176 

177 base_query = delete(User) 

178 filtered_query = base_query.where( 

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

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

181 ) 

182 all_query = base_query.all() 

183 

184 assert_ne(filtered_query, base_query) 

185 assert_is(all_query.all(), all_query) 

186 

187 with assert_raises(QueryConstructionError): 

188 _ = base_query.where() 

189 

190 with assert_raises(QueryConstructionError): 

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

192 

193 with assert_raises(QueryConstructionError): 

194 _ = filtered_query.all() 

195 

196 with assert_raises(QueryCompilationError): 

197 _ = compile_write_sql(base_query) 

198 

199 sql, params = compile_write_sql(filtered_query) 

200 

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

202 assert_eq(sql, expected_sql) 

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

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

205 

206 

207@test(mark="medium") 

208async def update_and_delete_execute_against_sqlite() -> None: 

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

210 

211 class User[S = Pending](Model[S, "User[object]"]): 

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

213 

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

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

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

217 

218 with TemporaryDirectory() as directory: 

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

220 database = await Database.initialize( 

221 NULL_LOGGER, database=database_path, models=[User] 

222 ) 

223 try: 

224 async with database.transaction() as transaction: 

225 await transaction.execute(insert(User(email="a@example.com"))) 

226 await transaction.execute(insert(User(email="b@example.com"))) 

227 await transaction.execute( 

228 update(User) 

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

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

231 ) 

232 await transaction.execute( 

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

234 ) 

235 finally: 

236 await database.close() 

237 

238 rows = _fetch_rows( 

239 database_path, 

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

241 ) 

242 

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