simplify Athena runtime and centralize MCP management
This commit is contained in:
1 parent
104c9904a5
commit
ab671a6396
31 files changed
+918
-419
No files matched your search
@@ -0,0 +1,81 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Regression tests for declarative, update-safe MCPHub settings."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).parents[1]
|
||||
SOURCE = ROOT / "platform/mcphub/configure-settings.py"
|
||||
|
||||
|
||||
def load_module():
|
||||
spec = importlib.util.spec_from_file_location("configure_settings", SOURCE)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class MCPHubSettingsTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.module = load_module()
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.root = Path(self.temp.name)
|
||||
self.secrets = self.root / "secrets"
|
||||
self.secrets.mkdir()
|
||||
(self.secrets / "remote.env").write_text(
|
||||
"REMOTE_URL=http://example.test/mcp\nTOKEN=secret-value\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
self.registry = self.root / "registry.json"
|
||||
self.registry.write_text(json.dumps({
|
||||
"version": 1,
|
||||
"servers": [
|
||||
{
|
||||
"id": "remote-local",
|
||||
"hermes_id": "remote",
|
||||
"hub": {
|
||||
"type": "streamable-http",
|
||||
"secret_file": "remote.env",
|
||||
"url": "${REMOTE_URL}",
|
||||
"headers": {"Authorization": "Bearer ${TOKEN}"},
|
||||
"enabled": True,
|
||||
},
|
||||
},
|
||||
{"id": "client-only", "url": "http://unused/mcp"},
|
||||
],
|
||||
}), encoding="utf-8")
|
||||
|
||||
def tearDown(self):
|
||||
self.temp.cleanup()
|
||||
|
||||
def test_registry_renders_only_hub_servers_and_expands_secrets(self):
|
||||
servers = self.module.registry_servers(self.registry, self.secrets, {})
|
||||
self.assertEqual(list(servers), ["remote"])
|
||||
self.assertEqual(servers["remote"]["url"], "http://example.test/mcp")
|
||||
self.assertEqual(
|
||||
servers["remote"]["headers"]["Authorization"],
|
||||
"Bearer secret-value",
|
||||
)
|
||||
|
||||
def test_existing_enabled_toggle_survives_reconciliation(self):
|
||||
servers = self.module.registry_servers(
|
||||
self.registry, self.secrets, {"remote": {"enabled": False}}
|
||||
)
|
||||
self.assertFalse(servers["remote"]["enabled"])
|
||||
|
||||
def test_missing_secret_fails_closed(self):
|
||||
os.unlink(self.secrets / "remote.env")
|
||||
with self.assertRaises(SystemExit):
|
||||
self.module.registry_servers(self.registry, self.secrets, {})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user