92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SOURCE = pathlib.Path(__file__).parents[1] / "platform/mcphub/deploy-extension.py"
|
|
SPEC = importlib.util.spec_from_file_location("deploy_extension", SOURCE)
|
|
assert SPEC and SPEC.loader
|
|
deploy = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(deploy)
|
|
|
|
|
|
class DeployExtensionTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = pathlib.Path(self.temp.name)
|
|
self.appdata = self.root / "appdata"
|
|
self.work = self.appdata / "work/example"
|
|
self.secrets = self.root / "secrets"
|
|
self.registry = self.appdata / "config/mcp-registry.json"
|
|
self.work.mkdir(parents=True)
|
|
self.secrets.mkdir()
|
|
self.registry.parent.mkdir(parents=True)
|
|
self.registry.write_text('{"version":1,"servers":[]}\n')
|
|
artifact = self.work / "index.js"
|
|
artifact.write_text("console.log('ok')\n")
|
|
digest = hashlib.sha256(artifact.read_bytes()).hexdigest()
|
|
self.manifest = self.work / "manifest.json"
|
|
self.manifest.write_text(json.dumps({
|
|
"server": {
|
|
"id": "example", "hermes_id": "example", "name": "Example",
|
|
"description": "Example MCP", "url": "http://host/mcp/example",
|
|
"clients": ["hermes"],
|
|
"deployment": {
|
|
"required_env": ["EXAMPLE_TOKEN"],
|
|
"required_files": ["example-config.toml"],
|
|
},
|
|
"hub": {
|
|
"type": "stdio", "secret_file": "example.env",
|
|
"command": "node", "args": ["/app/data/extensions/example/index.js"],
|
|
"enabled": True,
|
|
},
|
|
},
|
|
"artifacts": [{
|
|
"source": str(artifact), "path": "index.js",
|
|
"sha256": digest, "mode": "0644",
|
|
}],
|
|
}))
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp.cleanup()
|
|
|
|
def args(self, **extra: object) -> argparse.Namespace:
|
|
values = {
|
|
"appdata": self.appdata, "registry": self.registry,
|
|
"secrets": self.secrets, "manifest": self.manifest, "id": "example",
|
|
"skip_api": True,
|
|
}
|
|
values.update(extra)
|
|
return argparse.Namespace(**values)
|
|
|
|
def registered(self) -> dict:
|
|
return json.loads(self.registry.read_text())["servers"][0]
|
|
|
|
def test_missing_secret_forces_disabled_and_unpublished(self) -> None:
|
|
deploy.stage(self.args())
|
|
server = self.registered()
|
|
self.assertFalse(server["hub"]["enabled"])
|
|
self.assertEqual(server["clients"], [])
|
|
self.assertTrue((self.appdata / "extensions/example/index.js").is_file())
|
|
with self.assertRaises(SystemExit):
|
|
deploy.set_enabled(self.args(), True)
|
|
|
|
def test_complete_secret_allows_activation(self) -> None:
|
|
(self.secrets / "example.env").write_text("EXAMPLE_TOKEN=value\n")
|
|
(self.secrets / "example-config.toml").write_text("account = 'example'\n")
|
|
deploy.stage(self.args())
|
|
deploy.set_enabled(self.args(), True)
|
|
server = self.registered()
|
|
self.assertTrue(server["hub"]["enabled"])
|
|
self.assertEqual(server["clients"], ["hermes"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|