Codex
ef2e44656d
Resolve and write an absolute OpenCode path when enabling the ingest tool group so systemd services without an interactive PATH can pass ingest readiness checks. Preserve an existing OBSIDIAN_MCP_INGEST_OPENCODE_PATH across start, token rotation, and tool enable/disable updates instead of dropping it when rewriting the runtime env file. Validation: python3 -m unittest discover -v; python3 -m py_compile scripts/manage_mcp_service.py tests/test_mcp_service.py; git diff --check; live MCP tools/list and check_ingest_opencode_readiness calls passed after setting the service config.
383 lines
19 KiB
Python
383 lines
19 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",
|
|
enable_ingest=True,
|
|
)
|
|
|
|
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.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", content)
|
|
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\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('"wiki-file-assets"', output)
|
|
self.assertNotIn('"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)
|
|
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets\n", env_content)
|
|
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.shutil, "which", return_value="/home/nac/.opencode/bin/opencode"),
|
|
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",
|
|
enable_ingest=True,
|
|
)
|
|
|
|
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("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_PATH=/home/nac/.opencode/bin/opencode\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_start_service_preserves_saved_enabled_tools_when_flag_is_omitted(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",
|
|
enabled_tools=("file-assets", "ingest"),
|
|
)
|
|
env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env"
|
|
env_path.write_text(
|
|
env_path.read_text(encoding="utf-8")
|
|
+ "OBSIDIAN_MCP_INGEST_OPENCODE_PATH=/custom/bin/opencode\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
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"),
|
|
patch.object(manage_mcp_service, "run_systemctl"),
|
|
):
|
|
manage_mcp_service.start_service(project_root=project_root)
|
|
|
|
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_PATH=/custom/bin/opencode\n", env_content)
|
|
|
|
def test_main_tools_enable_disable_updates_enabled_tools_in_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.2"]),
|
|
patch("builtins.print") as print_fn,
|
|
):
|
|
enable_status = manage_mcp_service.main(
|
|
["tools", "enable", "ingest", "--project-root", str(project_root)]
|
|
)
|
|
|
|
self.assertEqual(enable_status, 0)
|
|
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", env_content)
|
|
self.assertIn("ingest: enabled", print_fn.call_args.args[0])
|
|
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch("builtins.print") as print_fn,
|
|
):
|
|
disable_status = manage_mcp_service.main(
|
|
["tools", "disable", "ingest", "--project-root", str(project_root)]
|
|
)
|
|
|
|
self.assertEqual(disable_status, 0)
|
|
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets\n", env_content)
|
|
self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=0\n", env_content)
|
|
self.assertIn("ingest: disabled", print_fn.call_args.args[0])
|
|
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch("builtins.print") as print_fn,
|
|
):
|
|
disable_files_status = manage_mcp_service.main(
|
|
["tools", "disable", "file-assets", "--project-root", str(project_root)]
|
|
)
|
|
|
|
self.assertEqual(disable_files_status, 0)
|
|
env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8")
|
|
self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=\n", env_content)
|
|
self.assertIn("file-assets: disabled", print_fn.call_args.args[0])
|
|
|
|
def test_main_start_passes_enable_ingest_to_start_service(self):
|
|
with (
|
|
patch.object(manage_mcp_service, "start_service", return_value="config\n") as start_service,
|
|
patch("builtins.print") as print_fn,
|
|
):
|
|
status = manage_mcp_service.main(["start", "--enable-ingest"])
|
|
|
|
self.assertEqual(status, 0)
|
|
start_service.assert_called_once()
|
|
self.assertTrue(start_service.call_args.kwargs["enable_ingest"])
|
|
print_fn.assert_called_once_with("config\n", end="")
|
|
|
|
def test_main_new_token_does_not_expect_enable_ingest_argument(self):
|
|
with (
|
|
patch.object(manage_mcp_service, "new_token", return_value="config\n") as new_token,
|
|
patch("builtins.print") as print_fn,
|
|
):
|
|
status = manage_mcp_service.main(["new-token"])
|
|
|
|
self.assertEqual(status, 0)
|
|
new_token.assert_called_once()
|
|
self.assertNotIn("enable_ingest", new_token.call_args.kwargs)
|
|
print_fn.assert_called_once_with("config\n", end="")
|
|
|
|
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()
|