Coverage for tests/test_structured_logging.py: 98%

1"""Structured Query Runtime logging tests.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from sqlite3 import connect 

7from tempfile import TemporaryDirectory 

8from typing import Any, cast 

9 

10from snektest import assert_eq, assert_raises, assert_true, test 

11 

12from snekql import ( 

13 MISSING, 

14 Database, 

15 ExecutionError, 

16 Fetched, 

17 Integer, 

18 Model, 

19 Pending, 

20 PoolTimeoutError, 

21 Text, 

22 insert, 

23 select, 

24) 

25 

26 

27class _RecordingStructuredLogger: 

28 """Structured logger fake that records event calls for assertions.""" 

29 

30 def __init__(self) -> None: 

31 self.events: list[tuple[str, str, dict[str, object]]] = [] 

32 

33 def debug(self, event: str, **fields: object) -> None: 

34 self.events.append(("debug", event, fields)) 

35 

36 def info(self, event: str, **fields: object) -> None: 

37 self.events.append(("info", event, fields)) 

38 

39 def warning(self, event: str, **fields: object) -> None: 

40 self.events.append(("warning", event, fields)) 

41 

42 def error(self, event: str, **fields: object) -> None: 

43 self.events.append(("error", event, fields)) 

44 

45 def find(self, event: str) -> dict[str, object]: 

46 """Return the first recorded event fields for the named event.""" 

47 

48 for _level, recorded_event, fields in self.events: 

49 if recorded_event == event: 

50 return fields 

51 msg = f"event not recorded: {event}" 

52 raise AssertionError(msg) 

53 

54 def has(self, level: str, event: str) -> bool: 

55 """Return whether a level/event pair was recorded.""" 

56 

57 return any( 

58 recorded_level == level and recorded_event == event 

59 for recorded_level, recorded_event, _fields in self.events 

60 ) 

61 

62 

63def _execute_sql(database_path: Path, sql: str) -> None: 

64 connection = connect(database_path) 

65 try: 

66 _ = connection.execute(sql) 

67 connection.commit() 

68 finally: 

69 connection.close() 

70 

71 

72@test() 

73def database_initialization_requires_a_logger() -> None: 

74 """The public initialization path requires explicit structured logging.""" 

75 

76 initialize = cast("Any", Database.initialize) 

77 

78 with assert_raises(TypeError): 

79 _ = initialize(database=":memory:") 

80 

81 with assert_raises(TypeError): 

82 _ = initialize(_RecordingStructuredLogger(), database=":memory:") 

83 

84 

85@test(mark="medium") 

86async def database_initialization_emits_structured_events() -> None: 

87 """Database initialization logs backend and schema startup context.""" 

88 

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

90 """Table model used to observe schema startup logging.""" 

91 

92 id: User.GenCol[int] = Integer( 

93 primary_key=True, 

94 auto_increment=True, 

95 default=MISSING, 

96 ) 

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

98 

99 logger = _RecordingStructuredLogger() 

100 with TemporaryDirectory() as directory: 

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

102 database = await Database.initialize( 

103 logger=logger, 

104 database=database_path, 

105 models=[User], 

106 ) 

107 await database.close() 

108 

109 started = logger.find("database initialization started") 

110 completed = logger.find("database initialization completed") 

111 created = logger.find("schema table created") 

112 

113 assert_eq(started["backend"], "sqlite") 

114 assert_eq(started["model_count"], 1) 

115 assert_eq(started["table_names"], ("user",)) 

116 assert_eq(completed["backend"], "sqlite") 

117 assert_eq(created["table_name"], "user") 

118 assert_true(logger.has("info", "database initialization started")) 

119 assert_true(logger.has("info", "database initialization completed")) 

120 

121 

122@test(mark="medium") 

123async def warn_schema_policy_uses_injected_structured_logger() -> None: 

124 """Warn schema verification reports drift through the supplied logger.""" 

125 

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

127 """Table model used for warn policy drift logging.""" 

128 

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

130 

131 logger = _RecordingStructuredLogger() 

132 with TemporaryDirectory() as directory: 

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

134 _execute_sql(database_path, 'CREATE TABLE "user" ("email" TEXT NOT NULL)') 

135 

136 database = await Database.initialize( 

137 logger=logger, 

138 database=database_path, 

139 models=[User], 

140 schema_policy="warn", 

141 ) 

142 await database.close() 

143 

144 drift = logger.find("schema drift detected") 

145 assert_eq(drift["table_name"], "user") 

146 assert_true(logger.has("warning", "schema drift detected")) 

147 

148 

149@test(mark="medium") 

150async def transaction_execution_emits_query_context() -> None: 

151 """Transaction logging includes SQL and params without redaction.""" 

152 

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

154 """Table model used to observe query execution logging.""" 

155 

156 id: User.GenCol[int] = Integer( 

157 primary_key=True, 

158 auto_increment=True, 

159 default=MISSING, 

160 ) 

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

162 

163 logger = _RecordingStructuredLogger() 

164 database = await Database.initialize( 

165 logger=logger, 

166 database=":memory:", 

167 models=[User], 

168 ) 

169 try: 

170 async with database.transaction() as tx: 

171 await tx.execute(insert(User(email="secret@example.com"))) 

172 row = await tx.fetch_one( 

173 select(User.email).where(User.email.eq("secret@example.com")) 

174 ) 

175 finally: 

176 await database.close() 

177 

178 assert_eq(row, "secret@example.com") 

179 write = next( 

180 fields 

181 for _level, event, fields in logger.events 

182 if event == "query executed" and fields["operation"] == "write" 

183 ) 

184 select_event = next( 

185 fields 

186 for _level, event, fields in logger.events 

187 if event == "query executed" and fields["operation"] == "fetch_one" 

188 ) 

189 

190 assert_eq(write["params"], ("secret@example.com",)) 

191 assert_eq(select_event["params"], ("secret@example.com",)) 

192 assert_true(logger.has("debug", "transaction begin")) 

193 assert_true(logger.has("debug", "transaction commit")) 

194 

195 

196@test(mark="medium") 

197async def query_failure_emits_structured_error_context() -> None: 

198 """Execution failures log SQL and params before raising ExecutionError.""" 

199 

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

201 """Table model with a unique field used to force a write failure.""" 

202 

203 email: User.Col[str] = Text(nullable=False, unique=True) 

204 

205 logger = _RecordingStructuredLogger() 

206 database = await Database.initialize( 

207 logger=logger, 

208 database=":memory:", 

209 models=[User], 

210 ) 

211 try: 

212 async with database.transaction() as tx: 

213 await tx.execute(insert(User(email="duplicate@example.com"))) 

214 with assert_raises(ExecutionError): 

215 await tx.execute(insert(User(email="duplicate@example.com"))) 

216 finally: 

217 await database.close() 

218 

219 failure = logger.find("query failed") 

220 assert_eq(failure["operation"], "write") 

221 assert_eq(failure["params"], ("duplicate@example.com",)) 

222 assert_true(logger.has("error", "query failed")) 

223 

224 

225@test(mark="medium") 

226async def pool_timeout_emits_structured_warning() -> None: 

227 """Pool acquisition timeouts are logged while preserving the public error.""" 

228 

229 logger = _RecordingStructuredLogger() 

230 database = await Database.initialize( 

231 logger=logger, 

232 database=":memory:", 

233 acquire_timeout=0.0, 

234 pool_size=1, 

235 ) 

236 try: 

237 async with database.transaction(): 

238 with assert_raises(PoolTimeoutError): 

239 async with database.transaction(timeout=0.0): 

240 pass 

241 finally: 

242 await database.close() 

243 

244 timeout = logger.find("connection acquisition timed out") 

245 assert_eq(timeout["backend"], "sqlite") 

246 assert_eq(timeout["timeout"], 0.0) 

247 assert_true(logger.has("warning", "connection acquisition timed out"))