""" Tests for Oncall/Runbook Tool """ import pytest import os import sys import json import tempfile from unittest.mock import AsyncMock, patch, MagicMock sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from services.router.tool_manager import ToolManager, ToolResult class TestOncallToolServicesList: """Test services_list action""" @pytest.mark.asyncio async def test_services_list(self): """Test getting services list from docker-compose""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: # Create a mock docker-compose file compose_file = os.path.join(tmpdir, "docker-compose.yml") with open(compose_file, 'w') as f: f.write("""services: router: image: router:latest gateway: image: gateway:latest """) with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "services_list" }, agent_id="sofiia") assert result.success is True assert "services" in result.result class TestOncallToolRunbook: """Test runbook actions""" @pytest.mark.asyncio async def test_runbook_search(self): """Test searching runbooks""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: # Create ops directory with runbook ops_dir = os.path.join(tmpdir, "ops") os.makedirs(ops_dir) runbook_file = os.path.join(ops_dir, "deploy.md") with open(runbook_file, 'w') as f: f.write("# Deployment Runbook\n\nInstructions for deploying.") with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "runbook_search", "params": {"query": "deploy"} }, agent_id="sofiia") assert result.success is True assert "results" in result.result @pytest.mark.asyncio async def test_runbook_read_path_traversal_blocked(self): """Test that path traversal is blocked""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "runbook_read", "params": {"runbook_path": "../etc/passwd"} }, agent_id="sofiia") assert result.success is False assert "traversal" in result.error.lower() or "not found" in result.error.lower() @pytest.mark.asyncio async def test_runbook_read_not_in_allowed_dir(self): """Test that reading outside allowed dirs is blocked""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: # Create a file outside allowed dirs outside_file = os.path.join(tmpdir, "secret.txt") with open(outside_file, 'w') as f: f.write("API_KEY=secret123") with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "runbook_read", "params": {"runbook_path": "secret.txt"} }, agent_id="sofiia") # Should fail because not in allowed dirs assert result.success is False @pytest.mark.asyncio async def test_runbook_secret_masking(self): """Test that secrets are masked in runbook content""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: # Create ops directory with runbook containing secrets ops_dir = os.path.join(tmpdir, "ops") os.makedirs(ops_dir) runbook_file = os.path.join(ops_dir, "config.md") with open(runbook_file, 'w') as f: f.write("""# Config API_KEY = sk-live-secret123 password = mypassword token = abc123 """) with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "runbook_read", "params": {"runbook_path": "ops/config.md"} }, agent_id="sofiia") assert result.success is True content = result.result.get("content", "") # Secrets should be masked assert "sk-live-secret123" not in content assert "***" in content class TestOncallToolIncidents: """Test incident log actions""" @pytest.mark.asyncio async def test_incident_log_list(self): """Test listing incidents""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: # Create ops directory with incidents ops_dir = os.path.join(tmpdir, "ops") os.makedirs(ops_dir) incidents_file = os.path.join(ops_dir, "incidents.jsonl") with open(incidents_file, 'w') as f: f.write(json.dumps({"ts": "2024-01-01", "severity": "sev1", "title": "Test"}) + "\n") with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "incident_log_list", "params": {} }, agent_id="sofiia") assert result.success is True assert "incidents" in result.result @pytest.mark.asyncio async def test_incident_log_append_allowed_for_sofiia(self): """Test that Sofiia can append incidents""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: ops_dir = os.path.join(tmpdir, "ops") with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "incident_log_append", "params": { "incident_title": "Test incident", "incident_severity": "sev2", "incident_details": "Test details" } }, agent_id="sofiia") assert result.success is True @pytest.mark.asyncio async def test_incident_log_append_blocked_for_regular_agent(self): """Test that regular agents cannot append incidents""" tool_mgr = ToolManager({}) result = await tool_mgr._oncall_tool({ "action": "incident_log_append", "params": { "incident_title": "Test incident", "incident_severity": "sev2" } }, agent_id="regular_agent") assert result.success is False assert "entitlement" in result.error.lower() or "require" in result.error.lower() class TestOncallToolDeployments: """Test deployment actions""" @pytest.mark.asyncio async def test_deployments_recent(self): """Test getting recent deployments""" tool_mgr = ToolManager({}) with tempfile.TemporaryDirectory() as tmpdir: # Create ops directory with deployments ops_dir = os.path.join(tmpdir, "ops") os.makedirs(ops_dir) deploy_file = os.path.join(ops_dir, "deployments.jsonl") with open(deploy_file, 'w') as f: f.write(json.dumps({"ts": "2024-01-01", "service": "router", "version": "1.0.0"}) + "\n") with patch.dict(os.environ, {"REPO_ROOT": tmpdir}): result = await tool_mgr._oncall_tool({ "action": "deployments_recent", "params": {} }, agent_id="sofiia") assert result.success is True assert "deployments" in result.result class TestOncallToolHealth: """Test health check action""" @pytest.mark.asyncio async def test_service_health_not_in_allowlist(self): """Test that health check to non-allowlisted host is blocked""" tool_mgr = ToolManager({}) result = await tool_mgr._oncall_tool({ "action": "service_health", "params": { "service_name": "evil-service", "health_endpoint": "http://evil.com/health" } }, agent_id="sofiia") assert result.success is False assert "allowlist" in result.error.lower() if __name__ == "__main__": pytest.main([__file__, "-v"])