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

16 Integer, 

17 Model, 

18 Pending, 

19 Text, 

20 insert, 

21) 

22from snekql.query import compile_write_sql 

23from tests.helpers import NULL_LOGGER 

24 

25 

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

27 connection = connect(database_path) 

28 try: 

29 cursor = connection.execute(sql) 

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

31 finally: 

32 connection.close() 

33 

34 

35@test(mark="fast") 

36def insert_compilation_omits_missing_and_quotes_identifiers() -> None: 

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

38 

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

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

41 

42 __tablename__ = "select" 

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

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

45 

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

47 

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

49 assert_eq(params, ("x",)) 

50 

51 

52@test(mark="fast") 

53def insert_compilation_uses_default_values_when_every_field_is_missing() -> None: 

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

55 

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

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

58 

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

60 

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

62 

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

64 assert_eq(params, ()) 

65 

66 

67@test(mark="medium") 

68async def insert_execution_includes_defaults_and_returns_none() -> None: 

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

70 

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

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

73 

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

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

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

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

78 

79 with TemporaryDirectory() as directory: 

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

81 database = await Database.initialize( 

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

83 ) 

84 try: 

85 async with database.transaction() as tx: 

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

87 finally: 

88 await database.close() 

89 

90 rows = _fetch_rows( 

91 database_path, 

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

93 ) 

94 

95 assert_is_none(result) 

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

97 

98 

99@test(mark="medium") 

100async def execution_errors_preserve_insert_sql_and_params() -> None: 

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

102 

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

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

105 

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

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

108 

109 with TemporaryDirectory() as directory: 

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

111 database = await Database.initialize( 

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

113 ) 

114 try: 

115 async with database.transaction() as tx: 

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

117 with assert_raises(ExecutionError) as caught_error: 

118 await tx.execute( 

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

120 ) 

121 finally: 

122 await database.close() 

123 

124 error = caught_error.exception 

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

126 assert_in("?", error.sql) 

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