PlanOpticon

planopticon / tests / test_agent.py
Source Blame History 571 lines
0981a08… noreply 1 """Tests for the planning agent, skill registry, KB context, and agent loop."""
9b34c98… leo 2
9b34c98… leo 3 import json
0981a08… noreply 4 from pathlib import Path
0981a08… noreply 5 from unittest.mock import MagicMock, patch
0981a08… noreply 6
0981a08… noreply 7 import pytest
0981a08… noreply 8
0981a08… noreply 9 from video_processor.agent.skills.base import (
0981a08… noreply 10 AgentContext,
0981a08… noreply 11 Artifact,
0981a08… noreply 12 Skill,
0981a08… noreply 13 _skills,
0981a08… noreply 14 get_skill,
0981a08… noreply 15 list_skills,
0981a08… noreply 16 register_skill,
0981a08… noreply 17 )
0981a08… noreply 18
0981a08… noreply 19 # ---------------------------------------------------------------------------
0981a08… noreply 20 # Fixtures
0981a08… noreply 21 # ---------------------------------------------------------------------------
0981a08… noreply 22
0981a08… noreply 23
0981a08… noreply 24 @pytest.fixture(autouse=True)
0981a08… noreply 25 def _clean_skill_registry():
0981a08… noreply 26 """Save and restore the global skill registry between tests."""
0981a08… noreply 27 original = dict(_skills)
0981a08… noreply 28 yield
0981a08… noreply 29 _skills.clear()
0981a08… noreply 30 _skills.update(original)
0981a08… noreply 31
0981a08… noreply 32
0981a08… noreply 33 class _DummySkill(Skill):
0981a08… noreply 34 name = "dummy_test_skill"
0981a08… noreply 35 description = "A dummy skill for testing"
0981a08… noreply 36
0981a08… noreply 37 def execute(self, context: AgentContext, **kwargs) -> Artifact:
0981a08… noreply 38 return Artifact(
0981a08… noreply 39 name="dummy artifact",
0981a08… noreply 40 content="dummy content",
0981a08… noreply 41 artifact_type="document",
0981a08… noreply 42 )
0981a08… noreply 43
0981a08… noreply 44
0981a08… noreply 45 class _NoLLMSkill(Skill):
0981a08… noreply 46 """Skill that doesn't require provider_manager."""
0981a08… noreply 47
0981a08… noreply 48 name = "nollm_skill"
0981a08… noreply 49 description = "Works without LLM"
0981a08… noreply 50
0981a08… noreply 51 def execute(self, context: AgentContext, **kwargs) -> Artifact:
0981a08… noreply 52 return Artifact(
0981a08… noreply 53 name="nollm artifact",
0981a08… noreply 54 content="generated",
0981a08… noreply 55 artifact_type="document",
0981a08… noreply 56 )
0981a08… noreply 57
0981a08… noreply 58 def can_execute(self, context: AgentContext) -> bool:
0981a08… noreply 59 return context.knowledge_graph is not None
0981a08… noreply 60
0981a08… noreply 61
0981a08… noreply 62 # ---------------------------------------------------------------------------
0981a08… noreply 63 # Skill registry
0981a08… noreply 64 # ---------------------------------------------------------------------------
0981a08… noreply 65
0981a08… noreply 66
0981a08… noreply 67 class TestSkillRegistry:
0981a08… noreply 68 def test_register_and_get(self):
0981a08… noreply 69 skill = _DummySkill()
0981a08… noreply 70 register_skill(skill)
0981a08… noreply 71 assert get_skill("dummy_test_skill") is skill
0981a08… noreply 72
0981a08… noreply 73 def test_get_unknown_returns_none(self):
0981a08… noreply 74 assert get_skill("no_such_skill_xyz") is None
0981a08… noreply 75
0981a08… noreply 76 def test_list_skills(self):
0981a08… noreply 77 s1 = _DummySkill()
0981a08… noreply 78 register_skill(s1)
0981a08… noreply 79 skills = list_skills()
0981a08… noreply 80 assert any(s.name == "dummy_test_skill" for s in skills)
0981a08… noreply 81
0981a08… noreply 82 def test_list_skills_empty(self):
0981a08… noreply 83 _skills.clear()
0981a08… noreply 84 assert list_skills() == []
0981a08… noreply 85
0981a08… noreply 86
0981a08… noreply 87 # ---------------------------------------------------------------------------
0981a08… noreply 88 # AgentContext dataclass
0981a08… noreply 89 # ---------------------------------------------------------------------------
0981a08… noreply 90
0981a08… noreply 91
0981a08… noreply 92 class TestAgentContext:
0981a08… noreply 93 def test_defaults(self):
0981a08… noreply 94 ctx = AgentContext()
0981a08… noreply 95 assert ctx.knowledge_graph is None
0981a08… noreply 96 assert ctx.query_engine is None
0981a08… noreply 97 assert ctx.provider_manager is None
0981a08… noreply 98 assert ctx.planning_entities == []
0981a08… noreply 99 assert ctx.user_requirements == {}
0981a08… noreply 100 assert ctx.conversation_history == []
0981a08… noreply 101 assert ctx.artifacts == []
0981a08… noreply 102 assert ctx.config == {}
0981a08… noreply 103
0981a08… noreply 104 def test_with_values(self):
0981a08… noreply 105 mock_kg = MagicMock()
0981a08… noreply 106 mock_qe = MagicMock()
0981a08… noreply 107 mock_pm = MagicMock()
0981a08… noreply 108 ctx = AgentContext(
0981a08… noreply 109 knowledge_graph=mock_kg,
0981a08… noreply 110 query_engine=mock_qe,
0981a08… noreply 111 provider_manager=mock_pm,
0981a08… noreply 112 config={"key": "value"},
0981a08… noreply 113 )
0981a08… noreply 114 assert ctx.knowledge_graph is mock_kg
0981a08… noreply 115 assert ctx.config == {"key": "value"}
0981a08… noreply 116
0981a08… noreply 117 def test_conversation_history_is_mutable(self):
0981a08… noreply 118 ctx = AgentContext()
0981a08… noreply 119 ctx.conversation_history.append({"role": "user", "content": "hello"})
0981a08… noreply 120 assert len(ctx.conversation_history) == 1
0981a08… noreply 121
0981a08… noreply 122
0981a08… noreply 123 # ---------------------------------------------------------------------------
0981a08… noreply 124 # Artifact dataclass
0981a08… noreply 125 # ---------------------------------------------------------------------------
0981a08… noreply 126
0981a08… noreply 127
0981a08… noreply 128 class TestArtifact:
0981a08… noreply 129 def test_basic(self):
0981a08… noreply 130 a = Artifact(name="Plan", content="# Plan\n...", artifact_type="project_plan")
0981a08… noreply 131 assert a.name == "Plan"
0981a08… noreply 132 assert a.format == "markdown" # default
0981a08… noreply 133 assert a.metadata == {}
0981a08… noreply 134
0981a08… noreply 135 def test_with_metadata(self):
0981a08… noreply 136 a = Artifact(
0981a08… noreply 137 name="Tasks",
0981a08… noreply 138 content="[]",
0981a08… noreply 139 artifact_type="task_list",
0981a08… noreply 140 format="json",
0981a08… noreply 141 metadata={"source": "kg"},
0981a08… noreply 142 )
0981a08… noreply 143 assert a.format == "json"
0981a08… noreply 144 assert a.metadata["source"] == "kg"
0981a08… noreply 145
0981a08… noreply 146
0981a08… noreply 147 # ---------------------------------------------------------------------------
0981a08… noreply 148 # Skill.can_execute
0981a08… noreply 149 # ---------------------------------------------------------------------------
0981a08… noreply 150
0981a08… noreply 151
0981a08… noreply 152 class TestSkillCanExecute:
0981a08… noreply 153 def test_default_requires_kg_and_pm(self):
0981a08… noreply 154 skill = _DummySkill()
0981a08… noreply 155 ctx_no_kg = AgentContext(provider_manager=MagicMock())
0981a08… noreply 156 assert not skill.can_execute(ctx_no_kg)
0981a08… noreply 157
0981a08… noreply 158 ctx_no_pm = AgentContext(knowledge_graph=MagicMock())
0981a08… noreply 159 assert not skill.can_execute(ctx_no_pm)
0981a08… noreply 160
0981a08… noreply 161 ctx_both = AgentContext(knowledge_graph=MagicMock(), provider_manager=MagicMock())
0981a08… noreply 162 assert skill.can_execute(ctx_both)
0981a08… noreply 163
0981a08… noreply 164
0981a08… noreply 165 # ---------------------------------------------------------------------------
0981a08… noreply 166 # KBContext
0981a08… noreply 167 # ---------------------------------------------------------------------------
0981a08… noreply 168
0981a08… noreply 169
0981a08… noreply 170 class TestKBContext:
0981a08… noreply 171 def test_add_source_nonexistent_raises(self, tmp_path):
0981a08… noreply 172 from video_processor.agent.kb_context import KBContext
0981a08… noreply 173
0981a08… noreply 174 ctx = KBContext()
0981a08… noreply 175 with pytest.raises(FileNotFoundError, match="Not found"):
0981a08… noreply 176 ctx.add_source(tmp_path / "nonexistent.json")
0981a08… noreply 177
0981a08… noreply 178 def test_add_source_file(self, tmp_path):
0981a08… noreply 179 from video_processor.agent.kb_context import KBContext
0981a08… noreply 180
0981a08… noreply 181 f = tmp_path / "kg.json"
0981a08… noreply 182 f.write_text("{}")
0981a08… noreply 183 ctx = KBContext()
0981a08… noreply 184 ctx.add_source(f)
0981a08… noreply 185 assert len(ctx.sources) == 1
0981a08… noreply 186 assert ctx.sources[0] == f.resolve()
0981a08… noreply 187
0981a08… noreply 188 def test_add_source_directory(self, tmp_path):
0981a08… noreply 189 from video_processor.agent.kb_context import KBContext
0981a08… noreply 190
0981a08… noreply 191 with patch(
0981a08… noreply 192 "video_processor.integrators.graph_discovery.find_knowledge_graphs",
0981a08… noreply 193 return_value=[tmp_path / "a.db"],
0981a08… noreply 194 ):
0981a08… noreply 195 ctx = KBContext()
0981a08… noreply 196 ctx.add_source(tmp_path)
0981a08… noreply 197 assert len(ctx.sources) == 1
0981a08… noreply 198
0981a08… noreply 199 def test_knowledge_graph_before_load_raises(self):
0981a08… noreply 200 from video_processor.agent.kb_context import KBContext
0981a08… noreply 201
0981a08… noreply 202 ctx = KBContext()
0981a08… noreply 203 with pytest.raises(RuntimeError, match="Call load"):
0981a08… noreply 204 _ = ctx.knowledge_graph
0981a08… noreply 205
0981a08… noreply 206 def test_query_engine_before_load_raises(self):
0981a08… noreply 207 from video_processor.agent.kb_context import KBContext
0981a08… noreply 208
0981a08… noreply 209 ctx = KBContext()
0981a08… noreply 210 with pytest.raises(RuntimeError, match="Call load"):
0981a08… noreply 211 _ = ctx.query_engine
0981a08… noreply 212
0981a08… noreply 213 def test_summary_no_data(self):
0981a08… noreply 214 from video_processor.agent.kb_context import KBContext
0981a08… noreply 215
0981a08… noreply 216 ctx = KBContext()
0981a08… noreply 217 assert ctx.summary() == "No knowledge base loaded."
0981a08… noreply 218
0981a08… noreply 219 def test_load_json_and_summary(self, tmp_path):
0981a08… noreply 220 from video_processor.agent.kb_context import KBContext
0981a08… noreply 221
0981a08… noreply 222 kg_data = {"nodes": [], "relationships": []}
0981a08… noreply 223 f = tmp_path / "kg.json"
0981a08… noreply 224 f.write_text(json.dumps(kg_data))
0981a08… noreply 225
0981a08… noreply 226 ctx = KBContext()
0981a08… noreply 227 ctx.add_source(f)
0981a08… noreply 228 ctx.load()
0981a08… noreply 229
0981a08… noreply 230 summary = ctx.summary()
0981a08… noreply 231 assert "Knowledge base" in summary
0981a08… noreply 232 assert "Entities" in summary
0981a08… noreply 233 assert "Relationships" in summary
0981a08… noreply 234
0981a08… noreply 235
0981a08… noreply 236 # ---------------------------------------------------------------------------
0981a08… noreply 237 # PlanningAgent
0981a08… noreply 238 # ---------------------------------------------------------------------------
0981a08… noreply 239
0981a08… noreply 240
0981a08… noreply 241 class TestPlanningAgent:
0981a08… noreply 242 def test_from_kb_paths(self, tmp_path):
0981a08… noreply 243 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 244
0981a08… noreply 245 kg_data = {"nodes": [], "relationships": []}
0981a08… noreply 246 f = tmp_path / "kg.json"
0981a08… noreply 247 f.write_text(json.dumps(kg_data))
0981a08… noreply 248
0981a08… noreply 249 agent = PlanningAgent.from_kb_paths([f], provider_manager=None)
0981a08… noreply 250 assert agent.context.knowledge_graph is not None
0981a08… noreply 251 assert agent.context.provider_manager is None
0981a08… noreply 252
0981a08… noreply 253 def test_execute_with_mock_provider(self, tmp_path):
0981a08… noreply 254 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 255
0981a08… noreply 256 # Register a dummy skill
0981a08… noreply 257 skill = _DummySkill()
0981a08… noreply 258 register_skill(skill)
0981a08… noreply 259
0981a08… noreply 260 mock_pm = MagicMock()
0981a08… noreply 261 mock_pm.chat.return_value = json.dumps([{"skill": "dummy_test_skill", "params": {}}])
0981a08… noreply 262
0981a08… noreply 263 ctx = AgentContext(
0981a08… noreply 264 knowledge_graph=MagicMock(),
0981a08… noreply 265 query_engine=MagicMock(),
0981a08… noreply 266 provider_manager=mock_pm,
0981a08… noreply 267 )
0981a08… noreply 268 # Mock stats().to_text()
0981a08… noreply 269 ctx.query_engine.stats.return_value.to_text.return_value = "3 entities"
0981a08… noreply 270
0981a08… noreply 271 agent = PlanningAgent(context=ctx)
0981a08… noreply 272 artifacts = agent.execute("generate a plan")
0981a08… noreply 273
0981a08… noreply 274 assert len(artifacts) == 1
0981a08… noreply 275 assert artifacts[0].name == "dummy artifact"
0981a08… noreply 276 mock_pm.chat.assert_called_once()
0981a08… noreply 277
0981a08… noreply 278 def test_execute_no_provider_keyword_match(self):
0981a08… noreply 279 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 280
0981a08… noreply 281 skill = _DummySkill()
0981a08… noreply 282 register_skill(skill)
0981a08… noreply 283
0981a08… noreply 284 ctx = AgentContext(
0981a08… noreply 285 knowledge_graph=MagicMock(),
0981a08… noreply 286 provider_manager=None,
0981a08… noreply 287 )
0981a08… noreply 288
0981a08… noreply 289 agent = PlanningAgent(context=ctx)
0981a08… noreply 290 # "dummy" is a keyword in the skill name, but can_execute needs provider_manager
0981a08… noreply 291 # so it should return empty
0981a08… noreply 292 artifacts = agent.execute("dummy request")
0981a08… noreply 293 assert artifacts == []
0981a08… noreply 294
0981a08… noreply 295 def test_execute_keyword_match_nollm_skill(self):
0981a08… noreply 296 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 297
0981a08… noreply 298 skill = _NoLLMSkill()
0981a08… noreply 299 register_skill(skill)
0981a08… noreply 300
0981a08… noreply 301 ctx = AgentContext(
0981a08… noreply 302 knowledge_graph=MagicMock(),
0981a08… noreply 303 provider_manager=None,
0981a08… noreply 304 )
0981a08… noreply 305
0981a08… noreply 306 agent = PlanningAgent(context=ctx)
0981a08… noreply 307 # "nollm" is in the skill name
0981a08… noreply 308 artifacts = agent.execute("nollm stuff")
0981a08… noreply 309 assert len(artifacts) == 1
0981a08… noreply 310 assert artifacts[0].name == "nollm artifact"
0981a08… noreply 311
0981a08… noreply 312 def test_execute_skips_unknown_skills(self):
0981a08… noreply 313 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 314
0981a08… noreply 315 mock_pm = MagicMock()
0981a08… noreply 316 mock_pm.chat.return_value = json.dumps([{"skill": "nonexistent_skill_xyz", "params": {}}])
0981a08… noreply 317
0981a08… noreply 318 ctx = AgentContext(
0981a08… noreply 319 knowledge_graph=MagicMock(),
0981a08… noreply 320 query_engine=MagicMock(),
0981a08… noreply 321 provider_manager=mock_pm,
0981a08… noreply 322 )
0981a08… noreply 323 ctx.query_engine.stats.return_value.to_text.return_value = ""
0981a08… noreply 324
0981a08… noreply 325 agent = PlanningAgent(context=ctx)
0981a08… noreply 326 artifacts = agent.execute("do something")
0981a08… noreply 327 assert artifacts == []
0981a08… noreply 328
0981a08… noreply 329 def test_chat_no_provider(self):
0981a08… noreply 330 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 331
0981a08… noreply 332 ctx = AgentContext(provider_manager=None)
0981a08… noreply 333 agent = PlanningAgent(context=ctx)
0981a08… noreply 334
0981a08… noreply 335 reply = agent.chat("hello")
0981a08… noreply 336 assert "requires" in reply.lower() or "provider" in reply.lower()
0981a08… noreply 337
0981a08… noreply 338 def test_chat_with_provider(self):
0981a08… noreply 339 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 340
0981a08… noreply 341 mock_pm = MagicMock()
0981a08… noreply 342 mock_pm.chat.return_value = "I can help you plan."
0981a08… noreply 343
0981a08… noreply 344 ctx = AgentContext(
0981a08… noreply 345 knowledge_graph=MagicMock(),
0981a08… noreply 346 query_engine=MagicMock(),
0981a08… noreply 347 provider_manager=mock_pm,
0981a08… noreply 348 )
0981a08… noreply 349 ctx.query_engine.stats.return_value.to_text.return_value = "5 entities"
0981a08… noreply 350
0981a08… noreply 351 agent = PlanningAgent(context=ctx)
0981a08… noreply 352 reply = agent.chat("help me plan")
0981a08… noreply 353
0981a08… noreply 354 assert reply == "I can help you plan."
0981a08… noreply 355 assert len(ctx.conversation_history) == 2 # user + assistant
0981a08… noreply 356 assert ctx.conversation_history[0]["role"] == "user"
0981a08… noreply 357 assert ctx.conversation_history[1]["role"] == "assistant"
0981a08… noreply 358
0981a08… noreply 359 def test_chat_accumulates_history(self):
0981a08… noreply 360 from video_processor.agent.agent_loop import PlanningAgent
0981a08… noreply 361
0981a08… noreply 362 mock_pm = MagicMock()
0981a08… noreply 363 mock_pm.chat.side_effect = ["reply1", "reply2"]
0981a08… noreply 364
0981a08… noreply 365 ctx = AgentContext(provider_manager=mock_pm)
0981a08… noreply 366 agent = PlanningAgent(context=ctx)
0981a08… noreply 367
0981a08… noreply 368 agent.chat("msg1")
0981a08… noreply 369 agent.chat("msg2")
0981a08… noreply 370
0981a08… noreply 371 assert len(ctx.conversation_history) == 4 # 2 user + 2 assistant
0981a08… noreply 372 # The system message is constructed each time but not stored in history
0981a08… noreply 373 # Provider should receive progressively longer message lists
0981a08… noreply 374 second_call_messages = mock_pm.chat.call_args_list[1][0][0]
0981a08… noreply 375 # Should include system + 3 prior messages (user, assistant, user)
0981a08… noreply 376 assert len(second_call_messages) == 4 # system + user + assistant + user
0981a08… noreply 377
829e24a… leo 378
0981a08… noreply 379 # ---------------------------------------------------------------------------
0981a08… noreply 380 # Orchestrator tests (from existing test_agent.py — kept for coverage)
0981a08… noreply 381 # ---------------------------------------------------------------------------
9b34c98… leo 382
9b34c98… leo 383
9b34c98… leo 384 class TestPlanCreation:
9b34c98… leo 385 def test_basic_plan(self):
0981a08… noreply 386 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 387
9b34c98… leo 388 agent = AgentOrchestrator()
9b34c98… leo 389 plan = agent._create_plan("test.mp4", "basic")
9b34c98… leo 390 steps = [s["step"] for s in plan]
9b34c98… leo 391 assert "extract_frames" in steps
9b34c98… leo 392 assert "extract_audio" in steps
9b34c98… leo 393 assert "transcribe" in steps
9b34c98… leo 394 assert "extract_key_points" in steps
9b34c98… leo 395 assert "extract_action_items" in steps
9b34c98… leo 396 assert "generate_reports" in steps
9b34c98… leo 397 assert "detect_diagrams" not in steps
9b34c98… leo 398
9b34c98… leo 399 def test_standard_plan(self):
0981a08… noreply 400 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 401
9b34c98… leo 402 agent = AgentOrchestrator()
9b34c98… leo 403 plan = agent._create_plan("test.mp4", "standard")
9b34c98… leo 404 steps = [s["step"] for s in plan]
9b34c98… leo 405 assert "detect_diagrams" in steps
9b34c98… leo 406 assert "build_knowledge_graph" in steps
9b34c98… leo 407 assert "deep_analysis" not in steps
9b34c98… leo 408
9b34c98… leo 409 def test_comprehensive_plan(self):
0981a08… noreply 410 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 411
9b34c98… leo 412 agent = AgentOrchestrator()
9b34c98… leo 413 plan = agent._create_plan("test.mp4", "comprehensive")
9b34c98… leo 414 steps = [s["step"] for s in plan]
9b34c98… leo 415 assert "detect_diagrams" in steps
9b34c98… leo 416 assert "deep_analysis" in steps
9b34c98… leo 417 assert "cross_reference" in steps
9b34c98… leo 418
9b34c98… leo 419
9b34c98… leo 420 class TestAdaptPlan:
9b34c98… leo 421 def test_adapts_for_long_transcript(self):
0981a08… noreply 422 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 423
9b34c98… leo 424 agent = AgentOrchestrator()
9b34c98… leo 425 agent._plan = [{"step": "generate_reports", "priority": "required"}]
0981a08… noreply 426 long_text = "word " * 3000
9b34c98… leo 427 agent._adapt_plan("transcribe", {"text": long_text})
9b34c98… leo 428 steps = [s["step"] for s in agent._plan]
9b34c98… leo 429 assert "deep_analysis" in steps
9b34c98… leo 430
9b34c98… leo 431 def test_no_adapt_for_short_transcript(self):
0981a08… noreply 432 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 433
9b34c98… leo 434 agent = AgentOrchestrator()
9b34c98… leo 435 agent._plan = [{"step": "generate_reports", "priority": "required"}]
9b34c98… leo 436 agent._adapt_plan("transcribe", {"text": "Short text"})
9b34c98… leo 437 steps = [s["step"] for s in agent._plan]
9b34c98… leo 438 assert "deep_analysis" not in steps
9b34c98… leo 439
9b34c98… leo 440 def test_adapts_for_many_diagrams(self):
0981a08… noreply 441 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 442
9b34c98… leo 443 agent = AgentOrchestrator()
9b34c98… leo 444 agent._plan = [{"step": "generate_reports", "priority": "required"}]
9b34c98… leo 445 diagrams = [MagicMock() for _ in range(5)]
9b34c98… leo 446 agent._adapt_plan("detect_diagrams", {"diagrams": diagrams, "captures": []})
9b34c98… leo 447 steps = [s["step"] for s in agent._plan]
9b34c98… leo 448 assert "cross_reference" in steps
9b34c98… leo 449
9b34c98… leo 450 def test_insight_for_many_captures(self):
0981a08… noreply 451 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 452
9b34c98… leo 453 agent = AgentOrchestrator()
9b34c98… leo 454 agent._plan = []
9b34c98… leo 455 captures = [MagicMock() for _ in range(5)]
9b34c98… leo 456 diagrams = [MagicMock() for _ in range(2)]
9b34c98… leo 457 agent._adapt_plan("detect_diagrams", {"diagrams": diagrams, "captures": captures})
9b34c98… leo 458 assert len(agent._insights) == 1
9b34c98… leo 459 assert "uncertain frames" in agent._insights[0]
9b34c98… leo 460
9b34c98… leo 461 def test_no_duplicate_steps(self):
0981a08… noreply 462 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 463
9b34c98… leo 464 agent = AgentOrchestrator()
9b34c98… leo 465 agent._plan = [{"step": "deep_analysis", "priority": "comprehensive"}]
9b34c98… leo 466 long_text = "word " * 3000
9b34c98… leo 467 agent._adapt_plan("transcribe", {"text": long_text})
9b34c98… leo 468 deep_steps = [s for s in agent._plan if s["step"] == "deep_analysis"]
9b34c98… leo 469 assert len(deep_steps) == 1
9b34c98… leo 470
9b34c98… leo 471
9b34c98… leo 472 class TestFallbacks:
9b34c98… leo 473 def test_diagram_fallback(self):
0981a08… noreply 474 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 475
9b34c98… leo 476 agent = AgentOrchestrator()
9b34c98… leo 477 assert agent._get_fallback("detect_diagrams") == "screengrab_fallback"
9b34c98… leo 478
9b34c98… leo 479 def test_no_fallback_for_unknown(self):
0981a08… noreply 480 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 481
9b34c98… leo 482 agent = AgentOrchestrator()
9b34c98… leo 483 assert agent._get_fallback("transcribe") is None
9b34c98… leo 484
9b34c98… leo 485
9b34c98… leo 486 class TestInsights:
9b34c98… leo 487 def test_insights_property(self):
0981a08… noreply 488 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 489
9b34c98… leo 490 agent = AgentOrchestrator()
9b34c98… leo 491 agent._insights = ["Insight 1", "Insight 2"]
9b34c98… leo 492 assert agent.insights == ["Insight 1", "Insight 2"]
9b34c98… leo 493 agent.insights.append("should not modify internal")
9b34c98… leo 494 assert len(agent._insights) == 2
9b34c98… leo 495
9b34c98… leo 496 def test_deep_analysis_populates_insights(self):
0981a08… noreply 497 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 498
9b34c98… leo 499 pm = MagicMock()
829e24a… leo 500 pm.chat.return_value = json.dumps(
829e24a… leo 501 {
829e24a… leo 502 "decisions": ["Decided to use microservices"],
829e24a… leo 503 "risks": ["Timeline is tight"],
829e24a… leo 504 "follow_ups": [],
829e24a… leo 505 "tensions": [],
829e24a… leo 506 }
829e24a… leo 507 )
9b34c98… leo 508 agent = AgentOrchestrator(provider_manager=pm)
9b34c98… leo 509 agent._results["transcribe"] = {"text": "Some long transcript text here"}
9b34c98… leo 510 result = agent._deep_analysis("/tmp")
9b34c98… leo 511 assert "decisions" in result
9b34c98… leo 512 assert any("microservices" in i for i in agent._insights)
9b34c98… leo 513 assert any("Timeline" in i for i in agent._insights)
9b34c98… leo 514
9b34c98… leo 515 def test_deep_analysis_handles_error(self):
0981a08… noreply 516 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 517
9b34c98… leo 518 pm = MagicMock()
9b34c98… leo 519 pm.chat.side_effect = Exception("API error")
9b34c98… leo 520 agent = AgentOrchestrator(provider_manager=pm)
9b34c98… leo 521 agent._results["transcribe"] = {"text": "some text"}
9b34c98… leo 522 result = agent._deep_analysis("/tmp")
9b34c98… leo 523 assert result == {}
9b34c98… leo 524
9b34c98… leo 525 def test_deep_analysis_no_transcript(self):
0981a08… noreply 526 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 527
9b34c98… leo 528 agent = AgentOrchestrator()
9b34c98… leo 529 agent._results["transcribe"] = {"text": ""}
9b34c98… leo 530 result = agent._deep_analysis("/tmp")
9b34c98… leo 531 assert result == {}
9b34c98… leo 532
9b34c98… leo 533
9b34c98… leo 534 class TestBuildManifest:
9b34c98… leo 535 def test_builds_from_results(self):
0981a08… noreply 536 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 537
9b34c98… leo 538 agent = AgentOrchestrator()
9b34c98… leo 539 agent._results = {
9b34c98… leo 540 "extract_frames": {"frames": [1, 2, 3], "paths": ["/a.jpg", "/b.jpg"]},
9b34c98… leo 541 "extract_audio": {"audio_path": "/audio.wav", "properties": {"duration": 60.0}},
9b34c98… leo 542 "detect_diagrams": {"diagrams": [], "captures": []},
9b34c98… leo 543 "extract_key_points": {"key_points": []},
9b34c98… leo 544 "extract_action_items": {"action_items": []},
9b34c98… leo 545 }
9b34c98… leo 546 manifest = agent._build_manifest(Path("test.mp4"), Path("/out"), "Test", 5.0)
9b34c98… leo 547 assert manifest.video.title == "Test"
9b34c98… leo 548 assert manifest.stats.frames_extracted == 3
9b34c98… leo 549 assert manifest.stats.duration_seconds == 5.0
9b34c98… leo 550 assert manifest.video.duration_seconds == 60.0
9b34c98… leo 551
9b34c98… leo 552 def test_handles_missing_results(self):
0981a08… noreply 553 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 554
9b34c98… leo 555 agent = AgentOrchestrator()
9b34c98… leo 556 agent._results = {}
9b34c98… leo 557 manifest = agent._build_manifest(Path("test.mp4"), Path("/out"), None, 1.0)
9b34c98… leo 558 assert manifest.video.title == "Analysis of test"
9b34c98… leo 559 assert manifest.stats.frames_extracted == 0
9b34c98… leo 560
9b34c98… leo 561 def test_handles_error_results(self):
0981a08… noreply 562 from video_processor.agent.orchestrator import AgentOrchestrator
0981a08… noreply 563
9b34c98… leo 564 agent = AgentOrchestrator()
9b34c98… leo 565 agent._results = {
9b34c98… leo 566 "extract_frames": {"error": "failed"},
9b34c98… leo 567 "detect_diagrams": {"error": "also failed"},
9b34c98… leo 568 }
9b34c98… leo 569 manifest = agent._build_manifest(Path("vid.mp4"), Path("/out"), None, 2.0)
9b34c98… leo 570 assert manifest.stats.frames_extracted == 0
9b34c98… leo 571 assert len(manifest.diagrams) == 0

Keyboard Shortcuts

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