test_root_cause_trace.py
5.61 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
"""U-015 tests: judge_root_cause's 2/3-majority logic and the test_failed/
human_reported wiring, with fake perspective_fns (CLAUDE.md Testing
constraint -- real perspectives shell out to `claude`, mocked here). Real
coverage is in src/test_root_cause_trace_e2e.py (RUN_REAL_AGENT_E2E=1).
Run: python3 -m unittest src.test_root_cause_trace -v
"""
import unittest
from src import bug_record, root_cause_trace, schema
def _fixed(category, detail="detail"):
return lambda: {"category": category, "detail": detail}
class JudgeRootCauseTest(unittest.TestCase):
def test_unanimous_agreement(self):
result = root_cause_trace.judge_root_cause(
[_fixed("coder", "d1"), _fixed("coder", "d2"), _fixed("coder", "d3")]
)
self.assertEqual(result["root_cause_category"], "coder")
self.assertIn("d1", result["root_cause_detail"])
self.assertIn("d2", result["root_cause_detail"])
def test_two_of_three_agreement_wins(self):
result = root_cause_trace.judge_root_cause(
[_fixed("planning", "d1"), _fixed("planning", "d2"), _fixed("coder", "d3")]
)
self.assertEqual(result["root_cause_category"], "planning")
self.assertNotIn("d3", result["root_cause_detail"])
def test_three_way_split_is_inconclusive(self):
with self.assertRaises(root_cause_trace.InconclusiveEscalateError) as ctx:
root_cause_trace.judge_root_cause(
[_fixed("planning"), _fixed("coder"), _fixed("skill_defect")]
)
self.assertEqual(len(ctx.exception.judgments), 3)
def test_below_two_thirds_with_more_perspectives_is_inconclusive(self):
# 2 out of 5 agreeing is not >=2/3 (04-P: "至少 2/3 视角一致")
with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
root_cause_trace.judge_root_cause(
[_fixed("coder"), _fixed("coder"), _fixed("planning"), _fixed("planning"), _fixed("skill_defect")]
)
class TraceRootCauseTestFailedTest(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()
def test_agreement_creates_bug_record(self):
result = root_cause_trace.trace_root_cause_test_failed(
self.conn, "task-1", "wall collision misjudged",
[_fixed("coder", "SDD §3.2"), _fixed("coder", "code line 42"), _fixed("planning", "no")],
suggested_route="fast_lane",
)
self.assertEqual(result["root_cause_category"], "coder")
row = self.conn.execute(
"SELECT source, status, task_id FROM bug_record WHERE bug_id = ?", (result["bug_id"],)
).fetchone()
self.assertEqual(row, ("validator", "routed", "task-1"))
def test_disagreement_raises_and_creates_no_record(self):
with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
root_cause_trace.trace_root_cause_test_failed(
self.conn, "task-1", "x",
[_fixed("planning"), _fixed("coder"), _fixed("skill_defect")],
suggested_route="fast_lane",
)
count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0]
self.assertEqual(count, 0)
class TraceRootCauseHumanReportedTest(unittest.TestCase):
def setUp(self):
self.conn = schema.connect(":memory:")
schema.create_schema(self.conn)
self.conn.execute(
"INSERT INTO bug_record (bug_id, source, description, status) "
"VALUES ('bug-1', 'human', 'something looks broken', 'triaging')"
)
self.conn.commit()
def tearDown(self):
self.conn.close()
def test_completes_existing_record_not_new_one(self):
root_cause_trace.trace_root_cause_human_reported(
self.conn, "bug-1", [_fixed("planning", "gap"), _fixed("planning", "gap2"), _fixed("coder")],
suggested_route="full_replan",
)
count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0]
self.assertEqual(count, 1)
row = self.conn.execute("SELECT status, root_cause_category FROM bug_record WHERE bug_id = 'bug-1'").fetchone()
self.assertEqual(row, ("routed", "planning"))
class RealValidatorPerspectiveParsingTest(unittest.TestCase):
def test_parses_category_and_detail(self):
fake_trigger = lambda org_node, prompt, target, timeout=300: "CATEGORY: coder\nDETAIL: off-by-one in bounds check"
perspective_fn = root_cause_trace.real_validator_perspective(
"correctness", "bug desc", "context", "/tmp/p", trigger_fn=fake_trigger
)
result = perspective_fn()
self.assertEqual(result, {"category": "coder", "detail": "off-by-one in bounds check"})
def test_missing_category_line_raises(self):
fake_trigger = lambda org_node, prompt, target, timeout=300: "I have no idea what happened."
perspective_fn = root_cause_trace.real_validator_perspective(
"correctness", "bug desc", "context", "/tmp/p", trigger_fn=fake_trigger
)
with self.assertRaises(ValueError):
perspective_fn()
if __name__ == "__main__":
unittest.main()