PlanOpticon

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

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button