98 lines
4.6 KiB
Python
98 lines
4.6 KiB
Python
import unittest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "opax-mcp"))
|
|
|
|
from contracts.common import (
|
|
EmmaConversation, MemoryRecord, AuditEvent, ToolRiskLevel
|
|
)
|
|
from persistence.memory_store import InMemoryMemoryStore
|
|
from persistence.firestore_store import FirestoreMemoryStore
|
|
from persistence.memory_scope import calculate_effective_memory_scope
|
|
from policy.caller_context import derive_caller_context
|
|
|
|
|
|
class TestPhase2BFirestoreStore(unittest.IsolatedAsyncioTestCase):
|
|
|
|
def setUp(self):
|
|
self.mock_firestore_client = MagicMock()
|
|
self.mock_firestore_client.collection.return_value.document.return_value.set = AsyncMock()
|
|
self.mock_firestore_client.collection.return_value.add = AsyncMock()
|
|
self.mock_firestore_client.collection.return_value.document.return_value.create = AsyncMock()
|
|
|
|
self.store = FirestoreMemoryStore(project_id="test-project", client=self.mock_firestore_client)
|
|
|
|
async def test_firestore_client_is_injected(self):
|
|
"""(1) Verifies the Firestore client is used when provided."""
|
|
self.assertIs(self.store._client, self.mock_firestore_client)
|
|
|
|
async def test_create_conversation_uses_correct_path(self):
|
|
"""(2) Verifies conversation is stored under the correct collection/document ID."""
|
|
convo = await self.store.create_conversation("owner1", "ws1", "creator1")
|
|
|
|
self.mock_firestore_client.collection.assert_called_with("emma_conversations")
|
|
collection_mock = self.mock_firestore_client.collection.return_value
|
|
collection_mock.document.assert_called_with(convo.conversation_id)
|
|
document_mock = collection_mock.document.return_value
|
|
document_mock.set.assert_awaited_once()
|
|
|
|
async def test_append_message_uses_subcollection(self):
|
|
"""(3) Verifies messages are stored as a subcollection."""
|
|
convo_id = "convo-123"
|
|
message = {"role": "user", "content": "hello"}
|
|
await self.store.append_message(convo_id, message)
|
|
|
|
# Check path to messages subcollection
|
|
self.mock_firestore_client.collection.assert_called_with("emma_conversations")
|
|
convo_collection_mock = self.mock_firestore_client.collection.return_value
|
|
convo_collection_mock.document.assert_called_with(convo_id)
|
|
doc_mock = convo_collection_mock.document.return_value
|
|
doc_mock.collection.assert_called_with("messages")
|
|
|
|
# Check that add was called on the subcollection
|
|
messages_collection_mock = doc_mock.collection.return_value
|
|
messages_collection_mock.add.assert_awaited_once_with(message)
|
|
|
|
async def test_audit_event_is_append_only(self):
|
|
"""(E) Verifies audit events use create() to be append-only."""
|
|
event = AuditEvent(event_type="test", caller_id="c1", details={}, workspace_id="ws1")
|
|
await self.store.record_audit_event(event)
|
|
|
|
self.mock_firestore_client.collection.assert_called_with("emma_audit_events")
|
|
collection_mock = self.mock_firestore_client.collection.return_value
|
|
collection_mock.document.assert_called_with(event.event_id)
|
|
document_mock = collection_mock.document.return_value
|
|
document_mock.create.assert_awaited_once() # Must use create(), not set()
|
|
document_mock.set.assert_not_called()
|
|
|
|
|
|
class TestPhase2BMemoryScope(unittest.TestCase):
|
|
|
|
def test_effective_scope_logic(self):
|
|
"""(D) Tests the effective memory scope calculation."""
|
|
admin_caller = derive_caller_context("user:chris.christiansen@vauco.no")
|
|
operator_caller = derive_caller_context("agent:perplexity")
|
|
|
|
convo = MagicMock()
|
|
|
|
# Admin requests all, gets all allowed for admin
|
|
requested = {"current_conversation", "owner_private_memory", "workspace_operational_memory"}
|
|
effective = calculate_effective_memory_scope(requested, admin_caller, convo)
|
|
self.assertEqual(effective, {"current_conversation", "owner_private_memory", "workspace_operational_memory"})
|
|
|
|
# Operator requests all, gets only what operator is allowed
|
|
effective = calculate_effective_memory_scope(requested, operator_caller, convo)
|
|
self.assertEqual(effective, {"current_conversation", "workspace_operational_memory"})
|
|
|
|
# Operator requests only conversation, gets only conversation
|
|
requested = {"current_conversation"}
|
|
effective = calculate_effective_memory_scope(requested, operator_caller, convo)
|
|
self.assertEqual(effective, {"current_conversation"})
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|