david.delagneau
1ad707373a
- Created daily log entries for May 13, 14, 18, 19, 20, and 21, capturing work done, findings, and next steps. - Established a daily logs index for easy navigation of daily notes. - Developed templates for daily logs, decisions, meeting notes, people, systems, and work items to standardize documentation. - Introduced base files for filtering and displaying various types of project knowledge, including daily notes, decisions, people, systems, work items, and workstreams. - Added maps for current work, fidelity apps, and fidelity domain to enhance project navigation and context.
225 lines
11 KiB
Python
225 lines
11 KiB
Python
#!/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("memory_hybrid_search", 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 / "workspaces" / "fidelity" / "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 / "workspaces" / "fidelity" / "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 / "workspaces" / "fidelity" / "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 / "workspaces" / "fidelity" / "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 / "workspaces" / "fidelity" / "project-knowledge" / "03-context" / "project.md"
|
|
template = root / "workspaces" / "fidelity" / "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_memory_hybrid_search_uses_index_when_available(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
index = root / ".aiw" / "indexes" / "fidelity" / "project-knowledge.jsonl"
|
|
manifest = root / ".aiw" / "indexes" / "fidelity" / "manifest.json"
|
|
index.parent.mkdir(parents=True)
|
|
index.write_text(json.dumps({
|
|
"chunk_id": "abc",
|
|
"path": "workspaces/fidelity/project-knowledge/03-context/project.md",
|
|
"heading": "XFlow",
|
|
"text": "Dismissal lifecycle sequencing for XFlow",
|
|
"mtime": 1.0,
|
|
"sha256": "hash",
|
|
}) + "\n", encoding="utf-8")
|
|
manifest.write_text(json.dumps({"chunk_count": 1}), encoding="utf-8")
|
|
|
|
with patch.object(server, "ROOT", root):
|
|
result = server.memory_hybrid_search({"profile": "fidelity", "query": "dismissal lifecycle"})["structuredContent"]
|
|
|
|
self.assertTrue(result["index_available"])
|
|
self.assertEqual(result["source"], "derived-project-knowledge-index")
|
|
self.assertEqual(result["matches"][0]["chunk_id"], "abc")
|
|
|
|
def test_memory_hybrid_search_falls_back_without_index(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
real = root / "workspaces" / "fidelity" / "project-knowledge" / "03-context" / "project.md"
|
|
real.parent.mkdir(parents=True)
|
|
real.write_text("Important XFlow context", encoding="utf-8")
|
|
|
|
with patch.object(server, "ROOT", root):
|
|
result = server.memory_hybrid_search({"profile": "fidelity", "query": "XFlow"})["structuredContent"]
|
|
|
|
self.assertFalse(result["index_available"])
|
|
self.assertEqual(result["source"], "live-project-knowledge-fallback")
|
|
self.assertEqual(len(result["matches"]), 1)
|
|
|
|
def test_project_context_uses_workspace_json_paths(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
config = root / "profiles" / "demo" / "workspace.json"
|
|
current = root / "workspaces" / "demo" / "project-knowledge" / "01-current" / "current-work.md"
|
|
work_items = root / "workspaces" / "demo" / "project-knowledge" / "01-current" / "work-items.md"
|
|
config.parent.mkdir(parents=True)
|
|
current.parent.mkdir(parents=True)
|
|
config.write_text(json.dumps({"knowledge_dir": "workspaces/demo/project-knowledge"}), encoding="utf-8")
|
|
current.write_text("# Current\nDemo current work", encoding="utf-8")
|
|
work_items.write_text("# Work Items", encoding="utf-8")
|
|
|
|
with patch.object(server, "ROOT", root):
|
|
result = server.project_current_context({"profile": "demo"})["structuredContent"]
|
|
|
|
self.assertEqual(result["files"][0]["path"], "workspaces/demo/project-knowledge/01-current/current-work.md")
|
|
self.assertIn("Demo current work", result["files"][0]["text"])
|
|
|
|
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()
|