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

21 Model, 

22 Pending, 

23 QueryCompilationError, 

24 QueryConstructionError, 

25 Text, 

26 delete, 

27 insert, 

28 select, 

29) 

30from snekql.query import compile_write_sql 

31from tests.logging_helpers import NULL_LOGGER 

32 

33 

34@test(mark="fast") 

35def delete_compilation_quotes_identifiers_and_parameterizes_filters() -> None: 

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

37 

38 class Order[S = Pending](Model[S, "Order[object]"]): 

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

40 

41 __tablename__ = "select" 

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

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

44 

45 query = delete(Order).where( 

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

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

48 ) 

49 

50 sql, params = compile_write_sql(query) 

51 

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

53 assert_eq(sql, expected_sql) 

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

55 

56 

57@test(mark="fast") 

58def delete_predicates_must_belong_to_target_model() -> None: 

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

60 

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

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

63 

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

65 

66 class AuditLog[S = Pending](Model[S, "AuditLog[object]"]): 

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

68 

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

70 

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

72 

73 with assert_raises(QueryConstructionError): 

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

75 

76 

77@test(mark="fast") 

78def delete_requires_exactly_one_filter_intent() -> None: 

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

80 

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

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

83 

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

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

86 

87 base_query = delete(User) 

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

89 all_query = base_query.all() 

90 

91 assert_ne(filtered_query, base_query) 

92 assert_is(all_query.all(), all_query) 

93 

94 with assert_raises(QueryConstructionError): 

95 _ = base_query.where() 

96 

97 with assert_raises(QueryConstructionError): 

98 _ = filtered_query.all() 

99 

100 with assert_raises(QueryConstructionError): 

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

102 

103 with assert_raises(QueryCompilationError): 

104 _ = compile_write_sql(base_query) 

105 

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

107 

108 

109@test(mark="medium") 

110async def delete_execute_returns_none_for_filtered_and_explicit_all_forms() -> None: 

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

112 

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

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

115 

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

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

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

119 

120 database = await Database.initialize( 

121 NULL_LOGGER, database=":memory:", models=[User] 

122 ) 

123 try: 

124 async with database.transaction() as transaction: 

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

126 await transaction.execute( 

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

128 ) 

129 

130 filtered_result = await transaction.execute( 

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

132 ) 

133 remaining_after_filtered = await transaction.fetch_all( 

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

135 ) 

136 all_result = await transaction.execute(delete(User).all()) 

137 remaining_after_all = await transaction.fetch_all(select(User.email).all()) 

138 finally: 

139 await database.close() 

140 

141 assert_is_none(filtered_result) 

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

143 assert_is_none(all_result) 

144 assert_eq(remaining_after_all, [])