#!/usr/bin/env python3 from __future__ import annotations import importlib.util import json import sys import tempfile import unittest from datetime import date from pathlib import Path from unittest.mock import patch SERVER_PATH = Path(__file__).with_name("server.py") SPEC = importlib.util.spec_from_file_location("aiw_context_mcp", SERVER_PATH) server = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None sys.modules[SPEC.name] = server SPEC.loader.exec_module(server) class ContextMCPTests(unittest.TestCase): def test_initialize_response_declares_tools(self) -> None: response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": server.PROTOCOL_VERSION}}) self.assertEqual(response["result"]["protocolVersion"], server.PROTOCOL_VERSION) self.assertIn("tools", response["result"]["capabilities"]) def test_tools_list_includes_project_search(self) -> None: response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "tools/list"}) names = {tool["name"] for tool in response["result"]["tools"]} self.assertIn("project_search_memory", names) self.assertIn("communication_latest", names) def test_initialize_response_declares_resources(self) -> None: response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": server.PROTOCOL_VERSION}}) self.assertIn("resources", response["result"]["capabilities"]) def test_resources_list_includes_current_work(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) profile = root / "profiles" / "fidelity" / "profile.md" profile.parent.mkdir(parents=True) profile.write_text("# Fidelity", encoding="utf-8") with patch.object(server, "ROOT", root): response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "resources/list"}) uris = {item["uri"] for item in response["result"]["resources"]} self.assertIn("aiw://profiles/fidelity/current-work", uris) def test_resources_read_current_work(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) profile = root / "profiles" / "fidelity" / "profile.md" current = root / "project-knowledge" / "01-current" / "current-work.md" profile.parent.mkdir(parents=True) current.parent.mkdir(parents=True) profile.write_text("# Fidelity", encoding="utf-8") current.write_text("# Current\nImportant", encoding="utf-8") with patch.object(server, "ROOT", root): response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "resources/read", "params": {"uri": "aiw://profiles/fidelity/current-work"}}) self.assertEqual(response["result"]["contents"][0]["text"], "# Current\nImportant") def test_unknown_tool_returns_protocol_error(self) -> None: response = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "missing", "arguments": {}}}) self.assertEqual(response["error"]["code"], -32602) def test_notification_returns_none(self) -> None: response = server.handle_request({"jsonrpc": "2.0", "method": "notifications/initialized"}) self.assertIsNone(response) def test_communication_latest_reads_bounded_records(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) latest = root / "ai" / "inbox" / "mattermost-mirror" / "latest.jsonl" latest.parent.mkdir(parents=True) for index in range(3): latest.write_text("", encoding="utf-8") if index == 0 else None latest.write_text( "\n".join(json.dumps({"post_id": str(index), "message": f"m{index}"}) for index in range(3)) + "\n", encoding="utf-8", ) with patch.object(server, "ROOT", root): result = server.communication_latest({"profile": "fidelity", "limit": 2})["structuredContent"] self.assertEqual([item["message"] for item in result["records"]], ["m1", "m2"]) def test_communication_latest_filters_to_profile_channels_by_default(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) latest = root / "ai" / "inbox" / "mattermost-mirror" / "latest.jsonl" profile_config = root / "profiles" / "fidelity" / "context-sources.json" latest.parent.mkdir(parents=True) profile_config.parent.mkdir(parents=True) profile_config.write_text(json.dumps({ "communication_sources": { "mattermost": {"context_channels": ["fidelity-code-review", "dm-david--jeff"]} } }), encoding="utf-8") latest.write_text( "\n".join([ json.dumps({"post_id": "1", "channel_name": "design-team", "message": "ignore"}), json.dumps({"post_id": "2", "channel_name": "fidelity-code-review", "message": "keep"}), ]) + "\n", encoding="utf-8", ) with patch.object(server, "ROOT", root), patch.dict(server.os.environ, {"AIW_MATTERMOST_CONTEXT_CHANNELS": "", "AIW_MATTERMOST_PROJECT_CHANNELS": ""}, clear=False): result = server.communication_latest({"profile": "fidelity", "limit": 10})["structuredContent"] self.assertEqual([item["message"] for item in result["records"]], ["keep"]) self.assertEqual(result["channel_scope"], "profile") def test_communication_latest_can_include_all_channels(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) latest = root / "ai" / "inbox" / "mattermost-mirror" / "latest.jsonl" profile_config = root / "profiles" / "fidelity" / "context-sources.json" latest.parent.mkdir(parents=True) profile_config.parent.mkdir(parents=True) profile_config.write_text(json.dumps({ "communication_sources": {"mattermost": {"context_channels": ["fidelity-code-review"]}} }), encoding="utf-8") latest.write_text( "\n".join([ json.dumps({"post_id": "1", "channel_name": "design-team", "message": "include"}), json.dumps({"post_id": "2", "channel_name": "fidelity-code-review", "message": "keep"}), ]) + "\n", encoding="utf-8", ) with patch.object(server, "ROOT", root), patch.dict(server.os.environ, {"AIW_MATTERMOST_CONTEXT_CHANNELS": "", "AIW_MATTERMOST_PROJECT_CHANNELS": ""}, clear=False): result = server.communication_latest({"profile": "fidelity", "include_all_channels": True, "limit": 10})["structuredContent"] self.assertEqual([item["message"] for item in result["records"]], ["include", "keep"]) self.assertEqual(result["channel_scope"], "all") def test_project_search_skips_templates(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) real = root / "project-knowledge" / "03-context" / "project.md" template = root / "project-knowledge" / "09-templates" / "daily.md" real.parent.mkdir(parents=True) template.parent.mkdir(parents=True) real.write_text("Important XFlow context", encoding="utf-8") template.write_text("Important XFlow template", encoding="utf-8") with patch.object(server, "ROOT", root): result = server.project_search_memory({"profile": "fidelity", "query": "XFlow"})["structuredContent"] self.assertEqual(len(result["matches"]), 1) self.assertIn("03-context/project.md", result["matches"][0]["path"]) def test_previous_workday_skips_weekend(self) -> None: monday = date(2026, 5, 18) self.assertEqual(server.previous_workday(monday), date(2026, 5, 15)) if __name__ == "__main__": unittest.main()