72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for the single declarative MCP client registry."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).parents[1]
|
|
SOURCE = ROOT / "platform/mcp/sync-clients.py"
|
|
|
|
|
|
def load_module():
|
|
spec = importlib.util.spec_from_file_location("sync_clients", SOURCE)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class RegistryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.module = load_module()
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
self.registry = self.root / "registry.json"
|
|
self.registry.write_text(json.dumps({"version": 1, "servers": [{
|
|
"id": "one", "name": "One", "description": "Test", "url": "http://one/mcp",
|
|
"clients": ["hermes", "openwebui"], "timeout": 123,
|
|
}]}))
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def test_same_registry_generates_both_clients(self):
|
|
items = self.module.active(self.registry, "hermes")
|
|
block = self.module.hermes_block(items)
|
|
self.assertIn("one:", block)
|
|
db = self.root / "webui.db"
|
|
con = sqlite3.connect(db)
|
|
con.execute("create table config (key text primary key, value text, updated_at integer)")
|
|
con.commit(); con.close()
|
|
self.module.update_openwebui(db, self.module.active(self.registry, "openwebui"))
|
|
con = sqlite3.connect(db)
|
|
value = json.loads(con.execute("select value from config where key='tool_server.connections'").fetchone()[0])
|
|
con.close()
|
|
self.assertEqual(value[0]["info"]["id"], "one")
|
|
|
|
def test_old_platform_context_registration_is_removed(self):
|
|
db = self.root / "webui.db"
|
|
con = sqlite3.connect(db)
|
|
con.execute("create table config (key text primary key, value text, updated_at integer)")
|
|
con.execute("insert into config values (?,?,?)", ("tool_server.connections", json.dumps([
|
|
{"info": {"id": "athena-platform"}, "url": "http://old/mcp"},
|
|
{"info": {"id": "unmanaged"}, "url": "http://keep/mcp"},
|
|
]), 0))
|
|
con.commit(); con.close()
|
|
self.module.update_openwebui(db, self.module.active(self.registry, "openwebui"))
|
|
con = sqlite3.connect(db)
|
|
ids = [item["info"]["id"] for item in json.loads(con.execute("select value from config where key='tool_server.connections'").fetchone()[0])]
|
|
con.close()
|
|
self.assertEqual(ids, ["unmanaged", "one"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|