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

1"""Delete query construction, compilation, and execution acceptance tests.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import cast 

7 

8from snektest import ( 

9 assert_eq, 

10 assert_is, 

11 assert_is_none, 

12 assert_ne, 

13 assert_raises, 

14 test, 

15) 

16 

17from snekql import ( 

18 MISSING, 

19 Database, 

20 Fetched, 

21 Integer, 

22 Model, 

23 Pending, 

24 QueryCompilationError, 

25 QueryConstructionError, 

26 Text, 

27 delete, 

28 insert, 

29 select, 

30) 

31from snekql.query import compile_write_sql 

32from tests.helpers import NULL_LOGGER 

33 

34 

35@test(mark="fast") 

36def delete_compilation_quotes_identifiers_and_parameterizes_filters() -> None: 

37 """Delete compiles quoted table and column names with bound parameters.""" 

38 

39 class Order[S = Pending](Model[S, "Order[Fetched]"]): 

40 """Table model with identifiers requiring SQLite quoting.""" 

41 

42 __tablename__ = "select" 

43 status: Order.Col[str] = Text(nullable=False) 

44 where: Order.Col[str] = Text(nullable=False) 

45 

46 query = delete(Order).where( 

47 Order.where.ne("old"), 

48 Order.status.eq("done"), 

49 ) 

50 

51 sql, params = compile_write_sql(query) 

52 

53 expected_sql = 'DELETE FROM "select" WHERE ("where" != ?) AND ("status" = ?)' 

54 assert_eq(sql, expected_sql) 

55 assert_eq(params, ("old", "done")) 

56 

57 

58@test(mark="fast") 

59def delete_predicates_must_belong_to_target_model() -> None: 

60 """Delete where() rejects predicates built from another table model.""" 

61 

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

63 """Target table model for ownership checks.""" 

64 

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

66 

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

68 """Unrelated table model for ownership checks.""" 

69 

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

71 

72 where = cast("Callable[..., object]", delete(User).where) 

73 

74 with assert_raises(QueryConstructionError): 

75 _ = where(AuditLog.message.eq("wrong table")) 

76 

77 

78@test(mark="fast") 

79def delete_requires_exactly_one_filter_intent() -> None: 

80 """Delete requires exactly one of where() or all() before compilation.""" 

81 

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

83 """Table model used by explicit delete intent checks.""" 

84 

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

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

87 

88 base_query = delete(User) 

89 filtered_query = base_query.where(User.status.eq("disabled")) 

90 all_query = base_query.all() 

91 

92 assert_ne(filtered_query, base_query) 

93 assert_is(all_query.all(), all_query) 

94 

95 with assert_raises(QueryConstructionError): 

96 _ = base_query.where() 

97 

98 with assert_raises(QueryConstructionError): 

99 _ = filtered_query.all() 

100 

101 with assert_raises(QueryConstructionError): 

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

103 

104 with assert_raises(QueryCompilationError): 

105 _ = compile_write_sql(base_query) 

106 

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

108 

109 

110@test(mark="medium") 

111async def delete_execute_returns_none_for_filtered_and_explicit_all_forms() -> None: 

112 """tx.execute(delete(...)) returns None for filtered and full-table deletes.""" 

113 

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

115 """Table model deleted through the async runtime.""" 

116 

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

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 

131 filtered_result = await tx.execute( 

132 delete(User).where(User.status.eq("disabled")), 

133 ) 

134 remaining_after_filtered = await tx.fetch_all( 

135 select(User.email).all().order_by(User.email.asc()), 

136 ) 

137 all_result = await tx.execute(delete(User).all()) 

138 remaining_after_all = await tx.fetch_all(select(User.email).all()) 

139 finally: 

140 await database.close() 

141 

142 assert_is_none(filtered_result) 

143 assert_eq(remaining_after_filtered, ["a@example.com"]) 

144 assert_is_none(all_result) 

145 assert_eq(remaining_after_all, [])