Coverage for tests/test_insert_execution.py: 100%

1"""Insert query construction, compilation, and execution tests.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from sqlite3 import connect 

7from tempfile import TemporaryDirectory 

8 

9from snektest import assert_eq, assert_in, assert_is_none, assert_raises, test 

10 

11from snekql import ( 

12 MISSING, 

13 Database, 

14 ExecutionError, 

15 Integer, 

16 Model, 

17 Pending, 

18 Text, 

19 insert, 

20) 

21from snekql.query import compile_write_sql 

22from tests.logging_helpers import NULL_LOGGER 

23 

24 

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

26 connection = connect(database_path) 

27 try: 

28 cursor = connection.execute(sql) 

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

30 finally: 

31 connection.close() 

32 

33 

34@test(mark="fast") 

35def insert_compilation_omits_missing_and_quotes_identifiers() -> None: 

36 """Compiled insert SQL targets the model table and omits MISSING fields.""" 

37 

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

39 """Table model with identifiers that must be quoted.""" 

40 

41 __tablename__ = "select" 

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

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

44 

45 sql, params = compile_write_sql(insert(Order(where="x"))) 

46 

47 assert_eq(sql, 'INSERT INTO "select" ("where") VALUES (?)') 

48 assert_eq(params, ("x",)) 

49 

50 

51@test(mark="fast") 

52def insert_compilation_uses_default_values_when_every_field_is_missing() -> None: 

53 """An all-generated model compiles to SQLite DEFAULT VALUES.""" 

54 

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

56 """Table model with no explicit insertable values.""" 

57 

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

59 

60 sql, params = compile_write_sql(insert(AuditLog())) 

61 

62 assert_eq(sql, 'INSERT INTO "audit_log" DEFAULT VALUES') 

63 assert_eq(params, ()) 

64 

65 

66@test(mark="medium") 

67async def insert_execution_includes_defaults_and_returns_none() -> None: 

68 """Executing an insert persists Python defaults and default-factory values.""" 

69 

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

71 """Table model with explicit, default, and generated fields.""" 

72 

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

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

75 label: User.Col[str] = Text(nullable=False, default_factory=lambda: "fresh") 

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

77 

78 with TemporaryDirectory() as directory: 

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

80 database = await Database.initialize( 

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

82 ) 

83 try: 

84 async with database.transaction() as transaction: 

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

86 finally: 

87 await database.close() 

88 

89 rows = _fetch_rows( 

90 database_path, 

91 'SELECT "email", "label", "status" FROM "user"', 

92 ) 

93 

94 assert_is_none(result) 

95 assert_eq(rows, [("a@example.com", "fresh", "active")]) 

96 

97 

98@test(mark="medium") 

99async def execution_errors_preserve_insert_sql_and_params() -> None: 

100 """SQLite insert failures are wrapped with SQL and parameter context.""" 

101 

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

103 """Table model used to trigger a duplicate primary key failure.""" 

104 

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

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

107 

108 with TemporaryDirectory() as directory: 

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

110 database = await Database.initialize( 

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

112 ) 

113 try: 

114 async with database.transaction() as transaction: 

115 await transaction.execute(insert(User(id=1, email="first@example.com"))) 

116 with assert_raises(ExecutionError) as caught_error: 

117 await transaction.execute( 

118 insert(User(id=1, email="second@example.com")), 

119 ) 

120 finally: 

121 await database.close() 

122 

123 error = caught_error.exception 

124 assert_in('INSERT INTO "user"', error.sql) 

125 assert_in("?", error.sql) 

126 assert_eq(error.params, (1, "second@example.com"))