Simplify Athena runtime and document current architecture
This commit is contained in:
Mikei386
@@ -1,91 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Dependency-free safety checks for the hass_mcp YAML overlay."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import pathlib
|
||||
import sys
|
||||
import types
|
||||
|
||||
|
||||
def module(name: str, **attributes):
|
||||
value = types.ModuleType(name)
|
||||
for key, item in attributes.items():
|
||||
setattr(value, key, item)
|
||||
sys.modules[name] = value
|
||||
return value
|
||||
|
||||
|
||||
class ToolError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def decorator(**_kwargs):
|
||||
return lambda function: function
|
||||
|
||||
|
||||
module("homeassistant")
|
||||
module("homeassistant.core", HomeAssistant=object)
|
||||
module("homeassistant.util", slugify=lambda value: str(value).lower().replace(" ", "_"))
|
||||
module("guarded")
|
||||
module("guarded.tools")
|
||||
module("guarded.identity", user_context=lambda: None)
|
||||
module("guarded.protocol", ToolError=ToolError, internal_error=lambda message, error: RuntimeError(f"{message}: {error}"))
|
||||
module(
|
||||
"guarded.registry",
|
||||
LIMIT_FIELD={"type": "integer"},
|
||||
OFFSET_FIELD={"type": "integer"},
|
||||
paginate=lambda items, limit, offset: {"items": items[offset : offset + limit]},
|
||||
schema=lambda **kwargs: kwargs,
|
||||
tool=decorator,
|
||||
)
|
||||
|
||||
source = pathlib.Path(__file__).parents[1] / "platform/mcp/patches/hass_mcp/yaml_config.py"
|
||||
spec = importlib.util.spec_from_file_location("guarded.tools.yaml_config", source)
|
||||
assert spec and spec.loader
|
||||
guard = importlib.util.module_from_spec(spec)
|
||||
sys.modules[spec.name] = guard
|
||||
spec.loader.exec_module(guard)
|
||||
|
||||
assert set(guard._KINDS) == {"automation", "script", "scene", "configuration"}
|
||||
assert all("secret" not in filename for filename, _, _ in guard._KINDS.values())
|
||||
assert guard._redact_line("api_key: abc") == "api_key: <redacted>"
|
||||
assert guard._redact_line("token: abc") == "token: <redacted>"
|
||||
assert guard._redact_line("value: !secret private_name") == "value: !secret <redacted-reference>"
|
||||
assert guard._redact_line("alias: Safe automation") == "alias: Safe automation"
|
||||
|
||||
for sensitive in ("token: abc", "password: abc", "value: !secret private_name"):
|
||||
try:
|
||||
guard._reject_sensitive_replacement(sensitive)
|
||||
except ToolError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError(f"sensitive replacement was accepted: {sensitive}")
|
||||
|
||||
change = guard._change_record("automation", "update", "before", {"id": "demo"})
|
||||
preview = guard._preview(change)
|
||||
assert preview["changed"] is False
|
||||
assert preview["confirmation_required"] is True
|
||||
guard._consume_ticket(change, preview["approval_ticket"])
|
||||
try:
|
||||
guard._consume_ticket(change, preview["approval_ticket"])
|
||||
except ToolError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError("one-time approval ticket was reusable")
|
||||
|
||||
other = guard._change_record("automation", "update", "different", {"id": "demo"})
|
||||
ticket = guard._preview(change)["approval_ticket"]
|
||||
try:
|
||||
guard._consume_ticket(other, ticket)
|
||||
except ToolError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError("ticket accepted a different current file fingerprint")
|
||||
|
||||
diff = guard._source_diff("automations.yaml", "a\nb\n", "a\nc\n")
|
||||
assert any("-b" in line for line in diff)
|
||||
assert any("+c" in line for line in diff)
|
||||
|
||||
print("hass_mcp_yaml_guard_tests=ok")
|
||||
@@ -1,106 +0,0 @@
|
||||
#!/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"])
|
||||
|
||||
def test_production_registry_has_unique_ids_and_fritzbox(self):
|
||||
document = json.loads((ROOT / "config/mcp-registry.json").read_text())
|
||||
ids = [item["id"] for item in document["servers"]]
|
||||
hermes_ids = [
|
||||
item.get("hermes_id", item["id"])
|
||||
for item in document["servers"] if "hermes" in item.get("clients", [])
|
||||
]
|
||||
self.assertEqual(len(ids), len(set(ids)))
|
||||
self.assertEqual(len(hermes_ids), len(set(hermes_ids)))
|
||||
fritz = next(item for item in document["servers"] if item["id"] == "fritzbox-local")
|
||||
self.assertEqual(fritz["hub"]["type"], "stdio")
|
||||
self.assertIn("fritz-mcp", fritz["hub"]["args"][-1])
|
||||
self.assertEqual(len(fritz["tool_include"]), 4)
|
||||
|
||||
def test_hermes_tool_filter_is_generated(self):
|
||||
item = {
|
||||
"id": "wide", "name": "Wide", "description": "Test",
|
||||
"url": "http://wide/mcp", "clients": ["hermes"],
|
||||
"tool_include": ["list", "describe", "call"],
|
||||
}
|
||||
block = self.module.hermes_block([item])
|
||||
self.assertIn(" tools:\n include:", block)
|
||||
self.assertIn(' - "describe"', block)
|
||||
|
||||
def test_raw_mcphub_token_can_replace_host_specific_env_file(self):
|
||||
item = {
|
||||
"id": "hub", "name": "Hub", "description": "Test",
|
||||
"url": "http://hub/mcp/test", "clients": ["hermes"],
|
||||
"env_file": "/missing/client.env",
|
||||
"key_env": "MCPHUB_BEARER_TOKEN",
|
||||
}
|
||||
self.module.CLIENT_TOKEN = "local-token"
|
||||
self.assertTrue(self.module.enabled(item))
|
||||
self.assertEqual(self.module.resolved(item), ("http://hub/mcp/test", "local-token"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,91 +0,0 @@
|
||||
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()
|
||||
@@ -1,131 +0,0 @@
|
||||
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()
|
||||
@@ -1,94 +0,0 @@
|
||||
#!/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, {})
|
||||
|
||||
def test_hermes_group_is_recoverable_and_bounded(self):
|
||||
settings = {"groups": [{"id": "keep", "name": "other", "servers": []}]}
|
||||
self.module.ensure_hermes_group(settings)
|
||||
self.module.ensure_hermes_group(settings)
|
||||
groups = settings["groups"]
|
||||
self.assertEqual(len([group for group in groups if group["name"] == "hermes"]), 1)
|
||||
hermes = next(group for group in groups if group["name"] == "hermes")
|
||||
fritzbox = next(item for item in hermes["servers"] if item["name"] == "fritzbox")
|
||||
self.assertEqual(
|
||||
fritzbox["tools"],
|
||||
["list_services", "list_actions", "describe_action", "call_action"],
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,616 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Offline tests for the versioned OpenWebUI filters.
|
||||
|
||||
The production container provides pydantic. A tiny local stand-in keeps these
|
||||
logic tests dependency-free and prevents test setup from reaching the network.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class _BaseModel:
|
||||
def __init__(self, **values):
|
||||
annotations = {}
|
||||
for base in reversed(type(self).__mro__):
|
||||
annotations.update(getattr(base, "__annotations__", {}))
|
||||
for name in annotations:
|
||||
setattr(self, name, values.get(name, getattr(type(self), name, None)))
|
||||
|
||||
|
||||
fake_pydantic = types.ModuleType("pydantic")
|
||||
fake_pydantic.BaseModel = _BaseModel
|
||||
fake_pydantic.Field = lambda default=None, **kwargs: default
|
||||
sys.modules.setdefault("pydantic", fake_pydantic)
|
||||
|
||||
FILTER_DIR = Path(__file__).parents[1] / "platform" / "openwebui" / "filters"
|
||||
ACTION_DIR = Path(__file__).parents[1] / "platform" / "openwebui" / "actions"
|
||||
|
||||
|
||||
def _load(name: str):
|
||||
spec = importlib.util.spec_from_file_location(name, FILTER_DIR / f"{name}.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader is not None
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _load_action(name: str):
|
||||
spec = importlib.util.spec_from_file_location(name, ACTION_DIR / f"{name}.py")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader is not None
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class StabilityGuardTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def asyncSetUp(self):
|
||||
self.module = _load("stability_guard")
|
||||
self.guard = self.module.Filter()
|
||||
|
||||
async def test_large_tool_output_is_bounded(self):
|
||||
body = {
|
||||
"model": "qwen-fast",
|
||||
"messages": [
|
||||
{"role": "user", "content": "Prüfe das Log."},
|
||||
{"role": "tool", "tool_call_id": "x", "content": "A" * 50000},
|
||||
],
|
||||
}
|
||||
result = await self.guard.inlet(body)
|
||||
content = result["messages"][1]["content"]
|
||||
self.assertLessEqual(len(content), self.guard.valves.max_single_tool_chars)
|
||||
self.assertIn("Werkzeugausgabe gekürzt", content)
|
||||
|
||||
async def test_uncensored_uses_its_80k_context_limit(self):
|
||||
self.assertEqual(
|
||||
self.guard._context_limit("mikeai-uncensored"),
|
||||
self.guard.valves.uncensored_context_tokens,
|
||||
)
|
||||
|
||||
async def test_duplicate_calls_disable_tools(self):
|
||||
call = {
|
||||
"id": "call",
|
||||
"type": "function",
|
||||
"function": {"name": "search", "arguments": '{"q":"same"}'},
|
||||
}
|
||||
body = {
|
||||
"model": "qwen-fast",
|
||||
"tools": [{"type": "function", "function": {"name": "search"}}],
|
||||
"tool_ids": ["server:mcp:web"],
|
||||
"messages": [
|
||||
{"role": "user", "content": "Suche genau einmal."},
|
||||
{"role": "assistant", "tool_calls": [call]},
|
||||
{"role": "tool", "tool_call_id": "1", "content": "nichts"},
|
||||
{"role": "assistant", "tool_calls": [call]},
|
||||
{"role": "tool", "tool_call_id": "2", "content": "nichts"},
|
||||
{"role": "assistant", "tool_calls": [call]},
|
||||
],
|
||||
}
|
||||
result = await self.guard.inlet(body)
|
||||
self.assertEqual(result["tools"], [])
|
||||
self.assertEqual(result["tool_ids"], [])
|
||||
self.assertIn("Weitere Werkzeugaufrufe", result["messages"][0]["content"])
|
||||
|
||||
async def test_old_context_is_compacted_before_current_turn(self):
|
||||
self.guard.valves.default_context_tokens = 10000
|
||||
self.guard.valves.hard_context_ratio = 0.8
|
||||
self.guard.valves.reserved_output_tokens = 1000
|
||||
body = {
|
||||
"model": "unknown",
|
||||
"messages": [
|
||||
{"role": "system", "content": "Sicher arbeiten."},
|
||||
{"role": "user", "content": "alt " * 15000},
|
||||
{"role": "assistant", "content": "altantwort " * 8000},
|
||||
{"role": "tool", "tool_call_id": "old", "content": "log " * 20000},
|
||||
{"role": "user", "content": "Aktuelle wichtige Frage"},
|
||||
],
|
||||
}
|
||||
result = await self.guard.inlet(body)
|
||||
self.assertEqual(result["messages"][-1]["content"], "Aktuelle wichtige Frage")
|
||||
self.assertLess(len(result["messages"][1]["content"]), 3000)
|
||||
|
||||
async def test_private_csv_disables_web_and_requires_local_table_analysis(self):
|
||||
body = {
|
||||
"model": "qwen-fast",
|
||||
"features": {"web_search": True, "code_interpreter": True},
|
||||
"tool_ids": ["server:mcp:web-local", "server:mcp:arr-local"],
|
||||
"tools": [
|
||||
{"type": "function", "function": {"name": "search_web"}},
|
||||
{"type": "function", "function": {"name": "execute_code"}},
|
||||
],
|
||||
"metadata": {"files": [{"name": "private-bank.csv"}]},
|
||||
"messages": [
|
||||
{"role": "user", "content": "Sortiere Ein- und Ausgänge."},
|
||||
],
|
||||
}
|
||||
result = await self.guard.inlet(body)
|
||||
self.assertEqual(result["tool_ids"], [])
|
||||
self.assertFalse(result["features"]["web_search"])
|
||||
self.assertTrue(result["features"]["code_interpreter"])
|
||||
self.assertEqual(
|
||||
[tool["function"]["name"] for tool in result["tools"]],
|
||||
["execute_code"],
|
||||
)
|
||||
self.assertIn("private table rule", result["messages"][0]["content"])
|
||||
self.assertIn("sep=None", result["messages"][0]["content"])
|
||||
self.assertIn("delimiter", result["messages"][0]["content"])
|
||||
|
||||
async def test_private_csv_is_detected_from_user_text_without_metadata(self):
|
||||
body = {
|
||||
"model": "qwen-fast",
|
||||
"tools": [
|
||||
{"type": "function", "function": {"name": "search_web"}},
|
||||
{"type": "function", "function": {"name": "execute_code"}},
|
||||
],
|
||||
"messages": [
|
||||
{"role": "user", "content": "Werte bitte diese CSV meines Bankkontos aus."},
|
||||
],
|
||||
}
|
||||
result = await self.guard.inlet(body)
|
||||
self.assertEqual(
|
||||
[tool["function"]["name"] for tool in result["tools"]],
|
||||
["execute_code"],
|
||||
)
|
||||
|
||||
async def test_private_csv_stops_before_openwebui_hard_tool_limit(self):
|
||||
calls = []
|
||||
for index in range(8):
|
||||
calls.extend(
|
||||
[
|
||||
{
|
||||
"role": "assistant",
|
||||
"tool_calls": [
|
||||
{
|
||||
"function": {
|
||||
"name": "execute_code",
|
||||
"arguments": '{"code":"step %d"}' % index,
|
||||
}
|
||||
}
|
||||
],
|
||||
},
|
||||
{"role": "tool", "content": "ok", "tool_call_id": str(index)},
|
||||
]
|
||||
)
|
||||
body = {
|
||||
"model": "qwen-fast",
|
||||
"tools": [
|
||||
{"type": "function", "function": {"name": "execute_code"}},
|
||||
],
|
||||
"messages": [
|
||||
{"role": "user", "content": "Werte diese CSV aus."},
|
||||
*calls,
|
||||
],
|
||||
}
|
||||
result = await self.guard.inlet(body)
|
||||
self.assertEqual(result["tools"], [])
|
||||
self.assertIn("vorhandenen Ergebnisse", result["messages"][0]["content"])
|
||||
|
||||
|
||||
class AutoToolSelectorTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def asyncSetUp(self):
|
||||
self.module = _load("auto_tool_selector")
|
||||
self.selector = self.module.Filter()
|
||||
|
||||
async def _select(self, prompt: str, existing=None):
|
||||
body = {
|
||||
"model": "mikeai-medium",
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
}
|
||||
if existing is not None:
|
||||
body["tool_ids"] = existing
|
||||
return await self.selector.inlet(body)
|
||||
|
||||
async def test_homeassistant_is_selected_for_room_temperature(self):
|
||||
result = await self._select("Wie warm ist es gerade in der Küche?")
|
||||
self.assertEqual(
|
||||
result["tool_ids"], ["server:mcp:homeassistant-local"]
|
||||
)
|
||||
self.assertIn("Never call ha_list_states merely", result["messages"][0]["content"])
|
||||
self.assertIn("Never infer an automation entity_id", result["messages"][0]["content"])
|
||||
|
||||
async def test_hyphenated_homeassistant_and_tool_name_are_selected(self):
|
||||
result = await self._select(
|
||||
"Führe einen Home-Assistant-Test aus und nutze ha_list_states genau einmal."
|
||||
)
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:homeassistant-local"])
|
||||
|
||||
async def test_general_web_is_available_without_site_specific_rules(self):
|
||||
result = await self._select("Erkläre mir kurz, wie ein Fahrrad funktioniert.")
|
||||
self.assertTrue(result["features"]["web_search"])
|
||||
self.assertTrue(result["metadata"]["features"]["web_search"])
|
||||
self.assertNotIn("tool_ids", result)
|
||||
|
||||
async def test_weather_gets_native_web_and_compact_fallback(self):
|
||||
result = await self._select("Soll es heute in Rastatt regnen?")
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:web-general-local"])
|
||||
|
||||
async def test_youtube_channel_question_gets_general_web_fallback(self):
|
||||
result = await self._select(
|
||||
"Welches Video steht aktuell oben auf dem YouTube-Kanal The Proper People?"
|
||||
)
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:web-general-local"])
|
||||
|
||||
async def test_unraid_uses_readonly_not_mua(self):
|
||||
result = await self._select(
|
||||
"Welche Docker-Container laufen aktuell auf Unraid?"
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"], ["server:mcp:mua-readonly-local"]
|
||||
)
|
||||
self.assertNotIn("server:mcp:mua", result["tool_ids"])
|
||||
self.assertIn("bounded evidence ladder", result["messages"][0]["content"])
|
||||
self.assertIn("Do not dump complete configuration files", result["messages"][0]["content"])
|
||||
|
||||
async def test_explicit_unraid_update_gets_read_and_management_tools(self):
|
||||
result = await self._select(
|
||||
"Prüfe auf Unraid alle Docker-Updates, führe die Updates durch und kontrolliere danach den Zustand."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
["server:mcp:mua-readonly-local", "server:mcp:mua"],
|
||||
)
|
||||
self.assertIn("single batched update workflow", result["messages"][0]["content"])
|
||||
|
||||
async def test_unraid_update_question_stays_readonly(self):
|
||||
result = await self._select(
|
||||
"Gibt es auf Unraid Updates für Docker-Container? Bitte nur prüfen."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"], ["server:mcp:mua-readonly-local"]
|
||||
)
|
||||
|
||||
async def test_readonly_unraid_media_audit_does_not_attach_admin_or_arr(self):
|
||||
result = await self._select(
|
||||
"Prüfe auf Unraid, welche Folgen meiner Hörspielserie Die drei Fragezeichen fehlen. "
|
||||
"Nur lesen, ohne Downloads, Umbenennungen oder sonstige Änderungen."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"], ["server:mcp:mua-readonly-local"]
|
||||
)
|
||||
self.assertIn("unraid_files_inventory", result["messages"][0]["content"])
|
||||
|
||||
async def test_unraid_media_audit_with_deezer_enables_native_web(self):
|
||||
result = await self._select(
|
||||
"Prüfe auf Unraid, welche Folgen meiner Hörspielserie fehlen, ermittle die "
|
||||
"aktuelle offizielle Liste online und prüfe jede fehlende Folge bei Deezer. "
|
||||
"Nur lesen, ohne Downloads oder Änderungen."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
["server:mcp:mua-readonly-local", "server:mcp:web-general-local"],
|
||||
)
|
||||
self.assertTrue(result["features"]["web_search"])
|
||||
self.assertTrue(result["metadata"]["features"]["web_search"])
|
||||
self.assertIn("search_web/fetch_url", result["messages"][0]["content"])
|
||||
self.assertEqual(result["reasoning_effort"], "medium")
|
||||
|
||||
async def test_unraid_media_write_gets_portable_operator_and_web(self):
|
||||
result = await self._select(
|
||||
"Ermittle online das neueste Video, lade es mit yt-dlp auf dem Unraid-Host "
|
||||
"in einen temporären Ordner, konvertiere es mit ffmpeg und lege die fertige "
|
||||
"Datei im Share Transfer ab."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
[
|
||||
"server:mcp:athena-operator-local",
|
||||
"server:mcp:mua-readonly-local",
|
||||
"server:mcp:mua",
|
||||
"server:mcp:web-general-local",
|
||||
],
|
||||
)
|
||||
self.assertIn("direct MUA management or shell tool", result["messages"][0]["content"])
|
||||
self.assertIn("Do not ask the user to enable another tool", result["messages"][0]["content"])
|
||||
self.assertIn("start-status-result pattern", result["messages"][0]["content"])
|
||||
self.assertIn("transient-by-default dependency handling", result["messages"][0]["content"])
|
||||
self.assertIn("task-local copy under /tmp", result["messages"][0]["content"])
|
||||
self.assertTrue(result["metadata"]["mikeai_long_operator_task"])
|
||||
|
||||
async def test_generic_remote_host_file_operation_gets_operator(self):
|
||||
result = await self._select(
|
||||
"Führe auf dem Server ein vorhandenes Skript aus und speichere die neue Datei unter /data/export."
|
||||
)
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:athena-operator-local"])
|
||||
self.assertTrue(result["metadata"]["mikeai_long_operator_task"])
|
||||
|
||||
async def test_readonly_diagnostics_do_not_get_long_operator_budget(self):
|
||||
result = await self._select("Prüfe nur lesend den Zustand von Unraid.")
|
||||
self.assertNotIn("mikeai_long_operator_task", result["metadata"])
|
||||
|
||||
async def test_narrow_safety_clause_does_not_cancel_authorized_unraid_write(self):
|
||||
result = await self._select(
|
||||
"Lade das Video auf dem Unraid-Host herunter, konvertiere es und lege es "
|
||||
"im Transfer-Share ab. Installiere dabei kein Paket dauerhaft und ändere "
|
||||
"keine Unraid-Systemkonfiguration."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
[
|
||||
"server:mcp:athena-operator-local",
|
||||
"server:mcp:mua-readonly-local",
|
||||
"server:mcp:mua",
|
||||
],
|
||||
)
|
||||
|
||||
async def test_plain_download_advice_does_not_attach_operator(self):
|
||||
result = await self._select("Erkläre mir, wie ein Browser einen Download technisch durchführt.")
|
||||
self.assertNotIn("tool_ids", result)
|
||||
|
||||
async def test_voice_transcription_variants_select_unraid(self):
|
||||
result = await self._select(
|
||||
"Welche Dacher Contäner laufen aktuell auf dem Anrate Server?"
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"], ["server:mcp:mua-readonly-local"]
|
||||
)
|
||||
|
||||
async def test_navidrome_is_selected(self):
|
||||
result = await self._select(
|
||||
"Schau in Navidrome nach ähnlichen Titeln und meiner Playlist."
|
||||
)
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:navidrome-local"])
|
||||
|
||||
async def test_platform_context_is_selected(self):
|
||||
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-operator-local"])
|
||||
|
||||
async def test_athena_operator_is_selected_for_platform_work(self):
|
||||
result = await self._select(
|
||||
"Baue und deploye auf Athena einen neuen MCP-Container."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"], ["server:mcp:athena-operator-local"]
|
||||
)
|
||||
|
||||
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"
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
["server:mcp:github-local", "server:mcp:athena-operator-local"],
|
||||
)
|
||||
|
||||
async def test_existing_unraid_backend_gets_operator_and_runtime_evidence(self):
|
||||
result = await self._select(
|
||||
"Baue aus https://github.com/foo/deemix einen MCP. Deemix läuft bereits als Container auf Unraid; prüfe ihn zuerst."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
[
|
||||
"server:mcp:github-local",
|
||||
"server:mcp:athena-operator-local",
|
||||
"server:mcp:mua-readonly-local",
|
||||
],
|
||||
)
|
||||
self.assertIn("integrate or relay", result["messages"][0]["content"])
|
||||
|
||||
async def test_readonly_integration_plan_does_not_attach_unraid_admin(self):
|
||||
result = await self._select(
|
||||
"Analysiere https://github.com/foo/deemix und prüfe den auf Unraid "
|
||||
"laufenden Container und entwirf einen Athena-MCP. Erstelle keinen zweiten Container, ändere nichts "
|
||||
"und lies keine Secrets."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
[
|
||||
"server:mcp:github-local",
|
||||
"server:mcp:athena-operator-local",
|
||||
"server:mcp:mua-readonly-local",
|
||||
],
|
||||
)
|
||||
|
||||
async def test_github_and_explicit_web_adds_compact_general_web(self):
|
||||
result = await self._select(
|
||||
"Prüfe dieses GitHub Repository und suche zusätzlich im Netz nach Nutzerstimmen."
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
["server:mcp:github-local", "server:mcp:web-general-local"],
|
||||
)
|
||||
|
||||
async def test_plain_public_web_request_attaches_general_not_legacy_web(self):
|
||||
result = await self._select(
|
||||
"Suche im Netz auf MakerWorld einen Schlümpfe-Schlüsselanhänger."
|
||||
)
|
||||
self.assertNotIn("server:mcp:web-local", result.get("tool_ids", []))
|
||||
self.assertIn("server:mcp:web-general-local", result.get("tool_ids", []))
|
||||
|
||||
async def test_ebay_research_gets_general_marketplace_protocol(self):
|
||||
result = await self._select(
|
||||
"Suche auf eBay nach einem vollständigen Highscreen 386 PC mit Preis und Versand."
|
||||
)
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:web-general-local"])
|
||||
self.assertIn("Marketplace research protocol", result["messages"][0]["content"])
|
||||
self.assertIn("roughly three searches and five page fetches", result["messages"][0]["content"])
|
||||
self.assertTrue(result["metadata"]["mikeai_marketplace_research"])
|
||||
|
||||
async def test_generic_marketplace_research_does_not_need_site_rule(self):
|
||||
result = await self._select(
|
||||
"Finde auf einem Marktplatz aktuelle Angebote für einen gebrauchten Synthesizer."
|
||||
)
|
||||
self.assertEqual(result["tool_ids"], ["server:mcp:web-general-local"])
|
||||
self.assertIn("deduplicate by listing URL or item number", result["messages"][0]["content"])
|
||||
self.assertTrue(result["metadata"]["mikeai_marketplace_research"])
|
||||
|
||||
async def test_manual_tool_is_preserved(self):
|
||||
result = await self._select(
|
||||
"Prüfe Sonarr.", ["server:mcp:manually-selected"]
|
||||
)
|
||||
self.assertEqual(
|
||||
result["tool_ids"],
|
||||
["server:mcp:manually-selected", "server:mcp:arr-local"],
|
||||
)
|
||||
|
||||
async def test_plain_chat_gets_no_tools(self):
|
||||
result = await self._select("Erkläre mir den Unterschied zwischen RAM und SSD.")
|
||||
self.assertNotIn("tool_ids", result)
|
||||
|
||||
async def test_selection_adds_write_safety_rule(self):
|
||||
result = await self._select("Zeige mir die Home Assistant Automatisierungen.")
|
||||
self.assertEqual(result["messages"][0]["role"], "system")
|
||||
self.assertIn("Availability is not authorization", result["messages"][0]["content"])
|
||||
|
||||
|
||||
class MetricsTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_metrics_file_contains_no_chat_content_or_ids(self):
|
||||
module = _load("local_performance_metrics")
|
||||
metrics = module.Filter()
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
path = Path(directory) / "metrics.jsonl"
|
||||
metrics.valves.metrics_path = str(path)
|
||||
metadata = {"message_id": "secret-message-id", "chat_id": "secret-chat-id"}
|
||||
await metrics.inlet(
|
||||
{
|
||||
"model": "qwen-fast",
|
||||
"messages": [{"role": "user", "content": "private prompt"}],
|
||||
"tools": [{"name": "tool"}],
|
||||
},
|
||||
__metadata__=metadata,
|
||||
)
|
||||
await metrics.outlet(
|
||||
{
|
||||
"model": "qwen-fast",
|
||||
"messages": [
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": "private answer",
|
||||
"usage": {
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 4,
|
||||
"total_tokens": 14,
|
||||
},
|
||||
}
|
||||
],
|
||||
},
|
||||
__metadata__=metadata,
|
||||
)
|
||||
raw = path.read_text()
|
||||
record = json.loads(raw)
|
||||
self.assertEqual(record["prompt_tokens"], 10)
|
||||
self.assertNotIn("private", raw)
|
||||
self.assertNotIn("secret", raw)
|
||||
|
||||
|
||||
class SecretRedactionTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_only_tool_and_assistant_content_is_redacted(self):
|
||||
module = _load("secret_redaction")
|
||||
guard = module.Filter()
|
||||
token = "eyJ" + "A" * 24 + "." + "B" * 24 + "." + "C" * 16
|
||||
body = {
|
||||
"messages": [
|
||||
{"role": "user", "content": f"Absichtlich lokal nutzen: {token}"},
|
||||
{"role": "tool", "content": f'{{"api_key":"1234567890abcdef"}} {token}'},
|
||||
]
|
||||
}
|
||||
result = await guard.inlet(body)
|
||||
self.assertIn(token, result["messages"][0]["content"])
|
||||
self.assertNotIn(token, result["messages"][1]["content"])
|
||||
self.assertNotIn("1234567890abcdef", result["messages"][1]["content"])
|
||||
|
||||
outlet = {
|
||||
"messages": [
|
||||
{"role": "assistant", "content": "Bearer " + "abcdefghijklmnopqrstuvwxyz"}
|
||||
]
|
||||
}
|
||||
result = await guard.outlet(outlet)
|
||||
self.assertNotIn("abcdefghijklmnopqrstuvwxyz", result["messages"][0]["content"])
|
||||
|
||||
async def test_documentation_placeholders_and_paths_are_not_redacted(self):
|
||||
module = _load("secret_redaction")
|
||||
guard = module.Filter()
|
||||
content = "\n".join(
|
||||
[
|
||||
'ROUTER_API_KEY=${ROUTER_API_KEY:?required}',
|
||||
'API_KEY=/etc/mike-ai/router-api-key',
|
||||
'access_token=<YOUR_TOKEN>',
|
||||
'password=[REDACTED]',
|
||||
'API-Key: Inhalt von /etc/mike-ai/router-api-key',
|
||||
]
|
||||
)
|
||||
redacted, counts = guard._redact(content)
|
||||
self.assertEqual(redacted, content)
|
||||
self.assertEqual(counts, {})
|
||||
|
||||
async def test_notification_reports_category_and_origin_but_not_value(self):
|
||||
module = _load("secret_redaction")
|
||||
guard = module.Filter()
|
||||
events = []
|
||||
|
||||
async def emit(event):
|
||||
events.append(event)
|
||||
|
||||
secret = "realistic-secret-value-123456"
|
||||
body = {"messages": [{"role": "tool", "content": f"api_key={secret}"}]}
|
||||
result = await guard.inlet(body, __event_emitter__=emit)
|
||||
self.assertNotIn(secret, result["messages"][0]["content"])
|
||||
description = events[0]["data"]["description"]
|
||||
self.assertIn("API-Key: 1", description)
|
||||
self.assertIn("Werkzeugausgaben", description)
|
||||
self.assertIn("nicht protokolliert", description)
|
||||
self.assertNotIn(secret, description)
|
||||
|
||||
|
||||
class QuickActionTests(unittest.IsolatedAsyncioTestCase):
|
||||
async def asyncSetUp(self):
|
||||
self.module = _load_action("quick_actions")
|
||||
self.action = self.module.Action()
|
||||
self.body = {
|
||||
"id": "assistant-1",
|
||||
"model": "qwen-fast",
|
||||
"messages": [
|
||||
{"id": "user-1", "role": "user", "content": "Frage"},
|
||||
{"id": "assistant-1", "role": "assistant", "content": "Lange Antwort"},
|
||||
],
|
||||
}
|
||||
|
||||
async def test_summary_appends_to_selected_message(self):
|
||||
async def fake_completion(*args, **kwargs):
|
||||
return "- Kurze Antwort"
|
||||
|
||||
self.action._completion = fake_completion
|
||||
result = await self.action.action(self.body, "summary")
|
||||
self.assertEqual(result["messages"][0]["id"], "assistant-1")
|
||||
self.assertIn("### Kurzfassung", result["messages"][0]["content"])
|
||||
self.assertIn("Kurze Antwort", result["messages"][0]["content"])
|
||||
|
||||
async def test_source_check_uses_web_evidence(self):
|
||||
completions = iter(["Qwen Fakten", "Bestätigt: Aussage [https://example.invalid]"])
|
||||
|
||||
async def fake_completion(*args, **kwargs):
|
||||
return next(completions)
|
||||
|
||||
async def fake_web(query):
|
||||
self.assertEqual(query, "Qwen Fakten")
|
||||
return '{"sources":[{"url":"https://example.invalid"}]}'
|
||||
|
||||
self.action._completion = fake_completion
|
||||
self.action._web_research = fake_web
|
||||
result = await self.action.action(self.body, "sources")
|
||||
self.assertIn("### Quellenprüfung", result["messages"][0]["content"])
|
||||
self.assertIn("example.invalid", result["messages"][0]["content"])
|
||||
|
||||
async def test_markdown_copy_uses_browser_clipboard(self):
|
||||
calls = []
|
||||
|
||||
async def event_call(event):
|
||||
calls.append(event)
|
||||
return True
|
||||
|
||||
await self.action.action(self.body, "copy_markdown", __event_call__=event_call)
|
||||
self.assertEqual(calls[0]["type"], "execute")
|
||||
self.assertIn("navigator.clipboard.writeText", calls[0]["data"]["code"])
|
||||
self.assertIn("Lange Antwort", calls[0]["data"]["code"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,84 +0,0 @@
|
||||
#!/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():
|
||||
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_surface_has_only_explicit_read_tools(self):
|
||||
class FakeMcp:
|
||||
def __init__(self):
|
||||
self.names = []
|
||||
|
||||
def tool(self, **_kwargs):
|
||||
def decorate(function):
|
||||
self.names.append(function.__name__)
|
||||
return function
|
||||
return decorate
|
||||
|
||||
mcp = FakeMcp()
|
||||
self.module.register_radarr_tools(mcp)
|
||||
self.assertEqual(
|
||||
mcp.names,
|
||||
["radarr_find_movie", "radarr_movie_codec_inventory", "radarr_search_releases"],
|
||||
)
|
||||
self.assertNotIn("radarr_action", mcp.names)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,195 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Focused offline tests for the model-oriented Sonarr overlay."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
SOURCE = Path(__file__).parents[1] / "platform/mcp/patches/mcp_sonarr.py"
|
||||
|
||||
|
||||
def load_module():
|
||||
async def run_blocking(function, *args, **kwargs):
|
||||
kwargs.pop("service", None)
|
||||
return function(*args, **kwargs)
|
||||
|
||||
def dispatch(client, action, kwargs, **_options):
|
||||
return getattr(client, action)(**kwargs)
|
||||
|
||||
utilities = types.ModuleType("agent_utilities.mcp_utilities")
|
||||
utilities.dispatch = dispatch
|
||||
utilities.run_blocking = run_blocking
|
||||
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_sonarr_client = lambda: None
|
||||
sys.modules["arr_mcp"] = types.ModuleType("arr_mcp")
|
||||
sys.modules["arr_mcp.auth"] = auth
|
||||
spec = importlib.util.spec_from_file_location("sonarr_patch", SOURCE)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class FakeSonarrClient:
|
||||
def __init__(self, *, rejected: bool = False) -> None:
|
||||
self.posted = []
|
||||
self.release = {
|
||||
"guid": "exact-guid",
|
||||
"title": "Murder.She.Wrote.S07.German.AC3D.DL.1080p.WebHD.x265-FuN",
|
||||
"indexer": "Test Indexer",
|
||||
"indexerId": 7,
|
||||
"size": 27_600_000_000,
|
||||
"protocol": "usenet",
|
||||
"downloadAllowed": not rejected,
|
||||
"releaseGroup": "FuN",
|
||||
"seasonNumber": 7,
|
||||
"fullSeason": True,
|
||||
"rejections": ["Existing file has equal or better quality"] if rejected else [],
|
||||
}
|
||||
|
||||
def get_release(self, **_kwargs):
|
||||
return [dict(self.release)]
|
||||
|
||||
def get_episode(self, **_kwargs):
|
||||
return [
|
||||
{"id": 1, "seasonNumber": 7, "episodeNumber": 1, "hasFile": True},
|
||||
{"id": 2, "seasonNumber": 7, "episodeNumber": 2, "hasFile": False},
|
||||
]
|
||||
|
||||
def post_release(self, data=None, **kwargs):
|
||||
payload = data if data is not None else kwargs
|
||||
self.posted.append(payload)
|
||||
return payload
|
||||
|
||||
|
||||
class ReleaseGrabTests(unittest.IsolatedAsyncioTestCase):
|
||||
def setUp(self) -> None:
|
||||
self.module = load_module()
|
||||
self.module._APPROVALS.clear()
|
||||
|
||||
async def test_exact_release_requires_preview_and_ticket(self) -> None:
|
||||
client = FakeSonarrClient()
|
||||
scope = {"series_id": 42, "season_number": 7, "guid": "exact-guid"}
|
||||
preview = await self.module._preview_release_grab(client, scope)
|
||||
self.assertFalse(preview["download_started"])
|
||||
self.assertEqual(preview["existing_episode_files_in_season"], 1)
|
||||
self.assertTrue(preview["approval_ticket"])
|
||||
|
||||
result = await self.module._grab_release(
|
||||
client,
|
||||
{**scope, "confirm": True, "approval_ticket": preview["approval_ticket"]},
|
||||
)
|
||||
self.assertTrue(result["download_started"])
|
||||
self.assertFalse(result["replacement_guaranteed"])
|
||||
self.assertEqual(len(client.posted), 1)
|
||||
self.assertEqual(client.posted[0]["guid"], "exact-guid")
|
||||
|
||||
async def test_rejected_release_needs_force_in_preview(self) -> None:
|
||||
client = FakeSonarrClient(rejected=True)
|
||||
scope = {"series_id": 42, "season_number": 7, "guid": "exact-guid"}
|
||||
blocked = await self.module._preview_release_grab(client, scope)
|
||||
self.assertTrue(blocked["force_required"])
|
||||
self.assertIsNone(blocked["approval_ticket"])
|
||||
|
||||
approved = await self.module._preview_release_grab(client, {**scope, "force": True})
|
||||
self.assertFalse(approved["force_required"])
|
||||
self.assertTrue(approved["approval_ticket"])
|
||||
|
||||
async def test_guid_must_still_match_current_sonarr_results(self) -> None:
|
||||
client = FakeSonarrClient()
|
||||
with self.assertRaisesRegex(ValueError, "no longer present"):
|
||||
await self.module._preview_release_grab(
|
||||
client,
|
||||
{"series_id": 42, "season_number": 7, "guid": "different-guid"},
|
||||
)
|
||||
|
||||
async def test_search_can_limit_results_to_group_and_season_pack(self) -> None:
|
||||
client = FakeSonarrClient()
|
||||
client.get_release = lambda **_kwargs: [
|
||||
dict(client.release),
|
||||
{
|
||||
**client.release,
|
||||
"guid": "episode-guid",
|
||||
"title": "Mord.ist.ihr.Hobby.S07E02.German.1080p-FuN",
|
||||
"fullSeason": False,
|
||||
},
|
||||
{
|
||||
**client.release,
|
||||
"guid": "other-group",
|
||||
"title": "Mord.ist.ihr.Hobby.S07.German.1080p-HQC",
|
||||
"releaseGroup": "HQC",
|
||||
},
|
||||
]
|
||||
result = await self.module._search_releases(
|
||||
client,
|
||||
{
|
||||
"series_id": 42,
|
||||
"season_number": 7,
|
||||
"release_group": "FuN",
|
||||
"season_pack_only": True,
|
||||
},
|
||||
)
|
||||
self.assertEqual(result["results"]["total"], 1)
|
||||
self.assertEqual(result["results"]["items"][0]["guid"], "exact-guid")
|
||||
|
||||
def test_read_only_surface_is_explicit_and_has_no_generic_action(self) -> None:
|
||||
class FakeMcp:
|
||||
def __init__(self):
|
||||
self.names = []
|
||||
|
||||
def tool(self, **_kwargs):
|
||||
def decorate(function):
|
||||
self.names.append(function.__name__)
|
||||
return function
|
||||
return decorate
|
||||
|
||||
with patch.dict("os.environ", {"ARR_MCP_WRITE": "0"}):
|
||||
mcp = FakeMcp()
|
||||
self.module.register_sonarr_tools(mcp)
|
||||
self.assertEqual(
|
||||
mcp.names,
|
||||
[
|
||||
"sonarr_find_series",
|
||||
"sonarr_get_season_summary",
|
||||
"sonarr_search_releases",
|
||||
"sonarr_system_status",
|
||||
],
|
||||
)
|
||||
self.assertNotIn("sonarr_action", mcp.names)
|
||||
|
||||
def test_write_tools_are_registered_only_when_enabled(self) -> None:
|
||||
class FakeMcp:
|
||||
def __init__(self):
|
||||
self.names = []
|
||||
|
||||
def tool(self, **_kwargs):
|
||||
def decorate(function):
|
||||
self.names.append(function.__name__)
|
||||
return function
|
||||
return decorate
|
||||
|
||||
with patch.dict("os.environ", {"ARR_MCP_WRITE": "1"}):
|
||||
mcp = FakeMcp()
|
||||
self.module.register_sonarr_tools(mcp)
|
||||
self.assertIn("sonarr_preview_release_grab", mcp.names)
|
||||
self.assertIn("sonarr_grab_release", mcp.names)
|
||||
self.assertIn("sonarr_preview_episode_search", mcp.names)
|
||||
self.assertIn("sonarr_start_episode_search", mcp.names)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,148 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Regression tests for the compact web MCP facade."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import pathlib
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
SPEC = importlib.util.spec_from_file_location(
|
||||
"web_search_mcp", ROOT / "platform/web-search/web_search_mcp.py"
|
||||
)
|
||||
WEB = importlib.util.module_from_spec(SPEC)
|
||||
assert SPEC.loader
|
||||
SPEC.loader.exec_module(WEB)
|
||||
|
||||
|
||||
class WebSearchMcpTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
WEB._search_attempts.clear()
|
||||
|
||||
def test_tool_surface_stays_small_and_explicit(self) -> None:
|
||||
self.assertEqual(
|
||||
[tool["name"] for tool in WEB.TOOLS],
|
||||
["web_search", "web_read", "web_youtube", "web_compare", "web_shop", "web_research"],
|
||||
)
|
||||
|
||||
def test_current_queries_do_not_get_wikipedia_noise(self) -> None:
|
||||
with (
|
||||
mock.patch.object(WEB, "SEARXNG_URL", "http://searxng:8080"),
|
||||
mock.patch.object(WEB, "searxng_json", return_value={"results": []}),
|
||||
mock.patch.object(WEB, "wikipedia_search") as wikipedia,
|
||||
):
|
||||
results, _ = WEB.general_discovery("latest video The Proper People", 4)
|
||||
self.assertEqual(results, [])
|
||||
wikipedia.assert_not_called()
|
||||
|
||||
def test_empty_fresh_search_relaxes_once_and_marks_result(self) -> None:
|
||||
hit = {"title": "Current page", "url": "https://example.com/current"}
|
||||
with mock.patch.object(
|
||||
WEB,
|
||||
"general_discovery",
|
||||
side_effect=[([], []), ([hit], [])],
|
||||
) as discovery:
|
||||
result = WEB.web_search({
|
||||
"query": "current test release",
|
||||
"freshness": "week",
|
||||
"max_results": 3,
|
||||
})
|
||||
self.assertTrue(result["task_complete"])
|
||||
self.assertFalse(result["freshness_applied"])
|
||||
self.assertIn("unfiltered", result["backend_warning"])
|
||||
self.assertEqual(discovery.call_count, 2)
|
||||
|
||||
def test_related_search_budget_is_enforced(self) -> None:
|
||||
with mock.patch.object(WEB, "SEARCH_BUDGET_MAX_RELATED_CALLS", 2):
|
||||
self.assertTrue(WEB.consume_search_budget("latest Proper People video")[0])
|
||||
self.assertTrue(WEB.consume_search_budget("Proper People newest video")[0])
|
||||
self.assertFalse(WEB.consume_search_budget("newest video by Proper People")[0])
|
||||
|
||||
def test_youtube_feed_provides_order_and_dates(self) -> None:
|
||||
feed = b'''<?xml version="1.0" encoding="UTF-8"?>
|
||||
<feed xmlns:yt="http://www.youtube.com/xml/schemas/2015"
|
||||
xmlns="http://www.w3.org/2005/Atom">
|
||||
<entry><yt:videoId>new123</yt:videoId><title>Newest</title>
|
||||
<published>2026-08-23T12:00:00+00:00</published>
|
||||
<author><name>The Proper People</name></author></entry>
|
||||
<entry><yt:videoId>old456</yt:videoId><title>Older</title>
|
||||
<published>2026-08-10T12:00:00+00:00</published>
|
||||
<author><name>The Proper People</name></author></entry>
|
||||
</feed>'''
|
||||
with mock.patch.object(WEB, "fetch_public_bytes", return_value=feed):
|
||||
rows = WEB.youtube_feed_records(
|
||||
"https://www.youtube.com/channel/UCcem9I78ybZLHLRUlkUO3sw", 2
|
||||
)
|
||||
self.assertEqual([row["title"] for row in rows], ["Newest", "Older"])
|
||||
self.assertEqual(rows[0]["published_at"], "2026-08-23T12:00:00+00:00")
|
||||
|
||||
def test_latest_youtube_is_one_bounded_specialist_operation(self) -> None:
|
||||
row = {
|
||||
"title": "Newest",
|
||||
"url": "https://www.youtube.com/watch?v=new123",
|
||||
"source_kind": "youtube_channel_feed",
|
||||
}
|
||||
with (
|
||||
mock.patch.object(WEB, "resolve_youtube_channel", return_value="https://www.youtube.com/channel/UCcem9I78ybZLHLRUlkUO3sw"),
|
||||
mock.patch.object(WEB, "youtube_feed_records", return_value=[row]),
|
||||
mock.patch.object(WEB, "run_ytdlp") as ytdlp,
|
||||
):
|
||||
result = WEB.web_youtube({"query": "The Proper People", "mode": "latest"})
|
||||
self.assertTrue(result["task_complete"])
|
||||
self.assertEqual(result["results"][0]["title"], "Newest")
|
||||
ytdlp.assert_not_called()
|
||||
|
||||
def test_latest_long_youtube_uses_verified_videos_tab(self) -> None:
|
||||
row = {
|
||||
"title": "Newest long video",
|
||||
"url": "https://www.youtube.com/watch?v=long123",
|
||||
"content_type": "long",
|
||||
"content_type_verified": True,
|
||||
"source_kind": "youtube_videos_tab",
|
||||
}
|
||||
with (
|
||||
mock.patch.object(
|
||||
WEB,
|
||||
"resolve_youtube_channel",
|
||||
return_value="https://www.youtube.com/channel/UCcem9I78ybZLHLRUlkUO3sw",
|
||||
),
|
||||
mock.patch.object(WEB, "youtube_tab_records", return_value=[row]) as tab,
|
||||
mock.patch.object(WEB, "youtube_feed_records") as feed,
|
||||
):
|
||||
result = WEB.web_youtube({
|
||||
"query": "The Proper People",
|
||||
"mode": "latest",
|
||||
"content_type": "long",
|
||||
})
|
||||
tab.assert_called_once_with(
|
||||
"https://www.youtube.com/channel/UCcem9I78ybZLHLRUlkUO3sw", "long", 5
|
||||
)
|
||||
feed.assert_not_called()
|
||||
self.assertEqual(result["content_type_filter"], "long")
|
||||
self.assertTrue(result["results"][0]["content_type_verified"])
|
||||
|
||||
def test_youtube_rejects_content_filter_outside_latest_mode(self) -> None:
|
||||
with self.assertRaisesRegex(ValueError, "only supported with mode=latest"):
|
||||
WEB.web_youtube({
|
||||
"query": "The Proper People",
|
||||
"mode": "search",
|
||||
"content_type": "long",
|
||||
})
|
||||
|
||||
def test_web_read_does_not_consume_search_loop_budget(self) -> None:
|
||||
page = {"url": "https://example.com/a", "page_evidence": ["Evidence"]}
|
||||
with mock.patch.object(WEB, "scrape", return_value=[page]):
|
||||
result = json.loads(WEB.call_tool("web_read", {
|
||||
"url": "https://example.com/a",
|
||||
"question": "What does this page say?",
|
||||
}))
|
||||
self.assertTrue(result["task_complete"])
|
||||
self.assertEqual(WEB._search_attempts, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user