mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
25ba6783b8
Extracts the JSON-RPC transport from stdio into an abstraction so the same
dispatcher drives Ink over stdio AND browser/iOS clients over WebSocket
without duplicating handler logic. Adds a Chat page to the existing web
dashboard that exercises the full surface — streaming, tool calls, slash
commands, model picker, session resume.
Backend
-------
* tui_gateway/transport.py — Transport protocol + contextvar binding + the
module-level StdioTransport. Stream is resolved through a callback so
tests that monkeypatch `_real_stdout` keep working.
* tui_gateway/server.py — write_json and dispatch are now transport-aware.
Backward compatible: no transport bound = legacy stdio path, so entry.py
(Ink's stdio entrypoint) is unchanged externally.
* tui_gateway/ws.py — WSTransport + handle_ws coroutine. Safe to call from
any thread: detects loop-thread deadlock and fire-and-forget schedules
when needed, blocking run_coroutine_threadsafe + future.result otherwise.
* hermes_cli/web_server.py — mounts /api/ws on the existing FastAPI app,
gated by the same ephemeral session token used for REST. Adds
HERMES_DASHBOARD_DEV_TOKEN env override so Vite HMR dev can share the
token with the backend.
Frontend
--------
* web/src/lib/gatewayClient.ts — browser WebSocket JSON-RPC client that
mirrors ui-tui/src/gatewayClient.ts.
* web/src/lib/slashExec.ts — slash command pipeline (slash.exec with
command.dispatch fallback + exec/plugin/alias/skill/send directive
handling), mirrors ui-tui/src/app/createSlashHandler.ts.
* web/src/pages/ChatPage.tsx — transcript + composer driven entirely by
the WS.
* web/src/components/SlashPopover.tsx — autocomplete popover above the
composer, debounced complete.slash.
* web/src/components/ModelPickerDialog.tsx — two-stage provider/model
picker; confirms by emitting /model through the slash pipeline.
* web/src/components/ToolCall.tsx — expandable tool call row (Ink-style
chevron + context + summary/error/diff).
* web/src/App.tsx — logo links to /, Chat entry added to nav.
* web/src/pages/SessionsPage.tsx — every session row gets an Open-in-chat
button that navigates to /chat?resume=<id> (uses session.resume).
* web/vite.config.ts — /api proxy configured with ws: true so WebSocket
upgrades forward in dev mode; injectDevToken plugin reads
HERMES_DASHBOARD_DEV_TOKEN and injects it into the served index.html so
Vite HMR can authenticate against FastAPI without a separate flow.
Tests
-----
tests/hermes_cli/test_web_server.py picks up three new classes:
* TestTuiGatewayWebSocket — handshake, auth rejection, parse errors,
unknown methods, inline + pool handler round-trips, session event
routing, disconnect cleanup.
* TestTuiGatewayTransportParity — byte-identical envelopes for the same
RPC over stdio vs WS (unknown method, inline handler, error envelope,
explicit stdio transport).
* TestTuiGatewayE2EAnyPort — scripted multi-RPC conversation driven
identically via handle_request and via WebSocket; order + shape must
match. This is the "hermes --tui in any port" check.
Existing tests under tests/tui_gateway/ and tests/test_tui_gateway_server.py
all still pass unchanged — backward compat preserved.
Try it
------
hermes dashboard # builds web, serves on :9119, click Chat
Dev with HMR:
export HERMES_DASHBOARD_DEV_TOKEN="dev-\$(openssl rand -hex 16)"
hermes dashboard --no-open
cd web && npm run dev # :5173, /api + /api/ws proxied to :9119
fix(chat): insert tool rows before the streaming assistant message
Transcript used to read "user → empty assistant bubble → tool → bubble
filling in", which is disorienting: the streaming cursor sits at the top
while the "work" rows appear below it chronologically.
Now tool.start inserts the row just before the current streaming
assistant message, so the order reads "user → tools → final message".
If no streaming assistant exists yet (rare), tools still append at the
end; tool.progress / tool.complete match by id regardless of position.
fix(web-chat): font, composer, streaming caret + port GoodVibesHeart
- ChatPage root opts out of App's `font-mondwest uppercase` (dashboard
chrome style) — adds `font-courier normal-case` so transcript prose is
readable mono mixed-case instead of pixel-display caps.
- Composer: textarea + send button wrapped as one bordered unit with
`focus-within` ring; `font-sans` dropped (it mapped to `Collapse`
display). Heights stretch together via `items-stretch`; button is a
flush cap with `border-l` divider.
- Streaming caret no longer wraps to a new line when the assistant
renders a block element. Markdown now takes a `streaming` prop and
injects the caret inside the last block (paragraph, list item, code)
so it hugs the trailing character. Caret sized in em units.
- EmptyState gets a blinking caret + <kbd> shortcut chips.
- Port ui-tui's GoodVibesHeart easter egg to the web: typing "thanks" /
"ty" / "ily" / "good bot" flashes a Lucide heart next to the
connection badge (same regex, same 650ms beat, same palette as
ui-tui/src/app/useMainApp.ts).
2130 lines
85 KiB
Python
2130 lines
85 KiB
Python
"""Tests for hermes_cli.web_server and related config utilities."""
|
|
|
|
import os
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
from hermes_cli.config import (
|
|
DEFAULT_CONFIG,
|
|
reload_env,
|
|
redact_key,
|
|
_EXTRA_ENV_KEYS,
|
|
OPTIONAL_ENV_VARS,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# reload_env tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestReloadEnv:
|
|
"""Tests for reload_env() — re-reads .env into os.environ."""
|
|
|
|
def test_adds_new_vars(self, tmp_path):
|
|
"""reload_env() adds vars from .env that are not in os.environ."""
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text("TEST_RELOAD_VAR=hello123\n")
|
|
with patch("hermes_cli.config.get_env_path", return_value=env_file):
|
|
os.environ.pop("TEST_RELOAD_VAR", None)
|
|
count = reload_env()
|
|
assert count >= 1
|
|
assert os.environ.get("TEST_RELOAD_VAR") == "hello123"
|
|
os.environ.pop("TEST_RELOAD_VAR", None)
|
|
|
|
def test_updates_changed_vars(self, tmp_path):
|
|
"""reload_env() updates vars whose value changed on disk."""
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text("TEST_RELOAD_VAR=old_value\n")
|
|
with patch("hermes_cli.config.get_env_path", return_value=env_file):
|
|
os.environ["TEST_RELOAD_VAR"] = "old_value"
|
|
# Now change the file
|
|
env_file.write_text("TEST_RELOAD_VAR=new_value\n")
|
|
count = reload_env()
|
|
assert count >= 1
|
|
assert os.environ.get("TEST_RELOAD_VAR") == "new_value"
|
|
os.environ.pop("TEST_RELOAD_VAR", None)
|
|
|
|
def test_removes_deleted_known_vars(self, tmp_path):
|
|
"""reload_env() removes known Hermes vars not present in .env."""
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text("") # empty .env
|
|
# Pick a known key from OPTIONAL_ENV_VARS
|
|
known_key = next(iter(OPTIONAL_ENV_VARS.keys()))
|
|
with patch("hermes_cli.config.get_env_path", return_value=env_file):
|
|
os.environ[known_key] = "stale_value"
|
|
count = reload_env()
|
|
assert known_key not in os.environ
|
|
assert count >= 1
|
|
|
|
def test_does_not_remove_unknown_vars(self, tmp_path):
|
|
"""reload_env() preserves non-Hermes env vars even when absent from .env."""
|
|
env_file = tmp_path / ".env"
|
|
env_file.write_text("")
|
|
with patch("hermes_cli.config.get_env_path", return_value=env_file):
|
|
os.environ["MY_CUSTOM_UNRELATED_VAR"] = "keep_me"
|
|
reload_env()
|
|
assert os.environ.get("MY_CUSTOM_UNRELATED_VAR") == "keep_me"
|
|
os.environ.pop("MY_CUSTOM_UNRELATED_VAR", None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# redact_key tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestRedactKey:
|
|
def test_long_key_shows_prefix_suffix(self):
|
|
result = redact_key("sk-1234567890abcdef")
|
|
assert result.startswith("sk-1")
|
|
assert result.endswith("cdef")
|
|
assert "..." in result
|
|
|
|
def test_short_key_fully_masked(self):
|
|
assert redact_key("short") == "***"
|
|
|
|
def test_empty_key(self):
|
|
result = redact_key("")
|
|
assert "not set" in result.lower() or result == "***" or "\x1b" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# web_server tests (FastAPI endpoints)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWebServerEndpoints:
|
|
"""Test the FastAPI REST endpoints using Starlette TestClient."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup_test_client(self, monkeypatch, _isolate_hermes_home):
|
|
"""Create a TestClient and isolate the state DB under the test HERMES_HOME."""
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
|
|
import hermes_state
|
|
from hermes_constants import get_hermes_home
|
|
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
|
|
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
|
|
|
self.client = TestClient(app)
|
|
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
|
|
|
def test_get_status(self):
|
|
resp = self.client.get("/api/status")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "version" in data
|
|
assert "hermes_home" in data
|
|
assert "active_sessions" in data
|
|
|
|
def test_get_status_filters_unconfigured_gateway_platforms(self, monkeypatch):
|
|
import gateway.config as gateway_config
|
|
import hermes_cli.web_server as web_server
|
|
|
|
class _Platform:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
class _GatewayConfig:
|
|
def get_connected_platforms(self):
|
|
return [_Platform("telegram")]
|
|
|
|
monkeypatch.setattr(web_server, "get_running_pid", lambda: 1234)
|
|
monkeypatch.setattr(
|
|
web_server,
|
|
"read_runtime_status",
|
|
lambda: {
|
|
"gateway_state": "running",
|
|
"updated_at": "2026-04-12T00:00:00+00:00",
|
|
"platforms": {
|
|
"telegram": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"},
|
|
"whatsapp": {"state": "retrying", "updated_at": "2026-04-12T00:00:00+00:00"},
|
|
"feishu": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"},
|
|
},
|
|
},
|
|
)
|
|
monkeypatch.setattr(web_server, "check_config_version", lambda: (1, 1))
|
|
monkeypatch.setattr(gateway_config, "load_gateway_config", lambda: _GatewayConfig())
|
|
|
|
resp = self.client.get("/api/status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["gateway_platforms"] == {
|
|
"telegram": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"},
|
|
}
|
|
|
|
def test_get_status_hides_stale_platforms_when_gateway_not_running(self, monkeypatch):
|
|
import gateway.config as gateway_config
|
|
import hermes_cli.web_server as web_server
|
|
|
|
class _GatewayConfig:
|
|
def get_connected_platforms(self):
|
|
return []
|
|
|
|
monkeypatch.setattr(web_server, "get_running_pid", lambda: None)
|
|
monkeypatch.setattr(
|
|
web_server,
|
|
"read_runtime_status",
|
|
lambda: {
|
|
"gateway_state": "startup_failed",
|
|
"updated_at": "2026-04-12T00:00:00+00:00",
|
|
"platforms": {
|
|
"whatsapp": {"state": "retrying", "updated_at": "2026-04-12T00:00:00+00:00"},
|
|
"feishu": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"},
|
|
},
|
|
},
|
|
)
|
|
monkeypatch.setattr(web_server, "check_config_version", lambda: (1, 1))
|
|
monkeypatch.setattr(gateway_config, "load_gateway_config", lambda: _GatewayConfig())
|
|
|
|
resp = self.client.get("/api/status")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["gateway_state"] == "startup_failed"
|
|
assert resp.json()["gateway_platforms"] == {}
|
|
|
|
def test_get_config_schema(self):
|
|
resp = self.client.get("/api/config/schema")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "fields" in data
|
|
assert "category_order" in data
|
|
schema = data["fields"]
|
|
assert len(schema) > 100 # Should have 150+ fields
|
|
assert "model" in schema
|
|
# Verify category_order is a non-empty list
|
|
assert isinstance(data["category_order"], list)
|
|
assert len(data["category_order"]) > 0
|
|
assert "general" in data["category_order"]
|
|
|
|
def test_get_config_defaults(self):
|
|
resp = self.client.get("/api/config/defaults")
|
|
assert resp.status_code == 200
|
|
defaults = resp.json()
|
|
assert "model" in defaults
|
|
|
|
def test_get_env_vars(self):
|
|
resp = self.client.get("/api/env")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# Should contain known env var names
|
|
assert any(k.endswith("_API_KEY") or k.endswith("_TOKEN") for k in data.keys())
|
|
|
|
def test_reveal_env_var(self, tmp_path):
|
|
"""POST /api/env/reveal should return the real unredacted value."""
|
|
from hermes_cli.config import save_env_value
|
|
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
save_env_value("TEST_REVEAL_KEY", "super-secret-value-12345")
|
|
resp = self.client.post(
|
|
"/api/env/reveal",
|
|
json={"key": "TEST_REVEAL_KEY"},
|
|
headers={_SESSION_HEADER_NAME: _SESSION_TOKEN},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["key"] == "TEST_REVEAL_KEY"
|
|
assert data["value"] == "super-secret-value-12345"
|
|
|
|
def test_reveal_env_var_not_found(self):
|
|
"""POST /api/env/reveal should 404 for unknown keys."""
|
|
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
resp = self.client.post(
|
|
"/api/env/reveal",
|
|
json={"key": "NONEXISTENT_KEY_XYZ"},
|
|
headers={_SESSION_HEADER_NAME: _SESSION_TOKEN},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_reveal_env_var_no_token(self, tmp_path):
|
|
"""POST /api/env/reveal without token should return 401."""
|
|
from starlette.testclient import TestClient
|
|
from hermes_cli.web_server import app
|
|
from hermes_cli.config import save_env_value
|
|
save_env_value("TEST_REVEAL_NOAUTH", "secret-value")
|
|
# Use a fresh client WITHOUT the dashboard session header
|
|
unauth_client = TestClient(app)
|
|
resp = unauth_client.post(
|
|
"/api/env/reveal",
|
|
json={"key": "TEST_REVEAL_NOAUTH"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_reveal_env_var_bad_token(self, tmp_path):
|
|
"""POST /api/env/reveal with wrong token should return 401."""
|
|
from hermes_cli.config import save_env_value
|
|
from hermes_cli.web_server import _SESSION_HEADER_NAME
|
|
save_env_value("TEST_REVEAL_BADAUTH", "secret-value")
|
|
resp = self.client.post(
|
|
"/api/env/reveal",
|
|
json={"key": "TEST_REVEAL_BADAUTH"},
|
|
headers={_SESSION_HEADER_NAME: "wrong-token-here"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
def test_reveal_env_var_custom_session_header_ignores_proxy_authorization(self, tmp_path):
|
|
"""A valid dashboard session header should coexist with proxy auth."""
|
|
from hermes_cli.config import save_env_value
|
|
from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
|
|
save_env_value("TEST_REVEAL_PROXY_AUTH", "secret-value")
|
|
resp = self.client.post(
|
|
"/api/env/reveal",
|
|
json={"key": "TEST_REVEAL_PROXY_AUTH"},
|
|
headers={
|
|
_SESSION_HEADER_NAME: _SESSION_TOKEN,
|
|
"Authorization": "Basic dXNlcjpwYXNz",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json()["value"] == "secret-value"
|
|
|
|
def test_reveal_env_var_legacy_authorization_header_still_works(self, tmp_path):
|
|
"""Keep old dashboard bundles working while the new header rolls out."""
|
|
from hermes_cli.config import save_env_value
|
|
from hermes_cli.web_server import _SESSION_TOKEN
|
|
|
|
save_env_value("TEST_REVEAL_LEGACY_AUTH", "secret-value")
|
|
resp = self.client.post(
|
|
"/api/env/reveal",
|
|
json={"key": "TEST_REVEAL_LEGACY_AUTH"},
|
|
headers={"Authorization": f"Bearer {_SESSION_TOKEN}"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
def test_session_token_endpoint_removed(self):
|
|
"""GET /api/auth/session-token should no longer exist (token injected via HTML)."""
|
|
resp = self.client.get("/api/auth/session-token")
|
|
# The endpoint is gone — the catch-all SPA route serves index.html
|
|
# or the middleware returns 401 for unauthenticated /api/ paths.
|
|
assert resp.status_code in (200, 404)
|
|
# Either way, it must NOT return the token as JSON
|
|
try:
|
|
data = resp.json()
|
|
assert "token" not in data
|
|
except Exception:
|
|
pass # Not JSON — that's fine (SPA HTML)
|
|
|
|
def test_unauthenticated_api_blocked(self):
|
|
"""API requests without the session token should be rejected."""
|
|
from starlette.testclient import TestClient
|
|
from hermes_cli.web_server import app
|
|
# Create a client WITHOUT the dashboard session header
|
|
unauth_client = TestClient(app)
|
|
resp = unauth_client.get("/api/env")
|
|
assert resp.status_code == 401
|
|
resp = unauth_client.get("/api/config")
|
|
assert resp.status_code == 401
|
|
# Public endpoints should still work
|
|
resp = unauth_client.get("/api/status")
|
|
assert resp.status_code == 200
|
|
|
|
def test_path_traversal_blocked(self):
|
|
"""Verify URL-encoded path traversal is blocked."""
|
|
# %2e%2e = ..
|
|
resp = self.client.get("/%2e%2e/%2e%2e/etc/passwd")
|
|
# Should return 200 with index.html (SPA fallback), not the actual file
|
|
assert resp.status_code in (200, 404)
|
|
if resp.status_code == 200:
|
|
# Should be the SPA fallback, not the system file
|
|
assert "root:" not in resp.text
|
|
|
|
def test_path_traversal_dotdot_blocked(self):
|
|
"""Direct .. path traversal via encoded sequences."""
|
|
resp = self.client.get("/%2e%2e/hermes_cli/web_server.py")
|
|
assert resp.status_code in (200, 404)
|
|
if resp.status_code == 200:
|
|
assert "FastAPI" not in resp.text # Should not serve the actual source
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _build_schema_from_config tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBuildSchemaFromConfig:
|
|
def test_produces_expected_field_count(self):
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
# DEFAULT_CONFIG has ~150+ leaf fields
|
|
assert len(CONFIG_SCHEMA) > 100
|
|
|
|
def test_schema_entries_have_required_fields(self):
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
for key, entry in list(CONFIG_SCHEMA.items())[:10]:
|
|
assert "type" in entry, f"Missing type for {key}"
|
|
assert "category" in entry, f"Missing category for {key}"
|
|
|
|
def test_overrides_applied(self):
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
# terminal.backend should be a select with options
|
|
if "terminal.backend" in CONFIG_SCHEMA:
|
|
entry = CONFIG_SCHEMA["terminal.backend"]
|
|
assert entry["type"] == "select"
|
|
assert "options" in entry
|
|
assert "local" in entry["options"]
|
|
|
|
def test_empty_prefix_produces_correct_keys(self):
|
|
from hermes_cli.web_server import _build_schema_from_config
|
|
test_config = {"model": "test", "nested": {"key": "val"}}
|
|
schema = _build_schema_from_config(test_config)
|
|
assert "model" in schema
|
|
assert "nested.key" in schema
|
|
|
|
def test_top_level_scalars_get_general_category(self):
|
|
"""Top-level scalar fields should be in 'general' category."""
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
assert CONFIG_SCHEMA["model"]["category"] == "general"
|
|
|
|
def test_nested_keys_get_parent_category(self):
|
|
"""Nested fields should use the top-level parent as their category."""
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
if "agent.max_turns" in CONFIG_SCHEMA:
|
|
assert CONFIG_SCHEMA["agent.max_turns"]["category"] == "agent"
|
|
|
|
def test_category_merge_applied(self):
|
|
"""Small categories should be merged into larger ones."""
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
categories = {e["category"] for e in CONFIG_SCHEMA.values()}
|
|
# These should be merged away
|
|
assert "privacy" not in categories # merged into security
|
|
assert "context" not in categories # merged into agent
|
|
|
|
def test_no_single_field_categories(self):
|
|
"""After merging, no category should have just 1 field."""
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
from collections import Counter
|
|
cats = Counter(e["category"] for e in CONFIG_SCHEMA.values())
|
|
for cat, count in cats.items():
|
|
assert count >= 2, f"Category '{cat}' has only {count} field(s) — should be merged"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config round-trip tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestConfigRoundTrip:
|
|
"""Verify config survives GET → edit → PUT without data loss."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup(self):
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
self.client = TestClient(app)
|
|
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
|
|
|
def test_get_config_no_internal_keys(self):
|
|
"""GET /api/config should not expose _config_version or _model_meta."""
|
|
config = self.client.get("/api/config").json()
|
|
internal = [k for k in config if k.startswith("_")]
|
|
assert not internal, f"Internal keys leaked to frontend: {internal}"
|
|
|
|
def test_get_config_model_is_string(self):
|
|
"""GET /api/config should normalize model dict to a string."""
|
|
config = self.client.get("/api/config").json()
|
|
assert isinstance(config.get("model"), str), \
|
|
f"model should be string, got {type(config.get('model'))}"
|
|
|
|
def test_round_trip_preserves_model_subkeys(self):
|
|
"""Save and reload should not lose model.provider, model.base_url, etc."""
|
|
from hermes_cli.config import load_config, save_config
|
|
|
|
# Set up a config with model as a dict (the common user config form)
|
|
save_config({
|
|
"model": {
|
|
"default": "anthropic/claude-sonnet-4",
|
|
"provider": "openrouter",
|
|
"base_url": "https://openrouter.ai/api/v1",
|
|
"api_mode": "openai",
|
|
}
|
|
})
|
|
|
|
before = load_config()
|
|
assert isinstance(before.get("model"), dict)
|
|
original_keys = set(before["model"].keys())
|
|
|
|
# GET → PUT unchanged
|
|
web_config = self.client.get("/api/config").json()
|
|
assert isinstance(web_config.get("model"), str), "GET should normalize model to string"
|
|
|
|
self.client.put("/api/config", json={"config": web_config})
|
|
|
|
after = load_config()
|
|
assert isinstance(after.get("model"), dict), "model should still be a dict after save"
|
|
assert set(after["model"].keys()) >= original_keys, \
|
|
f"Lost model subkeys: {original_keys - set(after['model'].keys())}"
|
|
|
|
def test_edit_model_name_preserved(self):
|
|
"""Changing the model string should update model.default on disk."""
|
|
from hermes_cli.config import load_config
|
|
|
|
web_config = self.client.get("/api/config").json()
|
|
original_model = web_config["model"]
|
|
|
|
# Change model
|
|
web_config["model"] = "test/editing-model"
|
|
self.client.put("/api/config", json={"config": web_config})
|
|
|
|
after = load_config()
|
|
if isinstance(after.get("model"), dict):
|
|
assert after["model"]["default"] == "test/editing-model"
|
|
else:
|
|
assert after["model"] == "test/editing-model"
|
|
|
|
# Restore
|
|
web_config["model"] = original_model
|
|
self.client.put("/api/config", json={"config": web_config})
|
|
|
|
def test_edit_nested_value(self):
|
|
"""Editing a nested config value should persist correctly."""
|
|
from hermes_cli.config import load_config
|
|
|
|
web_config = self.client.get("/api/config").json()
|
|
original_turns = web_config.get("agent", {}).get("max_turns")
|
|
|
|
# Change max_turns
|
|
if "agent" not in web_config:
|
|
web_config["agent"] = {}
|
|
web_config["agent"]["max_turns"] = 42
|
|
|
|
self.client.put("/api/config", json={"config": web_config})
|
|
|
|
after = load_config()
|
|
assert after.get("agent", {}).get("max_turns") == 42
|
|
|
|
# Restore
|
|
web_config["agent"]["max_turns"] = original_turns
|
|
self.client.put("/api/config", json={"config": web_config})
|
|
|
|
def test_schema_types_match_config_values(self):
|
|
"""Every schema field should have a matching-type value in the config."""
|
|
config = self.client.get("/api/config").json()
|
|
schema_resp = self.client.get("/api/config/schema").json()
|
|
schema = schema_resp["fields"]
|
|
|
|
def get_nested(obj, path):
|
|
parts = path.split(".")
|
|
cur = obj
|
|
for p in parts:
|
|
if cur is None or not isinstance(cur, dict):
|
|
return None
|
|
cur = cur.get(p)
|
|
return cur
|
|
|
|
mismatches = []
|
|
for key, entry in schema.items():
|
|
val = get_nested(config, key)
|
|
if val is None:
|
|
continue # not set in user config — fine
|
|
expected = entry["type"]
|
|
if expected in ("string", "select") and not isinstance(val, str):
|
|
mismatches.append(f"{key}: expected str, got {type(val).__name__}")
|
|
elif expected == "number" and not isinstance(val, (int, float)):
|
|
mismatches.append(f"{key}: expected number, got {type(val).__name__}")
|
|
elif expected == "boolean" and not isinstance(val, bool):
|
|
mismatches.append(f"{key}: expected bool, got {type(val).__name__}")
|
|
elif expected == "list" and not isinstance(val, list):
|
|
mismatches.append(f"{key}: expected list, got {type(val).__name__}")
|
|
assert not mismatches, f"Type mismatches:\n" + "\n".join(mismatches)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# New feature endpoint tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNewEndpoints:
|
|
"""Tests for session detail, logs, cron, skills, tools, raw config, analytics."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup(self, monkeypatch, _isolate_hermes_home):
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
|
|
import hermes_state
|
|
from hermes_constants import get_hermes_home
|
|
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
|
|
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
|
|
|
self.client = TestClient(app)
|
|
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
|
|
|
def test_get_logs_default(self):
|
|
resp = self.client.get("/api/logs")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "file" in data
|
|
assert "lines" in data
|
|
assert isinstance(data["lines"], list)
|
|
|
|
def test_get_logs_invalid_file(self):
|
|
resp = self.client.get("/api/logs?file=nonexistent")
|
|
assert resp.status_code == 400
|
|
|
|
def test_cron_list(self):
|
|
resp = self.client.get("/api/cron/jobs")
|
|
assert resp.status_code == 200
|
|
assert isinstance(resp.json(), list)
|
|
|
|
def test_cron_job_not_found(self):
|
|
resp = self.client.get("/api/cron/jobs/nonexistent-id")
|
|
assert resp.status_code == 404
|
|
|
|
def test_skills_list(self):
|
|
resp = self.client.get("/api/skills")
|
|
assert resp.status_code == 200
|
|
skills = resp.json()
|
|
assert isinstance(skills, list)
|
|
if skills:
|
|
assert "name" in skills[0]
|
|
assert "enabled" in skills[0]
|
|
|
|
def test_skills_list_includes_disabled_skills(self, monkeypatch):
|
|
import tools.skills_tool as skills_tool
|
|
import hermes_cli.skills_config as skills_config
|
|
import hermes_cli.web_server as web_server
|
|
|
|
def _fake_find_all_skills(*, skip_disabled=False):
|
|
if skip_disabled:
|
|
return [
|
|
{"name": "active-skill", "description": "active", "category": "demo"},
|
|
{"name": "disabled-skill", "description": "disabled", "category": "demo"},
|
|
]
|
|
return [
|
|
{"name": "active-skill", "description": "active", "category": "demo"},
|
|
]
|
|
|
|
monkeypatch.setattr(skills_tool, "_find_all_skills", _fake_find_all_skills)
|
|
monkeypatch.setattr(skills_config, "get_disabled_skills", lambda config: {"disabled-skill"})
|
|
monkeypatch.setattr(web_server, "load_config", lambda: {"skills": {"disabled": ["disabled-skill"]}})
|
|
|
|
resp = self.client.get("/api/skills")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == [
|
|
{
|
|
"name": "active-skill",
|
|
"description": "active",
|
|
"category": "demo",
|
|
"enabled": True,
|
|
},
|
|
{
|
|
"name": "disabled-skill",
|
|
"description": "disabled",
|
|
"category": "demo",
|
|
"enabled": False,
|
|
},
|
|
]
|
|
|
|
def test_toolsets_list(self):
|
|
resp = self.client.get("/api/tools/toolsets")
|
|
assert resp.status_code == 200
|
|
toolsets = resp.json()
|
|
assert isinstance(toolsets, list)
|
|
if toolsets:
|
|
assert "name" in toolsets[0]
|
|
assert "label" in toolsets[0]
|
|
assert "enabled" in toolsets[0]
|
|
|
|
def test_toolsets_list_matches_cli_enabled_state(self, monkeypatch):
|
|
import hermes_cli.tools_config as tools_config
|
|
import toolsets as toolsets_module
|
|
import hermes_cli.web_server as web_server
|
|
|
|
monkeypatch.setattr(
|
|
tools_config,
|
|
"_get_effective_configurable_toolsets",
|
|
lambda: [
|
|
("web", "🔍 Web Search & Scraping", "web_search, web_extract"),
|
|
("skills", "📚 Skills", "list, view, manage"),
|
|
("memory", "💾 Memory", "persistent memory across sessions"),
|
|
],
|
|
)
|
|
monkeypatch.setattr(
|
|
tools_config,
|
|
"_get_platform_tools",
|
|
lambda config, platform, include_default_mcp_servers=False: {"web", "skills"},
|
|
)
|
|
monkeypatch.setattr(
|
|
tools_config,
|
|
"_toolset_has_keys",
|
|
lambda ts_key, config=None: ts_key != "web",
|
|
)
|
|
monkeypatch.setattr(
|
|
toolsets_module,
|
|
"resolve_toolset",
|
|
lambda name: {
|
|
"web": ["web_search", "web_extract"],
|
|
"skills": ["skills_list", "skill_view"],
|
|
"memory": ["memory_read"],
|
|
}[name],
|
|
)
|
|
monkeypatch.setattr(web_server, "load_config", lambda: {"platform_toolsets": {"cli": ["web", "skills"]}})
|
|
|
|
resp = self.client.get("/api/tools/toolsets")
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == [
|
|
{
|
|
"name": "web",
|
|
"label": "🔍 Web Search & Scraping",
|
|
"description": "web_search, web_extract",
|
|
"enabled": True,
|
|
"available": True,
|
|
"configured": False,
|
|
"tools": ["web_extract", "web_search"],
|
|
},
|
|
{
|
|
"name": "skills",
|
|
"label": "📚 Skills",
|
|
"description": "list, view, manage",
|
|
"enabled": True,
|
|
"available": True,
|
|
"configured": True,
|
|
"tools": ["skill_view", "skills_list"],
|
|
},
|
|
{
|
|
"name": "memory",
|
|
"label": "💾 Memory",
|
|
"description": "persistent memory across sessions",
|
|
"enabled": False,
|
|
"available": False,
|
|
"configured": True,
|
|
"tools": ["memory_read"],
|
|
},
|
|
]
|
|
|
|
def test_config_raw_get(self):
|
|
resp = self.client.get("/api/config/raw")
|
|
assert resp.status_code == 200
|
|
assert "yaml" in resp.json()
|
|
|
|
def test_config_raw_put_valid(self):
|
|
resp = self.client.put(
|
|
"/api/config/raw",
|
|
json={"yaml_text": "model: test\ntoolsets:\n - all\n"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["ok"] is True
|
|
|
|
def test_config_raw_put_invalid(self):
|
|
resp = self.client.put(
|
|
"/api/config/raw",
|
|
json={"yaml_text": "- this is a list not a dict"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_analytics_usage(self):
|
|
resp = self.client.get("/api/analytics/usage?days=7")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "daily" in data
|
|
assert "by_model" in data
|
|
assert "totals" in data
|
|
assert "skills" in data
|
|
assert isinstance(data["daily"], list)
|
|
assert "total_sessions" in data["totals"]
|
|
assert "total_api_calls" in data["totals"]
|
|
assert data["skills"] == {
|
|
"summary": {
|
|
"total_skill_loads": 0,
|
|
"total_skill_edits": 0,
|
|
"total_skill_actions": 0,
|
|
"distinct_skills_used": 0,
|
|
},
|
|
"top_skills": [],
|
|
}
|
|
|
|
def test_analytics_usage_includes_skill_breakdown(self):
|
|
from hermes_state import SessionDB
|
|
|
|
db = SessionDB()
|
|
try:
|
|
db.create_session(
|
|
session_id="skills-analytics-test",
|
|
source="cli",
|
|
model="anthropic/claude-sonnet-4",
|
|
)
|
|
db.update_token_counts(
|
|
"skills-analytics-test",
|
|
input_tokens=120,
|
|
output_tokens=45,
|
|
)
|
|
db.append_message(
|
|
"skills-analytics-test",
|
|
role="assistant",
|
|
content="Loading and updating skills.",
|
|
tool_calls=[
|
|
{
|
|
"function": {
|
|
"name": "skill_view",
|
|
"arguments": '{"name":"github-pr-workflow"}',
|
|
}
|
|
},
|
|
{
|
|
"function": {
|
|
"name": "skill_manage",
|
|
"arguments": '{"name":"github-code-review"}',
|
|
}
|
|
},
|
|
],
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
resp = self.client.get("/api/analytics/usage?days=7")
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
assert data["skills"]["summary"] == {
|
|
"total_skill_loads": 1,
|
|
"total_skill_edits": 1,
|
|
"total_skill_actions": 2,
|
|
"distinct_skills_used": 2,
|
|
}
|
|
assert len(data["skills"]["top_skills"]) == 2
|
|
|
|
top_skill = data["skills"]["top_skills"][0]
|
|
assert top_skill["skill"] == "github-pr-workflow"
|
|
assert top_skill["view_count"] == 1
|
|
assert top_skill["manage_count"] == 0
|
|
assert top_skill["total_count"] == 1
|
|
assert top_skill["last_used_at"] is not None
|
|
|
|
def test_session_token_endpoint_removed(self):
|
|
"""GET /api/auth/session-token no longer exists."""
|
|
resp = self.client.get("/api/auth/session-token")
|
|
# Should not return a JSON token object
|
|
assert resp.status_code in (200, 404)
|
|
try:
|
|
data = resp.json()
|
|
assert "token" not in data
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Model context length: normalize/denormalize + /api/model/info
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestModelContextLength:
|
|
"""Tests for model_context_length in normalize/denormalize and /api/model/info."""
|
|
|
|
def test_normalize_extracts_context_length_from_dict(self):
|
|
"""normalize should surface context_length from model dict."""
|
|
from hermes_cli.web_server import _normalize_config_for_web
|
|
|
|
cfg = {
|
|
"model": {
|
|
"default": "anthropic/claude-opus-4.6",
|
|
"provider": "openrouter",
|
|
"context_length": 200000,
|
|
}
|
|
}
|
|
result = _normalize_config_for_web(cfg)
|
|
assert result["model"] == "anthropic/claude-opus-4.6"
|
|
assert result["model_context_length"] == 200000
|
|
|
|
def test_normalize_bare_string_model_yields_zero(self):
|
|
"""normalize should set model_context_length=0 for bare string model."""
|
|
from hermes_cli.web_server import _normalize_config_for_web
|
|
|
|
result = _normalize_config_for_web({"model": "anthropic/claude-sonnet-4"})
|
|
assert result["model"] == "anthropic/claude-sonnet-4"
|
|
assert result["model_context_length"] == 0
|
|
|
|
def test_normalize_dict_without_context_length_yields_zero(self):
|
|
"""normalize should default to 0 when model dict has no context_length."""
|
|
from hermes_cli.web_server import _normalize_config_for_web
|
|
|
|
cfg = {"model": {"default": "test/model", "provider": "openrouter"}}
|
|
result = _normalize_config_for_web(cfg)
|
|
assert result["model_context_length"] == 0
|
|
|
|
def test_normalize_non_int_context_length_yields_zero(self):
|
|
"""normalize should coerce non-int context_length to 0."""
|
|
from hermes_cli.web_server import _normalize_config_for_web
|
|
|
|
cfg = {"model": {"default": "test/model", "context_length": "invalid"}}
|
|
result = _normalize_config_for_web(cfg)
|
|
assert result["model_context_length"] == 0
|
|
|
|
def test_denormalize_writes_context_length_into_model_dict(self):
|
|
"""denormalize should write model_context_length back into model dict."""
|
|
from hermes_cli.web_server import _denormalize_config_from_web
|
|
from hermes_cli.config import save_config
|
|
|
|
# Set up disk config with model as a dict
|
|
save_config({
|
|
"model": {"default": "anthropic/claude-opus-4.6", "provider": "openrouter"}
|
|
})
|
|
|
|
result = _denormalize_config_from_web({
|
|
"model": "anthropic/claude-opus-4.6",
|
|
"model_context_length": 100000,
|
|
})
|
|
assert isinstance(result["model"], dict)
|
|
assert result["model"]["context_length"] == 100000
|
|
assert "model_context_length" not in result # virtual field removed
|
|
|
|
def test_denormalize_zero_removes_context_length(self):
|
|
"""denormalize with model_context_length=0 should remove context_length key."""
|
|
from hermes_cli.web_server import _denormalize_config_from_web
|
|
from hermes_cli.config import save_config
|
|
|
|
save_config({
|
|
"model": {
|
|
"default": "anthropic/claude-opus-4.6",
|
|
"provider": "openrouter",
|
|
"context_length": 50000,
|
|
}
|
|
})
|
|
|
|
result = _denormalize_config_from_web({
|
|
"model": "anthropic/claude-opus-4.6",
|
|
"model_context_length": 0,
|
|
})
|
|
assert isinstance(result["model"], dict)
|
|
assert "context_length" not in result["model"]
|
|
|
|
def test_denormalize_upgrades_bare_string_to_dict(self):
|
|
"""denormalize should upgrade bare string model to dict when context_length set."""
|
|
from hermes_cli.web_server import _denormalize_config_from_web
|
|
from hermes_cli.config import save_config
|
|
|
|
# Disk has model as bare string
|
|
save_config({"model": "anthropic/claude-sonnet-4"})
|
|
|
|
result = _denormalize_config_from_web({
|
|
"model": "anthropic/claude-sonnet-4",
|
|
"model_context_length": 65000,
|
|
})
|
|
assert isinstance(result["model"], dict)
|
|
assert result["model"]["default"] == "anthropic/claude-sonnet-4"
|
|
assert result["model"]["context_length"] == 65000
|
|
|
|
def test_denormalize_bare_string_stays_string_when_zero(self):
|
|
"""denormalize should keep bare string model as string when context_length=0."""
|
|
from hermes_cli.web_server import _denormalize_config_from_web
|
|
from hermes_cli.config import save_config
|
|
|
|
save_config({"model": "anthropic/claude-sonnet-4"})
|
|
|
|
result = _denormalize_config_from_web({
|
|
"model": "anthropic/claude-sonnet-4",
|
|
"model_context_length": 0,
|
|
})
|
|
assert result["model"] == "anthropic/claude-sonnet-4"
|
|
|
|
def test_denormalize_coerces_string_context_length(self):
|
|
"""denormalize should handle string model_context_length from frontend."""
|
|
from hermes_cli.web_server import _denormalize_config_from_web
|
|
from hermes_cli.config import save_config
|
|
|
|
save_config({
|
|
"model": {"default": "test/model", "provider": "openrouter"}
|
|
})
|
|
|
|
result = _denormalize_config_from_web({
|
|
"model": "test/model",
|
|
"model_context_length": "32000",
|
|
})
|
|
assert isinstance(result["model"], dict)
|
|
assert result["model"]["context_length"] == 32000
|
|
|
|
|
|
class TestModelContextLengthSchema:
|
|
"""Tests for model_context_length placement in CONFIG_SCHEMA."""
|
|
|
|
def test_schema_has_model_context_length(self):
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
assert "model_context_length" in CONFIG_SCHEMA
|
|
|
|
def test_schema_model_context_length_after_model(self):
|
|
"""model_context_length should appear immediately after model in schema."""
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
keys = list(CONFIG_SCHEMA.keys())
|
|
model_idx = keys.index("model")
|
|
assert keys[model_idx + 1] == "model_context_length"
|
|
|
|
def test_schema_model_context_length_is_number(self):
|
|
from hermes_cli.web_server import CONFIG_SCHEMA
|
|
entry = CONFIG_SCHEMA["model_context_length"]
|
|
assert entry["type"] == "number"
|
|
assert "category" in entry
|
|
|
|
|
|
class TestModelInfoEndpoint:
|
|
"""Tests for GET /api/model/info endpoint."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup(self):
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
from hermes_cli.web_server import app
|
|
self.client = TestClient(app)
|
|
|
|
def test_model_info_returns_200(self):
|
|
resp = self.client.get("/api/model/info")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "model" in data
|
|
assert "provider" in data
|
|
assert "auto_context_length" in data
|
|
assert "config_context_length" in data
|
|
assert "effective_context_length" in data
|
|
assert "capabilities" in data
|
|
|
|
def test_model_info_with_dict_config(self, monkeypatch):
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "load_config", lambda: {
|
|
"model": {
|
|
"default": "anthropic/claude-opus-4.6",
|
|
"provider": "openrouter",
|
|
"context_length": 100000,
|
|
}
|
|
})
|
|
|
|
with patch("agent.model_metadata.get_model_context_length", return_value=200000):
|
|
resp = self.client.get("/api/model/info")
|
|
|
|
data = resp.json()
|
|
assert data["model"] == "anthropic/claude-opus-4.6"
|
|
assert data["provider"] == "openrouter"
|
|
assert data["auto_context_length"] == 200000
|
|
assert data["config_context_length"] == 100000
|
|
assert data["effective_context_length"] == 100000 # override wins
|
|
|
|
def test_model_info_auto_detect_when_no_override(self, monkeypatch):
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "load_config", lambda: {
|
|
"model": {"default": "anthropic/claude-opus-4.6", "provider": "openrouter"}
|
|
})
|
|
|
|
with patch("agent.model_metadata.get_model_context_length", return_value=200000):
|
|
resp = self.client.get("/api/model/info")
|
|
|
|
data = resp.json()
|
|
assert data["auto_context_length"] == 200000
|
|
assert data["config_context_length"] == 0
|
|
assert data["effective_context_length"] == 200000 # auto wins
|
|
|
|
def test_model_info_empty_model(self, monkeypatch):
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "load_config", lambda: {"model": ""})
|
|
|
|
resp = self.client.get("/api/model/info")
|
|
data = resp.json()
|
|
assert data["model"] == ""
|
|
assert data["effective_context_length"] == 0
|
|
|
|
def test_model_info_bare_string_model(self, monkeypatch):
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "load_config", lambda: {
|
|
"model": "anthropic/claude-sonnet-4"
|
|
})
|
|
|
|
with patch("agent.model_metadata.get_model_context_length", return_value=200000):
|
|
resp = self.client.get("/api/model/info")
|
|
|
|
data = resp.json()
|
|
assert data["model"] == "anthropic/claude-sonnet-4"
|
|
assert data["provider"] == ""
|
|
assert data["config_context_length"] == 0
|
|
assert data["effective_context_length"] == 200000
|
|
|
|
def test_model_info_capabilities(self, monkeypatch):
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "load_config", lambda: {
|
|
"model": {"default": "anthropic/claude-opus-4.6", "provider": "openrouter"}
|
|
})
|
|
|
|
mock_caps = MagicMock()
|
|
mock_caps.supports_tools = True
|
|
mock_caps.supports_vision = True
|
|
mock_caps.supports_reasoning = True
|
|
mock_caps.context_window = 200000
|
|
mock_caps.max_output_tokens = 32000
|
|
mock_caps.model_family = "claude-opus"
|
|
|
|
with patch("agent.model_metadata.get_model_context_length", return_value=200000), \
|
|
patch("agent.models_dev.get_model_capabilities", return_value=mock_caps):
|
|
resp = self.client.get("/api/model/info")
|
|
|
|
caps = resp.json()["capabilities"]
|
|
assert caps["supports_tools"] is True
|
|
assert caps["supports_vision"] is True
|
|
assert caps["supports_reasoning"] is True
|
|
assert caps["max_output_tokens"] == 32000
|
|
assert caps["model_family"] == "claude-opus"
|
|
|
|
def test_model_info_graceful_on_metadata_error(self, monkeypatch):
|
|
"""Endpoint should return zeros on import/resolution errors, not 500."""
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "load_config", lambda: {
|
|
"model": "some/obscure-model"
|
|
})
|
|
|
|
with patch("agent.model_metadata.get_model_context_length", side_effect=Exception("boom")):
|
|
resp = self.client.get("/api/model/info")
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["auto_context_length"] == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Gateway health probe tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProbeGatewayHealth:
|
|
"""Tests for _probe_gateway_health() — cross-container gateway detection."""
|
|
|
|
def test_returns_false_when_no_url_configured(self, monkeypatch):
|
|
"""When GATEWAY_HEALTH_URL is unset, the probe returns (False, None)."""
|
|
import hermes_cli.web_server as ws
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None)
|
|
alive, body = ws._probe_gateway_health()
|
|
assert alive is False
|
|
assert body is None
|
|
|
|
def test_normalizes_url_with_health_suffix(self, monkeypatch):
|
|
"""If the user sets the URL to include /health, it's stripped to base."""
|
|
import hermes_cli.web_server as ws
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642/health")
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
|
|
# Both paths should fail (no server), but we verify they were constructed
|
|
# correctly by checking the URLs attempted.
|
|
calls = []
|
|
original_urlopen = ws.urllib.request.urlopen
|
|
|
|
def mock_urlopen(req, **kwargs):
|
|
calls.append(req.full_url)
|
|
raise ConnectionError("mock")
|
|
|
|
monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)
|
|
alive, body = ws._probe_gateway_health()
|
|
assert alive is False
|
|
assert "http://gw:8642/health/detailed" in calls
|
|
assert "http://gw:8642/health" in calls
|
|
|
|
def test_normalizes_url_with_health_detailed_suffix(self, monkeypatch):
|
|
"""If the user sets the URL to include /health/detailed, it's stripped to base."""
|
|
import hermes_cli.web_server as ws
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642/health/detailed")
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
|
|
calls = []
|
|
|
|
def mock_urlopen(req, **kwargs):
|
|
calls.append(req.full_url)
|
|
raise ConnectionError("mock")
|
|
|
|
monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)
|
|
ws._probe_gateway_health()
|
|
assert "http://gw:8642/health/detailed" in calls
|
|
assert "http://gw:8642/health" in calls
|
|
|
|
def test_successful_detailed_probe(self, monkeypatch):
|
|
"""Successful /health/detailed probe returns (True, body_dict)."""
|
|
import hermes_cli.web_server as ws
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
|
|
|
|
response_body = json.dumps({
|
|
"status": "ok",
|
|
"gateway_state": "running",
|
|
"pid": 42,
|
|
})
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status = 200
|
|
mock_resp.read.return_value = response_body.encode()
|
|
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
|
mock_resp.__exit__ = MagicMock(return_value=False)
|
|
|
|
monkeypatch.setattr(ws.urllib.request, "urlopen", lambda req, **kw: mock_resp)
|
|
alive, body = ws._probe_gateway_health()
|
|
assert alive is True
|
|
assert body["status"] == "ok"
|
|
assert body["pid"] == 42
|
|
|
|
def test_detailed_fails_falls_back_to_simple_health(self, monkeypatch):
|
|
"""If /health/detailed fails, falls back to /health."""
|
|
import hermes_cli.web_server as ws
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1)
|
|
|
|
call_count = [0]
|
|
|
|
def mock_urlopen(req, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
raise ConnectionError("detailed failed")
|
|
mock_resp = MagicMock()
|
|
mock_resp.status = 200
|
|
mock_resp.read.return_value = json.dumps({"status": "ok"}).encode()
|
|
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
|
mock_resp.__exit__ = MagicMock(return_value=False)
|
|
return mock_resp
|
|
|
|
monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen)
|
|
alive, body = ws._probe_gateway_health()
|
|
assert alive is True
|
|
assert body["status"] == "ok"
|
|
assert call_count[0] == 2
|
|
|
|
|
|
class TestStatusRemoteGateway:
|
|
"""Tests for /api/status with remote gateway health fallback."""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup_test_client(self):
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
|
|
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
|
self.client = TestClient(app)
|
|
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
|
|
|
def test_status_falls_back_to_remote_probe(self, monkeypatch):
|
|
"""When local PID check fails and remote probe succeeds, gateway shows running."""
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
|
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
|
|
monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, {
|
|
"status": "ok",
|
|
"gateway_state": "running",
|
|
"platforms": {"telegram": {"state": "connected"}},
|
|
"pid": 999,
|
|
}))
|
|
|
|
resp = self.client.get("/api/status")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["gateway_running"] is True
|
|
assert data["gateway_pid"] == 999
|
|
assert data["gateway_state"] == "running"
|
|
assert data["gateway_health_url"] == "http://gw:8642"
|
|
|
|
def test_status_remote_probe_not_attempted_when_local_pid_found(self, monkeypatch):
|
|
"""When local PID check succeeds, the remote probe is never called."""
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "get_running_pid", lambda: 1234)
|
|
monkeypatch.setattr(ws, "read_runtime_status", lambda: {
|
|
"gateway_state": "running",
|
|
"platforms": {},
|
|
})
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
|
|
probe_called = [False]
|
|
original = ws._probe_gateway_health
|
|
|
|
def track_probe():
|
|
probe_called[0] = True
|
|
return original()
|
|
|
|
monkeypatch.setattr(ws, "_probe_gateway_health", track_probe)
|
|
|
|
resp = self.client.get("/api/status")
|
|
assert resp.status_code == 200
|
|
assert not probe_called[0]
|
|
|
|
def test_status_remote_probe_not_attempted_when_no_url(self, monkeypatch):
|
|
"""When GATEWAY_HEALTH_URL is unset, no probe is attempted."""
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
|
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None)
|
|
|
|
resp = self.client.get("/api/status")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["gateway_running"] is False
|
|
assert data["gateway_health_url"] is None
|
|
|
|
def test_status_remote_running_null_pid(self, monkeypatch):
|
|
"""Remote gateway running but PID not in response — pid should be None."""
|
|
import hermes_cli.web_server as ws
|
|
|
|
monkeypatch.setattr(ws, "get_running_pid", lambda: None)
|
|
monkeypatch.setattr(ws, "read_runtime_status", lambda: None)
|
|
monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642")
|
|
monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, {
|
|
"status": "ok",
|
|
}))
|
|
|
|
resp = self.client.get("/api/status")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["gateway_running"] is True
|
|
assert data["gateway_pid"] is None
|
|
assert data["gateway_state"] == "running"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dashboard theme normaliser tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNormaliseThemeDefinition:
|
|
"""Tests for _normalise_theme_definition() — parses YAML theme files."""
|
|
|
|
def test_rejects_missing_name(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
assert _normalise_theme_definition({}) is None
|
|
assert _normalise_theme_definition({"name": ""}) is None
|
|
assert _normalise_theme_definition({"name": " "}) is None
|
|
|
|
def test_rejects_non_dict(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
assert _normalise_theme_definition("string") is None
|
|
assert _normalise_theme_definition(None) is None
|
|
assert _normalise_theme_definition([1, 2, 3]) is None
|
|
|
|
def test_loose_colors_shorthand(self):
|
|
"""Bare hex strings under `colors` parse as {hex, alpha=1.0}."""
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({
|
|
"name": "loose",
|
|
"colors": {"background": "#000000", "midground": "#ffffff"},
|
|
})
|
|
assert result is not None
|
|
assert result["palette"]["background"] == {"hex": "#000000", "alpha": 1.0}
|
|
assert result["palette"]["midground"] == {"hex": "#ffffff", "alpha": 1.0}
|
|
# foreground falls back to default (transparent white)
|
|
assert result["palette"]["foreground"]["hex"] == "#ffffff"
|
|
assert result["palette"]["foreground"]["alpha"] == 0.0
|
|
|
|
def test_full_palette_form(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({
|
|
"name": "full",
|
|
"palette": {
|
|
"background": {"hex": "#0a1628", "alpha": 1.0},
|
|
"midground": {"hex": "#a8d0ff", "alpha": 0.9},
|
|
"warmGlow": "rgba(255, 0, 0, 0.5)",
|
|
"noiseOpacity": 0.5,
|
|
},
|
|
})
|
|
assert result["palette"]["background"]["hex"] == "#0a1628"
|
|
assert result["palette"]["midground"]["alpha"] == 0.9
|
|
assert result["palette"]["warmGlow"] == "rgba(255, 0, 0, 0.5)"
|
|
assert result["palette"]["noiseOpacity"] == 0.5
|
|
|
|
def test_default_typography_applied_when_missing(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({"name": "minimal"})
|
|
typo = result["typography"]
|
|
assert "fontSans" in typo
|
|
assert "fontMono" in typo
|
|
assert typo["baseSize"] == "15px"
|
|
assert typo["lineHeight"] == "1.55"
|
|
assert typo["letterSpacing"] == "0"
|
|
|
|
def test_partial_typography_merges_with_defaults(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({
|
|
"name": "partial",
|
|
"typography": {
|
|
"fontSans": "MyFont, sans-serif",
|
|
"baseSize": "12px",
|
|
},
|
|
})
|
|
assert result["typography"]["fontSans"] == "MyFont, sans-serif"
|
|
assert result["typography"]["baseSize"] == "12px"
|
|
# fontMono defaulted
|
|
assert "monospace" in result["typography"]["fontMono"]
|
|
|
|
def test_layout_defaults(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({"name": "minimal"})
|
|
assert result["layout"]["radius"] == "0.5rem"
|
|
assert result["layout"]["density"] == "comfortable"
|
|
|
|
def test_invalid_density_falls_back(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({
|
|
"name": "bad",
|
|
"layout": {"density": "ultra-spacious"},
|
|
})
|
|
assert result["layout"]["density"] == "comfortable"
|
|
|
|
def test_valid_densities_accepted(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
for d in ("compact", "comfortable", "spacious"):
|
|
r = _normalise_theme_definition({"name": "x", "layout": {"density": d}})
|
|
assert r["layout"]["density"] == d
|
|
|
|
def test_color_overrides_filter_unknown_keys(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({
|
|
"name": "o",
|
|
"colorOverrides": {
|
|
"card": "#123456",
|
|
"fakeToken": "#abcdef",
|
|
"primary": 42, # non-string rejected
|
|
"destructive": "#ff0000",
|
|
},
|
|
})
|
|
assert result["colorOverrides"] == {
|
|
"card": "#123456",
|
|
"destructive": "#ff0000",
|
|
}
|
|
|
|
def test_color_overrides_omitted_when_empty(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({"name": "x"})
|
|
assert "colorOverrides" not in result
|
|
|
|
def test_alpha_clamped_to_unit_range(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({
|
|
"name": "c",
|
|
"palette": {"background": {"hex": "#000", "alpha": 99.5}},
|
|
})
|
|
assert r["palette"]["background"]["alpha"] == 1.0
|
|
r2 = _normalise_theme_definition({
|
|
"name": "c",
|
|
"palette": {"background": {"hex": "#000", "alpha": -5}},
|
|
})
|
|
assert r2["palette"]["background"]["alpha"] == 0.0
|
|
|
|
def test_invalid_alpha_uses_default(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({
|
|
"name": "c",
|
|
"palette": {"background": {"hex": "#000", "alpha": "not a number"}},
|
|
})
|
|
assert r["palette"]["background"]["alpha"] == 1.0
|
|
|
|
|
|
class TestDiscoverUserThemes:
|
|
"""Tests for _discover_user_themes() — scans ~/.hermes/dashboard-themes/."""
|
|
|
|
def test_returns_empty_when_dir_missing(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
from hermes_cli import web_server
|
|
assert web_server._discover_user_themes() == []
|
|
|
|
def test_loads_and_normalises_yaml(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
themes_dir = tmp_path / "dashboard-themes"
|
|
themes_dir.mkdir()
|
|
(themes_dir / "ocean.yaml").write_text(
|
|
"name: ocean\n"
|
|
"label: Ocean\n"
|
|
"palette:\n"
|
|
" background:\n"
|
|
" hex: \"#0a1628\"\n"
|
|
" alpha: 1.0\n"
|
|
"layout:\n"
|
|
" density: spacious\n"
|
|
)
|
|
from hermes_cli import web_server
|
|
results = web_server._discover_user_themes()
|
|
assert len(results) == 1
|
|
assert results[0]["name"] == "ocean"
|
|
assert results[0]["label"] == "Ocean"
|
|
assert results[0]["palette"]["background"]["hex"] == "#0a1628"
|
|
assert results[0]["layout"]["density"] == "spacious"
|
|
# defaults filled in
|
|
assert "fontSans" in results[0]["typography"]
|
|
|
|
def test_malformed_yaml_skipped(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
themes_dir = tmp_path / "dashboard-themes"
|
|
themes_dir.mkdir()
|
|
(themes_dir / "bad.yaml").write_text("::: not valid yaml :::\n\tindent wrong")
|
|
(themes_dir / "nameless.yaml").write_text("label: No Name Here\n")
|
|
(themes_dir / "ok.yaml").write_text("name: ok\n")
|
|
from hermes_cli import web_server
|
|
results = web_server._discover_user_themes()
|
|
names = [r["name"] for r in results]
|
|
assert "ok" in names
|
|
assert "bad" not in names # malformed YAML
|
|
assert len(results) == 1 # only the valid one
|
|
|
|
|
|
class TestNormaliseThemeExtensions:
|
|
"""Tests for the extended normaliser fields (assets, customCSS,
|
|
componentStyles, layoutVariant) — the surfaces themes use to reskin
|
|
the dashboard without shipping code."""
|
|
|
|
def test_layout_variant_defaults_to_standard(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
result = _normalise_theme_definition({"name": "t"})
|
|
assert result["layoutVariant"] == "standard"
|
|
|
|
def test_layout_variant_accepts_known_values(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
for variant in ("standard", "cockpit", "tiled"):
|
|
r = _normalise_theme_definition({"name": "t", "layoutVariant": variant})
|
|
assert r["layoutVariant"] == variant
|
|
|
|
def test_layout_variant_rejects_unknown(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({"name": "t", "layoutVariant": "warship"})
|
|
assert r["layoutVariant"] == "standard"
|
|
r2 = _normalise_theme_definition({"name": "t", "layoutVariant": 12})
|
|
assert r2["layoutVariant"] == "standard"
|
|
|
|
def test_assets_named_slots_passthrough(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({
|
|
"name": "t",
|
|
"assets": {
|
|
"bg": "https://example.com/bg.jpg",
|
|
"hero": "linear-gradient(180deg, red, blue)",
|
|
"crest": "/ds-assets/crest.svg",
|
|
"logo": " ", # whitespace-only — dropped
|
|
"notAKnownKey": "ignored",
|
|
},
|
|
})
|
|
assert r["assets"]["bg"] == "https://example.com/bg.jpg"
|
|
assert r["assets"]["hero"].startswith("linear-gradient")
|
|
assert r["assets"]["crest"] == "/ds-assets/crest.svg"
|
|
assert "logo" not in r["assets"] # whitespace-only rejected
|
|
assert "notAKnownKey" not in r["assets"] # unknown slot ignored
|
|
|
|
def test_assets_custom_block(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({
|
|
"name": "t",
|
|
"assets": {
|
|
"custom": {
|
|
"scan-lines": "/img/scan.png",
|
|
"my_overlay": "/img/ov.png",
|
|
"bad key!": "x", # non-alnum key — rejected
|
|
"empty": "", # empty value — rejected
|
|
},
|
|
},
|
|
})
|
|
assert r["assets"]["custom"] == {
|
|
"scan-lines": "/img/scan.png",
|
|
"my_overlay": "/img/ov.png",
|
|
}
|
|
|
|
def test_assets_absent_means_no_field(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({"name": "t"})
|
|
assert "assets" not in r
|
|
|
|
def test_custom_css_passthrough_and_capped(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
# Small CSS passes through verbatim.
|
|
r = _normalise_theme_definition({
|
|
"name": "t",
|
|
"customCSS": "body { color: red; }",
|
|
})
|
|
assert r["customCSS"] == "body { color: red; }"
|
|
|
|
# 40 KiB of CSS gets clipped to the 32 KiB cap.
|
|
huge = "/* x */ " * (40 * 1024 // 8 + 10)
|
|
r2 = _normalise_theme_definition({"name": "t", "customCSS": huge})
|
|
assert len(r2["customCSS"]) <= 32 * 1024
|
|
|
|
def test_custom_css_empty_dropped(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
for val in ("", " \n\t", None):
|
|
r = _normalise_theme_definition({"name": "t", "customCSS": val})
|
|
assert "customCSS" not in r
|
|
|
|
def test_component_styles_per_bucket(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({
|
|
"name": "t",
|
|
"componentStyles": {
|
|
"card": {
|
|
"clipPath": "polygon(0 0, 100% 0, 100% 100%, 0 100%)",
|
|
"boxShadow": "inset 0 0 0 1px red",
|
|
"bad prop!": "ignored", # non-alnum prop rejected
|
|
},
|
|
"header": {"background": "linear-gradient(red, blue)"},
|
|
"rogueBucket": {"foo": "bar"}, # not a known bucket — rejected
|
|
},
|
|
})
|
|
assert r["componentStyles"]["card"] == {
|
|
"clipPath": "polygon(0 0, 100% 0, 100% 100%, 0 100%)",
|
|
"boxShadow": "inset 0 0 0 1px red",
|
|
}
|
|
assert r["componentStyles"]["header"]["background"].startswith("linear-gradient")
|
|
assert "rogueBucket" not in r["componentStyles"]
|
|
|
|
def test_component_styles_empty_buckets_dropped(self):
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({
|
|
"name": "t",
|
|
"componentStyles": {
|
|
"card": {}, # empty — dropped entirely
|
|
"header": {"bad prop!": "ignored"}, # all props rejected — bucket dropped
|
|
"footer": {"background": "black"},
|
|
},
|
|
})
|
|
assert "card" not in r.get("componentStyles", {})
|
|
assert "header" not in r.get("componentStyles", {})
|
|
assert r["componentStyles"]["footer"]["background"] == "black"
|
|
|
|
def test_component_styles_accepts_numeric_values(self):
|
|
"""Numeric values (e.g. opacity: 0.8) are coerced to strings."""
|
|
from hermes_cli.web_server import _normalise_theme_definition
|
|
r = _normalise_theme_definition({
|
|
"name": "t",
|
|
"componentStyles": {"card": {"opacity": 0.8, "zIndex": 5}},
|
|
})
|
|
assert r["componentStyles"]["card"] == {"opacity": "0.8", "zIndex": "5"}
|
|
|
|
|
|
class TestDashboardPluginManifestExtensions:
|
|
"""Tests for the extended plugin manifest fields (tab.override,
|
|
tab.hidden, slots) read by _discover_dashboard_plugins()."""
|
|
|
|
def _write_plugin(self, tmp_path, name, manifest):
|
|
import json
|
|
plug_dir = tmp_path / "plugins" / name / "dashboard"
|
|
plug_dir.mkdir(parents=True)
|
|
(plug_dir / "manifest.json").write_text(json.dumps(manifest))
|
|
return plug_dir
|
|
|
|
def test_override_and_hidden_carried_through(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
self._write_plugin(tmp_path, "skin-home", {
|
|
"name": "skin-home",
|
|
"label": "Skin Home",
|
|
"tab": {"path": "/skin-home", "override": "/", "hidden": True},
|
|
"slots": ["sidebar", "header-left"],
|
|
"entry": "dist/index.js",
|
|
})
|
|
from hermes_cli import web_server
|
|
# Bust the process-level cache so the test plugin is picked up.
|
|
web_server._dashboard_plugins_cache = None
|
|
plugins = web_server._get_dashboard_plugins(force_rescan=True)
|
|
entry = next(p for p in plugins if p["name"] == "skin-home")
|
|
assert entry["tab"]["override"] == "/"
|
|
assert entry["tab"]["hidden"] is True
|
|
assert entry["slots"] == ["sidebar", "header-left"]
|
|
|
|
def test_override_requires_leading_slash(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
self._write_plugin(tmp_path, "bad-override", {
|
|
"name": "bad-override",
|
|
"label": "Bad",
|
|
"tab": {"path": "/bad", "override": "no-leading-slash"},
|
|
"entry": "dist/index.js",
|
|
})
|
|
from hermes_cli import web_server
|
|
web_server._dashboard_plugins_cache = None
|
|
plugins = web_server._get_dashboard_plugins(force_rescan=True)
|
|
entry = next(p for p in plugins if p["name"] == "bad-override")
|
|
assert "override" not in entry["tab"]
|
|
|
|
def test_slots_default_empty(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
self._write_plugin(tmp_path, "no-slots", {
|
|
"name": "no-slots",
|
|
"label": "No Slots",
|
|
"tab": {"path": "/no-slots"},
|
|
"entry": "dist/index.js",
|
|
})
|
|
from hermes_cli import web_server
|
|
web_server._dashboard_plugins_cache = None
|
|
plugins = web_server._get_dashboard_plugins(force_rescan=True)
|
|
entry = next(p for p in plugins if p["name"] == "no-slots")
|
|
assert entry["slots"] == []
|
|
assert "hidden" not in entry["tab"]
|
|
assert "override" not in entry["tab"]
|
|
|
|
def test_slots_filters_non_string_entries(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
self._write_plugin(tmp_path, "mixed-slots", {
|
|
"name": "mixed-slots",
|
|
"label": "Mixed",
|
|
"tab": {"path": "/mixed-slots"},
|
|
"slots": ["sidebar", "", 42, None, "header-right"],
|
|
"entry": "dist/index.js",
|
|
})
|
|
from hermes_cli import web_server
|
|
web_server._dashboard_plugins_cache = None
|
|
plugins = web_server._get_dashboard_plugins(force_rescan=True)
|
|
entry = next(p for p in plugins if p["name"] == "mixed-slots")
|
|
assert entry["slots"] == ["sidebar", "header-right"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# /api/ws — WebSocket wire-compatible with stdio tui_gateway
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTuiGatewayWebSocket:
|
|
"""E2E tests for /api/ws.
|
|
|
|
The WS endpoint multiplexes the same JSON-RPC protocol Ink speaks over
|
|
stdio onto a browser/iOS-friendly socket. These tests exercise the
|
|
transport boundary without booting a real AIAgent — handlers are
|
|
monkey-patched in for deterministic byte-level assertions.
|
|
"""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup(self):
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
from hermes_cli.web_server import app, _SESSION_TOKEN
|
|
self.client = TestClient(app)
|
|
self.token = _SESSION_TOKEN
|
|
|
|
def _url(self, token=None):
|
|
tok = self.token if token is None else token
|
|
return f"/api/ws?token={tok}" if tok else "/api/ws"
|
|
|
|
def _drain_ready(self, ws):
|
|
"""Skip the ``gateway.ready`` event emitted on accept."""
|
|
frame = ws.receive_json()
|
|
assert frame.get("method") == "event"
|
|
assert frame["params"]["type"] == "gateway.ready"
|
|
return frame
|
|
|
|
def test_handshake_emits_gateway_ready(self):
|
|
with self.client.websocket_connect(self._url()) as ws:
|
|
first = ws.receive_json()
|
|
assert first["jsonrpc"] == "2.0"
|
|
assert first["method"] == "event"
|
|
assert first["params"]["type"] == "gateway.ready"
|
|
assert "skin" in first["params"]["payload"]
|
|
|
|
def test_rejects_missing_token(self):
|
|
from starlette.websockets import WebSocketDisconnect
|
|
with pytest.raises(WebSocketDisconnect):
|
|
with self.client.websocket_connect(self._url(token="")) as ws:
|
|
ws.receive_json()
|
|
|
|
def test_rejects_bad_token(self):
|
|
from starlette.websockets import WebSocketDisconnect
|
|
with pytest.raises(WebSocketDisconnect):
|
|
with self.client.websocket_connect(self._url(token="bogus-token-xyz")) as ws:
|
|
ws.receive_json()
|
|
|
|
def test_parse_error_on_bad_frame(self):
|
|
with self.client.websocket_connect(self._url()) as ws:
|
|
self._drain_ready(ws)
|
|
ws.send_text("this is { not json")
|
|
resp = ws.receive_json()
|
|
assert resp["jsonrpc"] == "2.0"
|
|
assert resp["error"]["code"] == -32700
|
|
assert resp["error"]["message"] == "parse error"
|
|
|
|
def test_unknown_method_returns_rpc_error(self):
|
|
with self.client.websocket_connect(self._url()) as ws:
|
|
self._drain_ready(ws)
|
|
ws.send_json({"jsonrpc": "2.0", "id": "u1", "method": "does.not.exist"})
|
|
resp = ws.receive_json()
|
|
assert resp["id"] == "u1"
|
|
assert resp["error"]["code"] == -32601
|
|
assert "does.not.exist" in resp["error"]["message"]
|
|
|
|
def test_inline_handler_returns_response(self):
|
|
"""An inline handler's result round-trips via the WS transport."""
|
|
from tui_gateway import server
|
|
|
|
sentinel = "_ws_inline_test"
|
|
server._methods[sentinel] = lambda rid, params: server._ok(rid, {"pong": params.get("ping")})
|
|
try:
|
|
with self.client.websocket_connect(self._url()) as ws:
|
|
self._drain_ready(ws)
|
|
ws.send_json({"jsonrpc": "2.0", "id": "i1", "method": sentinel, "params": {"ping": "PONG"}})
|
|
resp = ws.receive_json()
|
|
assert resp == {"jsonrpc": "2.0", "id": "i1", "result": {"pong": "PONG"}}
|
|
finally:
|
|
server._methods.pop(sentinel, None)
|
|
|
|
def test_pool_handler_response_arrives_via_ws(self):
|
|
"""Long-handler responses written from the thread pool must reach the WS client."""
|
|
from tui_gateway import server
|
|
|
|
# Register a "slash.exec" replacement so we exercise the pool path
|
|
# (_LONG_HANDLERS includes "slash.exec").
|
|
original = server._methods.get("slash.exec")
|
|
server._methods["slash.exec"] = lambda rid, params: server._ok(rid, {"output": "async-ok"})
|
|
try:
|
|
with self.client.websocket_connect(self._url()) as ws:
|
|
self._drain_ready(ws)
|
|
ws.send_json({"jsonrpc": "2.0", "id": "p1", "method": "slash.exec", "params": {}})
|
|
resp = ws.receive_json()
|
|
assert resp["id"] == "p1"
|
|
assert resp["result"] == {"output": "async-ok"}
|
|
finally:
|
|
if original is not None:
|
|
server._methods["slash.exec"] = original
|
|
else:
|
|
server._methods.pop("slash.exec", None)
|
|
|
|
def test_session_events_route_to_owning_ws(self):
|
|
"""Events emitted for a session created over WS land on that WS."""
|
|
from tui_gateway import server
|
|
from tui_gateway.transport import current_transport
|
|
|
|
sentinel_create = "_ws_emit_test_create"
|
|
sentinel_emit = "_ws_emit_test_fire"
|
|
created_sid = {"value": ""}
|
|
|
|
def create(rid, params):
|
|
sid = f"ws-emit-test-{uuid_hex()}"
|
|
created_sid["value"] = sid
|
|
server._sessions[sid] = {
|
|
"session_key": sid,
|
|
"transport": current_transport(),
|
|
}
|
|
return server._ok(rid, {"session_id": sid})
|
|
|
|
def fire(rid, params):
|
|
sid = params["session_id"]
|
|
server._emit("demo.event", sid, {"n": params.get("n", 0)})
|
|
return server._ok(rid, {"ok": True})
|
|
|
|
def uuid_hex():
|
|
import uuid
|
|
return uuid.uuid4().hex[:8]
|
|
|
|
server._methods[sentinel_create] = create
|
|
server._methods[sentinel_emit] = fire
|
|
try:
|
|
with self.client.websocket_connect(self._url()) as ws:
|
|
self._drain_ready(ws)
|
|
|
|
ws.send_json({"jsonrpc": "2.0", "id": "c1", "method": sentinel_create})
|
|
create_resp = ws.receive_json()
|
|
assert create_resp["id"] == "c1"
|
|
sid = create_resp["result"]["session_id"]
|
|
assert sid == created_sid["value"]
|
|
|
|
ws.send_json({
|
|
"jsonrpc": "2.0",
|
|
"id": "e1",
|
|
"method": sentinel_emit,
|
|
"params": {"session_id": sid, "n": 7},
|
|
})
|
|
# Event fires synchronously inside the handler, so it should
|
|
# arrive before the response.
|
|
frame1 = ws.receive_json()
|
|
frame2 = ws.receive_json()
|
|
|
|
event_frame = frame1 if frame1.get("method") == "event" else frame2
|
|
resp_frame = frame2 if frame2.get("id") == "e1" else frame1
|
|
|
|
assert event_frame["params"]["type"] == "demo.event"
|
|
assert event_frame["params"]["session_id"] == sid
|
|
assert event_frame["params"]["payload"] == {"n": 7}
|
|
assert resp_frame["result"] == {"ok": True}
|
|
finally:
|
|
server._methods.pop(sentinel_create, None)
|
|
server._methods.pop(sentinel_emit, None)
|
|
server._sessions.pop(created_sid["value"], None)
|
|
|
|
def test_ws_disconnect_resets_session_transport(self):
|
|
"""After a WS hangs up, sessions it owned fall back to stdio so stray emits don't crash."""
|
|
from tui_gateway import server
|
|
from tui_gateway.transport import current_transport
|
|
|
|
sentinel = "_ws_disconnect_test"
|
|
captured = {"sid": "", "transport": None}
|
|
|
|
def create(rid, params):
|
|
sid = "ws-disconnect-sid"
|
|
captured["sid"] = sid
|
|
captured["transport"] = current_transport()
|
|
server._sessions[sid] = {
|
|
"session_key": sid,
|
|
"transport": captured["transport"],
|
|
}
|
|
return server._ok(rid, {"session_id": sid})
|
|
|
|
server._methods[sentinel] = create
|
|
try:
|
|
with self.client.websocket_connect(self._url()) as ws:
|
|
self._drain_ready(ws)
|
|
ws.send_json({"jsonrpc": "2.0", "id": "c1", "method": sentinel})
|
|
ws.receive_json()
|
|
|
|
# Give the server a moment to run the finally-block cleanup.
|
|
import time
|
|
for _ in range(50):
|
|
if server._sessions.get(captured["sid"], {}).get("transport") is not captured["transport"]:
|
|
break
|
|
time.sleep(0.02)
|
|
|
|
sess = server._sessions.get(captured["sid"])
|
|
assert sess is not None
|
|
assert sess["transport"] is server._stdio_transport
|
|
finally:
|
|
server._methods.pop(sentinel, None)
|
|
server._sessions.pop(captured["sid"], None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Transport parity — same RPC, stdio vs WS, byte-identical envelopes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTuiGatewayTransportParity:
|
|
"""The whole point of the transport abstraction is that handlers don't
|
|
know what's on the other end. These tests lock that in: the response
|
|
envelope produced by ``server.handle_request`` directly (stdio fast path)
|
|
must match what a WS client receives for the same request.
|
|
"""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup(self):
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
from hermes_cli.web_server import app, _SESSION_TOKEN
|
|
self.client = TestClient(app)
|
|
self.token = _SESSION_TOKEN
|
|
|
|
def _ws_roundtrip(self, req: dict) -> dict:
|
|
with self.client.websocket_connect(f"/api/ws?token={self.token}") as ws:
|
|
ready = ws.receive_json()
|
|
assert ready["params"]["type"] == "gateway.ready"
|
|
ws.send_json(req)
|
|
return ws.receive_json()
|
|
|
|
def test_parity_unknown_method(self):
|
|
from tui_gateway import server
|
|
req = {"jsonrpc": "2.0", "id": "p-unk", "method": "no.such.method"}
|
|
assert self._ws_roundtrip(req) == server.handle_request(req)
|
|
|
|
def test_parity_inline_handler(self):
|
|
from tui_gateway import server
|
|
|
|
sentinel = "_parity_inline"
|
|
server._methods[sentinel] = lambda rid, params: server._ok(rid, {
|
|
"echo": params,
|
|
"const": 42,
|
|
"nested": {"a": [1, 2, 3], "b": None},
|
|
})
|
|
try:
|
|
req = {
|
|
"jsonrpc": "2.0",
|
|
"id": "p-inline",
|
|
"method": sentinel,
|
|
"params": {"hello": "world", "n": 1},
|
|
}
|
|
assert self._ws_roundtrip(req) == server.handle_request(req)
|
|
finally:
|
|
server._methods.pop(sentinel, None)
|
|
|
|
def test_parity_error_envelope(self):
|
|
from tui_gateway import server
|
|
|
|
sentinel = "_parity_err"
|
|
server._methods[sentinel] = lambda rid, params: server._err(rid, 4242, "nope")
|
|
try:
|
|
req = {"jsonrpc": "2.0", "id": "p-err", "method": sentinel}
|
|
assert self._ws_roundtrip(req) == server.handle_request(req)
|
|
finally:
|
|
server._methods.pop(sentinel, None)
|
|
|
|
def test_parity_stdio_transport_also_works(self):
|
|
"""Calling dispatch() with the stdio transport explicitly must match the default."""
|
|
from tui_gateway import server
|
|
|
|
sentinel = "_parity_stdio"
|
|
server._methods[sentinel] = lambda rid, params: server._ok(rid, {"ok": True, "p": params})
|
|
try:
|
|
req = {"jsonrpc": "2.0", "id": "p-std", "method": sentinel, "params": {"x": 1}}
|
|
# Default (no transport arg)
|
|
default_resp = server.dispatch(dict(req))
|
|
# Explicit stdio transport
|
|
explicit_resp = server.dispatch(dict(req), server._stdio_transport)
|
|
assert default_resp == explicit_resp
|
|
assert default_resp["result"] == {"ok": True, "p": {"x": 1}}
|
|
finally:
|
|
server._methods.pop(sentinel, None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# E2E: drive the "Ink --tui" JSON-RPC surface over ANY transport
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTuiGatewayE2EAnyPort:
|
|
"""Scripted multi-message conversations that exercise the real dispatcher.
|
|
|
|
The same scripted sequence runs over (a) direct ``handle_request`` calls
|
|
and (b) a live WebSocket. Both must produce the same response envelopes
|
|
in the same order. This is the "hermes --tui in any port" check.
|
|
"""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _setup(self):
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip("fastapi/starlette not installed")
|
|
from hermes_cli.web_server import app, _SESSION_TOKEN
|
|
self.client = TestClient(app)
|
|
self.token = _SESSION_TOKEN
|
|
|
|
def _install_scripted_methods(self):
|
|
"""Install a tiny surface that mimics what Ink exercises on startup:
|
|
|
|
- commands.ping returns a deterministic pong
|
|
- session.sim_create creates a fake session (no real agent)
|
|
- session.sim_close tears down the session
|
|
- config.sim_get_value reads a key
|
|
"""
|
|
from tui_gateway import server
|
|
from tui_gateway.transport import current_transport
|
|
|
|
added = []
|
|
|
|
def ping(rid, params):
|
|
return server._ok(rid, {"pong": True, "id": rid})
|
|
server._methods["commands.ping"] = ping
|
|
added.append("commands.ping")
|
|
|
|
def sim_create(rid, params):
|
|
import uuid
|
|
sid = f"sim-{uuid.uuid4().hex[:6]}"
|
|
server._sessions[sid] = {
|
|
"session_key": sid,
|
|
"transport": current_transport(),
|
|
"agent": None,
|
|
}
|
|
return server._ok(rid, {"session_id": sid})
|
|
server._methods["session.sim_create"] = sim_create
|
|
added.append("session.sim_create")
|
|
|
|
def sim_close(rid, params):
|
|
sid = params.get("session_id", "")
|
|
removed = server._sessions.pop(sid, None) is not None
|
|
return server._ok(rid, {"closed": removed})
|
|
server._methods["session.sim_close"] = sim_close
|
|
added.append("session.sim_close")
|
|
|
|
def sim_get_value(rid, params):
|
|
return server._ok(rid, {"value": "deterministic", "key": params.get("key", "")})
|
|
server._methods["config.sim_get_value"] = sim_get_value
|
|
added.append("config.sim_get_value")
|
|
|
|
return added
|
|
|
|
def _uninstall(self, added):
|
|
from tui_gateway import server
|
|
for name in added:
|
|
server._methods.pop(name, None)
|
|
|
|
def _script(self):
|
|
return [
|
|
{"jsonrpc": "2.0", "id": "s1", "method": "commands.ping"},
|
|
{"jsonrpc": "2.0", "id": "s2", "method": "session.sim_create"},
|
|
{"jsonrpc": "2.0", "id": "s3", "method": "config.sim_get_value",
|
|
"params": {"key": "display.skin"}},
|
|
]
|
|
|
|
def test_script_over_direct_and_ws_match(self):
|
|
from tui_gateway import server
|
|
|
|
added = self._install_scripted_methods()
|
|
try:
|
|
script = self._script()
|
|
|
|
# Run over direct dispatch
|
|
direct_resps = [server.handle_request(dict(req)) for req in script]
|
|
# Clean up the session.create we just made so we don't leak into
|
|
# the WS run.
|
|
for r in direct_resps:
|
|
sid = (r.get("result") or {}).get("session_id")
|
|
if sid:
|
|
server._sessions.pop(sid, None)
|
|
|
|
# Run over WS
|
|
with self.client.websocket_connect(f"/api/ws?token={self.token}") as ws:
|
|
ready = ws.receive_json()
|
|
assert ready["params"]["type"] == "gateway.ready"
|
|
|
|
ws_resps = []
|
|
for req in script:
|
|
ws.send_json(req)
|
|
ws_resps.append(ws.receive_json())
|
|
|
|
# Result shapes (stripping session-identity fields) should match.
|
|
def normalize(r):
|
|
r = dict(r)
|
|
if "result" in r and isinstance(r["result"], dict):
|
|
result = dict(r["result"])
|
|
# session ids are random — compare only structure
|
|
if "session_id" in result:
|
|
result["session_id"] = "<random>"
|
|
r["result"] = result
|
|
return r
|
|
|
|
assert [normalize(r) for r in direct_resps] == [normalize(r) for r in ws_resps]
|
|
|
|
# And both surfaces ACTUALLY executed their handlers.
|
|
assert all("result" in r for r in ws_resps)
|
|
assert ws_resps[0]["result"]["pong"] is True
|
|
assert ws_resps[2]["result"]["value"] == "deterministic"
|
|
finally:
|
|
# Clean up any sessions created during the WS run.
|
|
for sid in [
|
|
sid for sid, sess in list(server._sessions.items()) if sid.startswith("sim-")
|
|
]:
|
|
server._sessions.pop(sid, None)
|
|
self._uninstall(added)
|
|
|
|
def test_session_lifecycle_over_ws(self):
|
|
"""Open a session, then close it — via WS only."""
|
|
from tui_gateway import server
|
|
|
|
added = self._install_scripted_methods()
|
|
try:
|
|
with self.client.websocket_connect(f"/api/ws?token={self.token}") as ws:
|
|
ready = ws.receive_json()
|
|
assert ready["params"]["type"] == "gateway.ready"
|
|
|
|
ws.send_json({"jsonrpc": "2.0", "id": "c1", "method": "session.sim_create"})
|
|
create = ws.receive_json()
|
|
sid = create["result"]["session_id"]
|
|
assert sid in server._sessions
|
|
|
|
ws.send_json({
|
|
"jsonrpc": "2.0", "id": "x1", "method": "session.sim_close",
|
|
"params": {"session_id": sid},
|
|
})
|
|
close = ws.receive_json()
|
|
assert close["result"] == {"closed": True}
|
|
assert sid not in server._sessions
|
|
finally:
|
|
self._uninstall(added)
|