feat(mcp): expose optional backend-local app discovery
This commit is contained in:
162
hermes_cli/mcp_app_detection.py
Normal file
162
hermes_cli/mcp_app_detection.py
Normal file
@@ -0,0 +1,162 @@
|
||||
"""Read-only, backend-local application signals for reviewed MCP catalog labels.
|
||||
|
||||
This module never runs apps, loads a terminal backend or contacts an MCP server.
|
||||
Only curated matches may leave it; inventory names and paths remain private.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import sys
|
||||
import time
|
||||
|
||||
_MAX_APPLICATIONS = 16
|
||||
_APP_LABEL = re.compile(r"[A-Za-z0-9][A-Za-z0-9 ._+-]{0,79}")
|
||||
|
||||
|
||||
def validate_applications(labels: object) -> list[str]:
|
||||
"""Accept bounded display labels/aliases, never paths, commands or regexes."""
|
||||
if not isinstance(labels, list) or len(labels) > _MAX_APPLICATIONS:
|
||||
raise ValueError("suggest.applications must be a list of at most 16 app labels")
|
||||
for label in labels:
|
||||
if (not isinstance(label, str) or not _APP_LABEL.fullmatch(label)
|
||||
or label != label.strip() or ".." in label or " --" in label):
|
||||
raise ValueError("suggest.applications must contain safe app labels (1–80 characters)")
|
||||
return list(labels)
|
||||
|
||||
|
||||
def _application_roots() -> list[Path]:
|
||||
# OS home, NOT HERMES_HOME: profiles do not change the backend machine's apps.
|
||||
home = Path.home()
|
||||
if sys.platform == "darwin":
|
||||
return [Path("/Applications"), Path("/System/Applications"), home / "Applications"]
|
||||
if sys.platform == "linux":
|
||||
return [Path("/usr/share/applications"), Path("/usr/local/share/applications"),
|
||||
home / ".local/share/applications"]
|
||||
if sys.platform == "win32":
|
||||
return [Path(os.environ.get("ProgramFiles", "C:/Program Files")),
|
||||
Path(os.environ.get("ProgramFiles(x86)", "C:/Program Files (x86)")),
|
||||
Path(os.environ.get("LOCALAPPDATA", str(home / "AppData/Local"))) / "Programs"]
|
||||
return []
|
||||
|
||||
|
||||
class _DiscoveryLimit(Exception):
|
||||
"""A partial scan cannot claim absence."""
|
||||
|
||||
|
||||
class _Scan:
|
||||
def __init__(self, labels: set[str]):
|
||||
self.patterns = {label: re.compile(r"(?<!\w)" + re.escape(label) + r"(?!\w)", re.I)
|
||||
for label in labels}
|
||||
self.matched: set[str] = set()
|
||||
self.readable = False
|
||||
self.unavailable = False
|
||||
self.remaining = 4096
|
||||
self.directories = 128
|
||||
self.deadline = time.monotonic() + 2.0
|
||||
|
||||
def tick(self):
|
||||
self.remaining -= 1
|
||||
if self.remaining < 0 or time.monotonic() > self.deadline:
|
||||
raise _DiscoveryLimit
|
||||
|
||||
def record(self, name: str):
|
||||
# Never retain the inventory, only which reviewed labels matched it.
|
||||
self.matched.update(label for label, pattern in self.patterns.items() if pattern.search(name))
|
||||
|
||||
def directory(self, root: Path, depth: int = 0):
|
||||
self.directories -= 1
|
||||
if self.directories < 0:
|
||||
raise _DiscoveryLimit
|
||||
try:
|
||||
with os.scandir(root) as children:
|
||||
self.readable = True
|
||||
for child in children:
|
||||
self.tick()
|
||||
if child.is_symlink():
|
||||
continue
|
||||
is_dir = child.is_dir(follow_symlinks=False)
|
||||
if sys.platform == "darwin" and is_dir and child.name.lower().endswith(".app"):
|
||||
self.record(child.name[:-4])
|
||||
continue # Never descend into bundles or run their executables.
|
||||
if sys.platform == "win32" and is_dir:
|
||||
self.record(child.name)
|
||||
if sys.platform == "linux" and child.name.endswith(".desktop") and child.is_file(follow_symlinks=False):
|
||||
self.desktop(Path(child.path))
|
||||
if is_dir and depth < 2:
|
||||
self.directory(Path(child.path), depth + 1)
|
||||
except FileNotFoundError:
|
||||
pass # An optional standard root need not exist.
|
||||
except OSError:
|
||||
self.unavailable = True
|
||||
|
||||
def desktop(self, path: Path):
|
||||
self.record(path.stem)
|
||||
# Name is data; deliberately ignore Exec, TryExec, URLs and localized commands.
|
||||
with path.open(encoding="utf-8", errors="replace") as stream:
|
||||
text = stream.read(32769)
|
||||
if len(text) > 32768:
|
||||
self.unavailable = True
|
||||
return
|
||||
in_entry = False
|
||||
for line in text.splitlines():
|
||||
if line.startswith("["):
|
||||
in_entry = line.strip() == "[Desktop Entry]"
|
||||
elif in_entry and line.startswith("Name="):
|
||||
self.record(line[5:])
|
||||
|
||||
def path_candidates(self):
|
||||
# Exact single-token candidates only. No multiword-to-command heuristics.
|
||||
candidates = {label: tuple(dict.fromkeys((label, label.lower()))) for label in self.patterns
|
||||
if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,79}", label)}
|
||||
parts = os.environ.get("PATH", "").split(os.pathsep)
|
||||
if len(parts) > 64:
|
||||
self.unavailable = True
|
||||
for part in dict.fromkeys(parts[:64]):
|
||||
if not part or len(part) > 4096 or not Path(part).is_absolute():
|
||||
continue # Never search the working directory.
|
||||
try:
|
||||
self.tick()
|
||||
with os.scandir(part):
|
||||
self.readable = True
|
||||
for label, spellings in candidates.items():
|
||||
self.tick()
|
||||
# Passing an absolute candidate also prevents Windows which() from
|
||||
# silently prepending the current directory to an explicit PATH.
|
||||
if any(shutil.which(str(Path(part) / candidate)) for candidate in spellings):
|
||||
self.matched.add(label)
|
||||
except FileNotFoundError:
|
||||
continue
|
||||
except OSError:
|
||||
self.unavailable = True
|
||||
|
||||
|
||||
def discover_catalog_apps(applications: dict[str, list[str]]) -> dict:
|
||||
"""Bounded backend-native signals, not proof an integration is configured.
|
||||
|
||||
``ok`` means the supported scan completed, not an exhaustive OS inventory.
|
||||
A missing permission, exhausted budget or unsupported OS is ``unavailable``;
|
||||
positive matches still stand, but empty arrays then mean unknown.
|
||||
"""
|
||||
criteria = {name: validate_applications(labels) for name, labels in applications.items()}
|
||||
labels = {label for values in criteria.values() for label in values}
|
||||
supported = sys.platform in {"darwin", "linux", "win32"} and len(labels) <= 256
|
||||
scan = _Scan(labels if supported else set())
|
||||
if supported:
|
||||
try:
|
||||
for root in _application_roots():
|
||||
scan.directory(root)
|
||||
scan.path_candidates()
|
||||
except (OSError, RuntimeError, _DiscoveryLimit):
|
||||
scan.unavailable = True
|
||||
else:
|
||||
scan.unavailable = True
|
||||
return {
|
||||
"matches": {name: list(dict.fromkeys(label for label in labels if label in scan.matched))
|
||||
for name, labels in criteria.items()},
|
||||
"discovery": {"scope": "backend", "status": "unavailable" if scan.unavailable or not scan.readable else "ok",
|
||||
"platform": sys.platform},
|
||||
}
|
||||
@@ -90,6 +90,9 @@ class SuggestSpec:
|
||||
|
||||
keywords: List[str] = field(default_factory=list) # lowercase whole-word/phrase triggers
|
||||
hosts: List[str] = field(default_factory=list) # hostname suffixes ("atlassian.net")
|
||||
applications: List[str] = field(default_factory=list) # reviewed local app labels/aliases
|
||||
examples: List[str] = field(default_factory=list) # capability examples, not executable instructions
|
||||
requires_app: bool = False # local app prerequisite, unlike cloud services with desktop clients
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -209,12 +212,26 @@ def _parse_suggest(path: Path, suggest_raw: Any) -> Optional[SuggestSpec]:
|
||||
hosts_raw = suggest_raw.get("hosts") or []
|
||||
_require_str_list(path, "suggest.keywords", kw_raw, non_empty=True)
|
||||
_require_str_list(path, "suggest.hosts", hosts_raw, non_empty=True)
|
||||
if not kw_raw and not hosts_raw:
|
||||
raise CatalogError(f"{path}: 'suggest' requires at least one keyword or host")
|
||||
from hermes_cli.mcp_app_detection import validate_applications
|
||||
|
||||
try:
|
||||
applications = validate_applications(suggest_raw.get("applications", []))
|
||||
except ValueError as exc:
|
||||
raise CatalogError(f"{path}: {exc}") from exc
|
||||
examples = suggest_raw.get("examples", [])
|
||||
_require_str_list(path, "suggest.examples", examples, non_empty=True)
|
||||
if len(examples) > 6 or any(len(e) > 240 or not e.isprintable() for e in examples):
|
||||
raise CatalogError(f"{path}: suggest.examples allows at most 6 single-line examples of 240 characters")
|
||||
requires_app = suggest_raw.get("requires_app", False)
|
||||
if not isinstance(requires_app, bool) or (requires_app and not applications):
|
||||
raise CatalogError(f"{path}: suggest.requires_app must be a boolean, with applications when true")
|
||||
if not kw_raw and not hosts_raw and not applications:
|
||||
raise CatalogError(f"{path}: 'suggest' requires at least one keyword, host or application")
|
||||
# Matching is case-insensitive whole-word / host-suffix: store lowercase so UIs needn't re-normalize.
|
||||
return SuggestSpec(
|
||||
keywords=[k.strip().lower() for k in kw_raw],
|
||||
hosts=[h.strip().lower().lstrip(".") for h in hosts_raw])
|
||||
hosts=[h.strip().lower().lstrip(".") for h in hosts_raw],
|
||||
applications=applications, examples=examples, requires_app=requires_app)
|
||||
|
||||
|
||||
def _parse_install(path: Path, install_raw: Any) -> Optional[InstallSpec]:
|
||||
|
||||
@@ -366,7 +366,12 @@ def _catalog_entry_json(entry: Any, installed: bool, enabled: bool) -> Dict[str,
|
||||
"post_install": entry.post_install or "",
|
||||
# Composer-suggestion triggers (desktop brand pills), only when the
|
||||
# manifest declares a `suggest` block.
|
||||
"suggest": {"keywords": list(entry.suggest.keywords), "hosts": list(entry.suggest.hosts)} if entry.suggest else None,
|
||||
"suggest": {
|
||||
"keywords": list(entry.suggest.keywords), "hosts": list(entry.suggest.hosts),
|
||||
"applications": list(getattr(entry.suggest, "applications", [])),
|
||||
"examples": list(getattr(entry.suggest, "examples", [])),
|
||||
"requires_app": getattr(entry.suggest, "requires_app", False),
|
||||
} if entry.suggest else None,
|
||||
"needs_install": install is not None,
|
||||
"installed": installed,
|
||||
"enabled": enabled,
|
||||
@@ -374,9 +379,10 @@ def _catalog_entry_json(entry: Any, installed: bool, enabled: bool) -> Dict[str,
|
||||
|
||||
|
||||
@router.get("/api/mcp/catalog")
|
||||
async def list_mcp_catalog(profile: Optional[str] = None):
|
||||
async def list_mcp_catalog(profile: Optional[str] = None, detect_apps: bool = False):
|
||||
"""Browse the Nous-approved MCP catalog (optional-mcps/ manifests), each
|
||||
entry annotated with installed/enabled state for ``profile``."""
|
||||
entry annotated with installed/enabled state for ``profile``. Opt-in app
|
||||
signals describe this backend machine, never the client or terminal sandbox."""
|
||||
with http_failure("mcp_catalog import failed", 500, "Catalog unavailable"):
|
||||
from hermes_cli import mcp_catalog
|
||||
|
||||
@@ -402,7 +408,27 @@ async def list_mcp_catalog(profile: Optional[str] = None):
|
||||
diagnostics = [{"name": n, "kind": k, "message": m} for (n, k, m) in mcp_catalog.catalog_diagnostics()]
|
||||
except Exception:
|
||||
pass
|
||||
return {"entries": entries, "diagnostics": diagnostics}
|
||||
result = {"entries": entries, "diagnostics": diagnostics}
|
||||
if detect_apps:
|
||||
import sys
|
||||
|
||||
try:
|
||||
from hermes_cli.mcp_app_detection import discover_catalog_apps
|
||||
|
||||
# Discovery is read-only and backend-local. Keep filesystem work off the
|
||||
# event loop and OUTSIDE the profile/skills lock used for config reads.
|
||||
detected = await asyncio.to_thread(discover_catalog_apps, {
|
||||
entry["name"]: (entry["suggest"] or {}).get("applications", []) for entry in entries
|
||||
})
|
||||
except Exception:
|
||||
_log.warning("Backend application discovery unavailable")
|
||||
detected = {"matches": {}, "discovery": {
|
||||
"scope": "backend", "status": "unavailable", "platform": sys.platform,
|
||||
}}
|
||||
for entry in entries:
|
||||
entry["detected_apps"] = detected["matches"].get(entry["name"], [])
|
||||
result["discovery"] = detected["discovery"]
|
||||
return result
|
||||
|
||||
|
||||
@router.post("/api/mcp/catalog/install")
|
||||
|
||||
68
tests/hermes_cli/test_mcp_app_detection.py
Normal file
68
tests/hermes_cli/test_mcp_app_detection.py
Normal file
@@ -0,0 +1,68 @@
|
||||
"""Backend-native, read-only catalog app discovery; no app or MCP is launched."""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_native_discovery_returns_only_curated_boundary_matches(tmp_path, monkeypatch):
|
||||
from hermes_cli import mcp_app_detection as detection
|
||||
|
||||
root = tmp_path / "applications"
|
||||
root.mkdir()
|
||||
if sys.platform == "darwin":
|
||||
(root / "Vendor" / "FIXTURE Paint 5.1.app").mkdir(parents=True)
|
||||
(root / "SuperFixtureCode.app").mkdir()
|
||||
(root / "Personal Secret.app").mkdir()
|
||||
elif sys.platform == "win32":
|
||||
(root / "Vendor" / "FIXTURE Paint 5.1").mkdir(parents=True)
|
||||
(root / "SuperFixtureCode").mkdir()
|
||||
(root / "Personal Secret").mkdir()
|
||||
else:
|
||||
(root / "org.example.editor.desktop").write_text(
|
||||
"[Desktop Entry]\nName=FIXTURE Paint 5.1\nExec=do-not-run\n", encoding="utf-8")
|
||||
(root / "SuperFixtureCode.desktop").write_text("[Desktop Entry]\nName=Personal Secret\n")
|
||||
monkeypatch.setattr(detection, "_application_roots", lambda: [root])
|
||||
monkeypatch.setenv("PATH", str(tmp_path / "bin"))
|
||||
(tmp_path / "bin").mkdir()
|
||||
criteria = {"paint": ["Fixture Paint", "Missing App"], "code": ["FixtureCode"], "legacy": []}
|
||||
result = detection.discover_catalog_apps(criteria)
|
||||
assert result == {"matches": {"paint": ["Fixture Paint"], "code": [], "legacy": []},
|
||||
"discovery": {"scope": "backend", "status": "ok", "platform": sys.platform}}
|
||||
# A PATH signal is exact, not a substring or shell invocation. Executable content is never run.
|
||||
candidate = tmp_path / "bin" / ("fixturecode.exe" if os.name == "nt" else "fixturecode")
|
||||
candidate.write_text("this is not an executable program", encoding="utf-8")
|
||||
candidate.chmod(0o755)
|
||||
assert detection.discover_catalog_apps(criteria)["matches"]["code"] == ["FixtureCode"]
|
||||
|
||||
|
||||
def test_discovery_failure_or_budget_is_unknown_not_absence(tmp_path, monkeypatch):
|
||||
from hermes_cli import mcp_app_detection as detection
|
||||
|
||||
monkeypatch.setattr(detection, "_application_roots", lambda: [tmp_path])
|
||||
monkeypatch.setenv("PATH", "")
|
||||
criteria = {f"entry{i}": [f"App{i}"] for i in range(300)}
|
||||
limited = detection.discover_catalog_apps(criteria)
|
||||
assert limited["discovery"]["status"] == "unavailable"
|
||||
assert all(matches == [] for matches in limited["matches"].values())
|
||||
|
||||
criteria = {"demo": ["Fixture Paint"]}
|
||||
assert detection.discover_catalog_apps(criteria)["discovery"]["status"] == "ok"
|
||||
(tmp_path / "ordinary-file").touch()
|
||||
monkeypatch.setattr(detection, "_application_roots", lambda: [tmp_path / "ordinary-file"])
|
||||
assert detection.discover_catalog_apps(criteria)["discovery"]["status"] == "unavailable"
|
||||
monkeypatch.setattr(detection, "_application_roots", lambda: [tmp_path / "missing"])
|
||||
assert detection.discover_catalog_apps(criteria)["discovery"]["status"] == "unavailable"
|
||||
|
||||
def denied(_path):
|
||||
raise PermissionError("private filesystem path must not escape")
|
||||
|
||||
monkeypatch.setattr(detection.os, "scandir", denied)
|
||||
assert detection.discover_catalog_apps(criteria) == {
|
||||
"matches": {"demo": []},
|
||||
"discovery": {"scope": "backend", "status": "unavailable", "platform": sys.platform},
|
||||
}
|
||||
with pytest.raises(ValueError, match="suggest.applications"):
|
||||
detection.discover_catalog_apps({"demo": ["/bin/echo"]})
|
||||
@@ -144,6 +144,57 @@ class TestManifestParsing:
|
||||
assert sg.keywords == ["jira", "confluence"]
|
||||
assert sg.hosts == ["atlassian.net", "atlassian.com"]
|
||||
|
||||
def test_suggest_onboarding_metadata_is_additive(self, catalog_dir):
|
||||
from hermes_cli.mcp_catalog import _build_server_config, _parse_manifest
|
||||
from hermes_cli.web_routers.mcp import _catalog_entry_json
|
||||
|
||||
triggers = {"keywords": ["Demo "], "hosts": [".Example.com"]}
|
||||
path = _write_manifest(catalog_dir, "demo", _basic_manifest(suggest=triggers))
|
||||
legacy = _parse_manifest(path)
|
||||
enriched = {**triggers, "applications": ["Blender", "Visual Studio Code"],
|
||||
"examples": ["Create a scene from this sketch."], "future_hint": "ignored"}
|
||||
_write_manifest(catalog_dir, "demo", _basic_manifest(suggest=enriched))
|
||||
entry = _parse_manifest(path)
|
||||
assert entry.suggest is not None and legacy.suggest is not None
|
||||
assert entry.suggest.applications == enriched["applications"]
|
||||
assert entry.suggest.examples == enriched["examples"]
|
||||
assert entry.suggest.keywords == legacy.suggest.keywords == ["demo"]
|
||||
assert entry.suggest.hosts == legacy.suggest.hosts == ["example.com"]
|
||||
assert legacy.suggest.applications == legacy.suggest.examples == []
|
||||
assert _catalog_entry_json(entry, False, False)["suggest"] == {
|
||||
"keywords": ["demo"], "hosts": ["example.com"],
|
||||
"applications": enriched["applications"], "examples": enriched["examples"],
|
||||
"requires_app": False,
|
||||
}
|
||||
assert _build_server_config(entry, None) == _build_server_config(legacy, None)
|
||||
_write_manifest(catalog_dir, "demo", _basic_manifest(suggest={"applications": ["Blender"]}))
|
||||
assert _parse_manifest(path).suggest.applications == ["Blender"]
|
||||
|
||||
def test_suggest_discovery_metadata_is_bounded_data(self, catalog_dir):
|
||||
from hermes_cli.mcp_catalog import CatalogError, _parse_manifest
|
||||
from hermes_cli.web_routers.mcp import _catalog_entry_json
|
||||
|
||||
def parse(**metadata):
|
||||
path = _write_manifest(catalog_dir, "demo", _basic_manifest(
|
||||
suggest={"keywords": ["demo"], **metadata}))
|
||||
return _parse_manifest(path)
|
||||
|
||||
entry = parse(applications=["Blender"], requires_app=True)
|
||||
assert entry.suggest is not None and entry.suggest.requires_app is True
|
||||
assert _catalog_entry_json(entry, False, False)["suggest"]["requires_app"] is True
|
||||
assert parse().suggest.requires_app is False
|
||||
for metadata in (
|
||||
{"applications": "Blender"}, {"applications": [None]}, {"applications": ["/Applications/Blender.app"]},
|
||||
{"applications": ["../blender"]}, {"applications": ["C:\\Blender"]}, {"applications": [".*"]},
|
||||
{"applications": ["blender; id"]}, {"applications": ["--help"]}, {"applications": ["x" * 81]},
|
||||
{"applications": ["Blender"] * 17}, {"applications": [" Blender"]}, {"applications": ["\n"]},
|
||||
{"examples": "example"}, {"examples": [""]}, {"examples": ["x" * 241]},
|
||||
{"examples": ["x"] * 7}, {"examples": ["one\ntwo"]}, {"requires_app": "true"},
|
||||
{"requires_app": 1}, {"requires_app": True},
|
||||
):
|
||||
with pytest.raises(CatalogError, match="suggest"):
|
||||
parse(**metadata)
|
||||
|
||||
def test_suggest_keywords_only_is_valid(self, catalog_dir):
|
||||
_write_manifest(catalog_dir, "demo", _basic_manifest(suggest={"keywords": ["demo"]}))
|
||||
from hermes_cli.mcp_catalog import list_catalog
|
||||
|
||||
130
tests/hermes_cli/test_mcp_catalog_discovery.py
Normal file
130
tests/hermes_cli/test_mcp_catalog_discovery.py
Normal file
@@ -0,0 +1,130 @@
|
||||
"""Catalog HTTP contract: opt-in backend signals with real A/B/A profile config I/O."""
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def catalog_client(tmp_path, monkeypatch):
|
||||
home = tmp_path / ".hermes"
|
||||
other = home / "profiles" / "b"
|
||||
other.mkdir(parents=True)
|
||||
monkeypatch.setattr(Path, "home", lambda: tmp_path)
|
||||
monkeypatch.setenv("HERMES_HOME", str(home))
|
||||
monkeypatch.setenv("PATH", "")
|
||||
for directory, servers in ((home, {"demo": {"command": "unused", "enabled": False}}),
|
||||
(other, {"demo": {"command": "unused", "enabled": True},
|
||||
"legacy": {"command": "unused"}})):
|
||||
(directory / "config.yaml").write_text(yaml.safe_dump(
|
||||
{"mcp_servers": servers, "terminal": {"backend": "docker"}}), encoding="utf-8")
|
||||
catalog = tmp_path / "catalog"
|
||||
for name, suggest in (("demo", {"keywords": ["demo"], "applications": ["Fixture Paint"],
|
||||
"examples": ["Draw a picture."], "requires_app": True}),
|
||||
("legacy", {"keywords": ["legacy"]})):
|
||||
entry = catalog / name
|
||||
entry.mkdir(parents=True)
|
||||
(entry / "manifest.yaml").write_text(yaml.safe_dump({
|
||||
"manifest_version": 1, "name": name, "description": "Fixture entry",
|
||||
"transport": {"type": "stdio", "command": "must-not-run"}, "suggest": suggest,
|
||||
}), encoding="utf-8")
|
||||
monkeypatch.setenv("HERMES_OPTIONAL_MCPS", str(catalog))
|
||||
|
||||
from agent import secret_scope
|
||||
from tui_gateway import launch_profile_policy
|
||||
monkeypatch.setattr(secret_scope, "_MULTIPLEX_ACTIVE", False)
|
||||
monkeypatch.setattr(launch_profile_policy, "_snapshot", None)
|
||||
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN, app
|
||||
client = TestClient(app)
|
||||
client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
return client, home, other
|
||||
|
||||
|
||||
def test_catalog_detection_is_opt_in_and_preserves_profile_state(catalog_client, tmp_path, monkeypatch):
|
||||
from hermes_cli import mcp_app_detection as detection
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
client, home, other = catalog_client
|
||||
import socket
|
||||
import subprocess
|
||||
|
||||
def forbidden(*_args, **_kwargs):
|
||||
pytest.fail("Catalog discovery must not execute apps, install or probe the network")
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", forbidden)
|
||||
monkeypatch.setattr(socket, "create_connection", forbidden)
|
||||
app_root = tmp_path / "Applications"
|
||||
app_root.mkdir()
|
||||
if sys.platform == "darwin":
|
||||
(app_root / "Fixture Paint.app").mkdir()
|
||||
elif sys.platform == "win32":
|
||||
(app_root / "Fixture Paint").mkdir()
|
||||
else:
|
||||
(app_root / "fixture.desktop").write_text("[Desktop Entry]\nName=Fixture Paint\nExec=must-not-run\n")
|
||||
monkeypatch.setattr(detection, "_application_roots", lambda: [app_root])
|
||||
calls = []
|
||||
discover = detection.discover_catalog_apps
|
||||
|
||||
def checked_discover(criteria):
|
||||
with pytest.raises(RuntimeError):
|
||||
asyncio.get_running_loop() # Scans stay off the event loop.
|
||||
assert get_hermes_home() == home # No named-profile scope/skills lock spans discovery.
|
||||
calls.append(criteria)
|
||||
return discover(criteria)
|
||||
|
||||
monkeypatch.setattr(detection, "discover_catalog_apps", checked_discover)
|
||||
snapshots = {directory: (directory / "config.yaml").read_bytes() for directory in (home, other)}
|
||||
for profile, enabled, legacy_installed in ((None, False, False), ("b", True, True), (None, False, False)):
|
||||
params = {"profile": profile} if profile else {}
|
||||
before = len(calls)
|
||||
old = client.get("/api/mcp/catalog", params=params)
|
||||
assert old.status_code == 200, old.text
|
||||
assert len(calls) == before
|
||||
assert "discovery" not in old.json()
|
||||
assert all("detected_apps" not in entry for entry in old.json()["entries"])
|
||||
new = client.get("/api/mcp/catalog", params={**params, "detect_apps": "true", "future_hint": "ignored"})
|
||||
assert new.status_code == 200, new.text
|
||||
data = new.json()
|
||||
assert data["discovery"] == {"scope": "backend", "status": "ok", "platform": sys.platform}
|
||||
entries = {entry["name"]: entry for entry in data["entries"]}
|
||||
assert entries["demo"]["detected_apps"] == ["Fixture Paint"]
|
||||
assert entries["demo"]["installed"] is True and entries["demo"]["enabled"] is enabled
|
||||
assert entries["legacy"]["detected_apps"] == []
|
||||
assert entries["legacy"]["installed"] is legacy_installed
|
||||
assert entries["legacy"]["suggest"]["applications"] == []
|
||||
assert entries["legacy"]["suggest"]["requires_app"] is False
|
||||
for entry in data["entries"]:
|
||||
entry.pop("detected_apps")
|
||||
data.pop("discovery")
|
||||
assert data == old.json()
|
||||
assert len(calls) == before + 1
|
||||
assert {directory: (directory / "config.yaml").read_bytes() for directory in snapshots} == snapshots
|
||||
assert client.get("/api/mcp/catalog", params={"profile": "missing", "detect_apps": True}).status_code == 404
|
||||
|
||||
|
||||
def test_catalog_discovery_failure_retains_entries_as_unknown(catalog_client, monkeypatch):
|
||||
from hermes_cli import mcp_app_detection as detection
|
||||
|
||||
client, _home, _other = catalog_client
|
||||
calls = []
|
||||
|
||||
def failed_discovery(criteria):
|
||||
calls.append(criteria)
|
||||
raise OSError("private path must not be returned")
|
||||
|
||||
monkeypatch.setattr(detection, "discover_catalog_apps", failed_discovery)
|
||||
old = client.get("/api/mcp/catalog?detect_apps=false").json()
|
||||
assert calls == []
|
||||
response = client.get("/api/mcp/catalog?detect_apps=true")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["discovery"] == {"scope": "backend", "status": "unavailable", "platform": sys.platform}
|
||||
assert all(entry.pop("detected_apps") == [] for entry in data["entries"])
|
||||
data.pop("discovery")
|
||||
assert data == old
|
||||
assert "private path" not in response.text
|
||||
assert client.get("/api/mcp/catalog?detect_apps=not-a-bool").status_code == 422
|
||||
Reference in New Issue
Block a user