import os import stat import subprocess 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 import mcp_service_runtime class McpServiceConfigTests(unittest.TestCase): def test_service_management_has_one_shell_entrypoint(self) -> None: entrypoint = SCRIPTS_DIR / "mcp_service.sh" self.assertTrue(entrypoint.exists()) content = entrypoint.read_text(encoding="utf-8") self.assertIn("manage_mcp_service.py", content) self.assertIn('exec python3 "$PROJECT_ROOT/scripts/manage_mcp_service.py"', content) self.assertFalse((SCRIPTS_DIR / "new_mcp_service_token.sh").exists()) self.assertFalse((SCRIPTS_DIR / "start_mcp_service.sh").exists()) self.assertFalse((SCRIPTS_DIR / "show_mcp_service_config.sh").exists()) self.assertFalse((SCRIPTS_DIR / "stop_mcp_service.sh").exists()) self.assertFalse((SCRIPTS_DIR / "uninstall_mcp_service.sh").exists()) def test_missing_env_message_points_to_start_and_token_rotation(self) -> None: with tempfile.TemporaryDirectory() as tmp: env_path = Path(tmp) / ".runtime" / "nocobase-crm-mcp.env" with self.assertRaisesRegex(RuntimeError, r"scripts/mcp_service\.sh start"): manage_mcp_service.read_service_env_file(env_path) def test_render_service_unit_uses_http_defaults_without_embedding_token(self) -> None: unit = manage_mcp_service.render_service_unit( project_root=Path("/opt/nocobase-mcp"), env_file=Path("/opt/nocobase-mcp/.runtime/nocobase-crm-mcp.env"), uv_path=Path("/home/nac/.local/bin/uv"), ) self.assertIn("WorkingDirectory=/opt/nocobase-mcp", unit) self.assertIn("EnvironmentFile=/opt/nocobase-mcp/.runtime/nocobase-crm-mcp.env", unit) self.assertIn("/home/nac/.local/bin/uv run --no-sync nocobase-crm-mcp", unit) self.assertIn("--transport streamable-http", unit) self.assertIn("--host 127.0.0.1", unit) self.assertIn("--port 10087", unit) self.assertNotIn("NOCOBASE_MCP_TOKEN=", unit) def test_write_token_env_file_is_private_and_contains_runtime_settings(self) -> None: with tempfile.TemporaryDirectory() as tmp: env_path = Path(tmp) / ".runtime" / "nocobase-crm-mcp.env" manage_mcp_service.write_token_env_file( env_path, "test-token", host="0.0.0.0", port=10087, path="/mcp", nb_env="nb-mcp", config_path="config.json", nb_bin="/opt/node/bin/nb", ) content = env_path.read_text(encoding="utf-8") self.assertIn("NOCOBASE_MCP_TOKEN=test-token\n", content) self.assertIn("NOCOBASE_MCP_HOST=0.0.0.0\n", content) self.assertIn("NOCOBASE_MCP_PORT=10087\n", content) self.assertIn("NOCOBASE_MCP_PATH=/mcp\n", content) self.assertIn("NOCOBASE_MCP_ENV=nb-mcp\n", content) self.assertIn("NOCOBASE_MCP_CONFIG=config.json\n", content) self.assertIn("NOCOBASE_MCP_NB_BIN=/opt/node/bin/nb\n", content) self.assertIn("NOCOBASE_MCP_ENABLE_UPLOADS=0\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_find_nb_path_uses_env_or_path_and_errors_when_missing(self) -> None: with tempfile.TemporaryDirectory() as tmp: nb_bin = Path(tmp) / "bin" / "nb" nb_bin.parent.mkdir() nb_bin.write_text("#!/bin/sh\n", encoding="utf-8") nb_bin.chmod(0o755) with patch.dict("os.environ", {"NOCOBASE_MCP_NB_BIN": str(nb_bin)}): self.assertEqual(manage_mcp_service.find_nb_path(), Path(os.path.abspath(nb_bin))) cli_script = Path(tmp) / "lib" / "node_modules" / "@nocobase" / "cli" / "bin" / "run.js" cli_script.parent.mkdir(parents=True) cli_script.write_text("#!/usr/bin/env node\n", encoding="utf-8") cli_script.chmod(0o755) nb_link = Path(tmp) / "node" / "bin" / "nb" nb_link.parent.mkdir(parents=True) nb_link.symlink_to(cli_script) with patch.dict("os.environ", {"NOCOBASE_MCP_NB_BIN": str(nb_link)}): self.assertEqual(manage_mcp_service.find_nb_path(), nb_link) missing_nb = Path(tmp) / "missing" / "nb" with patch.dict("os.environ", {"NOCOBASE_MCP_NB_BIN": str(missing_nb)}): with self.assertRaisesRegex(RuntimeError, "not an executable"): manage_mcp_service.find_nb_path() with ( patch.dict("os.environ", {"NOCOBASE_MCP_NB_BIN": "nb-custom"}), patch.object( mcp_service_runtime.shutil, "which", return_value="/usr/local/bin/nb-custom", ), ): self.assertEqual(manage_mcp_service.find_nb_path(), Path("/usr/local/bin/nb-custom")) with ( patch.dict("os.environ", {}, clear=True), patch.object(mcp_service_runtime.shutil, "which", return_value="/usr/local/bin/nb"), ): self.assertEqual(manage_mcp_service.find_nb_path(), Path("/usr/local/bin/nb")) with ( patch.dict("os.environ", {}, clear=True), patch.object(mcp_service_runtime.shutil, "which", return_value=None), ): with self.assertRaisesRegex(RuntimeError, "Could not find nb"): manage_mcp_service.find_nb_path() def test_generated_env_has_size_but_no_nocobase_api_credentials(self) -> None: with tempfile.TemporaryDirectory() as tmp: env_path = Path(tmp) / "mcp.env" manage_mcp_service.write_token_env_file(env_path, "token") content = env_path.read_text(encoding="utf-8") self.assertIn("NOCOBASE_MCP_UPLOAD_MAX_BYTES=26214400\n", content) self.assertIn("NOCOBASE_MCP_ENABLE_UPLOADS=0\n", content) self.assertNotIn("NOCOBASE_MCP_API_TOKEN", content) self.assertNotIn("NOCOBASE_MCP_API_BASE_URL", content) self.assertNotIn("NOCOBASE_MCP_UPLOAD_JWT", content) self.assertNotIn("NOCOBASE_MCP_UPLOAD_API_BASE_URL", content) def test_saved_service_config_returns_named_typed_values(self) -> None: with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) manage_mcp_service.write_token_env_file( project_root / ".runtime" / "nocobase-crm-mcp.env", "token", host="127.0.0.1", port=10088, path="/custom-mcp", nb_env="personal-env", config_path="custom.json", nb_bin="/opt/bin/nb", ) config = manage_mcp_service.saved_service_config(project_root) self.assertIsInstance(config, manage_mcp_service.ServiceConfig) self.assertEqual(config.token, "token") self.assertEqual(config.port, 10088) self.assertEqual(config.path, "/custom-mcp") self.assertEqual(config.nb_env, "personal-env") self.assertEqual(config.config_path, "custom.json") self.assertEqual(config.nb_bin, "/opt/bin/nb") def test_token_rotation_preserves_operator_credentials(self) -> None: with tempfile.TemporaryDirectory() as tmp: env_path = Path(tmp) / "mcp.env" env_path.write_text( "NOCOBASE_MCP_TOKEN=old-token\n" "NOCOBASE_MCP_API_BASE_URL=https://nocobase.example/api\n" "NOCOBASE_MCP_API_TOKEN=operator-api-key\n", encoding="utf-8", ) manage_mcp_service.write_token_env_file(env_path, "new-token", nb_bin="/opt/node/bin/nb") content = env_path.read_text(encoding="utf-8") self.assertIn("NOCOBASE_MCP_TOKEN=new-token\n", content) self.assertIn("NOCOBASE_MCP_NB_BIN=/opt/node/bin/nb\n", content) self.assertIn("NOCOBASE_MCP_UPLOAD_MAX_BYTES=26214400\n", content) self.assertIn("NOCOBASE_MCP_API_BASE_URL=https://nocobase.example/api\n", content) self.assertIn("NOCOBASE_MCP_API_TOKEN=operator-api-key\n", content) def test_token_rotation_preserves_explicit_upload_opt_in(self) -> None: with tempfile.TemporaryDirectory() as tmp: env_path = Path(tmp) / "mcp.env" manage_mcp_service.write_token_env_file(env_path, "old-token") content = env_path.read_text(encoding="utf-8").replace( "NOCOBASE_MCP_ENABLE_UPLOADS=0", "NOCOBASE_MCP_ENABLE_UPLOADS=1", ) env_path.write_text(content, encoding="utf-8") manage_mcp_service.write_token_env_file(env_path, "new-token") content = env_path.read_text(encoding="utf-8") self.assertIn("NOCOBASE_MCP_ENABLE_UPLOADS=1\n", content) self.assertEqual(content.count("NOCOBASE_MCP_ENABLE_UPLOADS="), 1) def test_render_http_config_prints_url_and_authorization_header(self) -> None: output = manage_mcp_service.render_remote_config( hosts=["192.168.1.10"], token="test-token", port=10087, path="/mcp", service_status="inactive", ) self.assertIn("http://192.168.1.10:10087/mcp", output) self.assertIn("Authorization: Bearer test-token", output) self.assertIn('"nocobase-crm"', output) self.assertIn('"type": "streamable-http"', output) self.assertIn("Service status: inactive", output) self.assertNotIn("MCP service is running.", output) self.assertIn("Use this loopback URL on this machine", output) def test_new_token_writes_token_without_touching_systemd_when_service_is_not_installed(self) -> None: 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(mcp_service_runtime, "generate_token", return_value="new-token"), patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(mcp_service_runtime, "find_nb_path", return_value=Path("/opt/node/bin/nb")), patch.object(mcp_service_runtime, "run_systemctl") as run_systemctl, ): output = manage_mcp_service.new_token(project_root=project_root) env_path = project_root / ".runtime" / "nocobase-crm-mcp.env" env_content = env_path.read_text(encoding="utf-8") self.assertIn("NOCOBASE_MCP_TOKEN=new-token\n", env_content) self.assertIn("NOCOBASE_MCP_PORT=10087\n", env_content) self.assertIn("NOCOBASE_MCP_NB_BIN=/opt/node/bin/nb\n", env_content) self.assertIn("http://127.0.0.1:10087/mcp", output) run_systemctl.assert_not_called() def test_new_token_preserves_operator_upload_settings_and_other_env_lines(self) -> None: with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() config_home = Path(tmp) / "config" env_path = project_root / ".runtime" / "nocobase-crm-mcp.env" manage_mcp_service.write_token_env_file(env_path, "old-token") with env_path.open("a", encoding="utf-8") as env_file: env_file.write("# operator-managed upload credentials\n") env_file.write("NOCOBASE_MCP_UPLOAD_API_BASE_URL=https://nb.example/api\n") env_file.write("NOCOBASE_MCP_UPLOAD_JWT=operator-jwt\n") env_file.write("EXTRA_OPERATOR_SETTING=preserve-me\n") with ( patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}), patch.object(mcp_service_runtime, "generate_token", return_value="new-token"), patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=[]), patch.object(mcp_service_runtime, "find_nb_path", return_value=Path("/opt/node/bin/nb")), patch.object(mcp_service_runtime, "run_systemctl") as run_systemctl, ): manage_mcp_service.new_token(project_root=project_root) content = env_path.read_text(encoding="utf-8") self.assertIn("NOCOBASE_MCP_TOKEN=new-token\n", content) self.assertIn("# operator-managed upload credentials\n", content) self.assertIn("NOCOBASE_MCP_UPLOAD_API_BASE_URL=https://nb.example/api\n", content) self.assertIn("NOCOBASE_MCP_UPLOAD_JWT=operator-jwt\n", content) self.assertIn("EXTRA_OPERATOR_SETTING=preserve-me\n", content) self.assertEqual(stat.S_IMODE(os.stat(env_path).st_mode), 0o600) run_systemctl.assert_not_called() def test_start_service_uses_saved_token_and_restarts_service(self) -> None: 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" / "nocobase-crm-mcp.env", "saved-token", host="0.0.0.0", port=10087, path="/mcp", nb_env="nb-mcp", config_path="config.json", ) with ( patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}), patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(mcp_service_runtime, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")), patch.object(mcp_service_runtime, "find_nb_path", return_value=Path("/opt/node/bin/nb")), patch.object(mcp_service_runtime, "run_systemctl") as run_systemctl, patch.object( mcp_service_runtime, "probe_http_service", return_value={"reachable": True, "status": 401}, create=True, ) as probe_http_service, ): 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") env_content = (project_root / ".runtime" / "nocobase-crm-mcp.env").read_text(encoding="utf-8") self.assertIn("/home/nac/.local/bin/uv run --no-sync nocobase-crm-mcp", unit_content) self.assertIn("EnvironmentFile=", unit_content) self.assertIn("NOCOBASE_MCP_NB_BIN=/opt/node/bin/nb\n", env_content) self.assertIn("http://127.0.0.1:10087/mcp", output) self.assertIn("Bearer saved-token", output) run_systemctl.assert_has_calls( [ call(["daemon-reload"]), call(["enable", manage_mcp_service.SERVICE_NAME]), call(["restart", manage_mcp_service.SERVICE_NAME]), call(["is-active", "--quiet", manage_mcp_service.SERVICE_NAME]), ] ) probe_http_service.assert_called_once_with("127.0.0.1", 10087, "/mcp", token="saved-token") def test_start_service_generates_token_on_first_run(self) -> None: 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(mcp_service_runtime, "generate_token", return_value="first-token"), patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(mcp_service_runtime, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")), patch.object(mcp_service_runtime, "find_nb_path", return_value=Path("/opt/node/bin/nb")), patch.object(mcp_service_runtime, "run_systemctl") as run_systemctl, patch.object( mcp_service_runtime, "probe_http_service", return_value={"reachable": True, "status": 401}, create=True, ) as probe_http_service, ): output = manage_mcp_service.start_service(project_root=project_root) env_path = project_root / ".runtime" / "nocobase-crm-mcp.env" self.assertIn("NOCOBASE_MCP_TOKEN=first-token\n", env_path.read_text(encoding="utf-8")) self.assertIn("NOCOBASE_MCP_NB_BIN=/opt/node/bin/nb\n", env_path.read_text(encoding="utf-8")) self.assertIn("Bearer first-token", output) unit_path = config_home / "systemd" / "user" / manage_mcp_service.SERVICE_NAME self.assertTrue(unit_path.exists()) run_systemctl.assert_has_calls( [ call(["daemon-reload"]), call(["enable", manage_mcp_service.SERVICE_NAME]), call(["restart", manage_mcp_service.SERVICE_NAME]), call(["is-active", "--quiet", manage_mcp_service.SERVICE_NAME]), ] ) probe_http_service.assert_called_once_with("127.0.0.1", 10087, "/mcp", token="first-token") def test_start_service_waits_for_http_probe_after_systemd_is_active(self) -> None: 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" / "nocobase-crm-mcp.env", "saved-token", ) with ( patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}), patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(mcp_service_runtime, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")), patch.object(mcp_service_runtime, "find_nb_path", return_value=Path("/opt/node/bin/nb")), patch.object(mcp_service_runtime, "run_systemctl"), patch.object(mcp_service_runtime.time, "sleep") as sleep, patch.object( mcp_service_runtime, "probe_http_service", side_effect=[ {"reachable": False, "error": "connection refused"}, {"reachable": True, "status": 406}, ], create=True, ) as probe_http_service, ): output = manage_mcp_service.start_service(project_root=project_root) self.assertIn("Service status: active", output) self.assertEqual(probe_http_service.call_count, 2) sleep.assert_called_once() def test_start_service_fails_when_systemd_does_not_become_active(self) -> None: 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" / "nocobase-crm-mcp.env", "saved-token", ) def fake_systemctl(args, *, check=True): if args[:2] == ["is-active", "--quiet"]: raise subprocess.CalledProcessError(3, ["systemctl", "--user", *args]) return subprocess.CompletedProcess(["systemctl", "--user", *args], 0) with ( patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}), patch.object(mcp_service_runtime, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")), patch.object(mcp_service_runtime, "find_nb_path", return_value=Path("/opt/node/bin/nb")), patch.object(mcp_service_runtime, "run_systemctl", side_effect=fake_systemctl), ): with self.assertRaisesRegex(RuntimeError, "did not become active"): manage_mcp_service.start_service(project_root=project_root) def test_show_config_reports_saved_config_not_running_claim(self) -> None: with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() manage_mcp_service.write_token_env_file( project_root / ".runtime" / "nocobase-crm-mcp.env", "saved-token", ) with ( patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(mcp_service_runtime, "service_state", return_value="inactive", create=True), ): output = manage_mcp_service.show_config(project_root) self.assertIn("Saved MCP client config.", output) self.assertIn("Service status: inactive", output) self.assertNotIn("MCP service is running.", output) def test_service_status_output_includes_paths_url_and_probe_state(self) -> None: with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() manage_mcp_service.write_token_env_file( project_root / ".runtime" / "nocobase-crm-mcp.env", "saved-token", port=10087, path="/mcp", ) with ( patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(mcp_service_runtime, "service_state", return_value="active", create=True), patch.object( mcp_service_runtime, "probe_http_service", return_value={"reachable": True, "status": 401}, create=True, ), ): output = manage_mcp_service.service_status(project_root=project_root) self.assertIn("Service status: active", output) self.assertIn("Unit file:", output) self.assertIn("Env file:", output) self.assertIn("http://127.0.0.1:10087/mcp", output) self.assertIn("NocoBase CLI env: nb-mcp", output) self.assertIn("Attachment uploads: disabled (safe default)", output) self.assertIn("HTTP probe: reachable", output) def test_service_status_reports_canonical_credentials_without_exposing_them(self) -> None: with tempfile.TemporaryDirectory() as tmp: project_root = Path(tmp) / "project" project_root.mkdir() env_path = project_root / ".runtime" / "nocobase-crm-mcp.env" manage_mcp_service.write_token_env_file(env_path, "mcp-token") with env_path.open("a", encoding="utf-8") as env_file: env_file.write("NOCOBASE_MCP_API_BASE_URL=https://nb.example/api\n") env_file.write("NOCOBASE_MCP_API_TOKEN=never-print-this\n") with ( patch.object(mcp_service_runtime, "detect_remote_hosts", return_value=[]), patch.object(mcp_service_runtime, "service_state", return_value="inactive"), patch.object( mcp_service_runtime, "probe_http_service", return_value={"reachable": False, "error": "stopped"}, ), ): output = manage_mcp_service.service_status(project_root=project_root) self.assertIn("Legacy NocoBase credential override: configured (canonical)", output) self.assertNotIn("never-print-this", output) self.assertNotIn("https://nb.example/api", output) if __name__ == "__main__": unittest.main()