my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Tests for Mnemosyne MCP Server (Phase 6)
#3
#4	Run with: pytest tests/test_mcp_server.py -v
#5	"""
#6
#7	import json
#8	import os
#9	import subprocess
#10	import sys
#11	import pytest
#12	from unittest.mock import MagicMock, patch
#13
#14	# Test tool schemas
#15	from mnemosyne.mcp_tools import (
#16	TOOLS, get_tool_definitions, handle_tool_call,
#17	_REMEMBER_SCHEMA, _RECALL_SCHEMA, _SLEEP_SCHEMA,
#18	_SCRATCHPAD_READ_SCHEMA, _SCRATCHPAD_WRITE_SCHEMA, _GET_STATS_SCHEMA
#19	)
#20
#21
#22	class TestToolSchemas:
#23	"""Verify tool schemas match MCP spec and are valid JSON."""
#24
#25	def test_all_tools_present(self):
#26	"""All 6 tools must be defined."""
#27	names = [t["name"] for t in TOOLS]
#28	assert len(names) == 6
#29	assert "mnemosyne_remember" in names
#30	assert "mnemosyne_recall" in names
#31	assert "mnemosyne_sleep" in names
#32	assert "mnemosyne_scratchpad_read" in names
#33	assert "mnemosyne_scratchpad_write" in names
#34	assert "mnemosyne_get_stats" in names
#35
#36	def test_tool_schemas_are_valid_json(self):
#37	"""Each tool schema must be valid JSON-serializable."""
#38	for tool in TOOLS:
#39	# Schema must be serializable
#40	dumped = json.dumps(tool["inputSchema"])
#41	loaded = json.loads(dumped)
#42	assert loaded["type"] == "object"
#43	assert "properties" in loaded
#44
#45	def test_remember_schema_has_required_fields(self):
#46	"""mnemosyne_remember requires 'content'."""
#47	schema = _REMEMBER_SCHEMA
#48	assert "required" in schema
#49	assert "content" in schema["required"]
#50	assert "properties" in schema
#51	assert "source" in schema["properties"]
#52	assert "importance" in schema["properties"]
#53	assert "metadata" in schema["properties"]
#54	assert "extract_entities" in schema["properties"]
#55	assert "extract" in schema["properties"]
#56	assert "bank" in schema["properties"]
#57
#58	def test_recall_schema_has_required_fields(self):
#59	"""mnemosyne_recall requires 'query'."""
#60	schema = _RECALL_SCHEMA
#61	assert "required" in schema
#62	assert "query" in schema["required"]
#63	assert "top_k" in schema["properties"]
#64	assert "bank" in schema["properties"]
#65	assert "temporal_weight" in schema["properties"]
#66
#67	def test_no_destructive_tools(self):
#68	"""No forget, invalidate, or export/import tools exposed."""
#69	names = [t["name"] for t in TOOLS]
#70	assert "mnemosyne_forget" not in names
#71	assert "mnemosyne_invalidate" not in names
#72	assert "mnemosyne_export" not in names
#73	assert "mnemosyne_import" not in names
#74
#75
#76	class TestToolHandlers:
#77	"""Test each handler with mocked Mnemosyne instance."""
#78
#79	@pytest.fixture
#80	def mock_mnemosyne(self):
#81	"""Create a mock Mnemosyne instance."""
#82	mock = MagicMock()
#83	mock.remember.return_value = "test-memory-id-123"
#84	mock.recall.return_value = [
#85	{"id": "mem1", "content": "Test content", "score": 0.95}
#86	]
#87	mock.sleep.return_value = {"consolidated": 3, "deleted": 1}
#88	mock.scratchpad_read.return_value = ["entry1", "entry2"]
#89	mock.scratchpad_write.return_value = "scratch-id-456"
#90	mock.get_stats.return_value = {
#91	"total_memories": 42,
#92	"total_sessions": 3,
#93	"sources": {"conversation": 30, "file": 12},
#94	"last_memory": "2026-04-29T01:00:00",
#95	"database": "/test/db",
#96	"mode": "beam",
#97	"beam": {"working_memory": {}, "episodic_memory": {}}
#98	}
#99	return mock
#100
#101	def test_handle_remember(self, mock_mnemosyne):
#102	"""handle_remember returns success with memory_id."""
#103	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#104	result = handle_tool_call("mnemosyne_remember", {
#105	"content": "Test memory",
#106	"source": "test",
#107	"importance": 0.9,
#108	"bank": "default"
#109	})
#110	assert result["status"] == "stored"
#111	assert result["memory_id"] == "test-memory-id-123"
#112	assert result["bank"] == "default"
#113	mock_mnemosyne.remember.assert_called_once()
#114
#115	def test_handle_remember_uses_mcp_bank_env_default(self, mock_mnemosyne, monkeypatch):
#116	"""MCP server bank default applies when tool call omits bank."""
#117	monkeypatch.setenv("MNEMOSYNE_MCP_BANK", "work")
#118
#119	with patch(
#120	"mnemosyne.mcp_tools._create_instance",
#121	return_value=mock_mnemosyne,
#122	) as create_instance:
#123	result = handle_tool_call("mnemosyne_remember", {
#124	"content": "Test memory",
#125	"source": "test",
#126	})
#127
#128	assert result["status"] == "stored"
#129	assert result["bank"] == "work"
#130	assert create_instance.call_args.kwargs["bank"] == "work"
#131
#132	def test_handle_remember_bank_arg_overrides_mcp_bank_env(self, mock_mnemosyne, monkeypatch):
#133	"""Explicit per-call bank should override the server default bank."""
#134	monkeypatch.setenv("MNEMOSYNE_MCP_BANK", "work")
#135
#136	with patch(
#137	"mnemosyne.mcp_tools._create_instance",
#138	return_value=mock_mnemosyne,
#139	) as create_instance:
#140	result = handle_tool_call("mnemosyne_remember", {
#141	"content": "Test memory",
#142	"source": "test",
#143	"bank": "personal",
#144	})
#145
#146	assert result["status"] == "stored"
#147	assert result["bank"] == "personal"
#148	assert create_instance.call_args.kwargs["bank"] == "personal"
#149
#150	def test_handle_recall_uses_mcp_bank_env_default(self, mock_mnemosyne, monkeypatch):
#151	"""MCP recall should use the server default bank when omitted."""
#152	monkeypatch.setenv("MNEMOSYNE_MCP_BANK", "work")
#153
#154	with patch(
#155	"mnemosyne.mcp_tools._create_instance",
#156	return_value=mock_mnemosyne,
#157	) as create_instance:
#158	result = handle_tool_call("mnemosyne_recall", {
#159	"query": "test query",
#160	})
#161
#162	assert result["status"] == "ok"
#163	assert result["bank"] == "work"
#164	assert create_instance.call_args.kwargs["bank"] == "work"
#165
#166	def test_handle_recall(self, mock_mnemosyne):
#167	"""handle_recall returns list of results."""
#168	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#169	result = handle_tool_call("mnemosyne_recall", {
#170	"query": "test query",
#171	"top_k": 5,
#172	"bank": "default"
#173	})
#174	assert result["status"] == "ok"
#175	assert result["count"] == 1
#176	assert len(result["results"]) == 1
#177	mock_mnemosyne.recall.assert_called_once()
#178
#179	def test_handle_recall_forwards_scoring_weights(self, mock_mnemosyne):
#180	"""Schema-advertised recall weights should be forwarded to Mnemosyne.recall()."""
#181	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#182	handle_tool_call("mnemosyne_recall", {
#183	"query": "test query",
#184	"top_k": 5,
#185	"bank": "default",
#186	"vec_weight": 0.6,
#187	"fts_weight": 0.3,
#188	"importance_weight": 0.1,
#189	})
#190
#191	_, kwargs = mock_mnemosyne.recall.call_args
#192	assert kwargs["vec_weight"] == 0.6
#193	assert kwargs["fts_weight"] == 0.3
#194	assert kwargs["importance_weight"] == 0.1
#195
#196	def test_handle_sleep(self, mock_mnemosyne):
#197	"""handle_sleep returns consolidation stats."""
#198	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#199	result = handle_tool_call("mnemosyne_sleep", {
#200	"dry_run": False,
#201	"bank": "default"
#202	})
#203	assert result["status"] == "ok"
#204	assert result["dry_run"] is False
#205	assert "result" in result
#206	mock_mnemosyne.sleep.assert_called_once_with(dry_run=False)
#207
#208	def test_handle_scratchpad_read(self, mock_mnemosyne):
#209	"""handle_scratchpad_read returns entries."""
#210	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#211	result = handle_tool_call("mnemosyne_scratchpad_read", {
#212	"bank": "default"
#213	})
#214	assert result["status"] == "ok"
#215	assert result["count"] == 2
#216	assert len(result["entries"]) == 2
#217
#218	def test_handle_scratchpad_write(self, mock_mnemosyne):
#219	"""handle_scratchpad_write returns entry_id."""
#220	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#221	result = handle_tool_call("mnemosyne_scratchpad_write", {
#222	"content": "New scratchpad entry",
#223	"bank": "default"
#224	})
#225	assert result["status"] == "stored"
#226	assert result["entry_id"] == "scratch-id-456"
#227
#228	def test_handle_get_stats(self, mock_mnemosyne):
#229	"""handle_get_stats returns JSON-serializable stats."""
#230	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#231	result = handle_tool_call("mnemosyne_get_stats", {
#232	"bank": "default"
#233	})
#234	assert result["status"] == "ok"
#235	assert "stats" in result
#236	# Must be JSON serializable
#237	dumped = json.dumps(result)
#238	loaded = json.loads(dumped)
#239	assert loaded["stats"]["total_memories"] == 42
#240
#241	def test_error_handling(self, mock_mnemosyne):
#242	"""Error handling returns MCP-compliant error results."""
#243	mock_mnemosyne.remember.side_effect = RuntimeError("DB locked")
#244	with patch("mnemosyne.mcp_tools._create_instance", return_value=mock_mnemosyne):
#245	with pytest.raises(RuntimeError, match="DB locked"):
#246	handle_tool_call("mnemosyne_remember", {"content": "test"})
#247
#248	def test_unknown_tool(self):
#249	"""Unknown tool raises ValueError."""
#250	with pytest.raises(ValueError, match="Unknown tool"):
#251	handle_tool_call("mnemosyne_unknown", {})
#252
#253
#254	class TestMCPIntegration:
#255	"""Integration tests for MCP server lifecycle."""
#256
#257	def test_mcp_server_imports(self):
#258	"""MCP server module imports successfully."""
#259	from mnemosyne.mcp_server import run_mcp_server, main
#260	assert callable(run_mcp_server)
#261	assert callable(main)
#262
#263	def test_mcp_tools_import_guard(self):
#264	"""mcp_tools imports even if mcp package not available."""
#265	# The module should load regardless
#266	from mnemosyne import mcp_tools
#267	assert hasattr(mcp_tools, "TOOLS")
#268	assert hasattr(mcp_tools, "handle_tool_call")
#269
#270	def test_get_tool_definitions_returns_all(self):
#271	"""get_tool_definitions returns all 6 tools."""
#272	tools = get_tool_definitions()
#273	assert len(tools) == 6
#274	names = [t["name"] for t in tools]
#275	assert "mnemosyne_remember" in names
#276
#277	def test_top_level_cli_forwards_mcp_arguments(self, tmp_path):
#278	"""`mnemosyne mcp ...` must pass subcommand args to the MCP parser."""
#279	env = os.environ.copy()
#280	env["HOME"] = str(tmp_path / "home")
#281	env["MNEMOSYNE_DATA_DIR"] = str(tmp_path / "mnemosyne-data")
#282	script = """
#283	import json
#284	import sys
#285	import mnemosyne.mcp_server
#286
#287	def fake_main(argv):
#288	print(json.dumps({"argv": argv}))
#289
#290	mnemosyne.mcp_server.main = fake_main
#291	sys.argv = [
#292	"mnemosyne",
#293	"mcp",
#294	"--transport",
#295	"sse",
#296	"--port",
#297	"19090",
#298	"--bank",
#299	"work",
#300	]
#301	from mnemosyne.cli import run_cli
#302	run_cli()
#303	"""
#304	result = subprocess.run(
#305	[sys.executable, "-c", script],
#306	text=True,
#307	capture_output=True,
#308	env=env,
#309	check=False,
#310	)
#311
#312	assert result.returncode == 0, result.stderr
#313	assert json.loads(result.stdout) == {
#314	"argv": ["--transport", "sse", "--port", "19090", "--bank", "work"]
#315	}
#316
#317	def test_mcp_server_main_accepts_explicit_argv(self):
#318	"""MCP server parser should parse caller-provided argv, not global sys.argv."""
#319	from mnemosyne.mcp_server import main
#320
#321	with patch("mnemosyne.mcp_server.run_mcp_server") as run_mcp_server:
#322	main(["--transport", "sse", "--port", "19090", "--bank", "work"])
#323
#324	run_mcp_server.assert_called_once_with(
#325	transport="sse", port=19090, bank="work"
#326	)
#327
#328
#329	class TestImportGuard:
#330	"""Verify MCP is truly optional."""
#331
#332	def test_core_imports_without_mcp(self):
#333	"""Core mnemosyne imports work without mcp installed."""
#334	from mnemosyne import remember, recall, get_stats
#335	assert callable(remember)
#336	assert callable(recall)
#337	assert callable(get_stats)
#338
#339	def test_mcp_server_raises_without_mcp(self):
#340	"""MCP server raises helpful error if mcp not installed."""
#341	from mnemosyne.mcp_server import _MCP_AVAILABLE, _run_stdio
#342
#343	if _MCP_AVAILABLE:
#344	# mcp is installed — verify the server function exists and the flag is True
#345	assert _MCP_AVAILABLE is True
#346	else:
#347	# mcp is NOT installed — verify _run_stdio raises RuntimeError
#348	import asyncio
#349	with pytest.raises(RuntimeError, match="MCP not installed"):
#350	asyncio.get_event_loop().run_until_complete(_run_stdio())
#351

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public