120 lines
5.5 KiB
Python
120 lines
5.5 KiB
Python
import unittest
|
|
from unittest.mock import MagicMock, AsyncMock
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
# Add the opax-mcp directory to the path to simulate the Docker container's layout
|
|
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
|
|
|
from contracts.common import (
|
|
EmmaConversation, MemoryRecord, ProposedToolAction, ApprovalRecord, ToolRiskLevel, MemoryType
|
|
)
|
|
from policy.caller_context import derive_caller_context
|
|
from policy.tool_policy import ToolPolicyEngine
|
|
from persistence.memory_store import InMemoryMemoryStore
|
|
|
|
|
|
class TestPhase2AContracts(unittest.TestCase):
|
|
|
|
def test_models_have_schema_version(self):
|
|
"""(C.4) Test that contracts have the schema_version field."""
|
|
convo = EmmaConversation(owner_id="o", workspace_id="w", created_by="c")
|
|
self.assertEqual(convo.schema_version, "1.0.0")
|
|
|
|
action = ProposedToolAction(
|
|
correlation_id="c1", tool_name="t1", parameters={},
|
|
human_readable_summary="s1", risk_class="read_only", requires_approval=False
|
|
)
|
|
self.assertEqual(action.schema_version, "1.0.0")
|
|
|
|
def test_no_mutable_defaults_in_lists(self):
|
|
"""Verifies that list fields are unique to each model instance."""
|
|
c1 = EmmaConversation(owner_id="o", workspace_id="w", created_by="c")
|
|
c2 = EmmaConversation(owner_id="o", workspace_id="w", created_by="c")
|
|
c1.message_references.append("test")
|
|
self.assertNotEqual(c1.message_references, c2.message_references)
|
|
self.assertEqual(c2.message_references, [])
|
|
|
|
def test_datetimes_are_timezone_aware(self):
|
|
"""Verifies that default datetimes are timezone-aware."""
|
|
c = EmmaConversation(owner_id="o", workspace_id="w", created_by="c")
|
|
self.assertIsNotNone(c.created_at.tzinfo)
|
|
|
|
|
|
class TestPhase2APolicy(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.policy_engine = ToolPolicyEngine()
|
|
|
|
def test_caller_context_derivation_and_least_privilege(self):
|
|
"""Tests that known callers get correct profiles and unknown callers get a safe, read-only default."""
|
|
op_context = derive_caller_context("agent:perplexity")
|
|
self.assertEqual(op_context.profile, "operator")
|
|
|
|
unknown_context = derive_caller_context("some-random-key")
|
|
self.assertEqual(unknown_context.profile, "readonly")
|
|
self.assertEqual(unknown_context.allowed_tool_policy['default'], "read_only")
|
|
|
|
def test_tool_policy_engine_enforces_all_levels(self):
|
|
"""Tests that the policy engine correctly maps decisions for different user profiles and tool risks."""
|
|
admin = derive_caller_context("user:chris.christiansen@vauco.no")
|
|
operator = derive_caller_context("agent:perplexity")
|
|
readonly = derive_caller_context("unknown:caller")
|
|
|
|
# Test a read_only tool
|
|
self.assertEqual(self.policy_engine.get_access_decision(admin, "get_file"), "allowed_directly")
|
|
self.assertEqual(self.policy_engine.get_access_decision(operator, "get_file"), "allowed_directly")
|
|
self.assertEqual(self.policy_engine.get_access_decision(readonly, "get_file"), "allowed_directly")
|
|
|
|
# Test a requires_approval tool
|
|
self.assertEqual(self.policy_engine.get_access_decision(admin, "create_issue"), "requires_approval")
|
|
self.assertEqual(self.policy_engine.get_access_decision(operator, "create_issue"), "requires_approval")
|
|
self.assertEqual(self.policy_engine.get_access_decision(readonly, "create_issue"), "denied")
|
|
|
|
# Test a requires_high_approval tool
|
|
self.assertEqual(self.policy_engine.get_access_decision(admin, "deploy_revision"), "requires_high_approval")
|
|
self.assertEqual(self.policy_engine.get_access_decision(operator, "deploy_revision"), "denied")
|
|
|
|
# Test a propose_only tool
|
|
self.assertEqual(self.policy_engine.get_access_decision(operator, "run_terminal"), "propose_only")
|
|
self.assertEqual(self.policy_engine.get_access_decision(admin, "run_terminal"), "denied") # Admins are not configured for this
|
|
|
|
# Test a forbidden tool
|
|
self.assertEqual(self.policy_engine.get_access_decision(admin, "read_secret_value"), "denied")
|
|
|
|
|
|
class TestPhase2AMemoryStore(unittest.IsolatedAsyncioTestCase):
|
|
|
|
def setUp(self):
|
|
self.store = InMemoryMemoryStore()
|
|
self.action = ProposedToolAction(
|
|
correlation_id="c1", tool_name="t1", parameters={},
|
|
human_readable_summary="s1", risk_class="read_only", requires_approval=False
|
|
)
|
|
|
|
async def test_in_memory_store_enforces_owner_id(self):
|
|
"""Tests that data is strictly partitioned by owner_id."""
|
|
convo1 = EmmaConversation(owner_id="owner1", workspace_id="ws1", created_by="owner1")
|
|
await self.store.save_conversation(convo1)
|
|
|
|
# Owner 1 can retrieve
|
|
self.assertIsNotNone(await self.store.get_conversation(convo1.conversation_id, "owner1"))
|
|
# Owner 2 cannot retrieve
|
|
self.assertIsNone(await self.store.get_conversation(convo1.conversation_id, "owner2"))
|
|
|
|
async def test_approval_record_instantiation(self):
|
|
"""(C.2) Tests that ApprovalRecord can be created with a real ProposedToolAction."""
|
|
record = ApprovalRecord(
|
|
action=self.action,
|
|
expires_at=datetime.now(timezone.utc),
|
|
requested_by_caller_id="caller1",
|
|
workspace_id="ws1"
|
|
)
|
|
self.assertEqual(record.action.tool_name, "t1")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|