Something went wrong. Try again.
The agent that grows with you
Something went wrong. Try again.
80 kB · 1966 lines
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967"""Tests for hermes_cli.web_server and related config utilities."""
import osimport jsonimport tempfilefrom pathlib import Pathfrom 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"]
def test_page_scoped_slots_preserved(self, tmp_path, monkeypatch): """Page-scoped slot names (e.g. ``sessions:top``) round-trip through the manifest loader untouched. The backend has no allowlist — the frontend ``<PluginSlot name="...">`` placements decide what actually renders — but the loader must not mangle colons in slot names.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path)) self._write_plugin(tmp_path, "page-slots", { "name": "page-slots", "label": "Page Slots", "tab": {"path": "/page-slots", "hidden": True}, "slots": [ "sessions:top", "analytics:bottom", "logs:top", "skills:bottom", "config:top", "env:bottom", "docs:top", "cron:bottom", "chat:top", ], "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"] == "page-slots") assert entry["slots"] == [ "sessions:top", "analytics:bottom", "logs:top", "skills:bottom", "config:top", "env:bottom", "docs:top", "cron:bottom", "chat:top", ]
# ---------------------------------------------------------------------------# /api/pty WebSocket — terminal bridge for the dashboard "Chat" tab.## These tests drive the endpoint with a tiny fake command (typically ``cat``# or ``sh -c 'printf …'``) instead of the real ``hermes --tui`` binary. The# endpoint resolves its argv through ``_resolve_chat_argv``, so tests# monkeypatch that hook.# ---------------------------------------------------------------------------
import sys
skip_on_windows = pytest.mark.skipif( sys.platform.startswith("win"), reason="PTY bridge is POSIX-only")
@skip_on_windowsclass TestPtyWebSocket: @pytest.fixture(autouse=True) def _setup(self, monkeypatch, _isolate_hermes_home): from starlette.testclient import TestClient
import hermes_cli.web_server as ws
# Avoid exec'ing the actual TUI in tests: every test below installs # its own fake argv via ``ws._resolve_chat_argv``. self.ws_module = ws monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True) self.token = ws._SESSION_TOKEN self.client = TestClient(ws.app)
def _url(self, token: str | None = None, **params: str) -> str: tok = token if token is not None else self.token # TestClient.websocket_connect takes the path; it reconstructs the # query string, so we pass it inline. from urllib.parse import urlencode
q = {"token": tok, **params} return f"/api/pty?{urlencode(q)}"
def test_rejects_when_embedded_chat_disabled(self, monkeypatch): monkeypatch.setattr(self.ws_module, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", False) from starlette.websockets import WebSocketDisconnect
with pytest.raises(WebSocketDisconnect) as exc: with self.client.websocket_connect(self._url()): pass assert exc.value.code == 4403
def test_rejects_missing_token(self, monkeypatch): monkeypatch.setattr( self.ws_module, "_resolve_chat_argv", lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), ) from starlette.websockets import WebSocketDisconnect
with pytest.raises(WebSocketDisconnect) as exc: with self.client.websocket_connect("/api/pty"): pass assert exc.value.code == 4401
def test_rejects_bad_token(self, monkeypatch): monkeypatch.setattr( self.ws_module, "_resolve_chat_argv", lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), ) from starlette.websockets import WebSocketDisconnect
with pytest.raises(WebSocketDisconnect) as exc: with self.client.websocket_connect(self._url(token="wrong")): pass assert exc.value.code == 4401
def test_streams_child_stdout_to_client(self, monkeypatch): monkeypatch.setattr( self.ws_module, "_resolve_chat_argv", lambda resume=None, sidecar_url=None: ( ["/bin/sh", "-c", "printf hermes-ws-ok"], None, None, ), ) with self.client.websocket_connect(self._url()) as conn: # Drain frames until we see the needle or time out. TestClient's # recv_bytes blocks; loop until we have the signal byte string. buf = b"" import time
deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: try: frame = conn.receive_bytes() except Exception: break if frame: buf += frame if b"hermes-ws-ok" in buf: break assert b"hermes-ws-ok" in buf
def test_client_input_reaches_child_stdin(self, monkeypatch): # ``cat`` echoes stdin back, so a write → read round-trip proves # the full duplex path. monkeypatch.setattr( self.ws_module, "_resolve_chat_argv", lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), ) with self.client.websocket_connect(self._url()) as conn: conn.send_bytes(b"round-trip-payload\n") buf = b"" import time
deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: frame = conn.receive_bytes() if frame: buf += frame if b"round-trip-payload" in buf: break assert b"round-trip-payload" in buf
def test_resize_escape_is_forwarded(self, monkeypatch): # Resize escape gets intercepted and applied via TIOCSWINSZ, # then ``tput cols/lines`` reports the new dimensions back. monkeypatch.setattr( self.ws_module, "_resolve_chat_argv", # sleep gives the test time to push the resize before tput runs lambda resume=None, sidecar_url=None: ( ["/bin/sh", "-c", "sleep 0.15; tput cols; tput lines"], None, None, ), ) with self.client.websocket_connect(self._url()) as conn: conn.send_text("\x1b[RESIZE:99;41]") buf = b"" import time
deadline = time.monotonic() + 5.0 while time.monotonic() < deadline: frame = conn.receive_bytes() if frame: buf += frame if b"99" in buf and b"41" in buf: break assert b"99" in buf and b"41" in buf
def test_unavailable_platform_closes_with_message(self, monkeypatch): from hermes_cli.pty_bridge import PtyUnavailableError
def _raise(argv, **kwargs): raise PtyUnavailableError("pty missing for tests")
monkeypatch.setattr( self.ws_module, "_resolve_chat_argv", lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), ) # Patch PtyBridge.spawn at the web_server module's binding. import hermes_cli.web_server as ws_mod
monkeypatch.setattr(ws_mod.PtyBridge, "spawn", classmethod(lambda cls, *a, **k: _raise(*a, **k)))
with self.client.websocket_connect(self._url()) as conn: # Expect a final text frame with the error message, then close. msg = conn.receive_text() assert "pty missing" in msg or "unavailable" in msg.lower() or "pty" in msg.lower()
def test_resume_parameter_is_forwarded_to_argv(self, monkeypatch): captured: dict = {}
def fake_resolve(resume=None, sidecar_url=None): captured["resume"] = resume return (["/bin/sh", "-c", "printf resume-arg-ok"], None, None)
monkeypatch.setattr(self.ws_module, "_resolve_chat_argv", fake_resolve)
with self.client.websocket_connect(self._url(resume="sess-42")) as conn: # Drain briefly so the handler actually invokes the resolver. try: conn.receive_bytes() except Exception: pass assert captured.get("resume") == "sess-42"
def test_channel_param_propagates_sidecar_url(self, monkeypatch): """When /api/pty is opened with ?channel=, the PTY child gets a HERMES_TUI_SIDECAR_URL env var pointing back at /api/pub on the same channel — which is how tool events reach the dashboard sidebar.""" captured: dict = {}
def fake_resolve(resume=None, sidecar_url=None): captured["sidecar_url"] = sidecar_url return (["/bin/sh", "-c", "printf sidecar-ok"], None, None)
monkeypatch.setattr(self.ws_module, "_resolve_chat_argv", fake_resolve) monkeypatch.setattr( self.ws_module.app.state, "bound_host", "127.0.0.1", raising=False ) monkeypatch.setattr( self.ws_module.app.state, "bound_port", 9119, raising=False )
with self.client.websocket_connect(self._url(channel="abc-123")) as conn: try: conn.receive_bytes() except Exception: pass
url = captured.get("sidecar_url") or "" assert url.startswith("ws://127.0.0.1:9119/api/pub?") assert "channel=abc-123" in url assert "token=" in url
def test_pub_broadcasts_to_events_subscribers(self, monkeypatch): """Frame written to /api/pub is rebroadcast verbatim to every /api/events subscriber on the same channel.""" from urllib.parse import urlencode
qs = urlencode({"token": self.token, "channel": "broadcast-test"}) pub_path = f"/api/pub?{qs}" sub_path = f"/api/events?{qs}"
with self.client.websocket_connect(sub_path) as sub: with self.client.websocket_connect(pub_path) as pub: pub.send_text('{"type":"tool.start","payload":{"tool_id":"t1"}}') received = sub.receive_text()
assert "tool.start" in received assert '"tool_id":"t1"' in received
def test_events_rejects_missing_channel(self): from starlette.websockets import WebSocketDisconnect
with pytest.raises(WebSocketDisconnect) as exc: with self.client.websocket_connect( f"/api/events?token={self.token}" ): pass assert exc.value.code == 4400