132 lines
6.0 KiB
Python
132 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import pathlib
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
SOURCE = pathlib.Path(__file__).parents[1] / "platform/mcphub/mcphub_git_installer.py"
|
|
SPEC = importlib.util.spec_from_file_location("mcphub_git_installer", SOURCE)
|
|
assert SPEC and SPEC.loader
|
|
installer = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = installer
|
|
SPEC.loader.exec_module(installer)
|
|
|
|
|
|
class GitInstallerTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = pathlib.Path(self.temp.name)
|
|
self.appdata = self.root / "appdata"
|
|
self.secrets = self.root / "secrets"
|
|
self.secrets.mkdir()
|
|
self.commits = iter(["a" * 40, "b" * 40, "c" * 40])
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp.cleanup()
|
|
|
|
def spec(self, **values: object):
|
|
data = {
|
|
"name": "example",
|
|
"repository": "https://github.com/example/mcp",
|
|
"ref": "main",
|
|
"runtime": "python",
|
|
"entrypoint": "example-mcp",
|
|
"arguments": ("--stdio",),
|
|
"required_env": ("EXAMPLE_TOKEN",),
|
|
}
|
|
data.update(values)
|
|
return installer.GitInstallSpec(**data)
|
|
|
|
def clone(self, _spec, destination: pathlib.Path) -> str:
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
(destination / "pyproject.toml").write_text("[project]\nname='example'\n")
|
|
return next(self.commits)
|
|
|
|
@staticmethod
|
|
def clone_same(_spec, destination: pathlib.Path) -> str:
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
(destination / "pyproject.toml").write_text("[project]\nname='example'\n")
|
|
return "a" * 40
|
|
|
|
@staticmethod
|
|
def build(_source: pathlib.Path, release: pathlib.Path, _entrypoint: str) -> list[str]:
|
|
release.mkdir(parents=True)
|
|
executable = release / ".venv/bin/example-mcp"
|
|
executable.parent.mkdir(parents=True)
|
|
executable.write_text("ok")
|
|
return [str(executable)]
|
|
|
|
def test_install_is_disabled_and_reports_only_missing_key_names(self) -> None:
|
|
with mock.patch.object(installer, "_clone", self.clone), mock.patch.object(
|
|
installer, "_python_release", self.build
|
|
):
|
|
result = installer.prepare_release(self.spec(), self.appdata, self.secrets)
|
|
self.assertFalse(result["config"]["enabled"])
|
|
self.assertFalse(result["credentials_ready"])
|
|
self.assertEqual(result["missing_env"], ["EXAMPLE_TOKEN"])
|
|
self.assertEqual(result["config"]["command"], "/usr/local/bin/run-with-env")
|
|
self.assertEqual(result["config"]["args"][-1], "--stdio")
|
|
|
|
def test_same_release_reuses_build_without_duplicating_arguments(self) -> None:
|
|
with mock.patch.object(installer, "_clone", self.clone_same), mock.patch.object(
|
|
installer, "_python_release", side_effect=self.build
|
|
) as build:
|
|
first = installer.prepare_release(self.spec(), self.appdata, self.secrets)
|
|
second = installer.prepare_release(self.spec(), self.appdata, self.secrets)
|
|
self.assertEqual(build.call_count, 1)
|
|
self.assertEqual(first["config"]["args"], second["config"]["args"])
|
|
self.assertEqual(second["config"]["args"].count("--stdio"), 1)
|
|
|
|
def test_update_and_rollback_preserve_both_releases(self) -> None:
|
|
with mock.patch.object(installer, "_clone", self.clone), mock.patch.object(
|
|
installer, "_python_release", self.build
|
|
):
|
|
first = installer.prepare_release(self.spec(), self.appdata, self.secrets)
|
|
second = installer.prepare_release(self.spec(ref="v2"), self.appdata, self.secrets)
|
|
self.assertEqual(second["previous_release"], first["release"])
|
|
rolled = installer.rollback_release("example", self.appdata)
|
|
self.assertEqual(rolled["release"], first["release"])
|
|
self.assertEqual(installer.current_release("example", self.appdata, self.secrets)["release"], first["release"])
|
|
|
|
def test_failed_update_leaves_previous_state_current(self) -> None:
|
|
with mock.patch.object(installer, "_clone", self.clone), mock.patch.object(
|
|
installer, "_python_release", self.build
|
|
):
|
|
first = installer.prepare_release(self.spec(), self.appdata, self.secrets)
|
|
with mock.patch.object(installer, "_clone", self.clone), mock.patch.object(
|
|
installer, "_python_release", side_effect=installer.GitInstallError("build failed")
|
|
):
|
|
with self.assertRaises(installer.GitInstallError):
|
|
installer.prepare_release(self.spec(ref="broken"), self.appdata, self.secrets)
|
|
current = installer.current_release("example", self.appdata, self.secrets)
|
|
self.assertEqual(current["release"], first["release"])
|
|
|
|
def test_registry_updates_only_matching_server(self) -> None:
|
|
registry = self.appdata / "config/mcp-registry.json"
|
|
registry.parent.mkdir(parents=True)
|
|
registry.write_text(json.dumps({"version": 1, "servers": [{"id": "keep", "hermes_id": "keep"}]}))
|
|
result = {
|
|
"name": "example", "repository": "https://github.com/example/mcp.git",
|
|
"requested_ref": "main", "commit": "a" * 40, "release": "a" * 12,
|
|
"required_env": [], "secret_file": None,
|
|
"config": {"type": "stdio", "command": "example", "args": [], "enabled": False},
|
|
}
|
|
installer.update_registry(registry, installer.registry_entry(result, "Example"))
|
|
servers = json.loads(registry.read_text())["servers"]
|
|
self.assertEqual({item["id"] for item in servers}, {"keep", "example-local"})
|
|
|
|
def test_rejects_non_github_and_escaping_subdirectory(self) -> None:
|
|
with self.assertRaises(installer.GitInstallError):
|
|
installer.normalize_spec(self.spec(repository="https://evil.example/repo"))
|
|
with self.assertRaises(installer.GitInstallError):
|
|
installer.normalize_spec(self.spec(subdirectory="../escape"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|