170 lines
7.5 KiB
Python
170 lines
7.5 KiB
Python
import unittest
|
|
import ast
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
# Ensure the project root is in the path for imports
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
|
|
|
from capability_bridge import build_capability_system_context
|
|
from capability_registry import (
|
|
list_capabilities,
|
|
Capability,
|
|
Availability,
|
|
)
|
|
|
|
|
|
class TestCapabilityBridge(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.context_output = build_capability_system_context()
|
|
|
|
def test_output_is_deterministic(self):
|
|
self.assertEqual(self.context_output, build_capability_system_context())
|
|
|
|
def test_output_has_all_three_headings(self):
|
|
self.assertIn("\nAvailable now:", self.context_output)
|
|
self.assertIn("\nPlanned, but not active in this chat:", self.context_output)
|
|
self.assertIn("\nUnavailable:", self.context_output)
|
|
|
|
def test_available_section_contains_correct_capabilities(self):
|
|
active_capabilities = [
|
|
c for c in list_capabilities() if c.availability == Availability.ACTIVE
|
|
]
|
|
self.assertIn("- Emma chat: Conversational reasoning through Emma.", self.context_output)
|
|
self.assertIn("- Analyze user-provided text: Analyze, summarize, and structure text provided in the current request.", self.context_output)
|
|
|
|
# Isolate the "Available now" section to count items accurately
|
|
start_index = self.context_output.find("Available now:")
|
|
end_index = self.context_output.find("\n\nPlanned, but not active in this chat:")
|
|
available_section = self.context_output[start_index:end_index]
|
|
|
|
self.assertEqual(available_section.count("- "), len(active_capabilities))
|
|
|
|
def test_planned_section_contains_all_planned_capabilities(self):
|
|
planned_capabilities = [
|
|
c for c in list_capabilities() if c.availability == Availability.PLANNED
|
|
]
|
|
self.assertEqual(len(planned_capabilities), 17)
|
|
|
|
planned_capability_ids = {c.id for c in planned_capabilities}
|
|
self.assertIn("gitea.list_repo_files", planned_capability_ids)
|
|
|
|
for capability in planned_capabilities:
|
|
self.assertIn(f"- {capability.display_name}: {capability.description}", self.context_output)
|
|
|
|
def test_unavailable_section_is_correct(self):
|
|
self.assertIn("- Arbitrary terminal access is not available.", self.context_output)
|
|
# Ensure no other forbidden capabilities are listed by name
|
|
self.assertNotIn("Arbitrary terminal shell:", self.context_output)
|
|
|
|
def test_bridge_text_does_not_contain_forbidden_internal_details(self):
|
|
forbidden_substrings = [
|
|
# IDs
|
|
"emma.chat", "terminal.arbitrary_shell",
|
|
# Backend tool names
|
|
"run_emma", "list_commits", "get_file", "create_issue", "push_file",
|
|
"build_and_deploy_service",
|
|
# Internal metadata
|
|
"required_actor_scope", "approval_binding", "audit_required",
|
|
"rollback_required", "self_approval_forbidden",
|
|
# Target names / infrastructure
|
|
"git.vauco.no", "opax-mcp", "propane-will-491900-m5", "us-central1",
|
|
]
|
|
for substring in forbidden_substrings:
|
|
self.assertNotIn(substring, self.context_output)
|
|
|
|
def test_bridge_module_has_no_prohibited_imports(self):
|
|
bridge_path = REPO_ROOT / "opax-mcp" / "capability_bridge.py"
|
|
with open(bridge_path, "r", encoding="utf-8") as f:
|
|
tree = ast.parse(f.read())
|
|
|
|
forbidden_imports = {
|
|
"os", "subprocess", "requests", "httpx", "google", "gitea",
|
|
"firebase", "firestore",
|
|
}
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
self.assertNotIn(alias.name.split('.')[0], forbidden_imports)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.module:
|
|
self.assertNotIn(node.module.split('.')[0], forbidden_imports)
|
|
|
|
def test_bridge_does_not_mutate_registry(self):
|
|
# This is implicitly tested by the deterministic output and other tests,
|
|
# but a direct check confirms the source objects are not modified.
|
|
initial_capabilities = list_capabilities()
|
|
build_capability_system_context()
|
|
self.assertEqual(initial_capabilities, list_capabilities())
|
|
|
|
def test_bridge_output_includes_all_rules(self):
|
|
rules_block = (
|
|
"Rules:\n"
|
|
"- Planned capabilities are not available in this chat and must not be claimed as active.\n"
|
|
"- No tool, repository, deployment, memory, document retrieval, external service or state-changing action is activated by this capability summary.\n"
|
|
"- Future write or high-impact capabilities require a concrete, explicit, bound user approval before execution."
|
|
)
|
|
self.assertIn(rules_block, self.context_output)
|
|
|
|
def test_bridge_does_not_expose_backend_tool_names(self):
|
|
all_capabilities = list_capabilities()
|
|
tool_names = {c.backend_tool for c in all_capabilities if c.backend_tool}
|
|
for tool_name in tool_names:
|
|
self.assertNotIn(tool_name, self.context_output)
|
|
|
|
|
|
class TestServerIntegrationContract(unittest.TestCase):
|
|
|
|
def test_run_emma_integration_in_server_py(self):
|
|
server_path = REPO_ROOT / "opax-mcp" / "server.py"
|
|
with open(server_path, "r", encoding="utf-8") as f:
|
|
tree = ast.parse(f.read())
|
|
|
|
run_emma_func = None
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.AsyncFunctionDef) and node.name == "run_emma":
|
|
run_emma_func = node
|
|
break
|
|
|
|
self.assertIsNotNone(run_emma_func, "async def run_emma not found in server.py")
|
|
|
|
# 1. References build_capability_system_context
|
|
call_to_build_context = any(
|
|
isinstance(n, ast.Call) and isinstance(n.func, ast.Name) and n.func.id == 'build_capability_system_context'
|
|
for n in ast.walk(run_emma_func)
|
|
)
|
|
self.assertTrue(call_to_build_context, "run_emma does not call build_capability_system_context")
|
|
|
|
# 2. Passes system_context to canonical_emma.run
|
|
call_to_run = None
|
|
for node in ast.walk(run_emma_func):
|
|
if isinstance(node, ast.Call) and hasattr(node.func, 'value') and hasattr(node.func.value, 'id') and node.func.value.id == 'canonical_emma' and node.func.attr == 'run':
|
|
call_to_run = node
|
|
break
|
|
|
|
self.assertIsNotNone(call_to_run, "run_emma does not call canonical_emma.run")
|
|
|
|
has_system_context_kw = any(
|
|
kw.arg == 'system_context' for kw in call_to_run.keywords
|
|
)
|
|
self.assertTrue(has_system_context_kw, "system_context is not passed to canonical_emma.run")
|
|
|
|
# 3. Preserves history=[]
|
|
has_history_kw = any(
|
|
kw.arg == 'history' and isinstance(kw.value, ast.List) and not kw.value.elts
|
|
for kw in call_to_run.keywords
|
|
)
|
|
self.assertTrue(has_history_kw, "history=[] is not preserved in call to canonical_emma.run")
|
|
|
|
# 4. Preserves prompt argument structure
|
|
prompt_kw = next((kw for kw in call_to_run.keywords if kw.arg == 'prompt'), None)
|
|
self.assertIsNotNone(prompt_kw, "prompt argument missing in call to canonical_emma.run")
|
|
# A basic check to ensure it's calling p.get, not a simple variable
|
|
self.assertIsInstance(prompt_kw.value, ast.Call, "prompt argument is not a function call")
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|