PlanOpticon

planopticon / tests / test_action_detector.py
Source Blame History 249 lines
ccf32cc… leo 1 """Tests for enhanced action item detection."""
ccf32cc… leo 2
ccf32cc… leo 3 import json
ccf32cc… leo 4 from unittest.mock import MagicMock
ccf32cc… leo 5
ccf32cc… leo 6 from video_processor.analyzers.action_detector import ActionDetector
ccf32cc… leo 7 from video_processor.models import ActionItem, TranscriptSegment
ccf32cc… leo 8
ccf32cc… leo 9
ccf32cc… leo 10 class TestPatternExtract:
ccf32cc… leo 11 def test_detects_need_to(self):
ccf32cc… leo 12 detector = ActionDetector()
829e24a… leo 13 items = detector.detect_from_transcript(
829e24a… leo 14 "We need to update the database schema before release."
829e24a… leo 15 )
ccf32cc… leo 16 assert len(items) >= 1
ccf32cc… leo 17 assert any("database" in i.action.lower() for i in items)
ccf32cc… leo 18
ccf32cc… leo 19 def test_detects_should(self):
ccf32cc… leo 20 detector = ActionDetector()
ccf32cc… leo 21 items = detector.detect_from_transcript("Alice should review the pull request by Friday.")
ccf32cc… leo 22 assert len(items) >= 1
ccf32cc… leo 23
ccf32cc… leo 24 def test_detects_action_item_keyword(self):
ccf32cc… leo 25 detector = ActionDetector()
829e24a… leo 26 items = detector.detect_from_transcript(
829e24a… leo 27 "Action item: set up monitoring for the new service."
829e24a… leo 28 )
ccf32cc… leo 29 assert len(items) >= 1
ccf32cc… leo 30
ccf32cc… leo 31 def test_detects_follow_up(self):
ccf32cc… leo 32 detector = ActionDetector()
ccf32cc… leo 33 items = detector.detect_from_transcript("Follow up with the client about requirements.")
ccf32cc… leo 34 assert len(items) >= 1
ccf32cc… leo 35
ccf32cc… leo 36 def test_detects_lets(self):
ccf32cc… leo 37 detector = ActionDetector()
ccf32cc… leo 38 items = detector.detect_from_transcript("Let's schedule a meeting to discuss the roadmap.")
ccf32cc… leo 39 assert len(items) >= 1
ccf32cc… leo 40
ccf32cc… leo 41 def test_ignores_short_sentences(self):
ccf32cc… leo 42 detector = ActionDetector()
ccf32cc… leo 43 items = detector.detect_from_transcript("Do it.")
ccf32cc… leo 44 assert len(items) == 0
ccf32cc… leo 45
ccf32cc… leo 46 def test_no_action_patterns(self):
ccf32cc… leo 47 detector = ActionDetector()
829e24a… leo 48 items = detector.detect_from_transcript("The weather was nice today. We had lunch at noon.")
ccf32cc… leo 49 assert len(items) == 0
ccf32cc… leo 50
ccf32cc… leo 51 def test_multiple_sentences(self):
ccf32cc… leo 52 detector = ActionDetector()
829e24a… leo 53 text = "We need to deploy the fix. Alice should test it first. The sky is blue."
ccf32cc… leo 54 items = detector.detect_from_transcript(text)
ccf32cc… leo 55 assert len(items) == 2
ccf32cc… leo 56
ccf32cc… leo 57 def test_source_is_transcript(self):
ccf32cc… leo 58 detector = ActionDetector()
ccf32cc… leo 59 items = detector.detect_from_transcript("We need to fix the authentication module.")
ccf32cc… leo 60 for item in items:
ccf32cc… leo 61 assert item.source == "transcript"
ccf32cc… leo 62
ccf32cc… leo 63
ccf32cc… leo 64 class TestLLMExtract:
ccf32cc… leo 65 def test_llm_extraction(self):
ccf32cc… leo 66 pm = MagicMock()
829e24a… leo 67 pm.chat.return_value = json.dumps(
829e24a… leo 68 [
829e24a… leo 69 {
829e24a… leo 70 "action": "Deploy new version",
829e24a… leo 71 "assignee": "Bob",
829e24a… leo 72 "deadline": "Friday",
829e24a… leo 73 "priority": "high",
829e24a… leo 74 "context": "Production release",
829e24a… leo 75 }
829e24a… leo 76 ]
829e24a… leo 77 )
ccf32cc… leo 78 detector = ActionDetector(provider_manager=pm)
ccf32cc… leo 79 items = detector.detect_from_transcript("Deploy new version by Friday.")
ccf32cc… leo 80 assert len(items) == 1
ccf32cc… leo 81 assert items[0].action == "Deploy new version"
ccf32cc… leo 82 assert items[0].assignee == "Bob"
ccf32cc… leo 83 assert items[0].deadline == "Friday"
ccf32cc… leo 84 assert items[0].priority == "high"
ccf32cc… leo 85 assert items[0].source == "transcript"
ccf32cc… leo 86
ccf32cc… leo 87 def test_llm_returns_empty(self):
ccf32cc… leo 88 pm = MagicMock()
ccf32cc… leo 89 pm.chat.return_value = "[]"
ccf32cc… leo 90 detector = ActionDetector(provider_manager=pm)
ccf32cc… leo 91 items = detector.detect_from_transcript("No action items here.")
ccf32cc… leo 92 assert items == []
ccf32cc… leo 93
ccf32cc… leo 94 def test_llm_error_returns_empty(self):
ccf32cc… leo 95 pm = MagicMock()
ccf32cc… leo 96 pm.chat.side_effect = Exception("API error")
ccf32cc… leo 97 detector = ActionDetector(provider_manager=pm)
ccf32cc… leo 98 items = detector.detect_from_transcript("We need to fix this.")
ccf32cc… leo 99 assert items == []
ccf32cc… leo 100
ccf32cc… leo 101 def test_llm_bad_json(self):
ccf32cc… leo 102 pm = MagicMock()
ccf32cc… leo 103 pm.chat.return_value = "not valid json"
ccf32cc… leo 104 detector = ActionDetector(provider_manager=pm)
ccf32cc… leo 105 items = detector.detect_from_transcript("Update the docs.")
ccf32cc… leo 106 assert items == []
ccf32cc… leo 107
ccf32cc… leo 108 def test_llm_skips_items_without_action(self):
ccf32cc… leo 109 pm = MagicMock()
829e24a… leo 110 pm.chat.return_value = json.dumps(
829e24a… leo 111 [
829e24a… leo 112 {"action": "Valid action", "assignee": None},
829e24a… leo 113 {"assignee": "Alice"}, # No action field
829e24a… leo 114 {"action": "", "assignee": "Bob"}, # Empty action
829e24a… leo 115 ]
829e24a… leo 116 )
ccf32cc… leo 117 detector = ActionDetector(provider_manager=pm)
ccf32cc… leo 118 items = detector.detect_from_transcript("Some text.")
ccf32cc… leo 119 assert len(items) == 1
ccf32cc… leo 120 assert items[0].action == "Valid action"
ccf32cc… leo 121
ccf32cc… leo 122
ccf32cc… leo 123 class TestDetectFromDiagrams:
ccf32cc… leo 124 def test_dict_diagrams(self):
ccf32cc… leo 125 pm = MagicMock()
829e24a… leo 126 pm.chat.return_value = json.dumps(
829e24a… leo 127 [
829e24a… leo 128 {
829e24a… leo 129 "action": "Migrate database",
829e24a… leo 130 "assignee": None,
829e24a… leo 131 "deadline": None,
829e24a… leo 132 "priority": None,
829e24a… leo 133 "context": None,
829e24a… leo 134 },
829e24a… leo 135 ]
829e24a… leo 136 )
ccf32cc… leo 137 detector = ActionDetector(provider_manager=pm)
ccf32cc… leo 138 diagrams = [
ccf32cc… leo 139 {"text_content": "Step 1: Migrate database", "elements": ["DB", "Migration"]},
ccf32cc… leo 140 ]
ccf32cc… leo 141 items = detector.detect_from_diagrams(diagrams)
ccf32cc… leo 142 assert len(items) == 1
ccf32cc… leo 143 assert items[0].source == "diagram"
ccf32cc… leo 144
ccf32cc… leo 145 def test_object_diagrams(self):
ccf32cc… leo 146 pm = MagicMock()
829e24a… leo 147 pm.chat.return_value = json.dumps(
829e24a… leo 148 [
829e24a… leo 149 {
829e24a… leo 150 "action": "Update API",
829e24a… leo 151 "assignee": None,
829e24a… leo 152 "deadline": None,
829e24a… leo 153 "priority": None,
829e24a… leo 154 "context": None,
829e24a… leo 155 },
829e24a… leo 156 ]
829e24a… leo 157 )
ccf32cc… leo 158 detector = ActionDetector(provider_manager=pm)
ccf32cc… leo 159
ccf32cc… leo 160 class FakeDiagram:
ccf32cc… leo 161 text_content = "Update API endpoints"
ccf32cc… leo 162 elements = ["API", "Gateway"]
ccf32cc… leo 163
ccf32cc… leo 164 items = detector.detect_from_diagrams([FakeDiagram()])
ccf32cc… leo 165 assert len(items) >= 1
ccf32cc… leo 166 assert items[0].source == "diagram"
ccf32cc… leo 167
ccf32cc… leo 168 def test_empty_diagram_skipped(self):
ccf32cc… leo 169 detector = ActionDetector()
ccf32cc… leo 170 diagrams = [{"text_content": "", "elements": []}]
ccf32cc… leo 171 items = detector.detect_from_diagrams(diagrams)
ccf32cc… leo 172 assert items == []
ccf32cc… leo 173
ccf32cc… leo 174 def test_pattern_fallback_for_diagrams(self):
ccf32cc… leo 175 detector = ActionDetector() # No provider
ccf32cc… leo 176 diagrams = [
829e24a… leo 177 {
829e24a… leo 178 "text_content": "We need to update the configuration before deployment.",
829e24a… leo 179 "elements": [],
829e24a… leo 180 },
ccf32cc… leo 181 ]
ccf32cc… leo 182 items = detector.detect_from_diagrams(diagrams)
ccf32cc… leo 183 assert len(items) >= 1
ccf32cc… leo 184 assert items[0].source == "diagram"
ccf32cc… leo 185
ccf32cc… leo 186
ccf32cc… leo 187 class TestMergeActionItems:
ccf32cc… leo 188 def test_deduplicates(self):
ccf32cc… leo 189 detector = ActionDetector()
ccf32cc… leo 190 t_items = [ActionItem(action="Deploy fix", source="transcript")]
ccf32cc… leo 191 d_items = [ActionItem(action="Deploy fix", source="diagram")]
ccf32cc… leo 192 merged = detector.merge_action_items(t_items, d_items)
ccf32cc… leo 193 assert len(merged) == 1
ccf32cc… leo 194
ccf32cc… leo 195 def test_case_insensitive_dedup(self):
ccf32cc… leo 196 detector = ActionDetector()
ccf32cc… leo 197 t_items = [ActionItem(action="deploy fix", source="transcript")]
ccf32cc… leo 198 d_items = [ActionItem(action="Deploy Fix", source="diagram")]
ccf32cc… leo 199 merged = detector.merge_action_items(t_items, d_items)
ccf32cc… leo 200 assert len(merged) == 1
ccf32cc… leo 201
ccf32cc… leo 202 def test_keeps_unique(self):
ccf32cc… leo 203 detector = ActionDetector()
ccf32cc… leo 204 t_items = [ActionItem(action="Task A", source="transcript")]
ccf32cc… leo 205 d_items = [ActionItem(action="Task B", source="diagram")]
ccf32cc… leo 206 merged = detector.merge_action_items(t_items, d_items)
ccf32cc… leo 207 assert len(merged) == 2
ccf32cc… leo 208
ccf32cc… leo 209 def test_empty_inputs(self):
ccf32cc… leo 210 detector = ActionDetector()
ccf32cc… leo 211 merged = detector.merge_action_items([], [])
ccf32cc… leo 212 assert merged == []
ccf32cc… leo 213
ccf32cc… leo 214
ccf32cc… leo 215 class TestAttachTimestamps:
ccf32cc… leo 216 def test_attaches_matching_segment(self):
ccf32cc… leo 217 detector = ActionDetector()
829e24a… leo 218 [
ccf32cc… leo 219 ActionItem(action="We need to update the database schema before release"),
ccf32cc… leo 220 ]
ccf32cc… leo 221 segments = [
ccf32cc… leo 222 TranscriptSegment(start=0.0, end=5.0, text="Welcome to the meeting."),
829e24a… leo 223 TranscriptSegment(
829e24a… leo 224 start=5.0, end=15.0, text="We need to update the database schema before release."
829e24a… leo 225 ),
ccf32cc… leo 226 TranscriptSegment(start=15.0, end=20.0, text="Any questions?"),
ccf32cc… leo 227 ]
ccf32cc… leo 228 detector.detect_from_transcript(
ccf32cc… leo 229 "We need to update the database schema before release.",
ccf32cc… leo 230 segments=segments,
ccf32cc… leo 231 )
ccf32cc… leo 232 # Pattern extract will create items; check them
ccf32cc… leo 233 result = detector.detect_from_transcript(
ccf32cc… leo 234 "We need to update the database schema before release.",
ccf32cc… leo 235 segments=segments,
ccf32cc… leo 236 )
ccf32cc… leo 237 assert len(result) >= 1
ccf32cc… leo 238 # Context should be set with timestamp
ccf32cc… leo 239 assert any(i.context and "5s" in i.context for i in result)
ccf32cc… leo 240
ccf32cc… leo 241 def test_no_match_no_context(self):
ccf32cc… leo 242 detector = ActionDetector()
ccf32cc… leo 243 items = [ActionItem(action="Completely unrelated action")]
ccf32cc… leo 244 segments = [
ccf32cc… leo 245 TranscriptSegment(start=0.0, end=5.0, text="Hello world."),
ccf32cc… leo 246 ]
ccf32cc… leo 247 # Manually test the private method
ccf32cc… leo 248 detector._attach_timestamps(items, segments)
ccf32cc… leo 249 assert items[0].context is None

Keyboard Shortcuts

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