Simplify Athena stack and recovery
This commit is contained in:
Mikei386
@@ -0,0 +1,71 @@
|
||||
#!/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()
|
||||
@@ -361,7 +361,7 @@ class AutoToolSelectorTests(unittest.IsolatedAsyncioTestCase):
|
||||
result = await self._select(
|
||||
"Wie ist der KI-Host Athena aufgebaut und wo liegt der Recovery-Koffer?"
|
||||
)
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:athena-platform"])
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:athena-operator-local"])
|
||||
|
||||
async def test_athena_operator_is_selected_for_platform_work(self):
|
||||
result = await self._select(
|
||||
@@ -371,7 +371,7 @@ class AutoToolSelectorTests(unittest.IsolatedAsyncioTestCase):
|
||||
result["tool_ids"], ["server:mcp:athena-operator-local"]
|
||||
)
|
||||
|
||||
async def test_mcp_build_from_github_gets_source_and_platform_context(self):
|
||||
async def test_mcp_build_from_github_gets_source_and_operator(self):
|
||||
result = await self._select(
|
||||
"Ich möchte hierfür einen MCP bauen: https://github.com/foo/bar"
|
||||
)
|
||||
|
||||
@@ -1,61 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def load_module(root: Path, runtime: Path):
|
||||
os.environ.update({
|
||||
"ATHENA_REPO_ROOT": str(root),
|
||||
"ATHENA_RUNTIME_FILE": str(runtime),
|
||||
})
|
||||
source = Path(__file__).parents[1] / "platform/mcp/platform_context_mcp.py"
|
||||
spec = importlib.util.spec_from_file_location("platform_context_mcp_test", source)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def main():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp) / "repo"
|
||||
runtime = Path(tmp) / "runtime.json"
|
||||
(root / "docs").mkdir(parents=True)
|
||||
(root / "config").mkdir(parents=True)
|
||||
(root / "ATHENA.md").write_text("# Athena\nOne short source of truth.\n")
|
||||
(root / "docs/OPERATIONS.md").write_text("# Operations\nRecovery detail.\n")
|
||||
(root / "config/service-catalog.json").write_text(json.dumps({
|
||||
"services": [{"id": "test", "name": "Test", "address": "127.0.0.1", "port": 9, "protocol": "tcp"}],
|
||||
}))
|
||||
runtime.write_text(json.dumps({
|
||||
"generated_at": "2100-01-01T00:00:00Z", "source_commit": "abc",
|
||||
"containers": [{"name": "mike-ai-test", "status": "Up"}],
|
||||
"active_inference_profiles": ["fast"], "gpus": [],
|
||||
}))
|
||||
m = load_module(root, runtime)
|
||||
|
||||
assert len(m.TOOLS) == 5
|
||||
assert m.overview()["source"] == "ATHENA.md"
|
||||
assert m.current_state()["active_inference_profiles"] == ["fast"]
|
||||
assert m.external_services()["services"][0]["id"] == "test"
|
||||
assert m.search_reference({"query": "Recovery"})["matches"]
|
||||
assert "Operations" in m.read_reference({"path": "docs/OPERATIONS.md"})["content"]
|
||||
|
||||
missing = m.read_reference({"path": "docs/MISSING.md"})
|
||||
assert missing["ok"] is False and missing["retry"] is False
|
||||
blocked = m.read_reference({"path": "config/secret.env"})
|
||||
assert blocked["ok"] is False and blocked["retry"] is False
|
||||
|
||||
for tool in m.TOOLS:
|
||||
for prop in tool["inputSchema"].get("properties", {}).values():
|
||||
pattern = prop.get("pattern")
|
||||
if pattern:
|
||||
assert pattern.startswith("^") and pattern.endswith("$")
|
||||
print("PLATFORM_CONTEXT_MCP_TEST_OK")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,75 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Focused offline tests for the model-oriented Radarr overlay."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SOURCE = Path(__file__).parents[1] / "platform/mcp/patches/mcp_radarr.py"
|
||||
|
||||
|
||||
def load_module():
|
||||
utilities = types.ModuleType("agent_utilities.mcp_utilities")
|
||||
utilities.dispatch = lambda *args, **kwargs: None
|
||||
utilities.public_actions = lambda client: client.actions
|
||||
utilities.run_blocking = lambda *args, **kwargs: None
|
||||
sys.modules["agent_utilities"] = types.ModuleType("agent_utilities")
|
||||
sys.modules["agent_utilities.mcp_utilities"] = utilities
|
||||
fastmcp = types.ModuleType("fastmcp")
|
||||
fastmcp.FastMCP = object
|
||||
sys.modules["fastmcp"] = fastmcp
|
||||
pydantic = types.ModuleType("pydantic")
|
||||
pydantic.Field = lambda *args, **kwargs: kwargs.get("default")
|
||||
sys.modules["pydantic"] = pydantic
|
||||
auth = types.ModuleType("arr_mcp.auth")
|
||||
auth.get_radarr_client = lambda: None
|
||||
sys.modules["arr_mcp"] = types.ModuleType("arr_mcp")
|
||||
sys.modules["arr_mcp.auth"] = auth
|
||||
spec = importlib.util.spec_from_file_location("radarr_patch", SOURCE)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class RadarrPatchTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.module = load_module()
|
||||
self.movies = [{
|
||||
"id": 12, "title": "Example", "year": 2024, "hasFile": True,
|
||||
"alternateTitles": [{"title": "large unwanted block"}],
|
||||
"movieFile": {
|
||||
"id": 44, "relativePath": "Example.mkv", "size": 2147483648,
|
||||
"quality": {"quality": {"name": "Bluray-1080p"}},
|
||||
"mediaInfo": {
|
||||
"videoCodec": "x264", "resolution": "1920x1080",
|
||||
"videoBitDepth": 8, "audioCodec": "EAC3",
|
||||
"audioLanguages": "ger/eng", "subtitles": "ger",
|
||||
},
|
||||
},
|
||||
}]
|
||||
|
||||
def test_inventory_is_compact_and_alias_aware(self):
|
||||
result = self.module._compact_inventory(self.movies, codecs="h264")
|
||||
self.assertEqual(result["totalMatched"], 1)
|
||||
self.assertEqual(result["movies"][0]["videoCodec"], "x264")
|
||||
self.assertEqual(result["movies"][0]["sizeGiB"], 2.0)
|
||||
self.assertNotIn("alternateTitles", result["movies"][0])
|
||||
|
||||
def test_filter_and_pagination_return_valid_bounded_data(self):
|
||||
result = self.module._compact_inventory(self.movies * 5, query="example", offset=1, limit=2)
|
||||
self.assertEqual(result["returned"], 2)
|
||||
self.assertTrue(result["hasMore"])
|
||||
|
||||
def test_internal_transport_and_power_actions_are_hidden(self):
|
||||
client = types.SimpleNamespace(actions=["get_movie", "request", "post_system_shutdown", "get_queue"])
|
||||
self.assertEqual(self.module._safe_actions(client), ["get_movie", "get_queue"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -14,8 +14,6 @@ verify() {
|
||||
docker exec "$OWUI_CONTAINER" python3 "$VERIFY_REMOTE" "$@"
|
||||
}
|
||||
|
||||
verify http://mike-ai-mcp-platform-context:8000/mcp \
|
||||
--max-tools 12 --max-schema-chars 12000
|
||||
verify http://mike-ai-mcp-athena-operator:8000/mcp \
|
||||
--max-tools 8 --max-schema-chars 12000
|
||||
verify http://tinysearch:8000/mcp \
|
||||
|
||||
Reference in New Issue
Block a user