|
1
|
"""Tests for the planning taxonomy classifier.""" |
|
2
|
|
|
3
|
from unittest.mock import MagicMock |
|
4
|
|
|
5
|
from video_processor.integrators.taxonomy import TaxonomyClassifier |
|
6
|
from video_processor.models import ( |
|
7
|
PlanningEntity, |
|
8
|
PlanningEntityType, |
|
9
|
PlanningRelationshipType, |
|
10
|
) |
|
11
|
|
|
12
|
# ── Fixtures ────────────────────────────────────────────────────────── |
|
13
|
|
|
14
|
|
|
15
|
def _entity(name, descriptions=None, entity_type="concept"): |
|
16
|
return { |
|
17
|
"name": name, |
|
18
|
"type": entity_type, |
|
19
|
"descriptions": descriptions or [], |
|
20
|
} |
|
21
|
|
|
22
|
|
|
23
|
# ── PlanningEntityType enum ────────────────────────────────────────── |
|
24
|
|
|
25
|
|
|
26
|
class TestPlanningEntityType: |
|
27
|
def test_all_values(self): |
|
28
|
expected = { |
|
29
|
"goal", |
|
30
|
"requirement", |
|
31
|
"constraint", |
|
32
|
"decision", |
|
33
|
"risk", |
|
34
|
"assumption", |
|
35
|
"dependency", |
|
36
|
"milestone", |
|
37
|
"task", |
|
38
|
"feature", |
|
39
|
} |
|
40
|
assert {t.value for t in PlanningEntityType} == expected |
|
41
|
|
|
42
|
def test_str_enum(self): |
|
43
|
assert PlanningEntityType.GOAL == "goal" |
|
44
|
assert PlanningEntityType.RISK.value == "risk" |
|
45
|
|
|
46
|
|
|
47
|
class TestPlanningRelationshipType: |
|
48
|
def test_all_values(self): |
|
49
|
expected = { |
|
50
|
"requires", |
|
51
|
"blocked_by", |
|
52
|
"has_risk", |
|
53
|
"depends_on", |
|
54
|
"addresses", |
|
55
|
"has_tradeoff", |
|
56
|
"delivers", |
|
57
|
"implements", |
|
58
|
"parent_of", |
|
59
|
} |
|
60
|
assert {t.value for t in PlanningRelationshipType} == expected |
|
61
|
|
|
62
|
|
|
63
|
# ── PlanningEntity model ───────────────────────────────────────────── |
|
64
|
|
|
65
|
|
|
66
|
class TestPlanningEntity: |
|
67
|
def test_minimal(self): |
|
68
|
pe = PlanningEntity(name="Ship v2", planning_type=PlanningEntityType.GOAL) |
|
69
|
assert pe.description == "" |
|
70
|
assert pe.priority is None |
|
71
|
assert pe.status is None |
|
72
|
assert pe.source_entities == [] |
|
73
|
assert pe.metadata == {} |
|
74
|
|
|
75
|
def test_full(self): |
|
76
|
pe = PlanningEntity( |
|
77
|
name="Ship v2", |
|
78
|
planning_type=PlanningEntityType.GOAL, |
|
79
|
description="Release version 2", |
|
80
|
priority="high", |
|
81
|
status="identified", |
|
82
|
source_entities=["v2 release"], |
|
83
|
metadata={"quarter": "Q3"}, |
|
84
|
) |
|
85
|
assert pe.priority == "high" |
|
86
|
assert pe.metadata["quarter"] == "Q3" |
|
87
|
|
|
88
|
def test_round_trip(self): |
|
89
|
pe = PlanningEntity( |
|
90
|
name="Auth module", |
|
91
|
planning_type=PlanningEntityType.FEATURE, |
|
92
|
priority="medium", |
|
93
|
source_entities=["Auth"], |
|
94
|
) |
|
95
|
restored = PlanningEntity.model_validate_json(pe.model_dump_json()) |
|
96
|
assert restored == pe |
|
97
|
|
|
98
|
|
|
99
|
# ── Heuristic classification ───────────────────────────────────────── |
|
100
|
|
|
101
|
|
|
102
|
class TestHeuristicClassify: |
|
103
|
def setup_method(self): |
|
104
|
self.classifier = TaxonomyClassifier() |
|
105
|
|
|
106
|
def test_goal_keyword(self): |
|
107
|
entities = [_entity("Ship v2", ["Our main goal is to ship v2"])] |
|
108
|
result = self.classifier.classify_entities(entities, []) |
|
109
|
assert len(result) == 1 |
|
110
|
assert result[0].planning_type == PlanningEntityType.GOAL |
|
111
|
|
|
112
|
def test_requirement_keyword(self): |
|
113
|
entities = [_entity("Auth", ["System must support SSO"])] |
|
114
|
result = self.classifier.classify_entities(entities, []) |
|
115
|
assert result[0].planning_type == PlanningEntityType.REQUIREMENT |
|
116
|
|
|
117
|
def test_constraint_keyword(self): |
|
118
|
entities = [_entity("Budget", ["Budget limitation of $50k"])] |
|
119
|
result = self.classifier.classify_entities(entities, []) |
|
120
|
assert result[0].planning_type == PlanningEntityType.CONSTRAINT |
|
121
|
|
|
122
|
def test_decision_keyword(self): |
|
123
|
entities = [_entity("DB choice", ["Team decided to use Postgres"])] |
|
124
|
result = self.classifier.classify_entities(entities, []) |
|
125
|
assert result[0].planning_type == PlanningEntityType.DECISION |
|
126
|
|
|
127
|
def test_risk_keyword(self): |
|
128
|
entities = [_entity("Vendor lock-in", ["There is a risk of vendor lock-in"])] |
|
129
|
result = self.classifier.classify_entities(entities, []) |
|
130
|
assert result[0].planning_type == PlanningEntityType.RISK |
|
131
|
|
|
132
|
def test_assumption_keyword(self): |
|
133
|
entities = [_entity("Team size", ["We assume the team stays at 5"])] |
|
134
|
result = self.classifier.classify_entities(entities, []) |
|
135
|
assert result[0].planning_type == PlanningEntityType.ASSUMPTION |
|
136
|
|
|
137
|
def test_dependency_keyword(self): |
|
138
|
entities = [_entity("API v3", ["This depends on API v3 being ready"])] |
|
139
|
result = self.classifier.classify_entities(entities, []) |
|
140
|
assert result[0].planning_type == PlanningEntityType.DEPENDENCY |
|
141
|
|
|
142
|
def test_milestone_keyword(self): |
|
143
|
entities = [_entity("Beta", ["Beta release milestone in March"])] |
|
144
|
result = self.classifier.classify_entities(entities, []) |
|
145
|
assert result[0].planning_type == PlanningEntityType.MILESTONE |
|
146
|
|
|
147
|
def test_task_keyword(self): |
|
148
|
entities = [_entity("Setup CI", ["Action item: set up CI pipeline"])] |
|
149
|
result = self.classifier.classify_entities(entities, []) |
|
150
|
assert result[0].planning_type == PlanningEntityType.TASK |
|
151
|
|
|
152
|
def test_feature_keyword(self): |
|
153
|
entities = [_entity("Search", ["Search feature with autocomplete"])] |
|
154
|
result = self.classifier.classify_entities(entities, []) |
|
155
|
assert result[0].planning_type == PlanningEntityType.FEATURE |
|
156
|
|
|
157
|
def test_no_match(self): |
|
158
|
entities = [_entity("Python", ["A programming language"])] |
|
159
|
result = self.classifier.classify_entities(entities, []) |
|
160
|
assert len(result) == 0 |
|
161
|
|
|
162
|
def test_multiple_entities(self): |
|
163
|
entities = [ |
|
164
|
_entity("Goal A", ["The goal is performance"]), |
|
165
|
_entity("Person B", ["Engineer on the team"], "person"), |
|
166
|
_entity("Risk C", ["Risk of data loss"]), |
|
167
|
] |
|
168
|
result = self.classifier.classify_entities(entities, []) |
|
169
|
assert len(result) == 2 |
|
170
|
types = {pe.planning_type for pe in result} |
|
171
|
assert PlanningEntityType.GOAL in types |
|
172
|
assert PlanningEntityType.RISK in types |
|
173
|
|
|
174
|
def test_description_joined(self): |
|
175
|
entities = [_entity("Perf", ["System must handle", "1000 req/s"])] |
|
176
|
result = self.classifier.classify_entities(entities, []) |
|
177
|
assert result[0].planning_type == PlanningEntityType.REQUIREMENT |
|
178
|
assert result[0].description == "System must handle; 1000 req/s" |
|
179
|
|
|
180
|
def test_source_entities_populated(self): |
|
181
|
entities = [_entity("Ship v2", ["Our main goal"])] |
|
182
|
result = self.classifier.classify_entities(entities, []) |
|
183
|
assert result[0].source_entities == ["Ship v2"] |
|
184
|
|
|
185
|
|
|
186
|
# ── LLM classification ─────────────────────────────────────────────── |
|
187
|
|
|
188
|
|
|
189
|
class TestLLMClassify: |
|
190
|
def test_llm_results_merged(self): |
|
191
|
mock_pm = MagicMock() |
|
192
|
mock_pm.chat.return_value = ( |
|
193
|
'[{"name": "Python", "planning_type": "feature", "priority": "medium"}]' |
|
194
|
) |
|
195
|
classifier = TaxonomyClassifier(provider_manager=mock_pm) |
|
196
|
entities = [_entity("Python", ["A programming language"])] |
|
197
|
result = classifier.classify_entities(entities, []) |
|
198
|
assert len(result) == 1 |
|
199
|
assert result[0].planning_type == PlanningEntityType.FEATURE |
|
200
|
assert result[0].priority == "medium" |
|
201
|
|
|
202
|
def test_llm_overrides_heuristic(self): |
|
203
|
mock_pm = MagicMock() |
|
204
|
# Heuristic would say REQUIREMENT ("must"), LLM says GOAL |
|
205
|
mock_pm.chat.return_value = ( |
|
206
|
'[{"name": "Perf", "planning_type": "goal", "priority": "high"}]' |
|
207
|
) |
|
208
|
classifier = TaxonomyClassifier(provider_manager=mock_pm) |
|
209
|
entities = [_entity("Perf", ["System must be fast"])] |
|
210
|
result = classifier.classify_entities(entities, []) |
|
211
|
assert len(result) == 1 |
|
212
|
assert result[0].planning_type == PlanningEntityType.GOAL |
|
213
|
|
|
214
|
def test_llm_invalid_type_skipped(self): |
|
215
|
mock_pm = MagicMock() |
|
216
|
mock_pm.chat.return_value = ( |
|
217
|
'[{"name": "X", "planning_type": "not_a_type", "priority": "low"}]' |
|
218
|
) |
|
219
|
classifier = TaxonomyClassifier(provider_manager=mock_pm) |
|
220
|
entities = [_entity("X", ["Something"])] |
|
221
|
result = classifier.classify_entities(entities, []) |
|
222
|
assert len(result) == 0 |
|
223
|
|
|
224
|
def test_llm_failure_falls_back(self): |
|
225
|
mock_pm = MagicMock() |
|
226
|
mock_pm.chat.side_effect = RuntimeError("API down") |
|
227
|
classifier = TaxonomyClassifier(provider_manager=mock_pm) |
|
228
|
entities = [_entity("Ship v2", ["Our goal"])] |
|
229
|
result = classifier.classify_entities(entities, []) |
|
230
|
# Should still get heuristic result |
|
231
|
assert len(result) == 1 |
|
232
|
assert result[0].planning_type == PlanningEntityType.GOAL |
|
233
|
|
|
234
|
def test_llm_empty_response(self): |
|
235
|
mock_pm = MagicMock() |
|
236
|
mock_pm.chat.return_value = "" |
|
237
|
classifier = TaxonomyClassifier(provider_manager=mock_pm) |
|
238
|
entities = [_entity("Ship v2", ["Our goal"])] |
|
239
|
result = classifier.classify_entities(entities, []) |
|
240
|
assert len(result) == 1 # heuristic still works |
|
241
|
|
|
242
|
|
|
243
|
# ── Workstream organization ────────────────────────────────────────── |
|
244
|
|
|
245
|
|
|
246
|
class TestOrganizeByWorkstream: |
|
247
|
def test_groups_by_type(self): |
|
248
|
classifier = TaxonomyClassifier() |
|
249
|
entities = [ |
|
250
|
PlanningEntity(name="A", planning_type=PlanningEntityType.GOAL), |
|
251
|
PlanningEntity(name="B", planning_type=PlanningEntityType.GOAL), |
|
252
|
PlanningEntity(name="C", planning_type=PlanningEntityType.RISK), |
|
253
|
] |
|
254
|
ws = classifier.organize_by_workstream(entities) |
|
255
|
assert len(ws["goals"]) == 2 |
|
256
|
assert len(ws["risks"]) == 1 |
|
257
|
|
|
258
|
def test_empty_input(self): |
|
259
|
classifier = TaxonomyClassifier() |
|
260
|
ws = classifier.organize_by_workstream([]) |
|
261
|
assert ws == {} |
|
262
|
|
|
263
|
|
|
264
|
# ── Merge classifications ──────────────────────────────────────────── |
|
265
|
|
|
266
|
|
|
267
|
class TestMergeClassifications: |
|
268
|
def test_llm_wins_conflict(self): |
|
269
|
h = [PlanningEntity(name="X", planning_type=PlanningEntityType.GOAL)] |
|
270
|
llm_list = [PlanningEntity(name="X", planning_type=PlanningEntityType.RISK)] |
|
271
|
merged = TaxonomyClassifier._merge_classifications(h, llm_list) |
|
272
|
assert len(merged) == 1 |
|
273
|
assert merged[0].planning_type == PlanningEntityType.RISK |
|
274
|
|
|
275
|
def test_case_insensitive_merge(self): |
|
276
|
h = [PlanningEntity(name="Auth", planning_type=PlanningEntityType.FEATURE)] |
|
277
|
llm_list = [PlanningEntity(name="auth", planning_type=PlanningEntityType.REQUIREMENT)] |
|
278
|
merged = TaxonomyClassifier._merge_classifications(h, llm_list) |
|
279
|
assert len(merged) == 1 |
|
280
|
assert merged[0].planning_type == PlanningEntityType.REQUIREMENT |
|
281
|
|
|
282
|
def test_union_of_distinct(self): |
|
283
|
h = [PlanningEntity(name="A", planning_type=PlanningEntityType.GOAL)] |
|
284
|
llm_list = [PlanningEntity(name="B", planning_type=PlanningEntityType.RISK)] |
|
285
|
merged = TaxonomyClassifier._merge_classifications(h, llm_list) |
|
286
|
assert len(merged) == 2 |
|
287
|
|