129e4ea1fc
New router intelligence modules (26 files): alert_ingest/store, audit_store, architecture_pressure, backlog_generator/store, cost_analyzer, data_governance, dependency_scanner, drift_analyzer, incident_* (5 files), llm_enrichment, platform_priority_digest, provider_budget, release_check_runner, risk_* (6 files), signature_state_store, sofiia_auto_router, tool_governance New services: - sofiia-console: Dockerfile, adapters/, monitor/nodes/ops/voice modules, launchd, react static - memory-service: integration_endpoints, integrations, voice_endpoints, static UI - aurora-service: full app suite (analysis, job_store, orchestrator, reporting, schemas, subagents) - sofiia-supervisor: new supervisor service - aistalk-bridge-lite: Telegram bridge lite - calendar-service: CalDAV calendar service with reminders - mlx-stt-service / mlx-tts-service: Apple Silicon speech services - binance-bot-monitor: market monitor service - node-worker: STT/TTS memory providers New tools (9): agent_email, browser_tool, contract_tool, observability_tool, oncall_tool, pr_reviewer_tool, repo_tool, safe_code_executor, secure_vault New crews: agromatrix_crew (10 modules: depth_classifier, doc_facts, doc_focus, farm_state, light_reply, llm_factory, memory_manager, proactivity, reflection_engine, session_context, style_adapter, telemetry) Tests: 85+ test files for all new modules Made-with: Cursor
175 lines
4.0 KiB
Python
175 lines
4.0 KiB
Python
"""
|
|
Unit tests for SafeCodeExecutor
|
|
"""
|
|
|
|
import sys
|
|
sys.path.insert(0, "..")
|
|
|
|
from safe_code_executor import SafeCodeExecutor
|
|
|
|
|
|
def test_basic_python_execution():
|
|
"""Test basic Python execution"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
result = executor.execute(
|
|
language="python",
|
|
code="print('Hello, World!')"
|
|
)
|
|
|
|
assert result["status"] == "succeeded"
|
|
assert "Hello, World!" in result["stdout"]
|
|
print("✓ Basic execution works")
|
|
|
|
|
|
def test_json_transform():
|
|
"""Test JSON transformation"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
code = '''
|
|
import json
|
|
data = {"a": 1, "b": 2, "c": 3}
|
|
result = {"sum": data["a"] + data["b"] + data["c"]}
|
|
print(json.dumps(result))
|
|
'''
|
|
|
|
result = executor.execute(language="python", code=code)
|
|
|
|
assert result["status"] == "succeeded"
|
|
assert result["result_json"]["sum"] == 6
|
|
print("✓ JSON transform works")
|
|
|
|
|
|
def test_regex_parse():
|
|
"""Test regex parsing"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
code = '''
|
|
import re
|
|
text = "Email: test@example.com"
|
|
match = re.search(r'[\w.-]+@[\w.-]+', text)
|
|
print(match.group() if match else "None")
|
|
'''
|
|
|
|
result = executor.execute(language="python", code=code)
|
|
|
|
assert result["status"] == "succeeded"
|
|
assert "test@example.com" in result["stdout"]
|
|
print("✓ Regex parse works")
|
|
|
|
|
|
def test_validation_blocks_os_import():
|
|
"""Test that os import is blocked"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
result = executor.execute(
|
|
language="python",
|
|
code="import os; print(os.getcwd())"
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert "import os" in result["error"] or "os" in result["error"]
|
|
print("✓ OS import blocked")
|
|
|
|
|
|
def test_validation_blocks_subprocess():
|
|
"""Test that subprocess is blocked"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
result = executor.execute(
|
|
language="python",
|
|
code="import subprocess; subprocess.run(['ls'])"
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
print("✓ Subprocess blocked")
|
|
|
|
|
|
def test_validation_blocks_socket():
|
|
"""Test that socket is blocked"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
result = executor.execute(
|
|
language="python",
|
|
code="import socket; s = socket.socket()"
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
print("✓ Socket blocked")
|
|
|
|
|
|
def test_validation_blocks_eval():
|
|
"""Test that eval is blocked"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
result = executor.execute(
|
|
language="python",
|
|
code="eval('1+1')"
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
print("✓ Eval blocked")
|
|
|
|
|
|
def test_validation_blocks_file_io():
|
|
"""Test that file I/O is blocked"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
result = executor.execute(
|
|
language="python",
|
|
code="open('/etc/passwd').read()"
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
print("✓ File I/O blocked")
|
|
|
|
|
|
def test_math_operations():
|
|
"""Test math operations work"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
code = '''
|
|
import math
|
|
print(math.sqrt(16))
|
|
print(math.pi)
|
|
'''
|
|
|
|
result = executor.execute(language="python", code=code)
|
|
|
|
assert result["status"] == "succeeded"
|
|
print("✓ Math operations work")
|
|
|
|
|
|
def test_statistics():
|
|
"""Test statistics module"""
|
|
executor = SafeCodeExecutor()
|
|
|
|
code = '''
|
|
import statistics
|
|
data = [1, 2, 3, 4, 5]
|
|
print(statistics.mean(data))
|
|
print(statistics.median(data))
|
|
'''
|
|
|
|
result = executor.execute(language="python", code=code)
|
|
|
|
assert result["status"] == "succeeded"
|
|
print("✓ Statistics module works")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("=== Running Unit Tests ===\n")
|
|
|
|
test_basic_python_execution()
|
|
test_json_transform()
|
|
test_regex_parse()
|
|
test_validation_blocks_os_import()
|
|
test_validation_blocks_subprocess()
|
|
test_validation_blocks_socket()
|
|
test_validation_blocks_eval()
|
|
test_validation_blocks_file_io()
|
|
test_math_operations()
|
|
test_statistics()
|
|
|
|
print("\n✅ All unit tests passed!")
|