"""Unit tests for Hermes coordinator model routing and profile generation.""" from __future__ import annotations import importlib.util import json import sys from pathlib import Path import yaml SCRIPT = ( Path(__file__).parents[2] / "services/hermes/scripts/hermes_coordinator.py" ) sys.path.insert(0, str(SCRIPT.parent)) routing = importlib.import_module("hermes_model_routing") SPEC = importlib.util.spec_from_file_location("hermes_coordinator", SCRIPT) assert SPEC and SPEC.loader coordinator = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = coordinator SPEC.loader.exec_module(coordinator) def _base_config() -> dict: """Return the minimal coordinator configuration used by routing tests.""" return { "model": { "provider": "openai-codex", "default": "gpt-5.6-terra", "model": "gpt-5.6-terra", }, "fallback_providers": [ {"provider": "anthropic", "model": "claude-opus-5"}, dict(routing.LOCAL_FALLBACK), ], "toolsets": ["kanban"], "terminal": {"cwd": "/opt/data/workspace"}, } def test_model_version_and_quality_selection_handle_new_and_small_models(): assert routing.model_version("claude-3-5-sonnet-20241022") == (3, 5) assert routing.model_version("gpt-5.7-terra") == (5, 7) codex = ["gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.7-luna", "gpt-5.3-codex-spark"] assert routing.choose_codex_model(codex) == "gpt-5.6-sol" assert routing.choose_codex_model(codex, balanced=True) == "gpt-5.6-terra" assert routing.choose_codex_model(codex + ["gpt-5.7-terra"]) == "gpt-5.7-terra" claude = ["claude-opus-4.8", "claude-haiku-5", "claude-sonnet-5"] assert routing.choose_claude_model(claude) == "claude-sonnet-5" assert routing.choose_codex_for_effort(codex, "low") == "gpt-5.7-luna" assert routing.choose_codex_for_effort(codex, "medium") == "gpt-5.6-terra" assert routing.choose_codex_for_effort(codex, "xhigh") == "gpt-5.6-sol" assert routing.choose_claude_for_effort(claude, "low") == "claude-haiku-5" assert routing.choose_claude_for_effort(claude, "medium") == "claude-sonnet-5" assert routing.choose_claude_for_effort(claude, "xhigh") == "claude-opus-4.8" def test_empty_catalog_retains_current_models(): assert routing.choose_codex_model([], "gpt-5.6-terra") == "gpt-5.6-terra" assert routing.choose_claude_model([], "claude-opus-5") == "claude-opus-5" def test_configure_routes_builds_cross_provider_fallback_profiles(tmp_path: Path): (tmp_path / "config.yaml").write_text( yaml.safe_dump(_base_config()), encoding="utf-8" ) (tmp_path / ".env").write_text( "API_SERVER_KEY=keep-root-only\n" "CLAUDE_CODE_OAUTH_TOKEN=claude-secret\n" "GITEA_TOKEN=gitea-secret\n" "GIT_ASKPASS=/opt/coordinator/gitea_askpass.sh\n", encoding="utf-8", ) codex = routing.Catalog( "openai-codex", ["gpt-5.6-sol", "gpt-5.6-terra"], True, True, "connected" ) claude = routing.Catalog( "anthropic", ["claude-sonnet-5", "claude-opus-5"], True, True, "connected" ) routes = routing.configure_routes(tmp_path, codex, claude) root = yaml.safe_load((tmp_path / "config.yaml").read_text(encoding="utf-8")) codex_profile = yaml.safe_load( (tmp_path / "profiles/codex-high/config.yaml").read_text(encoding="utf-8") ) claude_profile = yaml.safe_load( (tmp_path / "profiles/claude-high/config.yaml").read_text(encoding="utf-8") ) assert root["model"]["model"] == "gpt-5.6-terra" assert codex_profile["model"]["model"] == "gpt-5.6-sol" assert codex_profile["fallback_providers"][0] == { "provider": "anthropic", "model": "claude-opus-5", } assert claude_profile["model"]["model"] == "claude-opus-5" assert claude_profile["fallback_providers"][0]["provider"] == "openai-codex" assert codex_profile["toolsets"] == [] assert codex_profile["agent"]["reasoning_effort"] == "high" assert codex_profile["fallback_providers"][1] == routing.JETSON_FALLBACK assert routes["coordinator"][0] == "openai-codex/gpt-5.6-terra" assert "max" not in json.dumps(routes) profile_env = (tmp_path / "profiles/codex-high/.env").read_text(encoding="utf-8") assert "CLAUDE_CODE_OAUTH_TOKEN=claude-secret" in profile_env assert "GITEA_TOKEN=gitea-secret" in profile_env assert "API_SERVER_KEY" not in profile_env assert "claude-secret" not in json.dumps(routes) assert (tmp_path / "profiles/codex-high/.env").stat().st_mode & 0o777 == 0o600 def test_degraded_catalog_does_not_replace_known_working_route(tmp_path: Path): base = _base_config() base["model"]["model"] = base["model"]["default"] = "gpt-5.6-sol" (tmp_path / "config.yaml").write_text(yaml.safe_dump(base), encoding="utf-8") (tmp_path / ".env").write_text("", encoding="utf-8") codex = routing.Catalog("openai-codex", ["gpt-5.4"], False, True, "degraded") claude = routing.Catalog("anthropic", ["claude-haiku-4.5"], False, True, "degraded") routing.configure_routes(tmp_path, codex, claude) current = yaml.safe_load((tmp_path / "config.yaml").read_text(encoding="utf-8")) assert current["model"]["model"] == "gpt-5.6-sol" assert current["fallback_providers"][0]["model"] == "claude-opus-5" def test_degraded_refresh_preserves_last_worker_specific_model(tmp_path: Path): """A catalog outage must not collapse the Codex worker onto coordinator tier.""" (tmp_path / "config.yaml").write_text( yaml.safe_dump(_base_config()), encoding="utf-8" ) (tmp_path / ".env").write_text("", encoding="utf-8") live_codex = routing.Catalog( "openai-codex", ["gpt-5.6-sol", "gpt-5.6-terra"], True, True, "connected" ) live_claude = routing.Catalog( "anthropic", ["claude-opus-5"], True, True, "connected" ) routing.configure_routes(tmp_path, live_codex, live_claude) degraded_codex = routing.Catalog( "openai-codex", ["gpt-5.4"], False, True, "degraded" ) routing.configure_routes(tmp_path, degraded_codex, live_claude) worker = yaml.safe_load( (tmp_path / "profiles/codex-high/config.yaml").read_text(encoding="utf-8") ) assert worker["model"]["model"] == "gpt-5.6-sol" def test_refresh_writes_non_secret_routing_status(tmp_path: Path, monkeypatch): (tmp_path / "config.yaml").write_text( yaml.safe_dump(_base_config()), encoding="utf-8" ) (tmp_path / ".env").write_text("GITEA_TOKEN=do-not-report\n", encoding="utf-8") codex = routing.Catalog("openai-codex", ["gpt-5.6-terra"], True, True, "connected") claude = routing.Catalog("anthropic", ["claude-opus-5"], True, True, "connected") monkeypatch.setattr(coordinator, "discover_codex_models", lambda: codex) monkeypatch.setattr(coordinator, "discover_claude_models", lambda: claude) monkeypatch.setattr(coordinator, "bootstrap_cassandra", lambda root: None) monkeypatch.setattr(coordinator, "sync_cassandra_repo", lambda env: "ready") status = coordinator.refresh_once(tmp_path) status_path = tmp_path / "workspace/coordinator/model-routing.json" assert status_path.is_file() assert status["projects"]["cassandra"]["state"] == "ready" assert "do-not-report" not in status_path.read_text(encoding="utf-8")