test_bug_record.py
9.69 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
"""U-015 tests: c_bug_record_create + the human_reported claim/complete
mechanics. All real (no mocking needed -- sqlite is fast/free/local).
Run: python3 -m unittest src.test_bug_record -v
"""
import json
import unittest
from src import bug_record, schema
class BugRecordTestBase(unittest.TestCase):
def setUp(self):
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', 'goal', 'goal', 'validating', 'validator')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
class CreateBugRecordTest(BugRecordTestBase):
def test_happy_path_creates_routed_validator_record(self):
result = bug_record.create_bug_record(
self.conn, task_id="task-1", description="wall collision misjudged",
root_cause_category="coder", root_cause_detail="SDD §3.2 collision check off-by-one",
suggested_route="fast_lane",
)
row = self.conn.execute(
"SELECT source, status, root_cause_category, root_cause_detail, suggested_route "
"FROM bug_record WHERE bug_id = ?",
(result["bug_id"],),
).fetchone()
self.assertEqual(row, ("validator", "routed", "coder", "SDD §3.2 collision check off-by-one", "fast_lane"))
def test_related_artifact_ids_stored_as_json(self):
result = bug_record.create_bug_record(
self.conn, task_id="task-1", description="x", root_cause_category="planning",
root_cause_detail="d", suggested_route="full_replan",
related_artifact_ids=["art-1", "art-2"],
)
row = self.conn.execute(
"SELECT related_artifact_ids FROM bug_record WHERE bug_id = ?", (result["bug_id"],)
).fetchone()
self.assertEqual(json.loads(row[0]), ["art-1", "art-2"])
def test_invalid_root_cause_category_rejected(self):
with self.assertRaises(bug_record.InvalidRootCauseCategoryError):
bug_record.create_bug_record(
self.conn, task_id="task-1", description="x", root_cause_category="not_a_real_category",
root_cause_detail="d", suggested_route="fast_lane",
)
def test_core_engine_accepted_since_u035(self):
# U-034 widened the schema's CHECK constraint; U-035 widens this
# app-layer whitelist too, once src/core_engine_classify.py exists
# as a real (if separate) mechanism for determining WHEN a change is
# core_engine. create_bug_record() itself still doesn't classify
# anything -- a caller (e.g. a self-target-aware Validator FAIL path)
# decides the category and passes it explicitly, same as any other
# category.
result = bug_record.create_bug_record(
self.conn, task_id="task-1", description="x", root_cause_category="core_engine",
root_cause_detail="d", suggested_route="fast_lane",
)
row = self.conn.execute(
"SELECT root_cause_category FROM bug_record WHERE bug_id = ?", (result["bug_id"],)
).fetchone()
self.assertEqual(row[0], "core_engine")
class CreateBugReportTest(BugRecordTestBase):
"""U-023: c_bug_report_create (03-C §3.1.1) -- console-submitted human
bug reports, distinct from create_bug_record's Validator-judged path."""
def test_happy_path_creates_open_human_record(self):
result = bug_record.create_bug_report(
self.conn, project_id="proj-1", description="snake doesn't wrap around edges",
)
row = self.conn.execute(
"SELECT source, status, task_id, description FROM bug_record WHERE bug_id = ?",
(result["bug_id"],),
).fetchone()
self.assertEqual(row, ("human", "open", None, "snake doesn't wrap around edges"))
self.assertEqual(result["status"], "open")
def test_related_task_id_is_optional_and_stored_when_given(self):
result = bug_record.create_bug_report(
self.conn, project_id="proj-1", description="x", related_task_id="task-1",
)
row = self.conn.execute(
"SELECT task_id FROM bug_record WHERE bug_id = ?", (result["bug_id"],)
).fetchone()
self.assertEqual(row[0], "task-1")
def test_repro_steps_optional_and_stored_when_given(self):
result = bug_record.create_bug_report(
self.conn, project_id="proj-1", description="x", repro_steps="1. start game 2. hit wall",
)
row = self.conn.execute(
"SELECT repro_steps FROM bug_record WHERE bug_id = ?", (result["bug_id"],)
).fetchone()
self.assertEqual(row[0], "1. start game 2. hit wall")
def test_empty_description_rejected(self):
with self.assertRaises(bug_record.InvalidParamError):
bug_record.create_bug_report(self.conn, project_id="proj-1", description="")
def test_nonexistent_project_id_rejected(self):
# a raw FK violation surfacing as an uncaught IntegrityError instead
# of this is exactly what a real backend_server.py smoke test caught
# (a bad/stale project_id would otherwise 500 the request)
with self.assertRaises(bug_record.InvalidParamError):
bug_record.create_bug_report(self.conn, project_id="no-such-project", description="x")
def test_nonexistent_related_task_id_rejected(self):
with self.assertRaises(bug_record.InvalidParamError):
bug_record.create_bug_report(
self.conn, project_id="proj-1", description="x", related_task_id="no-such-task"
)
def test_writes_bug_recorded_execution_event(self):
result = bug_record.create_bug_report(self.conn, project_id="proj-1", description="x")
row = self.conn.execute(
"SELECT project_id, event_type FROM execution_event WHERE payload LIKE ?",
('%' + result["bug_id"] + '%',),
).fetchone()
self.assertEqual(row, ("proj-1", "bug_recorded"))
def test_no_related_task_still_writes_event_with_null_task_id(self):
# this is exactly the case 03-C's own input_contract can't satisfy
# without project_id -- see create_bug_report's docstring
result = bug_record.create_bug_report(self.conn, project_id="proj-1", description="x")
row = self.conn.execute(
"SELECT task_id FROM execution_event WHERE payload LIKE ?",
('%' + result["bug_id"] + '%',),
).fetchone()
self.assertIsNone(row[0])
class RecordCoreEngineChangeTest(BugRecordTestBase):
"""U-035: the PASS-path audit trail, distinct from create_bug_record's
FAIL-path root-cause attribution."""
def test_creates_resolved_core_engine_record_with_no_route(self):
result = bug_record.record_core_engine_change(
self.conn, task_id="task-1", description="self-target change touched src/scheduler.py"
)
row = self.conn.execute(
"SELECT source, status, root_cause_category, suggested_route, description "
"FROM bug_record WHERE bug_id = ?",
(result["bug_id"],),
).fetchone()
self.assertEqual(
row,
("validator", "resolved", "core_engine", None, "self-target change touched src/scheduler.py"),
)
class HumanReportedFlowTest(BugRecordTestBase):
def _insert_human_bug(self, bug_id="bug-1", status="open"):
self.conn.execute(
"INSERT INTO bug_record (bug_id, task_id, source, description, status) "
"VALUES (?, 'task-1', 'human', 'something looks broken', ?)",
(bug_id, status),
)
self.conn.commit()
def test_claim_transitions_open_to_triaging(self):
self._insert_human_bug()
bug_record.claim_human_reported_bug(self.conn, "bug-1")
status = self.conn.execute("SELECT status FROM bug_record WHERE bug_id = 'bug-1'").fetchone()[0]
self.assertEqual(status, "triaging")
def test_claim_rejects_already_claimed_bug(self):
self._insert_human_bug(status="triaging")
with self.assertRaises(ValueError):
bug_record.claim_human_reported_bug(self.conn, "bug-1")
def test_claim_rejects_unknown_bug_id(self):
with self.assertRaises(ValueError):
bug_record.claim_human_reported_bug(self.conn, "no-such-bug")
def test_complete_triage_updates_in_place_not_new_record(self):
self._insert_human_bug()
bug_record.claim_human_reported_bug(self.conn, "bug-1")
bug_record.complete_human_reported_triage(
self.conn, "bug-1", root_cause_category="coder",
root_cause_detail="found it", suggested_route="fast_lane",
)
count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0]
self.assertEqual(count, 1)
row = self.conn.execute(
"SELECT status, root_cause_category, root_cause_detail, suggested_route "
"FROM bug_record WHERE bug_id = 'bug-1'"
).fetchone()
self.assertEqual(row, ("routed", "coder", "found it", "fast_lane"))
def test_complete_triage_rejects_invalid_category(self):
self._insert_human_bug()
bug_record.claim_human_reported_bug(self.conn, "bug-1")
with self.assertRaises(bug_record.InvalidRootCauseCategoryError):
bug_record.complete_human_reported_triage(
self.conn, "bug-1", root_cause_category="nonsense",
root_cause_detail="x", suggested_route="fast_lane",
)
if __name__ == "__main__":
unittest.main()