test_execution_log.py
13.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
"""U-027 tests: c_execution_log_get + c_ws_log_tail's data layer +
p_log_retention_cleanup (REQ-13 happy/boundary/failure). Real sqlite + real
tempfile-backed log files, no mocking needed.
Run: python3 -m unittest src.test_execution_log -v
"""
import json
import os
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from src import execution_log, schema
class ExecutionLogTestBase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.conn.execute(
"INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
"VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
)
self.conn.execute(
"INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
"VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
)
self.conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
"VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
)
self.conn.execute(
"INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
"VALUES ('skill-1', 'coder', 'skills/skill-1/', 'sha', 'new', 'active')"
)
self.conn.execute(
"INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
"input_schema_ref, output_schema_ref, input_payload, status) "
"VALUES ('node-1', 'run-1', 'n1', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
def _log_path(self):
return os.path.join(self.tmp.name, "node-1.jsonl")
def _insert_log_file(self, status="active"):
self.conn.execute(
"INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
"VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', ?)",
(self._log_path(), status),
)
self.conn.commit()
class AppendLogTest(ExecutionLogTestBase):
def test_appends_content_to_file_and_bumps_size(self):
self._insert_log_file()
execution_log.append_log(self.conn, "node-1", "line one\n")
execution_log.append_log(self.conn, "node-1", "line two\n")
with open(self._log_path()) as f:
self.assertEqual(f.read(), "line one\nline two\n")
size = self.conn.execute("SELECT size_bytes FROM execution_log_file WHERE log_file_id = 'log-1'").fetchone()[0]
self.assertEqual(size, len("line one\nline two\n"))
def test_writes_log_appended_execution_event(self):
self._insert_log_file()
execution_log.append_log(self.conn, "node-1", "hello\n")
row = self.conn.execute(
"SELECT event_type, task_id, payload FROM execution_event WHERE node_exec_id = 'node-1'"
).fetchone()
self.assertEqual(row[0], "log_appended")
self.assertEqual(row[1], "task-1")
self.assertEqual(json.loads(row[2]), {"appended_content": "hello\n"})
def test_creates_parent_directory_if_missing(self):
nested_path = os.path.join(self.tmp.name, "a", "b", "node-1.jsonl")
self.conn.execute(
"INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
"VALUES ('log-1', 'node-1', 'proj-1', ?, 'jsonl', 'active')",
(nested_path,),
)
self.conn.commit()
execution_log.append_log(self.conn, "node-1", "x")
self.assertTrue(os.path.isfile(nested_path))
def test_nonexistent_node_exec_raises(self):
with self.assertRaises(execution_log.LogNotFoundError):
execution_log.append_log(self.conn, "no-such-node", "x")
class GetLogTest(ExecutionLogTestBase):
def test_returns_full_content_and_size(self):
self._insert_log_file(status="closed")
with open(self._log_path(), "w") as f:
f.write("all the log content")
result = execution_log.get_log(self.conn, "node-1")
self.assertEqual(result["content"], "all the log content")
self.assertEqual(result["total_size_bytes"], len("all the log content"))
self.assertEqual(result["status"], "closed")
def test_pagination_offset_and_limit(self):
self._insert_log_file(status="closed")
with open(self._log_path(), "w") as f:
f.write("0123456789")
result = execution_log.get_log(self.conn, "node-1", offset=3, limit=4)
self.assertEqual(result["content"], "3456")
self.assertEqual(result["total_size_bytes"], 10)
def test_missing_physical_file_returns_empty_not_error(self):
self._insert_log_file(status="closed")
result = execution_log.get_log(self.conn, "node-1")
self.assertEqual(result["content"], "")
self.assertEqual(result["total_size_bytes"], 0)
def test_nonexistent_node_exec_raises_log_not_found(self):
with self.assertRaises(execution_log.LogNotFoundError):
execution_log.get_log(self.conn, "no-such-node")
def test_purged_log_raises(self):
self._insert_log_file(status="purged")
with self.assertRaises(execution_log.LogPurgedError):
execution_log.get_log(self.conn, "node-1")
class NodeExecStatusTest(ExecutionLogTestBase):
def test_returns_status(self):
self.assertEqual(execution_log.node_exec_status(self.conn, "node-1"), "running")
def test_nonexistent_raises(self):
with self.assertRaises(execution_log.NodeExecNotFoundError):
execution_log.node_exec_status(self.conn, "no-such-node")
def test_is_terminal_status(self):
for status in ("done", "blocked", "failed"):
self.assertTrue(execution_log.is_terminal_status(status))
for status in ("pending", "running", "reflecting", "retrying", "waiting_for_quota"):
self.assertFalse(execution_log.is_terminal_status(status))
class PollLogEventsTest(ExecutionLogTestBase):
def test_polls_only_new_log_appended_events_for_this_node(self):
self._insert_log_file()
after = execution_log.latest_log_event_rowid(self.conn, "node-1")
execution_log.append_log(self.conn, "node-1", "chunk-1")
events, new_after = execution_log.poll_log_events(self.conn, "node-1", after)
self.assertEqual(len(events), 1)
self.assertEqual(events[0]["appended_content"], "chunk-1")
self.assertGreater(new_after, after)
events_again, after2 = execution_log.poll_log_events(self.conn, "node-1", new_after)
self.assertEqual(events_again, [])
self.assertEqual(after2, new_after)
def test_other_node_events_excluded(self):
self.conn.execute(
"INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
"input_schema_ref, output_schema_ref, input_payload, status) "
"VALUES ('node-2', 'run-1', 'n2', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'running')"
)
self.conn.execute(
"INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
"VALUES ('log-2', 'node-2', 'proj-1', ?, 'jsonl', 'active')",
(os.path.join(self.tmp.name, "node-2.jsonl"),),
)
self._insert_log_file()
self.conn.commit()
execution_log.append_log(self.conn, "node-2", "not mine")
events, _ = execution_log.poll_log_events(self.conn, "node-1", 0)
self.assertEqual(events, [])
class LogRetentionCleanupTest(ExecutionLogTestBase):
def setUp(self):
super().setUp()
self.conn.execute(
"INSERT INTO log_retention_config (config_id, retention_days, updated_by) "
"VALUES ('default', 1, 'test')"
)
self.conn.commit()
def _insert_closed_log(self, log_file_id, created_at, size_bytes=100, file_content="x"):
node_exec_id = log_file_id + "-node"
self.conn.execute(
"INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
"input_schema_ref, output_schema_ref, input_payload, status) "
"VALUES (?, 'run-1', ?, 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'done')",
(node_exec_id, log_file_id + "-n"),
)
path = os.path.join(self.tmp.name, log_file_id + ".jsonl")
with open(path, "w") as f:
f.write(file_content)
self.conn.execute(
"INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, "
"status, size_bytes, created_at) VALUES (?, ?, 'proj-1', ?, 'jsonl', 'closed', ?, ?)",
(log_file_id, node_exec_id, path, size_bytes, created_at),
)
self.conn.commit()
return path
def test_expired_closed_log_purged_and_file_deleted(self):
old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
path = self._insert_closed_log("log-old", old_ts)
result = execution_log.run_log_retention_cleanup(self.conn)
self.assertEqual(result["purged"], ["log-old"])
status, purged_at = self.conn.execute(
"SELECT status, purged_at FROM execution_log_file WHERE log_file_id = 'log-old'"
).fetchone()
self.assertEqual(status, "purged")
self.assertIsNotNone(purged_at)
self.assertFalse(os.path.isfile(path))
def test_not_yet_expired_log_untouched(self):
recent_ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
path = self._insert_closed_log("log-fresh", recent_ts)
result = execution_log.run_log_retention_cleanup(self.conn)
self.assertEqual(result["purged"], [])
status = self.conn.execute(
"SELECT status FROM execution_log_file WHERE log_file_id = 'log-fresh'"
).fetchone()[0]
self.assertEqual(status, "closed")
self.assertTrue(os.path.isfile(path))
def test_viewing_purged_log_raises_purged_not_error(self):
old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
self.conn.execute(
"INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
"input_schema_ref, output_schema_ref, input_payload, status) "
"VALUES ('node-old', 'run-1', 'n-old', 'coder', 'skill-1', 'in.json', 'out.json', '{}', 'done')"
)
path = os.path.join(self.tmp.name, "log-old2.jsonl")
with open(path, "w") as f:
f.write("x")
self.conn.execute(
"INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, "
"status, created_at) VALUES ('log-old2', 'node-old', 'proj-1', ?, 'jsonl', 'closed', ?)",
(path, old_ts),
)
self.conn.commit()
execution_log.run_log_retention_cleanup(self.conn)
with self.assertRaises(execution_log.LogPurgedError):
execution_log.get_log(self.conn, "node-old")
def test_max_total_size_pulls_in_oldest_non_expired_candidates(self):
self.conn.execute(
"UPDATE log_retention_config SET retention_days = 30, max_total_size_mb = 1 WHERE config_id = 'default'"
)
self.conn.commit()
now = datetime.now(timezone.utc)
one_mb = 1024 * 1024
# both within retention_days=30, but combined over the 1MB cap --
# the older of the two should get pulled in as a candidate too
self._insert_closed_log("log-a", (now - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S"), size_bytes=int(0.6 * one_mb))
self._insert_closed_log("log-b", (now - timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S"), size_bytes=int(0.6 * one_mb))
result = execution_log.run_log_retention_cleanup(self.conn, now=now)
self.assertEqual(result["purged"], ["log-a"])
status_b = self.conn.execute("SELECT status FROM execution_log_file WHERE log_file_id = 'log-b'").fetchone()[0]
self.assertEqual(status_b, "closed")
def test_delete_failure_recorded_and_others_still_purged(self):
old_ts = (datetime.now(timezone.utc) - timedelta(days=2)).strftime("%Y-%m-%d %H:%M:%S")
self._insert_closed_log("log-ok", old_ts)
bad_path = self._insert_closed_log("log-bad", old_ts)
os.remove(bad_path) # simulate "file missing/inaccessible" -- os.remove will raise
os.mkdir(bad_path) # replace it with a directory so os.remove() fails with OSError
self.addCleanup(lambda: os.path.isdir(bad_path) and os.rmdir(bad_path))
result = execution_log.run_log_retention_cleanup(self.conn)
self.assertEqual(result["purged"], ["log-ok"])
self.assertEqual(result["failed"], ["log-bad"])
status = self.conn.execute("SELECT status FROM execution_log_file WHERE log_file_id = 'log-bad'").fetchone()[0]
self.assertEqual(status, "closed")
def test_missing_config_raises(self):
self.conn.execute("DELETE FROM log_retention_config")
self.conn.commit()
with self.assertRaises(ValueError):
execution_log.run_log_retention_cleanup(self.conn)
if __name__ == "__main__":
unittest.main()