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('"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) 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()