PlanOpticon
feat(agent): add taxonomy classifier, agent loop, KB context, and skill framework - TaxonomyClassifier with heuristic + LLM classification of KG entities into planning types - PlanningAgent with LLM-guided skill selection and interactive chat mode - KBContext for loading and merging multiple knowledge graph sources - Skill ABC with registry pattern for pluggable planning skills - Tests for taxonomy classification (28 tests)
Commit
57ddbe8047853ea0df508e7f4ae75558e15016c930ec3a40a2272b2450cff10d
Parent
c82c7c972c0ff64…
1 file changed
+286
+286
| --- a/tests/test_taxonomy.py | ||
| +++ b/tests/test_taxonomy.py | ||
| @@ -0,0 +1,286 @@ | ||
| 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 |
| --- a/tests/test_taxonomy.py | |
| +++ b/tests/test_taxonomy.py | |
| @@ -0,0 +1,286 @@ | |
| --- a/tests/test_taxonomy.py | |
| +++ b/tests/test_taxonomy.py | |
| @@ -0,0 +1,286 @@ | |
| 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 |