# -*- coding: utf-8 -*- """Integration tests for EvoAgent system. These tests verify the integration between: - UnifiedAgentFactory - EvoAgent - ToolGuardMixin - Workspace-driven configuration """ import pytest from pathlib import Path from unittest.mock import MagicMock, AsyncMock class TestUnifiedAgentFactoryIntegration: """Test UnifiedAgentFactory creates agents correctly.""" def test_factory_creates_analyst_with_workspace_config(self, tmp_path, monkeypatch): """Test that factory creates EvoAgent with workspace config.""" from backend.agents.unified_factory import UnifiedAgentFactory # Setup mock skills manager class MockSkillsManager: def get_agent_asset_dir(self, config_name, agent_id): path = tmp_path / "runs" / config_name / "agents" / agent_id path.mkdir(parents=True, exist_ok=True) return path # Create workspace config workspace_dir = tmp_path / "runs" / "test_config" / "agents" / "fundamentals_analyst" workspace_dir.mkdir(parents=True, exist_ok=True) (workspace_dir / "agent.yaml").write_text( "prompt_files:\n - SOUL.md\n - CUSTOM.md\n", encoding="utf-8", ) (workspace_dir / "SOUL.md").write_text("System prompt content", encoding="utf-8") (workspace_dir / "CUSTOM.md").write_text("Custom instructions", encoding="utf-8") factory = UnifiedAgentFactory( config_name="test_config", skills_manager=MockSkillsManager(), ) # Mock EvoAgent creation by patching where it's imported created_kwargs = {} class MockEvoAgent: def __init__(self, **kwargs): created_kwargs.update(kwargs) self.toolkit = None # Patch at the location where EvoAgent is imported in unified_factory import backend.agents.base.evo_agent as evo_agent_module original_evo_agent = evo_agent_module.EvoAgent evo_agent_module.EvoAgent = MockEvoAgent try: monkeypatch.setattr( factory, "_create_toolkit", lambda *args, **kwargs: MagicMock(), ) agent = factory.create_analyst( analyst_type="fundamentals_analyst", model=MagicMock(), formatter=MagicMock(), ) assert isinstance(agent, MockEvoAgent) assert created_kwargs["agent_id"] == "fundamentals_analyst" assert created_kwargs["config_name"] == "test_config" assert "SOUL.md" in created_kwargs["prompt_files"] finally: evo_agent_module.EvoAgent = original_evo_agent def test_factory_creates_risk_manager(self, tmp_path, monkeypatch): """Test that factory creates risk manager EvoAgent.""" from backend.agents.unified_factory import UnifiedAgentFactory class MockSkillsManager: def get_agent_asset_dir(self, config_name, agent_id): path = tmp_path / "runs" / config_name / "agents" / agent_id path.mkdir(parents=True, exist_ok=True) return path factory = UnifiedAgentFactory( config_name="test_config", skills_manager=MockSkillsManager(), ) created_kwargs = {} class MockEvoAgent: def __init__(self, **kwargs): created_kwargs.update(kwargs) self.toolkit = None import backend.agents.base.evo_agent as evo_agent_module original_evo_agent = evo_agent_module.EvoAgent evo_agent_module.EvoAgent = MockEvoAgent try: monkeypatch.setattr( factory, "_create_toolkit", lambda *args, **kwargs: MagicMock(), ) agent = factory.create_risk_manager( model=MagicMock(), formatter=MagicMock(), ) assert isinstance(agent, MockEvoAgent) assert created_kwargs["agent_id"] == "risk_manager" finally: evo_agent_module.EvoAgent = original_evo_agent def test_factory_creates_portfolio_manager(self, tmp_path, monkeypatch): """Test that factory creates portfolio manager EvoAgent with financial params.""" from backend.agents.unified_factory import UnifiedAgentFactory class MockSkillsManager: def get_agent_asset_dir(self, config_name, agent_id): path = tmp_path / "runs" / config_name / "agents" / agent_id path.mkdir(parents=True, exist_ok=True) return path factory = UnifiedAgentFactory( config_name="test_config", skills_manager=MockSkillsManager(), ) created_kwargs = {} def mock_make_decision(*args, **kwargs): pass class MockEvoAgent: def __init__(self, **kwargs): created_kwargs.update(kwargs) self.toolkit = None # Add _make_decision for PM toolkit registration self._make_decision = mock_make_decision import backend.agents.base.evo_agent as evo_agent_module original_evo_agent = evo_agent_module.EvoAgent evo_agent_module.EvoAgent = MockEvoAgent try: agent = factory.create_portfolio_manager( model=MagicMock(), formatter=MagicMock(), initial_cash=50000.0, margin_requirement=0.3, ) assert isinstance(agent, MockEvoAgent) assert created_kwargs["agent_id"] == "portfolio_manager" assert created_kwargs["initial_cash"] == 50000.0 assert created_kwargs["margin_requirement"] == 0.3 finally: evo_agent_module.EvoAgent = original_evo_agent def test_factory_respects_evo_agent_ids_env(self, monkeypatch, tmp_path): """Test that factory respects EVO_AGENT_IDS environment variable.""" from backend.agents.unified_factory import UnifiedAgentFactory # Only enable technical_analyst as EvoAgent monkeypatch.setenv("EVO_AGENT_IDS", "technical_analyst") class MockSkillsManager: def get_agent_asset_dir(self, config_name, agent_id): path = tmp_path / "runs" / config_name / "agents" / agent_id path.mkdir(parents=True, exist_ok=True) return path factory = UnifiedAgentFactory( config_name="test_config", skills_manager=MockSkillsManager(), ) # technical_analyst should use EvoAgent assert factory._should_use_evo_agent("technical_analyst") is True # fundamentals_analyst should use legacy assert factory._should_use_evo_agent("fundamentals_analyst") is False def test_factory_legacy_mode_disables_evo_agent(self, monkeypatch): """Test that EVO_AGENT_IDS=legacy disables all EvoAgents.""" from backend.agents.unified_factory import UnifiedAgentFactory monkeypatch.setenv("EVO_AGENT_IDS", "legacy") factory = UnifiedAgentFactory( config_name="test_config", skills_manager=MagicMock(), ) assert factory._evo_agent_ids == set() assert factory._should_use_evo_agent("any_agent") is False class TestToolGuardIntegration: """Test ToolGuardMixin integration with EvoAgent.""" def test_tool_guard_intercepts_guarded_tools(self): """Test that ToolGuard intercepts tools requiring approval.""" from backend.agents.base.tool_guard import ToolGuardMixin class TestAgent(ToolGuardMixin): def __init__(self): self._init_tool_guard() self.agent_id = "test_agent" self.workspace_id = "test_workspace" self.session_id = "test_session" agent = TestAgent() # Verify place_order is in guarded tools assert agent._is_tool_guarded("place_order") is True assert agent._is_tool_denied("execute_shell_command") is True def test_tool_guard_approval_flow(self): """Test the full approval flow for a guarded tool.""" from backend.agents.base.tool_guard import ( ToolGuardStore, ApprovalStatus, ) store = ToolGuardStore() # Create a pending approval record record = store.create_pending( tool_name="place_order", tool_input={"ticker": "AAPL", "quantity": 100}, agent_id="test_agent", workspace_id="test_workspace", ) assert record.status == ApprovalStatus.PENDING assert record.tool_name == "place_order" # Approve the request with resolved_by updated = store.set_status(record.approval_id, ApprovalStatus.APPROVED, resolved_by="test_user") assert updated.status == ApprovalStatus.APPROVED assert updated.resolved_by == "test_user" def test_tool_guard_default_lists(self): """Test default guarded and denied tool lists.""" from backend.agents.base.tool_guard import ( DEFAULT_GUARDED_TOOLS, DEFAULT_DENIED_TOOLS, ) # Critical tools should be guarded assert "place_order" in DEFAULT_GUARDED_TOOLS assert "modify_position" in DEFAULT_GUARDED_TOOLS assert "write_file" in DEFAULT_GUARDED_TOOLS assert "edit_file" in DEFAULT_GUARDED_TOOLS # Dangerous tools should be denied assert "execute_shell_command" in DEFAULT_DENIED_TOOLS class TestEvoAgentWorkspaceIntegration: """Test EvoAgent workspace-driven configuration.""" def test_evo_agent_loads_prompt_files_from_workspace(self, tmp_path, monkeypatch): """Test that EvoAgent loads prompt files from workspace directory.""" from backend.agents.base.evo_agent import EvoAgent workspace_dir = tmp_path / "runs" / "demo" / "agents" / "test_analyst" workspace_dir.mkdir(parents=True, exist_ok=True) # Create prompt files (workspace_dir / "SOUL.md").write_text( "You are a test analyst.", encoding="utf-8" ) (workspace_dir / "INSTRUCTIONS.md").write_text( "Additional instructions.", encoding="utf-8" ) class MockToolkit: def __init__(self, *args, **kwargs): pass def register_agent_skill(self, path): pass monkeypatch.setattr( "backend.agents.base.evo_agent.Toolkit", MockToolkit, ) class MockSkillsManager: def get_agent_active_root(self, config_name, agent_id): return workspace_dir / "skills" / "active" def list_active_skill_metadata(self, config_name, agent_id): return [] agent = EvoAgent( agent_id="test_analyst", config_name="demo", workspace_dir=workspace_dir, model=MagicMock(), formatter=MagicMock(), skills_manager=MockSkillsManager(), prompt_files=["SOUL.md", "INSTRUCTIONS.md"], ) # Verify prompts are loaded into system prompt assert "You are a test analyst." in agent._sys_prompt assert "Additional instructions." in agent._sys_prompt class TestFactoryCaching: """Test UnifiedAgentFactory caching behavior.""" def test_factory_cache_per_config(self, monkeypatch): """Test that factory is cached per config name.""" from backend.agents.unified_factory import ( get_agent_factory, clear_factory_cache, ) # Clear any existing cache clear_factory_cache() mock_skills_manager = MagicMock() factory1 = get_agent_factory("config_a", mock_skills_manager) factory2 = get_agent_factory("config_a", mock_skills_manager) factory3 = get_agent_factory("config_b", mock_skills_manager) # Same config should return same instance assert factory1 is factory2 # Different config should return different instance assert factory1 is not factory3 def test_clear_factory_cache(self): """Test that clear_factory_cache removes all cached factories.""" from backend.agents.unified_factory import ( get_agent_factory, clear_factory_cache, ) mock_skills_manager = MagicMock() factory1 = get_agent_factory("config_c", mock_skills_manager) clear_factory_cache() factory2 = get_agent_factory("config_c", mock_skills_manager) # After clearing cache, should be new instance assert factory1 is not factory2 class TestDeprecationWarnings: """Test that legacy agents emit deprecation warnings.""" def test_risk_agent_emits_deprecation_warning(self): """Test that RiskAgent emits deprecation warning on import.""" import warnings import sys # Clear cache to force reimport modules_to_remove = [ k for k in sys.modules.keys() if k.endswith("risk_manager") and "backend.agents" in k ] for m in modules_to_remove: del sys.modules[m] with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") from backend.agents.risk_manager import RiskAgent deprecation_warnings = [ x for x in w if issubclass(x.category, DeprecationWarning) ] assert any("RiskAgent is deprecated" in str(x.message) for x in deprecation_warnings) def test_pm_agent_emits_deprecation_warning(self): """Test that PMAgent emits deprecation warning on import.""" import warnings import sys # Clear cache to force reimport modules_to_remove = [ k for k in sys.modules.keys() if k.endswith("portfolio_manager") and "backend.agents" in k ] for m in modules_to_remove: del sys.modules[m] with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") from backend.agents.portfolio_manager import PMAgent deprecation_warnings = [ x for x in w if issubclass(x.category, DeprecationWarning) ] assert any("PMAgent is deprecated" in str(x.message) for x in deprecation_warnings)