Katamari MCP Server

""" Tests for sandbox execution environment. """ import pytest import asyncio import sys from pathlib import Path # Add the project root to the path sys.path.insert(0, str(Path(__file__).parent.parent)) from katamari_mcp.acp.sandbox import SandboxConfig, PythonSandbox, CapabilitySandbox, SandboxResult class TestSandboxConfig: """Test sandbox configuration.""" def test_default_config(self): """Test default configuration values.""" config = SandboxConfig() assert config.max_cpu_time == 30 assert config.max_memory == 256 * 1024 * 1024 assert config.max_processes == 1 assert config.max_file_size == 10 * 1024 * 1024 assert not config.allow_network assert config.allow_file_write assert len(config.allowed_imports) > 0 assert 'asyncio' in config.allowed_imports assert 'os' in config.allowed_imports def test_allowed_imports(self): """Test allowed imports list.""" config = SandboxConfig() # Check core modules are allowed assert 'json' in config.allowed_imports assert 'pathlib' in config.allowed_imports assert 'datetime' in config.allowed_imports # Check dangerous modules are not allowed assert 'subprocess' not in config.allowed_imports assert 'socket' not in config.allowed_imports class TestSandboxResult: """Test sandbox result object.""" def test_success_result(self): """Test successful result creation.""" result = SandboxResult( success=True, data="test_result", execution_time=0.5, resource_usage={"memory": 1024} ) assert result.success is True assert result.data == "test_result" assert result.execution_time == 0.5 assert result.resource_usage["memory"] == 1024 assert result.error is None def test_error_result(self): """Test error result creation.""" result = SandboxResult( success=False, error="Test error", execution_time=0.1 ) assert result.success is False assert result.error == "Test error" assert result.execution_time == 0.1 assert result.data is None assert result.resource_usage == {} class TestPythonSandbox: """Test Python sandbox execution.""" @pytest.fixture def sandbox(self): """Create sandbox instance for testing.""" config = SandboxConfig() config.max_cpu_time = 5 # Shorter timeout for tests return PythonSandbox(config) @pytest.mark.asyncio async def test_simple_execution(self, sandbox): """Test simple code execution.""" code = """ result = "Hello from sandbox" """ result = await sandbox.execute(code) assert isinstance(result, SandboxResult) assert result.success is True assert result.data == "Hello from sandbox" assert result.execution_time > 0 @pytest.mark.asyncio async def test_math_execution(self, sandbox): """Test mathematical operations.""" code = """ import math result = math.sqrt(16) + math.pow(2, 3) """ result = await sandbox.execute(code) assert result.success is True assert result.data == 8.0 # sqrt(16)=4, pow(2,3)=8, total=12 @pytest.mark.asyncio async def test_syntax_error(self, sandbox): """Test handling of syntax errors.""" code = """ result = "unclosed string """ result = await sandbox.execute(code) assert result.success is False assert "SyntaxError" in result.error or "syntax error" in result.error.lower() @pytest.mark.asyncio async def test_runtime_error(self, sandbox): """Test handling of runtime errors.""" code = """ result = 1 / 0 """ result = await sandbox.execute(code) assert result.success is False assert "ZeroDivisionError" in result.error or "division by zero" in result.error.lower() @pytest.mark.asyncio async def test_import_restriction(self, sandbox): """Test that dangerous imports are blocked.""" code = """ import subprocess result = "should not reach here" """ result = await sandbox.execute(code) assert result.success is False assert "ImportError" in result.error or "not allowed" in result.error.lower() @pytest.mark.asyncio async def test_allowed_imports(self, sandbox): """Test that allowed imports work.""" code = """ import json import datetime result = {"json_loaded": json.loads('{"test": true}'), "now": str(datetime.datetime.now())} """ result = await sandbox.execute(code) assert result.success is True assert isinstance(result.data, dict) assert "json_loaded" in result.data assert "now" in result.data @pytest.mark.asyncio async def test_globals_access(self, sandbox): """Test access to provided globals.""" code = """ result = parameters["input"] * 2 """ globals_dict = {"parameters": {"input": 21}} result = await sandbox.execute(code, globals_dict) assert result.success is True assert result.data == 42 class TestCapabilitySandbox: """Test high-level capability sandbox interface.""" @pytest.fixture def capability_sandbox(self): """Create capability sandbox instance.""" config = SandboxConfig() config.max_cpu_time = 5 return CapabilitySandbox(config) @pytest.mark.asyncio async def test_capability_execution(self, capability_sandbox): """Test capability execution with parameters.""" code = """ def process_data(data): return [x * 2 for x in data] result = process_data(parameters.get("numbers", [])) """ result = await capability_sandbox.execute_capability( code, capability_name="test_doubler", parameters={"numbers": [1, 2, 3, 4]} ) assert result.success is True assert result.data == [2, 4, 6, 8] @pytest.mark.asyncio async def test_capability_validation(self, capability_sandbox): """Test capability code validation.""" # Valid code valid_code = """ import json result = json.dumps({"status": "ok"}) """ issues = await capability_sandbox.validate_capability(valid_code) assert len(issues) == 0 # Invalid code invalid_code = """ import subprocess result = subprocess.run(["ls"], capture_output=True) """ issues = await capability_sandbox.validate_capability(invalid_code) assert len(issues) > 0 assert any("subprocess" in issue for issue in issues) # Syntax error syntax_error_code = """ result = "unclosed string """ issues = await capability_sandbox.validate_capability(syntax_error_code) assert len(issues) > 0 assert any("syntax" in issue.lower() for issue in issues) def test_sandbox_status(self, capability_sandbox): """Test sandbox status reporting.""" status = capability_sandbox.get_sandbox_status() assert "config" in status assert "temp_dir" in status config = status["config"] assert config["max_cpu_time"] == 5 assert config["max_memory"] == 256 * 1024 * 1024 assert config["max_processes"] == 1 assert not config["allow_network"] assert config["allowed_imports"] > 0 class TestSandboxIntegration: """Integration tests for sandbox functionality.""" @pytest.mark.asyncio async def test_full_capability_lifecycle(self): """Test complete capability lifecycle from validation to execution.""" config = SandboxConfig() config.max_cpu_time = 10 sandbox = CapabilitySandbox(config) # Step 1: Define capability code capability_code = ''' import json from datetime import datetime def analyze_data(data): """Simple data analysis capability.""" if not isinstance(data, list): raise ValueError("Data must be a list") return { "count": len(data), "sum": sum(data) if all(isinstance(x, (int, float)) for x in data) else None, "average": sum(data) / len(data) if all(isinstance(x, (int, float)) for x in data) else None, "timestamp": datetime.now().isoformat() } # Execute the analysis result = analyze_data(parameters.get("data", [])) ''' # Step 2: Validate the capability issues = await sandbox.validate_capability(capability_code) assert len(issues) == 0, f"Validation failed: {issues}" # Step 3: Execute the capability execution_result = await sandbox.execute_capability( capability_code, capability_name="data_analyzer", parameters={"data": [1, 2, 3, 4, 5]} ) # Step 4: Verify results assert execution_result.success is True assert isinstance(execution_result.data, dict) assert execution_result.data["count"] == 5 assert execution_result.data["sum"] == 15 assert execution_result.data["average"] == 3.0 assert "timestamp" in execution_result.data assert execution_result.execution_time > 0 assert len(execution_result.resource_usage) > 0 @pytest.mark.asyncio async def test_error_handling_and_recovery(self): """Test sandbox error handling and recovery.""" sandbox = CapabilitySandbox() # Execute code that will fail failing_code = """ import os result = os.system("echo 'This should fail'") """ result = await sandbox.execute_capability(failing_code) assert result.success is False # Verify sandbox can still execute valid code after failure valid_code = """ result = "recovery test" """ recovery_result = await sandbox.execute_capability(valid_code) assert recovery_result.success is True assert recovery_result.data == "recovery test" if __name__ == "__main__": pytest.main([__file__, "-v"])