add autonomous GitHub MCP installer
This commit is contained in:
Mikei386
@@ -0,0 +1,131 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user