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_delegate_to_unified_user_ops_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/obsidian-wiki-ops.py" mcp "$@"', content) for wrapper in legacy_wrappers: self.assertFalse((SCRIPTS_DIR / wrapper).exists(), f"{wrapper} should be folded into obsidian-wiki-ops.py") 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.assertNotIn("--host", unit) self.assertNotIn("--port", unit) self.assertNotIn("--path", unit) self.assertIn("PassEnvironment=http_proxy https_proxy all_proxy no_proxy HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY", unit) self.assertIn("MemoryMax=2G", unit) self.assertIn("Restart=on-failure", unit) self.assertNotIn("OBSIDIAN_MCP_TOKEN=", unit) self.assertNotIn("readonly", unit.lower()) 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", public_base_url="http://192.168.1.10:8765/mcp", public_upload_url="https://uploads.example.test/custom/upload", enable_ingest=True, allow_source_files=False, archive_sources=True, ingest_job_timeout_seconds=111, query_job_timeout_seconds=22, ingest_max_request_bytes=333, ) 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_PUBLIC_BASE_URL=http://192.168.1.10:8765/mcp\n", content) self.assertIn("OBSIDIAN_MCP_INGEST_PUBLIC_UPLOAD_URL=https://uploads.example.test/custom/upload\n", content) self.assertIn("OBSIDIAN_MCP_ALLOW_SOURCE_FILES=0\n", content) self.assertIn("WIKI_ARCHIVE_SOURCES=1\n", content) self.assertIn("OBSIDIAN_MCP_ALLOWED_HOSTS=", content) self.assertIn("192.168.1.10:*", content) self.assertIn("OBSIDIAN_MCP_ALLOWED_ORIGINS=", content) self.assertIn("http://192.168.1.10:8765", content) self.assertIn("https://uploads.example.test", content) self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,ingest\n", content) self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=1\n", content) self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_MODEL=openai/gpt-5.5\n", content) self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT=high\n", content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_MODEL=openai/gpt-5.5\n", content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT=minimal\n", content) self.assertIn("OBSIDIAN_MCP_INGEST_JOB_TIMEOUT_SECONDS=111\n", content) self.assertIn("OBSIDIAN_MCP_QUERY_JOB_TIMEOUT_SECONDS=22\n", content) self.assertIn("OBSIDIAN_MCP_INGEST_MAX_REQUEST_BYTES=333\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_saved_service_config_returns_dataclass_and_preserves_explicit_source_file_setting(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="127.0.0.1", port=9988, path="/saved-mcp", public_base_url="https://wiki.example.test/saved-mcp", public_upload_url="https://uploads.example.test/saved", enabled_tools=("file-assets", "query"), allow_source_files=False, archive_sources=True, ingest_job_timeout_seconds=444, query_job_timeout_seconds=55, ingest_max_request_bytes=666, ) config = manage_mcp_service.saved_service_config(project_root) self.assertIsInstance(config, manage_mcp_service.ServiceConfig) self.assertEqual(config.token, "saved-token") self.assertEqual(config.host, "127.0.0.1") self.assertEqual(config.port, 9988) self.assertEqual(config.path, "/saved-mcp") self.assertEqual(config.public_base_url, "https://wiki.example.test/saved-mcp") self.assertEqual(config.public_upload_url, "https://uploads.example.test/saved") self.assertEqual(config.enabled_tools, ("file-assets", "query")) self.assertFalse(config.allow_source_files) self.assertTrue(config.archive_sources) self.assertEqual(config.ingest_job_timeout_seconds, 444) self.assertEqual(config.query_job_timeout_seconds, 55) self.assertEqual(config.ingest_max_request_bytes, 666) self.assertIn("wiki.example.test:*", config.allowed_hosts) self.assertIn("uploads.example.test:*", config.allowed_hosts) self.assertIn("https://wiki.example.test", config.allowed_origins) self.assertIn("https://uploads.example.test", config.allowed_origins) def test_saved_service_config_migrates_legacy_source_file_setting(self): with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env" env_path.parent.mkdir(parents=True) env_path.write_text( "OBSIDIAN_MCP_TOKEN=saved-token\n" "OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,query\n", encoding="utf-8", ) config = manage_mcp_service.saved_service_config(project_root) self.assertTrue(config.allow_source_files) env_path.write_text( "OBSIDIAN_MCP_TOKEN=saved-token\n" "OBSIDIAN_MCP_ENABLED_TOOLS=query\n", encoding="utf-8", ) config = manage_mcp_service.saved_service_config(project_root) self.assertFalse(config.allow_source_files) def test_ensure_service_config_returns_paths_and_default_service_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"]), ): paths, config = manage_mcp_service.ensure_service_config(project_root) self.assertIsInstance(paths, manage_mcp_service.ServicePaths) self.assertIsInstance(config, manage_mcp_service.ServiceConfig) self.assertEqual(config.host, "0.0.0.0") self.assertEqual(config.port, 8765) self.assertEqual(config.path, "/mcp") self.assertEqual(config.public_base_url, "http://10.0.0.2:8765/mcp") self.assertFalse(config.allow_source_files) self.assertIn("10.0.0.2:*", config.allowed_hosts) self.assertIn("http://10.0.0.2:8765", config.allowed_origins) self.assertIn( "OBSIDIAN_MCP_ALLOW_SOURCE_FILES=0\n", paths.env_path.read_text(encoding="utf-8"), ) 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_remote_urls_bracket_ipv6_hosts(self): self.assertEqual( manage_mcp_service.public_base_url_from_hosts(["2001:db8::1"]), "http://[2001:db8::1]:8765/mcp", ) output = manage_mcp_service.render_remote_config( hosts=["2001:db8::1"], token="secret-token", ) self.assertIn('"url": "http://[2001:db8::1]:8765/mcp"', output) def test_render_remote_config_prints_ingest_upload_endpoint_when_enabled(self): output = manage_mcp_service.render_remote_config( hosts=["192.168.1.10", "10.0.0.2"], token="test-token", port=8765, path="/mcp", enabled_tools=("file-assets", "ingest"), ) self.assertIn("http://192.168.1.10:8765/mcp", output) self.assertIn("http://192.168.1.10:8765/mcp-ingest/uploads", output) self.assertIn("http://10.0.0.2:8765/mcp-ingest/uploads", output) self.assertIn("Remote agents should upload with create_ingest_upload_ticket", output) self.assertIn("Bearer upload is for local/admin use", output) def test_render_remote_config_uses_public_base_url_for_upload_endpoint(self): output = manage_mcp_service.render_remote_config( hosts=["192.168.1.10"], token="test-token", port=8765, path="/mcp", public_base_url="https://wiki.example.test/custom-mcp", enabled_tools=("file-assets", "ingest"), ) self.assertIn("https://wiki.example.test/mcp-ingest/uploads", output) self.assertNotIn("http://192.168.1.10:8765/mcp-ingest/uploads", output) def test_render_remote_config_uses_normalized_public_url_as_primary_client_endpoint(self): output = manage_mcp_service.render_remote_config( hosts=["192.168.1.10", "10.0.0.2"], token="test-token", port=8765, path="/mcp", public_base_url=" https://wiki.example.test/proxy/mcp/?ignored=yes#fragment ", ) self.assertIn( "Use this URL from another machine:\nhttps://wiki.example.test/proxy/mcp\n", output, ) self.assertIn('"url": "https://wiki.example.test/proxy/mcp"', output) self.assertIn( "Other detected addresses:\n" "- http://192.168.1.10:8765/mcp\n" "- http://10.0.0.2:8765/mcp\n", output, ) def test_render_remote_config_prefers_explicit_upload_url(self): output = manage_mcp_service.render_remote_config( hosts=["192.168.1.10"], token="test-token", port=8765, path="/mcp", public_base_url="https://wiki.example.test/proxy/mcp", public_upload_url="https://uploads.example.test/custom/upload", enabled_tools=("file-assets", "ingest"), ) self.assertIn("https://uploads.example.test/custom/upload", output) self.assertNotIn("https://wiki.example.test/mcp-ingest/uploads", 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("OBSIDIAN_MCP_ALLOW_SOURCE_FILES=0\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_new_token_preserves_saved_service_config_when_endpoint_arguments_are_omitted(self): with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() config_home = Path(tmp) / "config" env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env" manage_mcp_service.write_token_env_file( env_path, "old-token", host="127.0.0.1", port=9988, path="/saved-mcp", public_base_url="https://wiki.example.test/saved-mcp", public_upload_url="https://uploads.example.test/saved", enabled_tools=("file-assets", "query"), allow_source_files=False, ) 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) config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.token, "new-token") self.assertEqual(config.host, "127.0.0.1") self.assertEqual(config.port, 9988) self.assertEqual(config.path, "/saved-mcp") self.assertEqual(config.public_base_url, "https://wiki.example.test/saved-mcp") self.assertEqual(config.public_upload_url, "https://uploads.example.test/saved") self.assertEqual(config.enabled_tools, ("file-assets", "query")) self.assertFalse(config.allow_source_files) self.assertIn('"url": "https://wiki.example.test/saved-mcp"', output) run_systemctl.assert_not_called() def test_new_token_only_changes_token_when_saved_public_urls_are_empty(self): with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() config_home = Path(tmp) / "config" env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env" manage_mcp_service.write_token_env_file( env_path, "old-token", host="127.0.0.1", port=9988, path="/saved-mcp", allow_source_files=False, ) original_lines = env_path.read_text(encoding="utf-8").splitlines() 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"), ): output = manage_mcp_service.new_token(project_root=project_root) updated_lines = env_path.read_text(encoding="utf-8").splitlines() self.assertEqual(updated_lines[1:], original_lines[1:]) self.assertEqual(updated_lines[0], "OBSIDIAN_MCP_TOKEN=new-token") self.assertNotIn("OBSIDIAN_MCP_PUBLIC_BASE_URL=", "\n".join(updated_lines)) self.assertIn('"url": "http://10.0.0.2:9988/saved-mcp"', output) def test_new_token_explicit_endpoint_and_public_urls_override_saved_config(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", "old-token", host="127.0.0.1", port=9988, path="/saved-mcp", public_base_url="https://wiki.example.test/saved-mcp", public_upload_url="https://uploads.example.test/saved", ) 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"), ): output = manage_mcp_service.new_token( project_root=project_root, host="0.0.0.0", port=8877, path="/new-mcp", public_base_url="https://new.example.test/api/mcp/", public_upload_url="https://uploads-new.example.test/api/upload/", ) config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.host, "0.0.0.0") self.assertEqual(config.port, 8877) self.assertEqual(config.path, "/new-mcp") self.assertEqual(config.public_base_url, "https://new.example.test/api/mcp") self.assertEqual(config.public_upload_url, "https://uploads-new.example.test/api/upload") self.assertIn("new.example.test", config.allowed_hosts) self.assertIn("uploads-new.example.test", config.allowed_hosts) self.assertIn("https://new.example.test", config.allowed_origins) self.assertIn("https://uploads-new.example.test", config.allowed_origins) self.assertIn('"url": "https://new.example.test/api/mcp"', output) def test_new_token_recomputes_public_url_when_only_endpoint_changes(self): with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() manage_mcp_service.write_token_env_file( project_root / ".runtime" / "obsidian-wiki-mcp.env", "old-token", host="127.0.0.1", port=9988, path="/saved-mcp", public_base_url="https://wiki.example.test/saved-mcp", ) with ( 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"), ): manage_mcp_service.new_token( project_root=project_root, port=8877, path="/new-mcp", ) config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.public_base_url, "http://127.0.0.1:8877/new-mcp") def test_new_token_first_install_uses_concrete_bind_host_as_public_url(self): with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() with ( 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"), ): manage_mcp_service.new_token(project_root=project_root, host="127.0.0.1") config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.public_base_url, "http://127.0.0.1:8765/mcp") def test_start_service_uses_saved_token_and_restarts_service(self): with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() (project_root / ".env").write_text( "OBSIDIAN_VAULT_PATH=/data/bjd-vault\n" "WIKI_ARCHIVE_SOURCES=true\n", encoding="utf-8", ) 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) self.assertIn("WIKI_ARCHIVE_SOURCES=1\n", env_content) self.assertNotIn("OBSIDIAN_MCP_PUBLIC_BASE_URL=", env_content) self.assertNotIn("OBSIDIAN_MCP_INGEST_PUBLIC_UPLOAD_URL=", 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_preserves_saved_endpoint_and_public_urls_when_arguments_are_omitted(self): with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() config_home = Path(tmp) / "config" env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env" manage_mcp_service.write_token_env_file( env_path, "saved-token", host="127.0.0.1", port=9988, path="/saved-mcp", public_base_url="https://wiki.example.test/saved-mcp", public_upload_url="https://uploads.example.test/saved", allow_source_files=False, ) 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"), ): output = manage_mcp_service.start_service(project_root=project_root) config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.host, "127.0.0.1") self.assertEqual(config.port, 9988) self.assertEqual(config.path, "/saved-mcp") self.assertEqual(config.public_base_url, "https://wiki.example.test/saved-mcp") self.assertEqual(config.public_upload_url, "https://uploads.example.test/saved") self.assertFalse(config.allow_source_files) unit_content = ( config_home / "systemd" / "user" / manage_mcp_service.SERVICE_NAME ).read_text(encoding="utf-8") self.assertNotIn("--host", unit_content) env_content = env_path.read_text(encoding="utf-8") self.assertIn("OBSIDIAN_MCP_HOST=127.0.0.1", env_content) self.assertIn("OBSIDIAN_MCP_PORT=9988", env_content) self.assertIn("OBSIDIAN_MCP_PATH=/saved-mcp", env_content) self.assertIn('"url": "https://wiki.example.test/saved-mcp"', output) def test_start_service_explicit_arguments_override_saved_config(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="127.0.0.1", port=9988, path="/saved-mcp", public_base_url="https://wiki.example.test/saved-mcp", public_upload_url="https://uploads.example.test/saved", allow_source_files=False, ) 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"), ): output = manage_mcp_service.start_service( project_root=project_root, host="0.0.0.0", port=8877, path="/new-mcp", public_base_url="https://new.example.test/api/mcp/", public_upload_url="https://uploads-new.example.test/api/upload/", allow_source_files=True, ) config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.host, "0.0.0.0") self.assertEqual(config.port, 8877) self.assertEqual(config.path, "/new-mcp") self.assertEqual(config.public_base_url, "https://new.example.test/api/mcp") self.assertEqual(config.public_upload_url, "https://uploads-new.example.test/api/upload") self.assertTrue(config.allow_source_files) self.assertIn("new.example.test", config.allowed_hosts) self.assertIn("uploads-new.example.test", config.allowed_hosts) self.assertIn("https://new.example.test", config.allowed_origins) self.assertIn("https://uploads-new.example.test", config.allowed_origins) unit_content = ( config_home / "systemd" / "user" / manage_mcp_service.SERVICE_NAME ).read_text(encoding="utf-8") self.assertNotIn("--host", unit_content) env_content = (project_root / ".runtime" / "obsidian-wiki-mcp.env").read_text(encoding="utf-8") self.assertIn("OBSIDIAN_MCP_HOST=0.0.0.0", env_content) self.assertIn("OBSIDIAN_MCP_PORT=8877", env_content) self.assertIn("OBSIDIAN_MCP_PATH=/new-mcp", env_content) self.assertIn('"url": "https://new.example.test/api/mcp"', output) def test_start_service_recomputes_public_url_when_only_endpoint_changes(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="127.0.0.1", port=9988, path="/saved-mcp", public_base_url="https://wiki.example.test/saved-mcp", ) 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, port=8877, path="/new-mcp", ) config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.public_base_url, "http://127.0.0.1:8877/new-mcp") def test_start_service_first_install_uses_concrete_bind_host_as_public_url(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.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, host="127.0.0.1") config = manage_mcp_service.saved_service_config(project_root) self.assertEqual(config.public_base_url, "http://127.0.0.1:8765/mcp") 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_HOST=0.0.0.0\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_PUBLIC_BASE_URL=http://10.0.0.4:9876/wiki-mcp\n", env_content) self.assertIn("OBSIDIAN_MCP_ALLOW_SOURCE_FILES=0\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("OBSIDIAN_MCP_INGEST_OPENCODE_MODEL=openai/gpt-5.5\n", env_content) self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT=high\n", env_content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_MODEL=openai/gpt-5.5\n", env_content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT=minimal\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", ) env_path.write_text( env_path.read_text(encoding="utf-8") + "OBSIDIAN_MCP_INGEST_OPENCODE_MODEL=openai/custom-strong\n" + "OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT=max\n" + "OBSIDIAN_MCP_QUERY_OPENCODE_MODEL=openai/custom-fast\n" + "OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT=low\n" + "OBSIDIAN_MCP_PUBLIC_BASE_URL=https://wiki.example.test/mcp\n", encoding="utf-8", ) env_path.write_text( env_path.read_text(encoding="utf-8") + "OBSIDIAN_MCP_INGEST_PUBLIC_UPLOAD_URL=https://uploads.example.test/custom/upload\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) self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_MODEL=openai/custom-strong\n", env_content) self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT=max\n", env_content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_MODEL=openai/custom-fast\n", env_content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT=low\n", env_content) self.assertIn("OBSIDIAN_MCP_PUBLIC_BASE_URL=https://wiki.example.test/mcp\n", env_content) self.assertIn("OBSIDIAN_MCP_INGEST_PUBLIC_UPLOAD_URL=https://uploads.example.test/custom/upload\n", env_content) def test_start_service_can_enable_query_without_ingest(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, enable_query=True, ) env_path = project_root / ".runtime" / "obsidian-wiki-mcp.env" env_content = env_path.read_text(encoding="utf-8") self.assertIn("OBSIDIAN_MCP_ENABLED_TOOLS=file-assets,query\n", env_content) self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=0\n", env_content) self.assertIn("OBSIDIAN_MCP_OPENCODE_PATH=/home/nac/.opencode/bin/opencode\n", env_content) self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_MODEL=openai/gpt-5.5\n", env_content) self.assertIn("OBSIDIAN_MCP_INGEST_OPENCODE_VARIANT=high\n", env_content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_MODEL=openai/gpt-5.5\n", env_content) self.assertIn("OBSIDIAN_MCP_QUERY_OPENCODE_VARIANT=minimal\n", env_content) self.assertNotIn("OBSIDIAN_MCP_INGEST_OPENCODE_PATH=", env_content) self.assertIn("Enabled tool groups:\nfile-assets,query", output) self.assertIn("Use ask_wiki for ordinary wiki, business facts, or hybrid questions.", 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_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_tools_enable_query_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.object(manage_mcp_service.shutil, "which", return_value="/home/nac/.opencode/bin/opencode"), patch("builtins.print") as print_fn, ): enable_status = manage_mcp_service.main( ["tools", "enable", "query", "--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,query\n", env_content) self.assertIn("OBSIDIAN_MCP_ENABLE_INGEST=0\n", env_content) self.assertIn("OBSIDIAN_MCP_OPENCODE_PATH=/home/nac/.opencode/bin/opencode\n", env_content) self.assertIn("query: 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", "query", "--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("query: disabled", print_fn.call_args.args[0]) def test_set_tool_enabled_imports_proxy_environment_when_restarting_query_service(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), "https_proxy": "http://proxy.local:8080"}, clear=True), patch.object(manage_mcp_service.shutil, "which", return_value="/home/nac/.opencode/bin/opencode"), patch.object(manage_mcp_service, "run_systemctl") as run_systemctl, ): output = manage_mcp_service.set_tool_enabled(project_root, "query", True) self.assertIn("query: enabled", output) run_systemctl.assert_has_calls( [ call(["import-environment", "https_proxy"], check=False), call(["restart", manage_mcp_service.SERVICE_NAME]), ] ) 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_start_passes_enable_query_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-query"]) self.assertEqual(status, 0) start_service.assert_called_once() self.assertTrue(start_service.call_args.kwargs["enable_query"]) print_fn.assert_called_once_with("config\n", end="") def test_main_start_passes_none_for_omitted_overrides_and_source_file_policy_flags(self): with ( patch.object(manage_mcp_service, "start_service", return_value="config\n") as start_service, patch("builtins.print"), ): status = manage_mcp_service.main(["start", "--deny-source-files"]) self.assertEqual(status, 0) kwargs = start_service.call_args.kwargs self.assertIsNone(kwargs["host"]) self.assertIsNone(kwargs["port"]) self.assertIsNone(kwargs["path"]) self.assertIsNone(kwargs["public_base_url"]) self.assertIsNone(kwargs["public_upload_url"]) self.assertFalse(kwargs["allow_source_files"]) with ( patch.object(manage_mcp_service, "start_service", return_value="config\n") as start_service, patch("builtins.print"), ): status = manage_mcp_service.main(["start", "--allow-source-files"]) self.assertEqual(status, 0) self.assertTrue(start_service.call_args.kwargs["allow_source_files"]) def test_main_passes_explicit_endpoint_and_public_url_overrides(self): endpoint_args = [ "--host", "127.0.0.1", "--port", "9988", "--path", "/custom-mcp", "--public-base-url", "https://wiki.example.test/custom-mcp", "--public-upload-url", "https://uploads.example.test/custom", ] with ( patch.object(manage_mcp_service, "start_service", return_value="config\n") as start_service, patch("builtins.print"), ): start_status = manage_mcp_service.main(["start", *endpoint_args]) self.assertEqual(start_status, 0) self.assertEqual(start_service.call_args.kwargs["host"], "127.0.0.1") self.assertEqual(start_service.call_args.kwargs["port"], 9988) self.assertEqual(start_service.call_args.kwargs["path"], "/custom-mcp") self.assertEqual( start_service.call_args.kwargs["public_base_url"], "https://wiki.example.test/custom-mcp", ) self.assertEqual( start_service.call_args.kwargs["public_upload_url"], "https://uploads.example.test/custom", ) with ( patch.object(manage_mcp_service, "new_token", return_value="config\n") as new_token, patch("builtins.print"), ): token_status = manage_mcp_service.main(["new-token", *endpoint_args]) self.assertEqual(token_status, 0) self.assertEqual(new_token.call_args.kwargs["host"], "127.0.0.1") self.assertEqual(new_token.call_args.kwargs["port"], 9988) self.assertEqual(new_token.call_args.kwargs["path"], "/custom-mcp") self.assertEqual( new_token.call_args.kwargs["public_base_url"], "https://wiki.example.test/custom-mcp", ) self.assertEqual( new_token.call_args.kwargs["public_upload_url"], "https://uploads.example.test/custom", ) 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) self.assertIsNone(new_token.call_args.kwargs["host"]) self.assertIsNone(new_token.call_args.kwargs["port"]) self.assertIsNone(new_token.call_args.kwargs["path"]) self.assertIsNone(new_token.call_args.kwargs["public_base_url"]) self.assertIsNone(new_token.call_args.kwargs["public_upload_url"]) print_fn.assert_called_once_with("config\n", end="") def test_main_cleanup_delegates_to_runtime_cleanup(self): with ( patch.object(manage_mcp_service, "cleanup_runtime", return_value="cleanup\n") as cleanup_runtime, patch("builtins.print") as print_fn, ): status = manage_mcp_service.main(["cleanup", "--project-root", "/tmp/project", "--max-age-days", "14"]) self.assertEqual(status, 0) cleanup_runtime.assert_called_once_with(Path("/tmp/project"), max_age_days=14) print_fn.assert_called_once_with("cleanup\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()