test_web_server.py
1 """Tests for hermes_cli.web_server and related config utilities.""" 2 3 import os 4 import json 5 import tempfile 6 from pathlib import Path 7 from unittest.mock import patch, MagicMock 8 9 import pytest 10 11 from hermes_cli.config import ( 12 DEFAULT_CONFIG, 13 reload_env, 14 redact_key, 15 _EXTRA_ENV_KEYS, 16 OPTIONAL_ENV_VARS, 17 ) 18 19 20 # --------------------------------------------------------------------------- 21 # reload_env tests 22 # --------------------------------------------------------------------------- 23 24 25 class TestReloadEnv: 26 """Tests for reload_env() — re-reads .env into os.environ.""" 27 28 def test_adds_new_vars(self, tmp_path): 29 """reload_env() adds vars from .env that are not in os.environ.""" 30 env_file = tmp_path / ".env" 31 env_file.write_text("TEST_RELOAD_VAR=hello123\n") 32 with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}): 33 os.environ.pop("TEST_RELOAD_VAR", None) 34 count = reload_env() 35 assert count >= 1 36 assert os.environ.get("TEST_RELOAD_VAR") == "hello123" 37 os.environ.pop("TEST_RELOAD_VAR", None) 38 39 def test_updates_changed_vars(self, tmp_path): 40 """reload_env() updates vars whose value changed on disk.""" 41 env_file = tmp_path / ".env" 42 env_file.write_text("TEST_RELOAD_VAR=old_value\n") 43 with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}): 44 os.environ["TEST_RELOAD_VAR"] = "old_value" 45 # Now change the file 46 env_file.write_text("TEST_RELOAD_VAR=new_value\n") 47 count = reload_env() 48 assert count >= 1 49 assert os.environ.get("TEST_RELOAD_VAR") == "new_value" 50 os.environ.pop("TEST_RELOAD_VAR", None) 51 52 def test_removes_deleted_known_vars(self, tmp_path): 53 """reload_env() removes known Hermes vars not present in .env.""" 54 env_file = tmp_path / ".env" 55 env_file.write_text("") # empty .env 56 # Pick a known key from OPTIONAL_ENV_VARS 57 known_key = next(iter(OPTIONAL_ENV_VARS.keys())) 58 with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}): 59 os.environ[known_key] = "stale_value" 60 count = reload_env() 61 assert known_key not in os.environ 62 assert count >= 1 63 64 def test_does_not_remove_unknown_vars(self, tmp_path): 65 """reload_env() preserves non-Hermes env vars even when absent from .env.""" 66 env_file = tmp_path / ".env" 67 env_file.write_text("") 68 with patch.dict(reload_env.__globals__, {"get_env_path": lambda: env_file}): 69 os.environ["MY_CUSTOM_UNRELATED_VAR"] = "keep_me" 70 reload_env() 71 assert os.environ.get("MY_CUSTOM_UNRELATED_VAR") == "keep_me" 72 os.environ.pop("MY_CUSTOM_UNRELATED_VAR", None) 73 74 75 # --------------------------------------------------------------------------- 76 # redact_key tests 77 # --------------------------------------------------------------------------- 78 79 80 class TestRedactKey: 81 def test_long_key_shows_prefix_suffix(self): 82 result = redact_key("sk-1234567890abcdef") 83 assert result.startswith("sk-1") 84 assert result.endswith("cdef") 85 assert "..." in result 86 87 def test_short_key_fully_masked(self): 88 assert redact_key("short") == "***" 89 90 def test_empty_key(self): 91 result = redact_key("") 92 assert "not set" in result.lower() or result == "***" or "\x1b" in result 93 94 95 # --------------------------------------------------------------------------- 96 # web_server tests (FastAPI endpoints) 97 # --------------------------------------------------------------------------- 98 99 100 class TestWebServerEndpoints: 101 """Test the FastAPI REST endpoints using Starlette TestClient.""" 102 103 @pytest.fixture(autouse=True) 104 def _setup_test_client(self, monkeypatch, _isolate_hermes_home): 105 """Create a TestClient and isolate the state DB under the test HERMES_HOME.""" 106 try: 107 from starlette.testclient import TestClient 108 except ImportError: 109 pytest.skip("fastapi/starlette not installed") 110 111 import hermes_state 112 from hermes_constants import get_hermes_home 113 from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN 114 115 monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db") 116 117 self.client = TestClient(app) 118 self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN 119 120 def test_get_status(self): 121 resp = self.client.get("/api/status") 122 assert resp.status_code == 200 123 data = resp.json() 124 assert "version" in data 125 assert "hermes_home" in data 126 assert "active_sessions" in data 127 128 def test_get_status_filters_unconfigured_gateway_platforms(self, monkeypatch): 129 import gateway.config as gateway_config 130 import hermes_cli.web_server as web_server 131 132 class _Platform: 133 def __init__(self, value): 134 self.value = value 135 136 class _GatewayConfig: 137 def get_connected_platforms(self): 138 return [_Platform("telegram")] 139 140 monkeypatch.setattr(web_server, "get_running_pid", lambda: 1234) 141 monkeypatch.setattr( 142 web_server, 143 "read_runtime_status", 144 lambda: { 145 "gateway_state": "running", 146 "updated_at": "2026-04-12T00:00:00+00:00", 147 "platforms": { 148 "telegram": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"}, 149 "whatsapp": {"state": "retrying", "updated_at": "2026-04-12T00:00:00+00:00"}, 150 "feishu": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"}, 151 }, 152 }, 153 ) 154 monkeypatch.setattr(web_server, "check_config_version", lambda: (1, 1)) 155 monkeypatch.setattr(gateway_config, "load_gateway_config", lambda: _GatewayConfig()) 156 157 resp = self.client.get("/api/status") 158 159 assert resp.status_code == 200 160 assert resp.json()["gateway_platforms"] == { 161 "telegram": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"}, 162 } 163 164 def test_get_status_hides_stale_platforms_when_gateway_not_running(self, monkeypatch): 165 import gateway.config as gateway_config 166 import hermes_cli.web_server as web_server 167 168 class _GatewayConfig: 169 def get_connected_platforms(self): 170 return [] 171 172 monkeypatch.setattr(web_server, "get_running_pid", lambda: None) 173 monkeypatch.setattr( 174 web_server, 175 "read_runtime_status", 176 lambda: { 177 "gateway_state": "startup_failed", 178 "updated_at": "2026-04-12T00:00:00+00:00", 179 "platforms": { 180 "whatsapp": {"state": "retrying", "updated_at": "2026-04-12T00:00:00+00:00"}, 181 "feishu": {"state": "connected", "updated_at": "2026-04-12T00:00:00+00:00"}, 182 }, 183 }, 184 ) 185 monkeypatch.setattr(web_server, "check_config_version", lambda: (1, 1)) 186 monkeypatch.setattr(gateway_config, "load_gateway_config", lambda: _GatewayConfig()) 187 188 resp = self.client.get("/api/status") 189 190 assert resp.status_code == 200 191 assert resp.json()["gateway_state"] == "startup_failed" 192 assert resp.json()["gateway_platforms"] == {} 193 194 def test_get_config_schema(self): 195 resp = self.client.get("/api/config/schema") 196 assert resp.status_code == 200 197 data = resp.json() 198 assert "fields" in data 199 assert "category_order" in data 200 schema = data["fields"] 201 assert len(schema) > 100 # Should have 150+ fields 202 assert "model" in schema 203 # Verify category_order is a non-empty list 204 assert isinstance(data["category_order"], list) 205 assert len(data["category_order"]) > 0 206 assert "general" in data["category_order"] 207 208 def test_get_config_defaults(self): 209 resp = self.client.get("/api/config/defaults") 210 assert resp.status_code == 200 211 defaults = resp.json() 212 assert "model" in defaults 213 214 def test_get_env_vars(self): 215 resp = self.client.get("/api/env") 216 assert resp.status_code == 200 217 data = resp.json() 218 # Should contain known env var names 219 assert any(k.endswith("_API_KEY") or k.endswith("_TOKEN") for k in data.keys()) 220 221 def test_reveal_env_var(self, tmp_path): 222 """POST /api/env/reveal should return the real unredacted value.""" 223 from hermes_cli.config import save_env_value 224 from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN 225 save_env_value("TEST_REVEAL_KEY", "super-secret-value-12345") 226 resp = self.client.post( 227 "/api/env/reveal", 228 json={"key": "TEST_REVEAL_KEY"}, 229 headers={_SESSION_HEADER_NAME: _SESSION_TOKEN}, 230 ) 231 assert resp.status_code == 200 232 data = resp.json() 233 assert data["key"] == "TEST_REVEAL_KEY" 234 assert data["value"] == "super-secret-value-12345" 235 236 def test_reveal_env_var_not_found(self): 237 """POST /api/env/reveal should 404 for unknown keys.""" 238 from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN 239 resp = self.client.post( 240 "/api/env/reveal", 241 json={"key": "NONEXISTENT_KEY_XYZ"}, 242 headers={_SESSION_HEADER_NAME: _SESSION_TOKEN}, 243 ) 244 assert resp.status_code == 404 245 246 def test_reveal_env_var_no_token(self, tmp_path): 247 """POST /api/env/reveal without token should return 401.""" 248 from starlette.testclient import TestClient 249 from hermes_cli.web_server import app 250 from hermes_cli.config import save_env_value 251 save_env_value("TEST_REVEAL_NOAUTH", "secret-value") 252 # Use a fresh client WITHOUT the dashboard session header 253 unauth_client = TestClient(app) 254 resp = unauth_client.post( 255 "/api/env/reveal", 256 json={"key": "TEST_REVEAL_NOAUTH"}, 257 ) 258 assert resp.status_code == 401 259 260 def test_reveal_env_var_bad_token(self, tmp_path): 261 """POST /api/env/reveal with wrong token should return 401.""" 262 from hermes_cli.config import save_env_value 263 from hermes_cli.web_server import _SESSION_HEADER_NAME 264 save_env_value("TEST_REVEAL_BADAUTH", "secret-value") 265 resp = self.client.post( 266 "/api/env/reveal", 267 json={"key": "TEST_REVEAL_BADAUTH"}, 268 headers={_SESSION_HEADER_NAME: "wrong-token-here"}, 269 ) 270 assert resp.status_code == 401 271 272 def test_reveal_env_var_custom_session_header_ignores_proxy_authorization(self, tmp_path): 273 """A valid dashboard session header should coexist with proxy auth.""" 274 from hermes_cli.config import save_env_value 275 from hermes_cli.web_server import _SESSION_HEADER_NAME, _SESSION_TOKEN 276 277 save_env_value("TEST_REVEAL_PROXY_AUTH", "secret-value") 278 resp = self.client.post( 279 "/api/env/reveal", 280 json={"key": "TEST_REVEAL_PROXY_AUTH"}, 281 headers={ 282 _SESSION_HEADER_NAME: _SESSION_TOKEN, 283 "Authorization": "Basic dXNlcjpwYXNz", 284 }, 285 ) 286 287 assert resp.status_code == 200 288 assert resp.json()["value"] == "secret-value" 289 290 def test_reveal_env_var_legacy_authorization_header_still_works(self, tmp_path): 291 """Keep old dashboard bundles working while the new header rolls out.""" 292 from hermes_cli.config import save_env_value 293 from hermes_cli.web_server import _SESSION_TOKEN 294 295 save_env_value("TEST_REVEAL_LEGACY_AUTH", "secret-value") 296 resp = self.client.post( 297 "/api/env/reveal", 298 json={"key": "TEST_REVEAL_LEGACY_AUTH"}, 299 headers={"Authorization": f"Bearer {_SESSION_TOKEN}"}, 300 ) 301 302 assert resp.status_code == 200 303 304 def test_session_token_endpoint_removed(self): 305 """GET /api/auth/session-token should no longer exist (token injected via HTML).""" 306 resp = self.client.get("/api/auth/session-token") 307 # The endpoint is gone — the catch-all SPA route serves index.html 308 # or the middleware returns 401 for unauthenticated /api/ paths. 309 assert resp.status_code in (200, 404) 310 # Either way, it must NOT return the token as JSON 311 try: 312 data = resp.json() 313 assert "token" not in data 314 except Exception: 315 pass # Not JSON — that's fine (SPA HTML) 316 317 def test_unauthenticated_api_blocked(self): 318 """API requests without the session token should be rejected.""" 319 from starlette.testclient import TestClient 320 from hermes_cli.web_server import app 321 # Create a client WITHOUT the dashboard session header 322 unauth_client = TestClient(app) 323 resp = unauth_client.get("/api/env") 324 assert resp.status_code == 401 325 resp = unauth_client.get("/api/config") 326 assert resp.status_code == 401 327 # Public endpoints should still work 328 resp = unauth_client.get("/api/status") 329 assert resp.status_code == 200 330 331 def test_path_traversal_blocked(self): 332 """Verify URL-encoded path traversal is blocked.""" 333 # %2e%2e = .. 334 resp = self.client.get("/%2e%2e/%2e%2e/etc/passwd") 335 # Should return 200 with index.html (SPA fallback), not the actual file 336 assert resp.status_code in (200, 404) 337 if resp.status_code == 200: 338 # Should be the SPA fallback, not the system file 339 assert "root:" not in resp.text 340 341 def test_path_traversal_dotdot_blocked(self): 342 """Direct .. path traversal via encoded sequences.""" 343 resp = self.client.get("/%2e%2e/hermes_cli/web_server.py") 344 assert resp.status_code in (200, 404) 345 if resp.status_code == 200: 346 assert "FastAPI" not in resp.text # Should not serve the actual source 347 348 349 # --------------------------------------------------------------------------- 350 # _build_schema_from_config tests 351 # --------------------------------------------------------------------------- 352 353 354 class TestBuildSchemaFromConfig: 355 def test_produces_expected_field_count(self): 356 from hermes_cli.web_server import CONFIG_SCHEMA 357 # DEFAULT_CONFIG has ~150+ leaf fields 358 assert len(CONFIG_SCHEMA) > 100 359 360 def test_schema_entries_have_required_fields(self): 361 from hermes_cli.web_server import CONFIG_SCHEMA 362 for key, entry in list(CONFIG_SCHEMA.items())[:10]: 363 assert "type" in entry, f"Missing type for {key}" 364 assert "category" in entry, f"Missing category for {key}" 365 366 def test_overrides_applied(self): 367 from hermes_cli.web_server import CONFIG_SCHEMA 368 # terminal.backend should be a select with options 369 if "terminal.backend" in CONFIG_SCHEMA: 370 entry = CONFIG_SCHEMA["terminal.backend"] 371 assert entry["type"] == "select" 372 assert "options" in entry 373 assert "local" in entry["options"] 374 assert "vercel_sandbox" in entry["options"] 375 runtime_entry = CONFIG_SCHEMA["terminal.vercel_runtime"] 376 assert runtime_entry["type"] == "select" 377 assert "node24" in runtime_entry["options"] 378 assert "python3.13" in runtime_entry["options"] 379 assert len(runtime_entry["options"]) >= 3 380 381 def test_empty_prefix_produces_correct_keys(self): 382 from hermes_cli.web_server import _build_schema_from_config 383 test_config = {"model": "test", "nested": {"key": "val"}} 384 schema = _build_schema_from_config(test_config) 385 assert "model" in schema 386 assert "nested.key" in schema 387 388 def test_top_level_scalars_get_general_category(self): 389 """Top-level scalar fields should be in 'general' category.""" 390 from hermes_cli.web_server import CONFIG_SCHEMA 391 assert CONFIG_SCHEMA["model"]["category"] == "general" 392 393 def test_nested_keys_get_parent_category(self): 394 """Nested fields should use the top-level parent as their category.""" 395 from hermes_cli.web_server import CONFIG_SCHEMA 396 if "agent.max_turns" in CONFIG_SCHEMA: 397 assert CONFIG_SCHEMA["agent.max_turns"]["category"] == "agent" 398 399 def test_category_merge_applied(self): 400 """Small categories should be merged into larger ones.""" 401 from hermes_cli.web_server import CONFIG_SCHEMA 402 categories = {e["category"] for e in CONFIG_SCHEMA.values()} 403 # These should be merged away 404 assert "privacy" not in categories # merged into security 405 assert "context" not in categories # merged into agent 406 407 def test_no_single_field_categories(self): 408 """After merging, no category should have just 1 field.""" 409 from hermes_cli.web_server import CONFIG_SCHEMA 410 from collections import Counter 411 cats = Counter(e["category"] for e in CONFIG_SCHEMA.values()) 412 for cat, count in cats.items(): 413 assert count >= 2, f"Category '{cat}' has only {count} field(s) — should be merged" 414 415 416 # --------------------------------------------------------------------------- 417 # Config round-trip tests 418 # --------------------------------------------------------------------------- 419 420 421 class TestConfigRoundTrip: 422 """Verify config survives GET → edit → PUT without data loss.""" 423 424 @pytest.fixture(autouse=True) 425 def _setup(self): 426 try: 427 from starlette.testclient import TestClient 428 except ImportError: 429 pytest.skip("fastapi/starlette not installed") 430 from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN 431 self.client = TestClient(app) 432 self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN 433 434 def test_get_config_no_internal_keys(self): 435 """GET /api/config should not expose _config_version or _model_meta.""" 436 config = self.client.get("/api/config").json() 437 internal = [k for k in config if k.startswith("_")] 438 assert not internal, f"Internal keys leaked to frontend: {internal}" 439 440 def test_get_config_model_is_string(self): 441 """GET /api/config should normalize model dict to a string.""" 442 config = self.client.get("/api/config").json() 443 assert isinstance(config.get("model"), str), \ 444 f"model should be string, got {type(config.get('model'))}" 445 446 def test_round_trip_preserves_model_subkeys(self): 447 """Save and reload should not lose model.provider, model.base_url, etc.""" 448 from hermes_cli.config import load_config, save_config 449 450 # Set up a config with model as a dict (the common user config form) 451 save_config({ 452 "model": { 453 "default": "anthropic/claude-sonnet-4", 454 "provider": "openrouter", 455 "base_url": "https://openrouter.ai/api/v1", 456 "api_mode": "openai", 457 } 458 }) 459 460 before = load_config() 461 assert isinstance(before.get("model"), dict) 462 original_keys = set(before["model"].keys()) 463 464 # GET → PUT unchanged 465 web_config = self.client.get("/api/config").json() 466 assert isinstance(web_config.get("model"), str), "GET should normalize model to string" 467 468 self.client.put("/api/config", json={"config": web_config}) 469 470 after = load_config() 471 assert isinstance(after.get("model"), dict), "model should still be a dict after save" 472 assert set(after["model"].keys()) >= original_keys, \ 473 f"Lost model subkeys: {original_keys - set(after['model'].keys())}" 474 475 def test_edit_model_name_preserved(self): 476 """Changing the model string should update model.default on disk.""" 477 from hermes_cli.config import load_config 478 479 web_config = self.client.get("/api/config").json() 480 original_model = web_config["model"] 481 482 # Change model 483 web_config["model"] = "test/editing-model" 484 self.client.put("/api/config", json={"config": web_config}) 485 486 after = load_config() 487 if isinstance(after.get("model"), dict): 488 assert after["model"]["default"] == "test/editing-model" 489 else: 490 assert after["model"] == "test/editing-model" 491 492 # Restore 493 web_config["model"] = original_model 494 self.client.put("/api/config", json={"config": web_config}) 495 496 def test_edit_nested_value(self): 497 """Editing a nested config value should persist correctly.""" 498 from hermes_cli.config import load_config 499 500 web_config = self.client.get("/api/config").json() 501 original_turns = web_config.get("agent", {}).get("max_turns") 502 503 # Change max_turns 504 if "agent" not in web_config: 505 web_config["agent"] = {} 506 web_config["agent"]["max_turns"] = 42 507 508 self.client.put("/api/config", json={"config": web_config}) 509 510 after = load_config() 511 assert after.get("agent", {}).get("max_turns") == 42 512 513 # Restore 514 web_config["agent"]["max_turns"] = original_turns 515 self.client.put("/api/config", json={"config": web_config}) 516 517 def test_schema_types_match_config_values(self): 518 """Every schema field should have a matching-type value in the config.""" 519 config = self.client.get("/api/config").json() 520 schema_resp = self.client.get("/api/config/schema").json() 521 schema = schema_resp["fields"] 522 523 def get_nested(obj, path): 524 parts = path.split(".") 525 cur = obj 526 for p in parts: 527 if cur is None or not isinstance(cur, dict): 528 return None 529 cur = cur.get(p) 530 return cur 531 532 mismatches = [] 533 for key, entry in schema.items(): 534 val = get_nested(config, key) 535 if val is None: 536 continue # not set in user config — fine 537 expected = entry["type"] 538 if expected in ("string", "select") and not isinstance(val, str): 539 mismatches.append(f"{key}: expected str, got {type(val).__name__}") 540 elif expected == "number" and not isinstance(val, (int, float)): 541 mismatches.append(f"{key}: expected number, got {type(val).__name__}") 542 elif expected == "boolean" and not isinstance(val, bool): 543 mismatches.append(f"{key}: expected bool, got {type(val).__name__}") 544 elif expected == "list" and not isinstance(val, list): 545 mismatches.append(f"{key}: expected list, got {type(val).__name__}") 546 assert not mismatches, f"Type mismatches:\n" + "\n".join(mismatches) 547 548 549 # --------------------------------------------------------------------------- 550 # New feature endpoint tests 551 # --------------------------------------------------------------------------- 552 553 554 class TestNewEndpoints: 555 """Tests for session detail, logs, cron, skills, tools, raw config, analytics.""" 556 557 @pytest.fixture(autouse=True) 558 def _setup(self, monkeypatch, _isolate_hermes_home): 559 try: 560 from starlette.testclient import TestClient 561 except ImportError: 562 pytest.skip("fastapi/starlette not installed") 563 564 import hermes_state 565 from hermes_constants import get_hermes_home 566 from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN 567 568 monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db") 569 570 self.client = TestClient(app) 571 self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN 572 573 def test_get_logs_default(self): 574 resp = self.client.get("/api/logs") 575 assert resp.status_code == 200 576 data = resp.json() 577 assert "file" in data 578 assert "lines" in data 579 assert isinstance(data["lines"], list) 580 581 def test_get_logs_invalid_file(self): 582 resp = self.client.get("/api/logs?file=nonexistent") 583 assert resp.status_code == 400 584 585 def test_cron_list(self): 586 resp = self.client.get("/api/cron/jobs") 587 assert resp.status_code == 200 588 assert isinstance(resp.json(), list) 589 590 def test_cron_job_not_found(self): 591 resp = self.client.get("/api/cron/jobs/nonexistent-id") 592 assert resp.status_code == 404 593 594 # --- Profiles --- 595 596 def test_profiles_list_includes_default(self): 597 from hermes_constants import get_hermes_home 598 get_hermes_home().mkdir(parents=True, exist_ok=True) 599 600 resp = self.client.get("/api/profiles") 601 assert resp.status_code == 200 602 names = [p["name"] for p in resp.json()["profiles"]] 603 assert "default" in names 604 605 def test_profiles_list_falls_back_when_profile_listing_fails(self, monkeypatch): 606 from hermes_constants import get_hermes_home 607 import hermes_cli.profiles as profiles_mod 608 609 hermes_home = get_hermes_home() 610 hermes_home.mkdir(parents=True, exist_ok=True) 611 (hermes_home / "config.yaml").write_text( 612 "model:\n provider: openrouter\n name: anthropic/claude-sonnet-4.6\n", 613 encoding="utf-8", 614 ) 615 named = hermes_home / "profiles" / "multi-agent" 616 named.mkdir(parents=True) 617 (named / ".env").write_text("EXAMPLE=1\n", encoding="utf-8") 618 (named / "skills" / "demo").mkdir(parents=True) 619 (named / "skills" / "demo" / "SKILL.md").write_text("---\nname: demo\n---\n", encoding="utf-8") 620 621 monkeypatch.setattr( 622 profiles_mod, 623 "list_profiles", 624 lambda: (_ for _ in ()).throw(RuntimeError("boom")), 625 ) 626 627 resp = self.client.get("/api/profiles") 628 629 assert resp.status_code == 200 630 profiles = {p["name"]: p for p in resp.json()["profiles"]} 631 assert profiles["default"]["is_default"] is True 632 assert profiles["default"]["provider"] == "openrouter" 633 assert profiles["multi-agent"]["has_env"] is True 634 assert profiles["multi-agent"]["skill_count"] == 1 635 636 def test_profiles_create_rename_delete_round_trip(self, monkeypatch): 637 # Stub gateway service teardown so the test doesn't shell out to 638 # launchctl/systemctl on the host. 639 import hermes_cli.profiles as profiles_mod 640 monkeypatch.setattr(profiles_mod, "_cleanup_gateway_service", lambda *a, **kw: None) 641 642 created = self.client.post("/api/profiles", json={"name": "test-prof"}) 643 assert created.status_code == 200 644 645 renamed = self.client.patch( 646 "/api/profiles/test-prof", 647 json={"new_name": "test-prof-2"}, 648 ) 649 assert renamed.status_code == 200 650 651 names = [p["name"] for p in self.client.get("/api/profiles").json()["profiles"]] 652 assert "test-prof" not in names 653 assert "test-prof-2" in names 654 655 deleted = self.client.delete("/api/profiles/test-prof-2") 656 assert deleted.status_code == 200 657 names = [p["name"] for p in self.client.get("/api/profiles").json()["profiles"]] 658 assert "test-prof-2" not in names 659 660 def test_profile_setup_command_uses_named_profile_wrapper(self): 661 from hermes_constants import get_hermes_home 662 663 (get_hermes_home() / "profiles" / "coder").mkdir(parents=True) 664 665 resp = self.client.get("/api/profiles/coder/setup-command") 666 667 assert resp.status_code == 200 668 assert resp.json()["command"] == "coder setup" 669 670 def test_profile_setup_command_uses_hermes_for_default_profile(self): 671 from hermes_constants import get_hermes_home 672 673 get_hermes_home().mkdir(parents=True, exist_ok=True) 674 675 resp = self.client.get("/api/profiles/default/setup-command") 676 677 assert resp.status_code == 200 678 assert resp.json()["command"] == "hermes setup" 679 680 def test_profiles_create_creates_wrapper_alias_when_safe(self, monkeypatch, tmp_path): 681 import hermes_cli.profiles as profiles_mod 682 683 wrapper_dir = tmp_path / "bin" 684 wrapper_dir.mkdir() 685 monkeypatch.setattr(profiles_mod, "_get_wrapper_dir", lambda: wrapper_dir) 686 687 resp = self.client.post( 688 "/api/profiles", 689 json={"name": "writer", "clone_from_default": False}, 690 ) 691 692 assert resp.status_code == 200 693 wrapper_path = wrapper_dir / "writer" 694 assert wrapper_path.exists() 695 assert wrapper_path.read_text() == '#!/bin/sh\nexec hermes -p writer "$@"\n' 696 697 def test_profiles_create_with_clone_from_default_copies_default_skills(self, monkeypatch): 698 from hermes_constants import get_hermes_home 699 import hermes_cli.profiles as profiles_mod 700 701 monkeypatch.setattr(profiles_mod, "create_wrapper_script", lambda name: None) 702 default_skill = get_hermes_home() / "skills" / "custom" / "new-skill" 703 default_skill.mkdir(parents=True) 704 (default_skill / "SKILL.md").write_text("---\nname: new-skill\n---\n", encoding="utf-8") 705 706 resp = self.client.post( 707 "/api/profiles", 708 json={"name": "cloned", "clone_from_default": True}, 709 ) 710 711 assert resp.status_code == 200 712 cloned_skill = get_hermes_home() / "profiles" / "cloned" / "skills" / "custom" / "new-skill" / "SKILL.md" 713 assert cloned_skill.exists() 714 profiles = {p["name"]: p for p in self.client.get("/api/profiles").json()["profiles"]} 715 assert profiles["cloned"]["skill_count"] == 1 716 717 def test_profiles_create_without_clone_seeds_bundled_skills(self, monkeypatch): 718 from hermes_constants import get_hermes_home 719 import hermes_cli.profiles as profiles_mod 720 721 monkeypatch.setattr(profiles_mod, "create_wrapper_script", lambda name: None) 722 723 def fake_seed(profile_dir, quiet=False): 724 skill_dir = profile_dir / "skills" / "software-development" / "plan" 725 skill_dir.mkdir(parents=True) 726 (skill_dir / "SKILL.md").write_text("---\nname: plan\n---\n", encoding="utf-8") 727 return {"copied": ["plan"]} 728 729 monkeypatch.setattr(profiles_mod, "seed_profile_skills", fake_seed) 730 731 resp = self.client.post( 732 "/api/profiles", 733 json={"name": "fresh", "clone_from_default": False}, 734 ) 735 736 assert resp.status_code == 200 737 seeded_skill = get_hermes_home() / "profiles" / "fresh" / "skills" / "software-development" / "plan" / "SKILL.md" 738 assert seeded_skill.exists() 739 profiles = {p["name"]: p for p in self.client.get("/api/profiles").json()["profiles"]} 740 assert profiles["fresh"]["skill_count"] == 1 741 742 def test_profile_open_terminal_uses_macos_terminal(self, monkeypatch): 743 from hermes_constants import get_hermes_home 744 import hermes_cli.web_server as web_server 745 746 (get_hermes_home() / "profiles" / "coder").mkdir(parents=True) 747 calls = [] 748 monkeypatch.setattr(web_server.sys, "platform", "darwin") 749 monkeypatch.setattr(web_server.subprocess, "Popen", lambda args, **kwargs: calls.append(args)) 750 751 resp = self.client.post("/api/profiles/coder/open-terminal") 752 753 assert resp.status_code == 200 754 assert calls 755 assert calls[0][0] == "osascript" 756 assert "coder setup" in " ".join(calls[0]) 757 758 def test_profile_open_terminal_uses_windows_cmd(self, monkeypatch): 759 from hermes_constants import get_hermes_home 760 import hermes_cli.web_server as web_server 761 762 (get_hermes_home() / "profiles" / "coder").mkdir(parents=True) 763 calls = [] 764 monkeypatch.setattr(web_server.sys, "platform", "win32") 765 monkeypatch.setattr(web_server.subprocess, "Popen", lambda args, **kwargs: calls.append(args)) 766 767 resp = self.client.post("/api/profiles/coder/open-terminal") 768 769 assert resp.status_code == 200 770 assert calls 771 assert calls[0][:4] == ["cmd.exe", "/c", "start", ""] 772 assert calls[0][-1] == "coder setup" 773 774 def test_profiles_create_rejects_invalid_name(self): 775 resp = self.client.post("/api/profiles", json={"name": "Has Spaces"}) 776 assert resp.status_code == 400 777 778 def test_profiles_delete_default_forbidden(self): 779 resp = self.client.delete("/api/profiles/default") 780 assert resp.status_code == 400 781 782 def test_profiles_delete_not_found(self): 783 resp = self.client.delete("/api/profiles/does-not-exist") 784 assert resp.status_code == 404 785 786 def test_profile_soul_round_trip(self, monkeypatch): 787 import hermes_cli.profiles as profiles_mod 788 monkeypatch.setattr(profiles_mod, "_cleanup_gateway_service", lambda *a, **kw: None) 789 790 self.client.post("/api/profiles", json={"name": "soul-prof"}) 791 get1 = self.client.get("/api/profiles/soul-prof/soul") 792 assert get1.status_code == 200 793 assert get1.json()["exists"] is True 794 795 put = self.client.put( 796 "/api/profiles/soul-prof/soul", 797 json={"content": "# Edited soul"}, 798 ) 799 assert put.status_code == 200 800 801 got = self.client.get("/api/profiles/soul-prof/soul").json() 802 assert got["content"] == "# Edited soul" 803 804 self.client.delete("/api/profiles/soul-prof") 805 806 def test_profile_soul_unknown_profile_404(self): 807 resp = self.client.get("/api/profiles/nonexistent/soul") 808 assert resp.status_code == 404 809 810 def test_skills_list(self): 811 resp = self.client.get("/api/skills") 812 assert resp.status_code == 200 813 skills = resp.json() 814 assert isinstance(skills, list) 815 if skills: 816 assert "name" in skills[0] 817 assert "enabled" in skills[0] 818 819 def test_skills_list_includes_disabled_skills(self, monkeypatch): 820 import tools.skills_tool as skills_tool 821 import hermes_cli.skills_config as skills_config 822 import hermes_cli.web_server as web_server 823 824 def _fake_find_all_skills(*, skip_disabled=False): 825 if skip_disabled: 826 return [ 827 {"name": "active-skill", "description": "active", "category": "demo"}, 828 {"name": "disabled-skill", "description": "disabled", "category": "demo"}, 829 ] 830 return [ 831 {"name": "active-skill", "description": "active", "category": "demo"}, 832 ] 833 834 monkeypatch.setattr(skills_tool, "_find_all_skills", _fake_find_all_skills) 835 monkeypatch.setattr(skills_config, "get_disabled_skills", lambda config: {"disabled-skill"}) 836 monkeypatch.setattr(web_server, "load_config", lambda: {"skills": {"disabled": ["disabled-skill"]}}) 837 838 resp = self.client.get("/api/skills") 839 840 assert resp.status_code == 200 841 assert resp.json() == [ 842 { 843 "name": "active-skill", 844 "description": "active", 845 "category": "demo", 846 "enabled": True, 847 }, 848 { 849 "name": "disabled-skill", 850 "description": "disabled", 851 "category": "demo", 852 "enabled": False, 853 }, 854 ] 855 856 def test_toolsets_list(self): 857 resp = self.client.get("/api/tools/toolsets") 858 assert resp.status_code == 200 859 toolsets = resp.json() 860 assert isinstance(toolsets, list) 861 if toolsets: 862 assert "name" in toolsets[0] 863 assert "label" in toolsets[0] 864 assert "enabled" in toolsets[0] 865 866 def test_toolsets_list_matches_cli_enabled_state(self, monkeypatch): 867 import hermes_cli.tools_config as tools_config 868 import toolsets as toolsets_module 869 import hermes_cli.web_server as web_server 870 871 monkeypatch.setattr( 872 tools_config, 873 "_get_effective_configurable_toolsets", 874 lambda: [ 875 ("web", "🔍 Web Search & Scraping", "web_search, web_extract"), 876 ("skills", "📚 Skills", "list, view, manage"), 877 ("memory", "💾 Memory", "persistent memory across sessions"), 878 ], 879 ) 880 monkeypatch.setattr( 881 tools_config, 882 "_get_platform_tools", 883 lambda config, platform, include_default_mcp_servers=False: {"web", "skills"}, 884 ) 885 monkeypatch.setattr( 886 tools_config, 887 "_toolset_has_keys", 888 lambda ts_key, config=None: ts_key != "web", 889 ) 890 monkeypatch.setattr( 891 toolsets_module, 892 "resolve_toolset", 893 lambda name: { 894 "web": ["web_search", "web_extract"], 895 "skills": ["skills_list", "skill_view"], 896 "memory": ["memory_read"], 897 }[name], 898 ) 899 monkeypatch.setattr(web_server, "load_config", lambda: {"platform_toolsets": {"cli": ["web", "skills"]}}) 900 901 resp = self.client.get("/api/tools/toolsets") 902 903 assert resp.status_code == 200 904 assert resp.json() == [ 905 { 906 "name": "web", 907 "label": "🔍 Web Search & Scraping", 908 "description": "web_search, web_extract", 909 "enabled": True, 910 "available": True, 911 "configured": False, 912 "tools": ["web_extract", "web_search"], 913 }, 914 { 915 "name": "skills", 916 "label": "📚 Skills", 917 "description": "list, view, manage", 918 "enabled": True, 919 "available": True, 920 "configured": True, 921 "tools": ["skill_view", "skills_list"], 922 }, 923 { 924 "name": "memory", 925 "label": "💾 Memory", 926 "description": "persistent memory across sessions", 927 "enabled": False, 928 "available": False, 929 "configured": True, 930 "tools": ["memory_read"], 931 }, 932 ] 933 934 def test_config_raw_get(self): 935 resp = self.client.get("/api/config/raw") 936 assert resp.status_code == 200 937 assert "yaml" in resp.json() 938 939 def test_config_raw_put_valid(self): 940 resp = self.client.put( 941 "/api/config/raw", 942 json={"yaml_text": "model: test\ntoolsets:\n - all\n"}, 943 ) 944 assert resp.status_code == 200 945 assert resp.json()["ok"] is True 946 947 def test_config_raw_put_invalid(self): 948 resp = self.client.put( 949 "/api/config/raw", 950 json={"yaml_text": "- this is a list not a dict"}, 951 ) 952 assert resp.status_code == 400 953 954 def test_analytics_usage(self): 955 resp = self.client.get("/api/analytics/usage?days=7") 956 assert resp.status_code == 200 957 data = resp.json() 958 assert "daily" in data 959 assert "by_model" in data 960 assert "totals" in data 961 assert "skills" in data 962 assert isinstance(data["daily"], list) 963 assert "total_sessions" in data["totals"] 964 assert "total_api_calls" in data["totals"] 965 assert data["skills"] == { 966 "summary": { 967 "total_skill_loads": 0, 968 "total_skill_edits": 0, 969 "total_skill_actions": 0, 970 "distinct_skills_used": 0, 971 }, 972 "top_skills": [], 973 } 974 975 def test_analytics_usage_includes_skill_breakdown(self): 976 from hermes_state import SessionDB 977 978 db = SessionDB() 979 try: 980 db.create_session( 981 session_id="skills-analytics-test", 982 source="cli", 983 model="anthropic/claude-sonnet-4", 984 ) 985 db.update_token_counts( 986 "skills-analytics-test", 987 input_tokens=120, 988 output_tokens=45, 989 ) 990 db.append_message( 991 "skills-analytics-test", 992 role="assistant", 993 content="Loading and updating skills.", 994 tool_calls=[ 995 { 996 "function": { 997 "name": "skill_view", 998 "arguments": '{"name":"github-pr-workflow"}', 999 } 1000 }, 1001 { 1002 "function": { 1003 "name": "skill_manage", 1004 "arguments": '{"name":"github-code-review"}', 1005 } 1006 }, 1007 ], 1008 ) 1009 finally: 1010 db.close() 1011 1012 resp = self.client.get("/api/analytics/usage?days=7") 1013 assert resp.status_code == 200 1014 1015 data = resp.json() 1016 assert data["skills"]["summary"] == { 1017 "total_skill_loads": 1, 1018 "total_skill_edits": 1, 1019 "total_skill_actions": 2, 1020 "distinct_skills_used": 2, 1021 } 1022 assert len(data["skills"]["top_skills"]) == 2 1023 1024 top_skill = data["skills"]["top_skills"][0] 1025 assert top_skill["skill"] == "github-pr-workflow" 1026 assert top_skill["view_count"] == 1 1027 assert top_skill["manage_count"] == 0 1028 assert top_skill["total_count"] == 1 1029 assert top_skill["last_used_at"] is not None 1030 1031 def test_session_token_endpoint_removed(self): 1032 """GET /api/auth/session-token no longer exists.""" 1033 resp = self.client.get("/api/auth/session-token") 1034 # Should not return a JSON token object 1035 assert resp.status_code in (200, 404) 1036 try: 1037 data = resp.json() 1038 assert "token" not in data 1039 except Exception: 1040 pass 1041 1042 1043 # --------------------------------------------------------------------------- 1044 # Model context length: normalize/denormalize + /api/model/info 1045 # --------------------------------------------------------------------------- 1046 1047 1048 class TestModelContextLength: 1049 """Tests for model_context_length in normalize/denormalize and /api/model/info.""" 1050 1051 def test_normalize_extracts_context_length_from_dict(self): 1052 """normalize should surface context_length from model dict.""" 1053 from hermes_cli.web_server import _normalize_config_for_web 1054 1055 cfg = { 1056 "model": { 1057 "default": "anthropic/claude-opus-4.6", 1058 "provider": "openrouter", 1059 "context_length": 200000, 1060 } 1061 } 1062 result = _normalize_config_for_web(cfg) 1063 assert result["model"] == "anthropic/claude-opus-4.6" 1064 assert result["model_context_length"] == 200000 1065 1066 def test_normalize_bare_string_model_yields_zero(self): 1067 """normalize should set model_context_length=0 for bare string model.""" 1068 from hermes_cli.web_server import _normalize_config_for_web 1069 1070 result = _normalize_config_for_web({"model": "anthropic/claude-sonnet-4"}) 1071 assert result["model"] == "anthropic/claude-sonnet-4" 1072 assert result["model_context_length"] == 0 1073 1074 def test_normalize_dict_without_context_length_yields_zero(self): 1075 """normalize should default to 0 when model dict has no context_length.""" 1076 from hermes_cli.web_server import _normalize_config_for_web 1077 1078 cfg = {"model": {"default": "test/model", "provider": "openrouter"}} 1079 result = _normalize_config_for_web(cfg) 1080 assert result["model_context_length"] == 0 1081 1082 def test_normalize_non_int_context_length_yields_zero(self): 1083 """normalize should coerce non-int context_length to 0.""" 1084 from hermes_cli.web_server import _normalize_config_for_web 1085 1086 cfg = {"model": {"default": "test/model", "context_length": "invalid"}} 1087 result = _normalize_config_for_web(cfg) 1088 assert result["model_context_length"] == 0 1089 1090 def test_denormalize_writes_context_length_into_model_dict(self): 1091 """denormalize should write model_context_length back into model dict.""" 1092 from hermes_cli.web_server import _denormalize_config_from_web 1093 from hermes_cli.config import save_config 1094 1095 # Set up disk config with model as a dict 1096 save_config({ 1097 "model": {"default": "anthropic/claude-opus-4.6", "provider": "openrouter"} 1098 }) 1099 1100 result = _denormalize_config_from_web({ 1101 "model": "anthropic/claude-opus-4.6", 1102 "model_context_length": 100000, 1103 }) 1104 assert isinstance(result["model"], dict) 1105 assert result["model"]["context_length"] == 100000 1106 assert "model_context_length" not in result # virtual field removed 1107 1108 def test_denormalize_zero_removes_context_length(self): 1109 """denormalize with model_context_length=0 should remove context_length key.""" 1110 from hermes_cli.web_server import _denormalize_config_from_web 1111 from hermes_cli.config import save_config 1112 1113 save_config({ 1114 "model": { 1115 "default": "anthropic/claude-opus-4.6", 1116 "provider": "openrouter", 1117 "context_length": 50000, 1118 } 1119 }) 1120 1121 result = _denormalize_config_from_web({ 1122 "model": "anthropic/claude-opus-4.6", 1123 "model_context_length": 0, 1124 }) 1125 assert isinstance(result["model"], dict) 1126 assert "context_length" not in result["model"] 1127 1128 def test_denormalize_upgrades_bare_string_to_dict(self): 1129 """denormalize should upgrade bare string model to dict when context_length set.""" 1130 from hermes_cli.web_server import _denormalize_config_from_web 1131 from hermes_cli.config import save_config 1132 1133 # Disk has model as bare string 1134 save_config({"model": "anthropic/claude-sonnet-4"}) 1135 1136 result = _denormalize_config_from_web({ 1137 "model": "anthropic/claude-sonnet-4", 1138 "model_context_length": 65000, 1139 }) 1140 assert isinstance(result["model"], dict) 1141 assert result["model"]["default"] == "anthropic/claude-sonnet-4" 1142 assert result["model"]["context_length"] == 65000 1143 1144 def test_denormalize_bare_string_stays_string_when_zero(self): 1145 """denormalize should keep bare string model as string when context_length=0.""" 1146 from hermes_cli.web_server import _denormalize_config_from_web 1147 from hermes_cli.config import save_config 1148 1149 save_config({"model": "anthropic/claude-sonnet-4"}) 1150 1151 result = _denormalize_config_from_web({ 1152 "model": "anthropic/claude-sonnet-4", 1153 "model_context_length": 0, 1154 }) 1155 assert result["model"] == "anthropic/claude-sonnet-4" 1156 1157 def test_denormalize_coerces_string_context_length(self): 1158 """denormalize should handle string model_context_length from frontend.""" 1159 from hermes_cli.web_server import _denormalize_config_from_web 1160 from hermes_cli.config import save_config 1161 1162 save_config({ 1163 "model": {"default": "test/model", "provider": "openrouter"} 1164 }) 1165 1166 result = _denormalize_config_from_web({ 1167 "model": "test/model", 1168 "model_context_length": "32000", 1169 }) 1170 assert isinstance(result["model"], dict) 1171 assert result["model"]["context_length"] == 32000 1172 1173 1174 class TestModelContextLengthSchema: 1175 """Tests for model_context_length placement in CONFIG_SCHEMA.""" 1176 1177 def test_schema_has_model_context_length(self): 1178 from hermes_cli.web_server import CONFIG_SCHEMA 1179 assert "model_context_length" in CONFIG_SCHEMA 1180 1181 def test_schema_model_context_length_after_model(self): 1182 """model_context_length should appear immediately after model in schema.""" 1183 from hermes_cli.web_server import CONFIG_SCHEMA 1184 keys = list(CONFIG_SCHEMA.keys()) 1185 model_idx = keys.index("model") 1186 assert keys[model_idx + 1] == "model_context_length" 1187 1188 def test_schema_model_context_length_is_number(self): 1189 from hermes_cli.web_server import CONFIG_SCHEMA 1190 entry = CONFIG_SCHEMA["model_context_length"] 1191 assert entry["type"] == "number" 1192 assert "category" in entry 1193 1194 1195 class TestModelInfoEndpoint: 1196 """Tests for GET /api/model/info endpoint.""" 1197 1198 @pytest.fixture(autouse=True) 1199 def _setup(self): 1200 try: 1201 from starlette.testclient import TestClient 1202 except ImportError: 1203 pytest.skip("fastapi/starlette not installed") 1204 from hermes_cli.web_server import app 1205 self.client = TestClient(app) 1206 1207 def test_model_info_returns_200(self): 1208 resp = self.client.get("/api/model/info") 1209 assert resp.status_code == 200 1210 data = resp.json() 1211 assert "model" in data 1212 assert "provider" in data 1213 assert "auto_context_length" in data 1214 assert "config_context_length" in data 1215 assert "effective_context_length" in data 1216 assert "capabilities" in data 1217 1218 def test_model_info_with_dict_config(self, monkeypatch): 1219 import hermes_cli.web_server as ws 1220 1221 monkeypatch.setattr(ws, "load_config", lambda: { 1222 "model": { 1223 "default": "anthropic/claude-opus-4.6", 1224 "provider": "openrouter", 1225 "context_length": 100000, 1226 } 1227 }) 1228 1229 with patch("agent.model_metadata.get_model_context_length", return_value=200000): 1230 resp = self.client.get("/api/model/info") 1231 1232 data = resp.json() 1233 assert data["model"] == "anthropic/claude-opus-4.6" 1234 assert data["provider"] == "openrouter" 1235 assert data["auto_context_length"] == 200000 1236 assert data["config_context_length"] == 100000 1237 assert data["effective_context_length"] == 100000 # override wins 1238 1239 def test_model_info_auto_detect_when_no_override(self, monkeypatch): 1240 import hermes_cli.web_server as ws 1241 1242 monkeypatch.setattr(ws, "load_config", lambda: { 1243 "model": {"default": "anthropic/claude-opus-4.6", "provider": "openrouter"} 1244 }) 1245 1246 with patch("agent.model_metadata.get_model_context_length", return_value=200000): 1247 resp = self.client.get("/api/model/info") 1248 1249 data = resp.json() 1250 assert data["auto_context_length"] == 200000 1251 assert data["config_context_length"] == 0 1252 assert data["effective_context_length"] == 200000 # auto wins 1253 1254 def test_model_info_empty_model(self, monkeypatch): 1255 import hermes_cli.web_server as ws 1256 1257 monkeypatch.setattr(ws, "load_config", lambda: {"model": ""}) 1258 1259 resp = self.client.get("/api/model/info") 1260 data = resp.json() 1261 assert data["model"] == "" 1262 assert data["effective_context_length"] == 0 1263 1264 def test_model_info_bare_string_model(self, monkeypatch): 1265 import hermes_cli.web_server as ws 1266 1267 monkeypatch.setattr(ws, "load_config", lambda: { 1268 "model": "anthropic/claude-sonnet-4" 1269 }) 1270 1271 with patch("agent.model_metadata.get_model_context_length", return_value=200000): 1272 resp = self.client.get("/api/model/info") 1273 1274 data = resp.json() 1275 assert data["model"] == "anthropic/claude-sonnet-4" 1276 assert data["provider"] == "" 1277 assert data["config_context_length"] == 0 1278 assert data["effective_context_length"] == 200000 1279 1280 def test_model_info_capabilities(self, monkeypatch): 1281 import hermes_cli.web_server as ws 1282 1283 monkeypatch.setattr(ws, "load_config", lambda: { 1284 "model": {"default": "anthropic/claude-opus-4.6", "provider": "openrouter"} 1285 }) 1286 1287 mock_caps = MagicMock() 1288 mock_caps.supports_tools = True 1289 mock_caps.supports_vision = True 1290 mock_caps.supports_reasoning = True 1291 mock_caps.context_window = 200000 1292 mock_caps.max_output_tokens = 32000 1293 mock_caps.model_family = "claude-opus" 1294 1295 with patch("agent.model_metadata.get_model_context_length", return_value=200000), \ 1296 patch("agent.models_dev.get_model_capabilities", return_value=mock_caps): 1297 resp = self.client.get("/api/model/info") 1298 1299 caps = resp.json()["capabilities"] 1300 assert caps["supports_tools"] is True 1301 assert caps["supports_vision"] is True 1302 assert caps["supports_reasoning"] is True 1303 assert caps["max_output_tokens"] == 32000 1304 assert caps["model_family"] == "claude-opus" 1305 1306 def test_model_info_graceful_on_metadata_error(self, monkeypatch): 1307 """Endpoint should return zeros on import/resolution errors, not 500.""" 1308 import hermes_cli.web_server as ws 1309 1310 monkeypatch.setattr(ws, "load_config", lambda: { 1311 "model": "some/obscure-model" 1312 }) 1313 1314 with patch("agent.model_metadata.get_model_context_length", side_effect=Exception("boom")): 1315 resp = self.client.get("/api/model/info") 1316 1317 assert resp.status_code == 200 1318 data = resp.json() 1319 assert data["auto_context_length"] == 0 1320 1321 1322 # --------------------------------------------------------------------------- 1323 # Gateway health probe tests 1324 # --------------------------------------------------------------------------- 1325 1326 1327 class TestProbeGatewayHealth: 1328 """Tests for _probe_gateway_health() — cross-container gateway detection.""" 1329 1330 def test_returns_false_when_no_url_configured(self, monkeypatch): 1331 """When GATEWAY_HEALTH_URL is unset, the probe returns (False, None).""" 1332 import hermes_cli.web_server as ws 1333 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None) 1334 alive, body = ws._probe_gateway_health() 1335 assert alive is False 1336 assert body is None 1337 1338 def test_normalizes_url_with_health_suffix(self, monkeypatch): 1339 """If the user sets the URL to include /health, it's stripped to base.""" 1340 import hermes_cli.web_server as ws 1341 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642/health") 1342 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1) 1343 # Both paths should fail (no server), but we verify they were constructed 1344 # correctly by checking the URLs attempted. 1345 calls = [] 1346 original_urlopen = ws.urllib.request.urlopen 1347 1348 def mock_urlopen(req, **kwargs): 1349 calls.append(req.full_url) 1350 raise ConnectionError("mock") 1351 1352 monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen) 1353 alive, body = ws._probe_gateway_health() 1354 assert alive is False 1355 assert "http://gw:8642/health/detailed" in calls 1356 assert "http://gw:8642/health" in calls 1357 1358 def test_normalizes_url_with_health_detailed_suffix(self, monkeypatch): 1359 """If the user sets the URL to include /health/detailed, it's stripped to base.""" 1360 import hermes_cli.web_server as ws 1361 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642/health/detailed") 1362 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1) 1363 calls = [] 1364 1365 def mock_urlopen(req, **kwargs): 1366 calls.append(req.full_url) 1367 raise ConnectionError("mock") 1368 1369 monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen) 1370 ws._probe_gateway_health() 1371 assert "http://gw:8642/health/detailed" in calls 1372 assert "http://gw:8642/health" in calls 1373 1374 def test_successful_detailed_probe(self, monkeypatch): 1375 """Successful /health/detailed probe returns (True, body_dict).""" 1376 import hermes_cli.web_server as ws 1377 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642") 1378 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1) 1379 1380 response_body = json.dumps({ 1381 "status": "ok", 1382 "gateway_state": "running", 1383 "pid": 42, 1384 }) 1385 1386 mock_resp = MagicMock() 1387 mock_resp.status = 200 1388 mock_resp.read.return_value = response_body.encode() 1389 mock_resp.__enter__ = MagicMock(return_value=mock_resp) 1390 mock_resp.__exit__ = MagicMock(return_value=False) 1391 1392 monkeypatch.setattr(ws.urllib.request, "urlopen", lambda req, **kw: mock_resp) 1393 alive, body = ws._probe_gateway_health() 1394 assert alive is True 1395 assert body["status"] == "ok" 1396 assert body["pid"] == 42 1397 1398 def test_detailed_fails_falls_back_to_simple_health(self, monkeypatch): 1399 """If /health/detailed fails, falls back to /health.""" 1400 import hermes_cli.web_server as ws 1401 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642") 1402 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_TIMEOUT", 1) 1403 1404 call_count = [0] 1405 1406 def mock_urlopen(req, **kwargs): 1407 call_count[0] += 1 1408 if call_count[0] == 1: 1409 raise ConnectionError("detailed failed") 1410 mock_resp = MagicMock() 1411 mock_resp.status = 200 1412 mock_resp.read.return_value = json.dumps({"status": "ok"}).encode() 1413 mock_resp.__enter__ = MagicMock(return_value=mock_resp) 1414 mock_resp.__exit__ = MagicMock(return_value=False) 1415 return mock_resp 1416 1417 monkeypatch.setattr(ws.urllib.request, "urlopen", mock_urlopen) 1418 alive, body = ws._probe_gateway_health() 1419 assert alive is True 1420 assert body["status"] == "ok" 1421 assert call_count[0] == 2 1422 1423 1424 class TestStatusRemoteGateway: 1425 """Tests for /api/status with remote gateway health fallback.""" 1426 1427 @pytest.fixture(autouse=True) 1428 def _setup_test_client(self): 1429 try: 1430 from starlette.testclient import TestClient 1431 except ImportError: 1432 pytest.skip("fastapi/starlette not installed") 1433 1434 from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN 1435 self.client = TestClient(app) 1436 self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN 1437 1438 def test_status_falls_back_to_remote_probe(self, monkeypatch): 1439 """When local PID check fails and remote probe succeeds, gateway shows running.""" 1440 import hermes_cli.web_server as ws 1441 1442 monkeypatch.setattr(ws, "get_running_pid", lambda: None) 1443 monkeypatch.setattr(ws, "read_runtime_status", lambda: None) 1444 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642") 1445 monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, { 1446 "status": "ok", 1447 "gateway_state": "running", 1448 "platforms": {"telegram": {"state": "connected"}}, 1449 "pid": 999, 1450 })) 1451 1452 resp = self.client.get("/api/status") 1453 assert resp.status_code == 200 1454 data = resp.json() 1455 assert data["gateway_running"] is True 1456 assert data["gateway_pid"] == 999 1457 assert data["gateway_state"] == "running" 1458 assert data["gateway_health_url"] == "http://gw:8642" 1459 1460 def test_status_remote_probe_not_attempted_when_local_pid_found(self, monkeypatch): 1461 """When local PID check succeeds, the remote probe is never called.""" 1462 import hermes_cli.web_server as ws 1463 1464 monkeypatch.setattr(ws, "get_running_pid", lambda: 1234) 1465 monkeypatch.setattr(ws, "read_runtime_status", lambda: { 1466 "gateway_state": "running", 1467 "platforms": {}, 1468 }) 1469 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642") 1470 probe_called = [False] 1471 original = ws._probe_gateway_health 1472 1473 def track_probe(): 1474 probe_called[0] = True 1475 return original() 1476 1477 monkeypatch.setattr(ws, "_probe_gateway_health", track_probe) 1478 1479 resp = self.client.get("/api/status") 1480 assert resp.status_code == 200 1481 assert not probe_called[0] 1482 1483 def test_status_remote_probe_not_attempted_when_no_url(self, monkeypatch): 1484 """When GATEWAY_HEALTH_URL is unset, no probe is attempted.""" 1485 import hermes_cli.web_server as ws 1486 1487 monkeypatch.setattr(ws, "get_running_pid", lambda: None) 1488 monkeypatch.setattr(ws, "read_runtime_status", lambda: None) 1489 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", None) 1490 1491 resp = self.client.get("/api/status") 1492 assert resp.status_code == 200 1493 data = resp.json() 1494 assert data["gateway_running"] is False 1495 assert data["gateway_health_url"] is None 1496 1497 def test_status_remote_running_null_pid(self, monkeypatch): 1498 """Remote gateway running but PID not in response — pid should be None.""" 1499 import hermes_cli.web_server as ws 1500 1501 monkeypatch.setattr(ws, "get_running_pid", lambda: None) 1502 monkeypatch.setattr(ws, "read_runtime_status", lambda: None) 1503 monkeypatch.setattr(ws, "_GATEWAY_HEALTH_URL", "http://gw:8642") 1504 monkeypatch.setattr(ws, "_probe_gateway_health", lambda: (True, { 1505 "status": "ok", 1506 })) 1507 1508 resp = self.client.get("/api/status") 1509 assert resp.status_code == 200 1510 data = resp.json() 1511 assert data["gateway_running"] is True 1512 assert data["gateway_pid"] is None 1513 assert data["gateway_state"] == "running" 1514 1515 1516 # --------------------------------------------------------------------------- 1517 # Dashboard theme normaliser tests 1518 # --------------------------------------------------------------------------- 1519 1520 1521 class TestNormaliseThemeDefinition: 1522 """Tests for _normalise_theme_definition() — parses YAML theme files.""" 1523 1524 def test_rejects_missing_name(self): 1525 from hermes_cli.web_server import _normalise_theme_definition 1526 assert _normalise_theme_definition({}) is None 1527 assert _normalise_theme_definition({"name": ""}) is None 1528 assert _normalise_theme_definition({"name": " "}) is None 1529 1530 def test_rejects_non_dict(self): 1531 from hermes_cli.web_server import _normalise_theme_definition 1532 assert _normalise_theme_definition("string") is None 1533 assert _normalise_theme_definition(None) is None 1534 assert _normalise_theme_definition([1, 2, 3]) is None 1535 1536 def test_loose_colors_shorthand(self): 1537 """Bare hex strings under `colors` parse as {hex, alpha=1.0}.""" 1538 from hermes_cli.web_server import _normalise_theme_definition 1539 result = _normalise_theme_definition({ 1540 "name": "loose", 1541 "colors": {"background": "#000000", "midground": "#ffffff"}, 1542 }) 1543 assert result is not None 1544 assert result["palette"]["background"] == {"hex": "#000000", "alpha": 1.0} 1545 assert result["palette"]["midground"] == {"hex": "#ffffff", "alpha": 1.0} 1546 # foreground falls back to default (transparent white) 1547 assert result["palette"]["foreground"]["hex"] == "#ffffff" 1548 assert result["palette"]["foreground"]["alpha"] == 0.0 1549 1550 def test_full_palette_form(self): 1551 from hermes_cli.web_server import _normalise_theme_definition 1552 result = _normalise_theme_definition({ 1553 "name": "full", 1554 "palette": { 1555 "background": {"hex": "#0a1628", "alpha": 1.0}, 1556 "midground": {"hex": "#a8d0ff", "alpha": 0.9}, 1557 "warmGlow": "rgba(255, 0, 0, 0.5)", 1558 "noiseOpacity": 0.5, 1559 }, 1560 }) 1561 assert result["palette"]["background"]["hex"] == "#0a1628" 1562 assert result["palette"]["midground"]["alpha"] == 0.9 1563 assert result["palette"]["warmGlow"] == "rgba(255, 0, 0, 0.5)" 1564 assert result["palette"]["noiseOpacity"] == 0.5 1565 1566 def test_default_typography_applied_when_missing(self): 1567 from hermes_cli.web_server import _normalise_theme_definition 1568 result = _normalise_theme_definition({"name": "minimal"}) 1569 typo = result["typography"] 1570 assert "fontSans" in typo 1571 assert "fontMono" in typo 1572 assert typo["baseSize"] == "15px" 1573 assert typo["lineHeight"] == "1.55" 1574 assert typo["letterSpacing"] == "0" 1575 1576 def test_partial_typography_merges_with_defaults(self): 1577 from hermes_cli.web_server import _normalise_theme_definition 1578 result = _normalise_theme_definition({ 1579 "name": "partial", 1580 "typography": { 1581 "fontSans": "MyFont, sans-serif", 1582 "baseSize": "12px", 1583 }, 1584 }) 1585 assert result["typography"]["fontSans"] == "MyFont, sans-serif" 1586 assert result["typography"]["baseSize"] == "12px" 1587 # fontMono defaulted 1588 assert "monospace" in result["typography"]["fontMono"] 1589 1590 def test_layout_defaults(self): 1591 from hermes_cli.web_server import _normalise_theme_definition 1592 result = _normalise_theme_definition({"name": "minimal"}) 1593 assert result["layout"]["radius"] == "0.5rem" 1594 assert result["layout"]["density"] == "comfortable" 1595 1596 def test_invalid_density_falls_back(self): 1597 from hermes_cli.web_server import _normalise_theme_definition 1598 result = _normalise_theme_definition({ 1599 "name": "bad", 1600 "layout": {"density": "ultra-spacious"}, 1601 }) 1602 assert result["layout"]["density"] == "comfortable" 1603 1604 def test_valid_densities_accepted(self): 1605 from hermes_cli.web_server import _normalise_theme_definition 1606 for d in ("compact", "comfortable", "spacious"): 1607 r = _normalise_theme_definition({"name": "x", "layout": {"density": d}}) 1608 assert r["layout"]["density"] == d 1609 1610 def test_color_overrides_filter_unknown_keys(self): 1611 from hermes_cli.web_server import _normalise_theme_definition 1612 result = _normalise_theme_definition({ 1613 "name": "o", 1614 "colorOverrides": { 1615 "card": "#123456", 1616 "fakeToken": "#abcdef", 1617 "primary": 42, # non-string rejected 1618 "destructive": "#ff0000", 1619 }, 1620 }) 1621 assert result["colorOverrides"] == { 1622 "card": "#123456", 1623 "destructive": "#ff0000", 1624 } 1625 1626 def test_color_overrides_omitted_when_empty(self): 1627 from hermes_cli.web_server import _normalise_theme_definition 1628 result = _normalise_theme_definition({"name": "x"}) 1629 assert "colorOverrides" not in result 1630 1631 def test_alpha_clamped_to_unit_range(self): 1632 from hermes_cli.web_server import _normalise_theme_definition 1633 r = _normalise_theme_definition({ 1634 "name": "c", 1635 "palette": {"background": {"hex": "#000", "alpha": 99.5}}, 1636 }) 1637 assert r["palette"]["background"]["alpha"] == 1.0 1638 r2 = _normalise_theme_definition({ 1639 "name": "c", 1640 "palette": {"background": {"hex": "#000", "alpha": -5}}, 1641 }) 1642 assert r2["palette"]["background"]["alpha"] == 0.0 1643 1644 def test_invalid_alpha_uses_default(self): 1645 from hermes_cli.web_server import _normalise_theme_definition 1646 r = _normalise_theme_definition({ 1647 "name": "c", 1648 "palette": {"background": {"hex": "#000", "alpha": "not a number"}}, 1649 }) 1650 assert r["palette"]["background"]["alpha"] == 1.0 1651 1652 1653 class TestDiscoverUserThemes: 1654 """Tests for _discover_user_themes() — scans ~/.hermes/dashboard-themes/.""" 1655 1656 def test_returns_empty_when_dir_missing(self, tmp_path, monkeypatch): 1657 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1658 from hermes_cli import web_server 1659 assert web_server._discover_user_themes() == [] 1660 1661 def test_loads_and_normalises_yaml(self, tmp_path, monkeypatch): 1662 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1663 themes_dir = tmp_path / "dashboard-themes" 1664 themes_dir.mkdir() 1665 (themes_dir / "ocean.yaml").write_text( 1666 "name: ocean\n" 1667 "label: Ocean\n" 1668 "palette:\n" 1669 " background:\n" 1670 " hex: \"#0a1628\"\n" 1671 " alpha: 1.0\n" 1672 "layout:\n" 1673 " density: spacious\n" 1674 ) 1675 from hermes_cli import web_server 1676 results = web_server._discover_user_themes() 1677 assert len(results) == 1 1678 assert results[0]["name"] == "ocean" 1679 assert results[0]["label"] == "Ocean" 1680 assert results[0]["palette"]["background"]["hex"] == "#0a1628" 1681 assert results[0]["layout"]["density"] == "spacious" 1682 # defaults filled in 1683 assert "fontSans" in results[0]["typography"] 1684 1685 def test_malformed_yaml_skipped(self, tmp_path, monkeypatch): 1686 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1687 themes_dir = tmp_path / "dashboard-themes" 1688 themes_dir.mkdir() 1689 (themes_dir / "bad.yaml").write_text("::: not valid yaml :::\n\tindent wrong") 1690 (themes_dir / "nameless.yaml").write_text("label: No Name Here\n") 1691 (themes_dir / "ok.yaml").write_text("name: ok\n") 1692 from hermes_cli import web_server 1693 results = web_server._discover_user_themes() 1694 names = [r["name"] for r in results] 1695 assert "ok" in names 1696 assert "bad" not in names # malformed YAML 1697 assert len(results) == 1 # only the valid one 1698 1699 1700 class TestNormaliseThemeExtensions: 1701 """Tests for the extended normaliser fields (assets, customCSS, 1702 componentStyles, layoutVariant) — the surfaces themes use to reskin 1703 the dashboard without shipping code.""" 1704 1705 def test_layout_variant_defaults_to_standard(self): 1706 from hermes_cli.web_server import _normalise_theme_definition 1707 result = _normalise_theme_definition({"name": "t"}) 1708 assert result["layoutVariant"] == "standard" 1709 1710 def test_layout_variant_accepts_known_values(self): 1711 from hermes_cli.web_server import _normalise_theme_definition 1712 for variant in ("standard", "cockpit", "tiled"): 1713 r = _normalise_theme_definition({"name": "t", "layoutVariant": variant}) 1714 assert r["layoutVariant"] == variant 1715 1716 def test_layout_variant_rejects_unknown(self): 1717 from hermes_cli.web_server import _normalise_theme_definition 1718 r = _normalise_theme_definition({"name": "t", "layoutVariant": "warship"}) 1719 assert r["layoutVariant"] == "standard" 1720 r2 = _normalise_theme_definition({"name": "t", "layoutVariant": 12}) 1721 assert r2["layoutVariant"] == "standard" 1722 1723 def test_assets_named_slots_passthrough(self): 1724 from hermes_cli.web_server import _normalise_theme_definition 1725 r = _normalise_theme_definition({ 1726 "name": "t", 1727 "assets": { 1728 "bg": "https://example.com/bg.jpg", 1729 "hero": "linear-gradient(180deg, red, blue)", 1730 "crest": "/ds-assets/crest.svg", 1731 "logo": " ", # whitespace-only — dropped 1732 "notAKnownKey": "ignored", 1733 }, 1734 }) 1735 assert r["assets"]["bg"] == "https://example.com/bg.jpg" 1736 assert r["assets"]["hero"].startswith("linear-gradient") 1737 assert r["assets"]["crest"] == "/ds-assets/crest.svg" 1738 assert "logo" not in r["assets"] # whitespace-only rejected 1739 assert "notAKnownKey" not in r["assets"] # unknown slot ignored 1740 1741 def test_assets_custom_block(self): 1742 from hermes_cli.web_server import _normalise_theme_definition 1743 r = _normalise_theme_definition({ 1744 "name": "t", 1745 "assets": { 1746 "custom": { 1747 "scan-lines": "/img/scan.png", 1748 "my_overlay": "/img/ov.png", 1749 "bad key!": "x", # non-alnum key — rejected 1750 "empty": "", # empty value — rejected 1751 }, 1752 }, 1753 }) 1754 assert r["assets"]["custom"] == { 1755 "scan-lines": "/img/scan.png", 1756 "my_overlay": "/img/ov.png", 1757 } 1758 1759 def test_assets_absent_means_no_field(self): 1760 from hermes_cli.web_server import _normalise_theme_definition 1761 r = _normalise_theme_definition({"name": "t"}) 1762 assert "assets" not in r 1763 1764 def test_custom_css_passthrough_and_capped(self): 1765 from hermes_cli.web_server import _normalise_theme_definition 1766 # Small CSS passes through verbatim. 1767 r = _normalise_theme_definition({ 1768 "name": "t", 1769 "customCSS": "body { color: red; }", 1770 }) 1771 assert r["customCSS"] == "body { color: red; }" 1772 1773 # 40 KiB of CSS gets clipped to the 32 KiB cap. 1774 huge = "/* x */ " * (40 * 1024 // 8 + 10) 1775 r2 = _normalise_theme_definition({"name": "t", "customCSS": huge}) 1776 assert len(r2["customCSS"]) <= 32 * 1024 1777 1778 def test_custom_css_empty_dropped(self): 1779 from hermes_cli.web_server import _normalise_theme_definition 1780 for val in ("", " \n\t", None): 1781 r = _normalise_theme_definition({"name": "t", "customCSS": val}) 1782 assert "customCSS" not in r 1783 1784 def test_component_styles_per_bucket(self): 1785 from hermes_cli.web_server import _normalise_theme_definition 1786 r = _normalise_theme_definition({ 1787 "name": "t", 1788 "componentStyles": { 1789 "card": { 1790 "clipPath": "polygon(0 0, 100% 0, 100% 100%, 0 100%)", 1791 "boxShadow": "inset 0 0 0 1px red", 1792 "bad prop!": "ignored", # non-alnum prop rejected 1793 }, 1794 "header": {"background": "linear-gradient(red, blue)"}, 1795 "rogueBucket": {"foo": "bar"}, # not a known bucket — rejected 1796 }, 1797 }) 1798 assert r["componentStyles"]["card"] == { 1799 "clipPath": "polygon(0 0, 100% 0, 100% 100%, 0 100%)", 1800 "boxShadow": "inset 0 0 0 1px red", 1801 } 1802 assert r["componentStyles"]["header"]["background"].startswith("linear-gradient") 1803 assert "rogueBucket" not in r["componentStyles"] 1804 1805 def test_component_styles_empty_buckets_dropped(self): 1806 from hermes_cli.web_server import _normalise_theme_definition 1807 r = _normalise_theme_definition({ 1808 "name": "t", 1809 "componentStyles": { 1810 "card": {}, # empty — dropped entirely 1811 "header": {"bad prop!": "ignored"}, # all props rejected — bucket dropped 1812 "footer": {"background": "black"}, 1813 }, 1814 }) 1815 assert "card" not in r.get("componentStyles", {}) 1816 assert "header" not in r.get("componentStyles", {}) 1817 assert r["componentStyles"]["footer"]["background"] == "black" 1818 1819 def test_component_styles_accepts_numeric_values(self): 1820 """Numeric values (e.g. opacity: 0.8) are coerced to strings.""" 1821 from hermes_cli.web_server import _normalise_theme_definition 1822 r = _normalise_theme_definition({ 1823 "name": "t", 1824 "componentStyles": {"card": {"opacity": 0.8, "zIndex": 5}}, 1825 }) 1826 assert r["componentStyles"]["card"] == {"opacity": "0.8", "zIndex": "5"} 1827 1828 1829 class TestDashboardPluginManifestExtensions: 1830 """Tests for the extended plugin manifest fields (tab.override, 1831 tab.hidden, slots) read by _discover_dashboard_plugins().""" 1832 1833 def _write_plugin(self, tmp_path, name, manifest): 1834 import json 1835 plug_dir = tmp_path / "plugins" / name / "dashboard" 1836 plug_dir.mkdir(parents=True) 1837 (plug_dir / "manifest.json").write_text(json.dumps(manifest)) 1838 return plug_dir 1839 1840 def test_override_and_hidden_carried_through(self, tmp_path, monkeypatch): 1841 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1842 self._write_plugin(tmp_path, "skin-home", { 1843 "name": "skin-home", 1844 "label": "Skin Home", 1845 "tab": {"path": "/skin-home", "override": "/", "hidden": True}, 1846 "slots": ["sidebar", "header-left"], 1847 "entry": "dist/index.js", 1848 }) 1849 from hermes_cli import web_server 1850 # Bust the process-level cache so the test plugin is picked up. 1851 web_server._dashboard_plugins_cache = None 1852 plugins = web_server._get_dashboard_plugins(force_rescan=True) 1853 entry = next(p for p in plugins if p["name"] == "skin-home") 1854 assert entry["tab"]["override"] == "/" 1855 assert entry["tab"]["hidden"] is True 1856 assert entry["slots"] == ["sidebar", "header-left"] 1857 1858 def test_override_requires_leading_slash(self, tmp_path, monkeypatch): 1859 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1860 self._write_plugin(tmp_path, "bad-override", { 1861 "name": "bad-override", 1862 "label": "Bad", 1863 "tab": {"path": "/bad", "override": "no-leading-slash"}, 1864 "entry": "dist/index.js", 1865 }) 1866 from hermes_cli import web_server 1867 web_server._dashboard_plugins_cache = None 1868 plugins = web_server._get_dashboard_plugins(force_rescan=True) 1869 entry = next(p for p in plugins if p["name"] == "bad-override") 1870 assert "override" not in entry["tab"] 1871 1872 def test_slots_default_empty(self, tmp_path, monkeypatch): 1873 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1874 self._write_plugin(tmp_path, "no-slots", { 1875 "name": "no-slots", 1876 "label": "No Slots", 1877 "tab": {"path": "/no-slots"}, 1878 "entry": "dist/index.js", 1879 }) 1880 from hermes_cli import web_server 1881 web_server._dashboard_plugins_cache = None 1882 plugins = web_server._get_dashboard_plugins(force_rescan=True) 1883 entry = next(p for p in plugins if p["name"] == "no-slots") 1884 assert entry["slots"] == [] 1885 assert "hidden" not in entry["tab"] 1886 assert "override" not in entry["tab"] 1887 1888 def test_slots_filters_non_string_entries(self, tmp_path, monkeypatch): 1889 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1890 self._write_plugin(tmp_path, "mixed-slots", { 1891 "name": "mixed-slots", 1892 "label": "Mixed", 1893 "tab": {"path": "/mixed-slots"}, 1894 "slots": ["sidebar", "", 42, None, "header-right"], 1895 "entry": "dist/index.js", 1896 }) 1897 from hermes_cli import web_server 1898 web_server._dashboard_plugins_cache = None 1899 plugins = web_server._get_dashboard_plugins(force_rescan=True) 1900 entry = next(p for p in plugins if p["name"] == "mixed-slots") 1901 assert entry["slots"] == ["sidebar", "header-right"] 1902 1903 def test_page_scoped_slots_preserved(self, tmp_path, monkeypatch): 1904 """Page-scoped slot names (e.g. ``sessions:top``) round-trip through 1905 the manifest loader untouched. The backend has no allowlist — the 1906 frontend ``<PluginSlot name="...">`` placements decide what actually 1907 renders — but the loader must not mangle colons in slot names.""" 1908 monkeypatch.setenv("HERMES_HOME", str(tmp_path)) 1909 self._write_plugin(tmp_path, "page-slots", { 1910 "name": "page-slots", 1911 "label": "Page Slots", 1912 "tab": {"path": "/page-slots", "hidden": True}, 1913 "slots": [ 1914 "sessions:top", 1915 "analytics:bottom", 1916 "logs:top", 1917 "skills:bottom", 1918 "config:top", 1919 "env:bottom", 1920 "docs:top", 1921 "cron:bottom", 1922 "chat:top", 1923 ], 1924 "entry": "dist/index.js", 1925 }) 1926 from hermes_cli import web_server 1927 web_server._dashboard_plugins_cache = None 1928 plugins = web_server._get_dashboard_plugins(force_rescan=True) 1929 entry = next(p for p in plugins if p["name"] == "page-slots") 1930 assert entry["slots"] == [ 1931 "sessions:top", 1932 "analytics:bottom", 1933 "logs:top", 1934 "skills:bottom", 1935 "config:top", 1936 "env:bottom", 1937 "docs:top", 1938 "cron:bottom", 1939 "chat:top", 1940 ] 1941 1942 1943 # --------------------------------------------------------------------------- 1944 # /api/pty WebSocket — terminal bridge for the dashboard "Chat" tab. 1945 # 1946 # These tests drive the endpoint with a tiny fake command (typically ``cat`` 1947 # or ``sh -c 'printf …'``) instead of the real ``hermes --tui`` binary. The 1948 # endpoint resolves its argv through ``_resolve_chat_argv``, so tests 1949 # monkeypatch that hook. 1950 # --------------------------------------------------------------------------- 1951 1952 import sys 1953 1954 1955 skip_on_windows = pytest.mark.skipif( 1956 sys.platform.startswith("win"), reason="PTY bridge is POSIX-only" 1957 ) 1958 1959 1960 @skip_on_windows 1961 class TestPtyWebSocket: 1962 @pytest.fixture(autouse=True) 1963 def _setup(self, monkeypatch, _isolate_hermes_home): 1964 from starlette.testclient import TestClient 1965 1966 import hermes_cli.web_server as ws 1967 1968 # Avoid exec'ing the actual TUI in tests: every test below installs 1969 # its own fake argv via ``ws._resolve_chat_argv``. 1970 self.ws_module = ws 1971 monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True) 1972 self.token = ws._SESSION_TOKEN 1973 self.client = TestClient(ws.app) 1974 1975 def _url(self, token: str | None = None, **params: str) -> str: 1976 tok = token if token is not None else self.token 1977 # TestClient.websocket_connect takes the path; it reconstructs the 1978 # query string, so we pass it inline. 1979 from urllib.parse import urlencode 1980 1981 q = {"token": tok, **params} 1982 return f"/api/pty?{urlencode(q)}" 1983 1984 def test_rejects_when_embedded_chat_disabled(self, monkeypatch): 1985 monkeypatch.setattr(self.ws_module, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", False) 1986 from starlette.websockets import WebSocketDisconnect 1987 1988 with pytest.raises(WebSocketDisconnect) as exc: 1989 with self.client.websocket_connect(self._url()): 1990 pass 1991 assert exc.value.code == 4403 1992 1993 def test_rejects_missing_token(self, monkeypatch): 1994 monkeypatch.setattr( 1995 self.ws_module, 1996 "_resolve_chat_argv", 1997 lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), 1998 ) 1999 from starlette.websockets import WebSocketDisconnect 2000 2001 with pytest.raises(WebSocketDisconnect) as exc: 2002 with self.client.websocket_connect("/api/pty"): 2003 pass 2004 assert exc.value.code == 4401 2005 2006 def test_rejects_bad_token(self, monkeypatch): 2007 monkeypatch.setattr( 2008 self.ws_module, 2009 "_resolve_chat_argv", 2010 lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), 2011 ) 2012 from starlette.websockets import WebSocketDisconnect 2013 2014 with pytest.raises(WebSocketDisconnect) as exc: 2015 with self.client.websocket_connect(self._url(token="wrong")): 2016 pass 2017 assert exc.value.code == 4401 2018 2019 def test_streams_child_stdout_to_client(self, monkeypatch): 2020 monkeypatch.setattr( 2021 self.ws_module, 2022 "_resolve_chat_argv", 2023 lambda resume=None, sidecar_url=None: ( 2024 ["/bin/sh", "-c", "printf hermes-ws-ok"], 2025 None, 2026 None, 2027 ), 2028 ) 2029 with self.client.websocket_connect(self._url()) as conn: 2030 # Drain frames until we see the needle or time out. TestClient's 2031 # recv_bytes blocks; loop until we have the signal byte string. 2032 buf = b"" 2033 import time 2034 2035 deadline = time.monotonic() + 5.0 2036 while time.monotonic() < deadline: 2037 try: 2038 frame = conn.receive_bytes() 2039 except Exception: 2040 break 2041 if frame: 2042 buf += frame 2043 if b"hermes-ws-ok" in buf: 2044 break 2045 assert b"hermes-ws-ok" in buf 2046 2047 def test_client_input_reaches_child_stdin(self, monkeypatch): 2048 # ``cat`` echoes stdin back, so a write → read round-trip proves 2049 # the full duplex path. 2050 monkeypatch.setattr( 2051 self.ws_module, 2052 "_resolve_chat_argv", 2053 lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), 2054 ) 2055 with self.client.websocket_connect(self._url()) as conn: 2056 conn.send_bytes(b"round-trip-payload\n") 2057 buf = b"" 2058 import time 2059 2060 deadline = time.monotonic() + 5.0 2061 while time.monotonic() < deadline: 2062 frame = conn.receive_bytes() 2063 if frame: 2064 buf += frame 2065 if b"round-trip-payload" in buf: 2066 break 2067 assert b"round-trip-payload" in buf 2068 2069 def test_resize_escape_is_forwarded(self, monkeypatch): 2070 # Resize escape gets intercepted and applied via TIOCSWINSZ, then the 2071 # child reads the TTY ioctl directly. Avoid tput because CI may not set 2072 # TERM for non-interactive shells. 2073 import sys 2074 2075 winsize_script = ( 2076 "import fcntl, struct, termios, time; " 2077 "time.sleep(0.15); " 2078 "rows, cols, *_ = struct.unpack('HHHH', " 2079 "fcntl.ioctl(0, termios.TIOCGWINSZ, b'\\0' * 8)); " 2080 "print(cols); print(rows)" 2081 ) 2082 monkeypatch.setattr( 2083 self.ws_module, 2084 "_resolve_chat_argv", 2085 # sleep gives the test time to push the resize before the child reads the ioctl. 2086 lambda resume=None, sidecar_url=None: ( 2087 [sys.executable, "-c", winsize_script], 2088 None, 2089 None, 2090 ), 2091 ) 2092 with self.client.websocket_connect(self._url()) as conn: 2093 conn.send_text("\x1b[RESIZE:99;41]") 2094 buf = b"" 2095 import time 2096 2097 deadline = time.monotonic() + 5.0 2098 while time.monotonic() < deadline: 2099 frame = conn.receive_bytes() 2100 if frame: 2101 buf += frame 2102 if b"99" in buf and b"41" in buf: 2103 break 2104 assert b"99" in buf and b"41" in buf 2105 2106 def test_unavailable_platform_closes_with_message(self, monkeypatch): 2107 from hermes_cli.pty_bridge import PtyUnavailableError 2108 2109 def _raise(argv, **kwargs): 2110 raise PtyUnavailableError("pty missing for tests") 2111 2112 monkeypatch.setattr( 2113 self.ws_module, 2114 "_resolve_chat_argv", 2115 lambda resume=None, sidecar_url=None: (["/bin/cat"], None, None), 2116 ) 2117 # Patch PtyBridge.spawn at the web_server module's binding. 2118 import hermes_cli.web_server as ws_mod 2119 2120 monkeypatch.setattr(ws_mod.PtyBridge, "spawn", classmethod(lambda cls, *a, **k: _raise(*a, **k))) 2121 2122 with self.client.websocket_connect(self._url()) as conn: 2123 # Expect a final text frame with the error message, then close. 2124 msg = conn.receive_text() 2125 assert "pty missing" in msg or "unavailable" in msg.lower() or "pty" in msg.lower() 2126 2127 def test_resume_parameter_is_forwarded_to_argv(self, monkeypatch): 2128 captured: dict = {} 2129 2130 def fake_resolve(resume=None, sidecar_url=None): 2131 captured["resume"] = resume 2132 return (["/bin/sh", "-c", "printf resume-arg-ok"], None, None) 2133 2134 monkeypatch.setattr(self.ws_module, "_resolve_chat_argv", fake_resolve) 2135 2136 with self.client.websocket_connect(self._url(resume="sess-42")) as conn: 2137 # Drain briefly so the handler actually invokes the resolver. 2138 try: 2139 conn.receive_bytes() 2140 except Exception: 2141 pass 2142 assert captured.get("resume") == "sess-42" 2143 2144 def test_channel_param_propagates_sidecar_url(self, monkeypatch): 2145 """When /api/pty is opened with ?channel=, the PTY child gets a 2146 HERMES_TUI_SIDECAR_URL env var pointing back at /api/pub on the 2147 same channel — which is how tool events reach the dashboard sidebar.""" 2148 captured: dict = {} 2149 2150 def fake_resolve(resume=None, sidecar_url=None): 2151 captured["sidecar_url"] = sidecar_url 2152 return (["/bin/sh", "-c", "printf sidecar-ok"], None, None) 2153 2154 monkeypatch.setattr(self.ws_module, "_resolve_chat_argv", fake_resolve) 2155 monkeypatch.setattr( 2156 self.ws_module.app.state, "bound_host", "127.0.0.1", raising=False 2157 ) 2158 monkeypatch.setattr( 2159 self.ws_module.app.state, "bound_port", 9119, raising=False 2160 ) 2161 2162 with self.client.websocket_connect(self._url(channel="abc-123")) as conn: 2163 try: 2164 conn.receive_bytes() 2165 except Exception: 2166 pass 2167 2168 url = captured.get("sidecar_url") or "" 2169 assert url.startswith("ws://127.0.0.1:9119/api/pub?") 2170 assert "channel=abc-123" in url 2171 assert "token=" in url 2172 2173 def test_pub_broadcasts_to_events_subscribers(self, monkeypatch): 2174 """Frame written to /api/pub is rebroadcast verbatim to every 2175 /api/events subscriber on the same channel.""" 2176 import time 2177 from urllib.parse import urlencode 2178 from hermes_cli import web_server as ws_mod 2179 2180 qs = urlencode({"token": self.token, "channel": "broadcast-test"}) 2181 pub_path = f"/api/pub?{qs}" 2182 sub_path = f"/api/events?{qs}" 2183 2184 with self.client.websocket_connect(sub_path) as sub: 2185 # Wait for the subscriber to be registered on the server side. 2186 # websocket_connect returns when ws.accept() completes, but the 2187 # server adds us to ``_event_channels`` in a follow-up await, 2188 # so a publish immediately after connect can race ahead of the 2189 # subscriber registration and the message is dropped. 2190 deadline = time.monotonic() + 5.0 2191 while time.monotonic() < deadline: 2192 if ws_mod._event_channels.get("broadcast-test"): 2193 break 2194 time.sleep(0.01) 2195 else: 2196 raise AssertionError( 2197 "subscriber did not register on channel within 5s" 2198 ) 2199 2200 with self.client.websocket_connect(pub_path) as pub: 2201 pub.send_text('{"type":"tool.start","payload":{"tool_id":"t1"}}') 2202 received = sub.receive_text() 2203 2204 assert "tool.start" in received 2205 assert '"tool_id":"t1"' in received 2206 2207 def test_events_rejects_missing_channel(self): 2208 from starlette.websockets import WebSocketDisconnect 2209 2210 with pytest.raises(WebSocketDisconnect) as exc: 2211 with self.client.websocket_connect( 2212 f"/api/events?token={self.token}" 2213 ): 2214 pass 2215 assert exc.value.code == 4400