from __future__ import annotations from pathlib import Path from django.test import TestCase, override_settings from core.mcp.tools import execute_tool, tool_specs from core.models import ( AIRequest, MCPToolAuditLog, MemoryItem, TaskProject, User, WorkspaceConversation, DerivedTask, DerivedTaskEvent, ) @override_settings(MEMORY_SEARCH_BACKEND="django") class MCPToolTests(TestCase): def setUp(self): self.user = User.objects.create_superuser( username="mcp-tools-user", email="mcp-tools@example.com", password="pw", ) self.conversation = WorkspaceConversation.objects.create( user=self.user, platform_type="signal", title="MCP Memory Scope", platform_thread_id="mcp-thread-1", ) request = AIRequest.objects.create( user=self.user, conversation=self.conversation, window_spec={}, operation="memory_propose", ) self.memory = MemoryItem.objects.create( user=self.user, conversation=self.conversation, memory_kind="fact", status="active", content={"text": "Prefers concise implementation notes."}, source_request=request, confidence_score=0.8, ) self.project = TaskProject.objects.create(user=self.user, name="MCP Project") self.task = DerivedTask.objects.create( user=self.user, project=self.project, title="Wire MCP server", source_service="signal", source_channel="team-chat", status_snapshot="open", immutable_payload={"scope": "memory"}, ) DerivedTaskEvent.objects.create( task=self.task, event_type="created", actor_identifier="agent", payload={"note": "task created"}, ) def test_tool_specs_include_memory_task_wiki_tools(self): names = {item["name"] for item in tool_specs()} self.assertIn("manticore.status", names) self.assertIn("memory.propose", names) self.assertIn("tasks.link_artifact", names) self.assertIn("wiki.create_article", names) self.assertIn("project.get_runbook", names) def test_manticore_query_and_tasks_tools(self): memory_payload = execute_tool( "manticore.query", {"user_id": self.user.id, "query": "concise implementation"}, ) self.assertGreaterEqual(int(memory_payload.get("count") or 0), 1) first_hit = (memory_payload.get("hits") or [{}])[0] self.assertEqual(str(self.memory.id), str(first_hit.get("memory_id"))) list_payload = execute_tool("tasks.list", {"user_id": self.user.id, "limit": 10}) self.assertEqual(1, int(list_payload.get("count") or 0)) self.assertEqual(str(self.task.id), str((list_payload.get("items") or [{}])[0].get("id"))) search_payload = execute_tool( "tasks.search", {"user_id": self.user.id, "query": "wire"}, ) self.assertEqual(1, int(search_payload.get("count") or 0)) events_payload = execute_tool("tasks.events", {"task_id": str(self.task.id), "limit": 5}) self.assertEqual(1, int(events_payload.get("count") or 0)) self.assertEqual("created", str((events_payload.get("items") or [{}])[0].get("event_type"))) def test_memory_proposal_review_flow(self): propose_payload = execute_tool( "memory.propose", { "user_id": self.user.id, "conversation_id": str(self.conversation.id), "memory_kind": "fact", "content": {"field": "likes", "text": "short status bullets"}, "reason": "Operator memory capture", "requested_by_identifier": "unit-test", }, ) request_id = str((propose_payload.get("request") or {}).get("id") or "") self.assertTrue(request_id) pending_payload = execute_tool("memory.pending", {"user_id": self.user.id}) self.assertGreaterEqual(int(pending_payload.get("count") or 0), 1) review_payload = execute_tool( "memory.review", { "user_id": self.user.id, "request_id": request_id, "decision": "approve", "reviewer_identifier": "admin", }, ) memory_data = review_payload.get("memory") or {} self.assertEqual("active", str(memory_data.get("status") or "")) list_payload = execute_tool( "memory.list", {"user_id": self.user.id, "query": "status bullets"}, ) self.assertGreaterEqual(int(list_payload.get("count") or 0), 1) def test_wiki_and_task_artifact_tools(self): article_create = execute_tool( "wiki.create_article", { "user_id": self.user.id, "title": "MCP Integration Notes", "markdown": "Initial setup steps.", "related_task_id": str(self.task.id), "tags": ["mcp", "memory"], "status": "published", }, ) article = article_create.get("article") or {} self.assertEqual("mcp-integration-notes", str(article.get("slug") or "")) article_update = execute_tool( "wiki.update_article", { "user_id": self.user.id, "article_id": str(article.get("id") or ""), "markdown": "Updated setup steps.", "approve_overwrite": True, "summary": "Expanded usage instructions.", }, ) revision = article_update.get("revision") or {} self.assertEqual(2, int(revision.get("revision") or 0)) list_payload = execute_tool( "wiki.list", {"user_id": self.user.id, "query": "integration"}, ) self.assertEqual(1, int(list_payload.get("count") or 0)) get_payload = execute_tool( "wiki.get", { "user_id": self.user.id, "article_id": str(article.get("id") or ""), "include_revisions": True, }, ) self.assertGreaterEqual(len(get_payload.get("revisions") or []), 2) note_payload = execute_tool( "tasks.create_note", { "task_id": str(self.task.id), "user_id": self.user.id, "note": "Implemented wiki tooling.", }, ) self.assertEqual("progress", str((note_payload.get("event") or {}).get("event_type"))) artifact_payload = execute_tool( "tasks.link_artifact", { "task_id": str(self.task.id), "user_id": self.user.id, "kind": "wiki", "path": "artifacts/wiki/mcp-integration-notes.md", "summary": "Reference docs", }, ) self.assertTrue(str((artifact_payload.get("artifact") or {}).get("id") or "")) task_payload = execute_tool( "tasks.get", {"task_id": str(self.task.id), "user_id": self.user.id}, ) self.assertGreaterEqual(len(task_payload.get("artifact_links") or []), 1) self.assertGreaterEqual(len(task_payload.get("knowledge_articles") or []), 1) def test_docs_append_run_note_writes_file(self): target = Path("/tmp/gia-mcp-test-notes.md") if target.exists(): target.unlink() payload = execute_tool( "docs.append_run_note", { "title": "MCP Integration", "content": "Connected manticore memory tools.", "task_id": str(self.task.id), "path": str(target), }, ) self.assertTrue(payload.get("ok")) content = target.read_text(encoding="utf-8") self.assertIn("MCP Integration", content) self.assertIn("Connected manticore memory tools.", content) target.unlink() def test_audit_logs_record_success_and_failures(self): execute_tool("tasks.list", {"user_id": self.user.id}) with self.assertRaises(ValueError): execute_tool("tasks.search", {"user_id": self.user.id}) success_row = MCPToolAuditLog.objects.filter( tool_name="tasks.list", ok=True, ).first() self.assertIsNotNone(success_row) failure_row = MCPToolAuditLog.objects.filter( tool_name="tasks.search", ok=False, ).first() self.assertIsNotNone(failure_row)