import os import stat import subprocess import tempfile import unittest from pathlib import Path from unittest.mock import call, patch import sys 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_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 0.0.0.0", 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", ) 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.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_render_remote_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) 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(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" / "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("http://10.0.0.2:10087/mcp", output) 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(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, "run_systemctl") as run_systemctl, patch.object( manage_mcp_service, "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") self.assertIn("/home/nac/.local/bin/uv run --no-sync nocobase-crm-mcp", unit_content) self.assertIn("EnvironmentFile=", unit_content) self.assertIn("http://10.0.0.2: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(manage_mcp_service, "generate_token", return_value="first-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, "run_systemctl") as run_systemctl, patch.object( manage_mcp_service, "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("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_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(manage_mcp_service, "find_uv_path", return_value=Path("/home/nac/.local/bin/uv")), patch.object(manage_mcp_service, "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(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(manage_mcp_service, "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(manage_mcp_service, "detect_remote_hosts", return_value=["10.0.0.2"]), patch.object(manage_mcp_service, "service_state", return_value="active", create=True), patch.object( manage_mcp_service, "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://10.0.0.2:10087/mcp", output) self.assertIn("HTTP probe: reachable", output) if __name__ == "__main__": unittest.main()