Codex
acff3c14db
Use the Python module entrypoint in the user service, prepare the MCP optional runtime before restart, and serve the configured vault path instead of the repo root. Verify binary asset resources use an octet-stream MIME type and cover service vault resolution in tests.
253 lines
12 KiB
Python
253 lines
12 KiB
Python
import os
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import call, patch
|
|
|
|
|
|
SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
|
|
if str(SCRIPTS_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS_DIR))
|
|
|
|
import manage_mcp_service
|
|
|
|
|
|
class McpServiceConfigTests(unittest.TestCase):
|
|
def test_service_commands_use_one_shell_entrypoint(self):
|
|
entrypoint = SCRIPTS_DIR / "mcp_service.sh"
|
|
legacy_wrappers = [
|
|
"new_mcp_service_token.sh",
|
|
"start_mcp_service.sh",
|
|
"show_mcp_service_config.sh",
|
|
"stop_mcp_service.sh",
|
|
"uninstall_mcp_service.sh",
|
|
]
|
|
|
|
self.assertTrue(entrypoint.is_file())
|
|
self.assertTrue(os.access(entrypoint, os.X_OK))
|
|
content = entrypoint.read_text(encoding="utf-8")
|
|
self.assertIn('exec python3 "$SCRIPT_DIR/manage_mcp_service.py" "$@"', content)
|
|
for wrapper in legacy_wrappers:
|
|
self.assertFalse((SCRIPTS_DIR / wrapper).exists(), f"{wrapper} should be folded into mcp_service.sh")
|
|
|
|
def test_render_service_unit_uses_http_transport_without_embedding_token(self):
|
|
unit = manage_mcp_service.render_service_unit(
|
|
project_root=Path("/opt/obsidian-wiki"),
|
|
vault_path=Path("/data/bjd-vault"),
|
|
env_file=Path("/opt/obsidian-wiki/.runtime/obsidian-wiki-mcp.env"),
|
|
uv_path=Path("/home/nac/.local/bin/uv"),
|
|
)
|
|
|
|
self.assertIn("WorkingDirectory=/opt/obsidian-wiki", unit)
|
|
self.assertIn("EnvironmentFile=/opt/obsidian-wiki/.runtime/obsidian-wiki-mcp.env", unit)
|
|
self.assertIn("/home/nac/.local/bin/uv run --no-sync python -m obsidian_wiki.mcp_server", unit)
|
|
self.assertIn("--vault /data/bjd-vault", unit)
|
|
self.assertIn("--transport streamable-http", unit)
|
|
self.assertIn("--host 0.0.0.0", unit)
|
|
self.assertIn("--port 8765", unit)
|
|
self.assertNotIn("OBSIDIAN_MCP_TOKEN=", unit)
|
|
|
|
def test_write_token_env_file_is_private(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
env_path = Path(tmp) / ".runtime" / "obsidian-wiki-mcp.env"
|
|
|
|
manage_mcp_service.write_token_env_file(env_path, "test-token", host="0.0.0.0", port=8765, path="/mcp")
|
|
|
|
content = env_path.read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_TOKEN=test-token\n", content)
|
|
self.assertIn("OBSIDIAN_MCP_HOST=0.0.0.0\n", content)
|
|
self.assertIn("OBSIDIAN_MCP_PORT=8765\n", content)
|
|
self.assertIn("OBSIDIAN_MCP_PATH=/mcp\n", content)
|
|
self.assertEqual(stat.S_IMODE(os.stat(env_path).st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(os.stat(env_path.parent).st_mode), 0o700)
|
|
|
|
def test_resolve_vault_for_service_uses_explicit_or_configured_vault(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
explicit_vault = Path(tmp) / "explicit-vault"
|
|
configured_vault = Path(tmp) / "configured-vault"
|
|
project_root.mkdir()
|
|
explicit_vault.mkdir()
|
|
configured_vault.mkdir()
|
|
|
|
self.assertEqual(
|
|
manage_mcp_service.resolve_vault_for_service(project_root, vault_path=explicit_vault),
|
|
explicit_vault.resolve(),
|
|
)
|
|
with patch.object(manage_mcp_service, "resolve_vault_path", return_value=str(configured_vault)):
|
|
self.assertEqual(
|
|
manage_mcp_service.resolve_vault_for_service(project_root),
|
|
configured_vault.resolve(),
|
|
)
|
|
|
|
def test_render_remote_config_prints_client_configuration(self):
|
|
output = manage_mcp_service.render_remote_config(
|
|
hosts=["192.168.1.10"],
|
|
token="test-token",
|
|
port=8765,
|
|
path="/mcp",
|
|
)
|
|
|
|
self.assertIn("http://192.168.1.10:8765/mcp", output)
|
|
self.assertIn("Authorization", output)
|
|
self.assertIn("Bearer test-token", output)
|
|
self.assertIn('"type": "streamable-http"', output)
|
|
self.assertIn('"obsidian-wiki"', output)
|
|
|
|
def test_new_token_writes_token_without_touching_systemd_when_unit_missing(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
project_root.mkdir()
|
|
config_home = Path(tmp) / "config"
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch.object(manage_mcp_service, "generate_token", return_value="new-token"),
|
|
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
|
|
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
|
|
):
|
|
output = manage_mcp_service.new_token(project_root=project_root)
|
|
|
|
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
|
|
env_content = env_path.read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_TOKEN=new-token\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_PORT=8765\n", env_content)
|
|
self.assertIn("http://10.0.0.2:8765/mcp", output)
|
|
run_systemctl.assert_not_called()
|
|
|
|
def test_new_token_restarts_service_when_unit_exists(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
project_root.mkdir()
|
|
config_home = Path(tmp) / "config"
|
|
paths = manage_mcp_service.resolve_service_paths(project_root, config_home=config_home)
|
|
paths.unit_path.parent.mkdir(parents=True)
|
|
paths.unit_path.write_text("unit\n", encoding="utf-8")
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch.object(manage_mcp_service, "generate_token", return_value="new-token"),
|
|
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
|
|
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
|
|
):
|
|
output = manage_mcp_service.new_token(project_root=project_root)
|
|
|
|
env_content = paths.env_path.read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_TOKEN=new-token\n", env_content)
|
|
self.assertIn("http://10.0.0.2:8765/mcp", output)
|
|
run_systemctl.assert_called_once_with(["restart", manage_mcp_service.SERVICE_NAME])
|
|
|
|
def test_start_service_uses_saved_token_and_restarts_service(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
project_root.mkdir()
|
|
config_home = Path(tmp) / "config"
|
|
manage_mcp_service.write_token_env_file(
|
|
project_root / ".runtime" / "obsidian-wiki-mcp.env",
|
|
"saved-token",
|
|
host="0.0.0.0",
|
|
port=8765,
|
|
path="/mcp",
|
|
)
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch.object(manage_mcp_service, "generate_token") as generate_token,
|
|
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]),
|
|
patch.object(manage_mcp_service, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")),
|
|
patch.object(manage_mcp_service, "resolve_vault_for_service", return_value=Path("/data/bjd-vault")),
|
|
patch.object(manage_mcp_service, "prepare_runtime") as prepare_runtime,
|
|
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
|
|
):
|
|
output = manage_mcp_service.start_service(project_root=project_root)
|
|
|
|
unit_path = config_home / "systemd" / "user" / manage_mcp_service.SERVICE_NAME
|
|
unit_content = unit_path.read_text(encoding="utf-8")
|
|
self.assertIn("/home/nac/.local/bin/uv run --no-sync python -m obsidian_wiki.mcp_server", unit_content)
|
|
self.assertIn("--vault /data/bjd-vault", unit_content)
|
|
self.assertIn("EnvironmentFile=", unit_content)
|
|
self.assertIn("http://10.0.0.2:8765/mcp", output)
|
|
self.assertIn("Bearer saved-token", output)
|
|
generate_token.assert_not_called()
|
|
prepare_runtime.assert_called_once_with(project_root, uv_path=Path("/home/nac/.local/bin/uv"))
|
|
run_systemctl.assert_has_calls(
|
|
[
|
|
call(["daemon-reload"]),
|
|
call(["enable", manage_mcp_service.SERVICE_NAME]),
|
|
call(["restart", manage_mcp_service.SERVICE_NAME]),
|
|
]
|
|
)
|
|
|
|
def test_start_service_generates_token_when_missing_and_prints_config(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
project_root.mkdir()
|
|
config_home = Path(tmp) / "config"
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch.object(manage_mcp_service, "generate_token", return_value="created-token"),
|
|
patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.4"]),
|
|
patch.object(manage_mcp_service, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")),
|
|
patch.object(manage_mcp_service, "prepare_runtime") as prepare_runtime,
|
|
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
|
|
):
|
|
output = manage_mcp_service.start_service(project_root=project_root, host="0.0.0.0", port=9876, path="/wiki-mcp")
|
|
|
|
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
|
|
env_content = env_path.read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_TOKEN=created-token\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_PORT=9876\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_PATH=/wiki-mcp\n", env_content)
|
|
self.assertIn("http://10.0.0.4:9876/wiki-mcp", output)
|
|
self.assertIn("Authorization: Bearer created-token", output)
|
|
self.assertIn('"type": "streamable-http"', output)
|
|
prepare_runtime.assert_called_once_with(project_root, uv_path=Path("/home/nac/.local/bin/uv"))
|
|
run_systemctl.assert_has_calls(
|
|
[
|
|
call(["daemon-reload"]),
|
|
call(["enable", manage_mcp_service.SERVICE_NAME]),
|
|
call(["restart", manage_mcp_service.SERVICE_NAME]),
|
|
]
|
|
)
|
|
|
|
def test_uninstall_service_removes_only_managed_files(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
config_home = Path(tmp) / "config"
|
|
paths = manage_mcp_service.resolve_service_paths(project_root, config_home=config_home)
|
|
paths.unit_path.parent.mkdir(parents=True)
|
|
paths.env_path.parent.mkdir(parents=True)
|
|
paths.unit_path.write_text("unit\n", encoding="utf-8")
|
|
paths.env_path.write_text("OBSIDIAN_MCP_TOKEN=old\n", encoding="utf-8")
|
|
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch.object(manage_mcp_service, "run_systemctl") as run_systemctl,
|
|
):
|
|
manage_mcp_service.uninstall_service(project_root)
|
|
|
|
self.assertFalse(paths.unit_path.exists())
|
|
self.assertFalse(paths.env_path.exists())
|
|
run_systemctl.assert_has_calls(
|
|
[
|
|
call(["disable", "--now", manage_mcp_service.SERVICE_NAME], check=False),
|
|
call(["daemon-reload"], check=False),
|
|
call(["reset-failed", manage_mcp_service.SERVICE_NAME], check=False),
|
|
]
|
|
)
|
|
|
|
def test_show_config_reads_saved_token_and_prints_remote_config(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
|
|
manage_mcp_service.write_token_env_file(env_path, "saved-token", host="0.0.0.0", port=9876, path="/wiki-mcp")
|
|
|
|
with patch.object(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.3"]):
|
|
output = manage_mcp_service.show_config(project_root)
|
|
|
|
self.assertIn("http://10.0.0.3:9876/wiki-mcp", output)
|
|
self.assertIn("Authorization: Bearer saved-token", output)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|