test_model_catalog.py
1 """Tests for hermes_cli.model_catalog — remote manifest fetch + cache + fallback.""" 2 3 from __future__ import annotations 4 5 import json 6 import time 7 from pathlib import Path 8 from unittest.mock import patch 9 10 import pytest 11 12 13 @pytest.fixture 14 def isolated_home(tmp_path, monkeypatch): 15 """Isolate HERMES_HOME + reset any module-level catalog cache per test.""" 16 home = tmp_path / ".hermes" 17 home.mkdir() 18 monkeypatch.setattr(Path, "home", lambda: tmp_path) 19 monkeypatch.setenv("HERMES_HOME", str(home)) 20 21 # Force a fresh catalog module state for each test. 22 import importlib 23 from hermes_cli import model_catalog 24 importlib.reload(model_catalog) 25 yield home 26 model_catalog.reset_cache() 27 28 29 def _valid_manifest() -> dict: 30 return { 31 "version": 1, 32 "updated_at": "2026-04-25T22:00:00Z", 33 "metadata": {"source": "test"}, 34 "providers": { 35 "openrouter": { 36 "metadata": {"display_name": "OpenRouter"}, 37 "models": [ 38 {"id": "anthropic/claude-opus-4.7", "description": "recommended"}, 39 {"id": "openai/gpt-5.4", "description": ""}, 40 {"id": "openrouter/elephant-alpha", "description": "free"}, 41 ], 42 }, 43 "nous": { 44 "metadata": {"display_name": "Nous Portal"}, 45 "models": [ 46 {"id": "anthropic/claude-opus-4.7"}, 47 {"id": "moonshotai/kimi-k2.6"}, 48 ], 49 }, 50 }, 51 } 52 53 54 class TestValidation: 55 def test_accepts_well_formed_manifest(self, isolated_home): 56 from hermes_cli.model_catalog import _validate_manifest 57 assert _validate_manifest(_valid_manifest()) is True 58 59 def test_rejects_non_dict(self, isolated_home): 60 from hermes_cli.model_catalog import _validate_manifest 61 assert _validate_manifest("string") is False 62 assert _validate_manifest([]) is False 63 assert _validate_manifest(None) is False 64 65 def test_rejects_missing_version(self, isolated_home): 66 from hermes_cli.model_catalog import _validate_manifest 67 m = _valid_manifest() 68 del m["version"] 69 assert _validate_manifest(m) is False 70 71 def test_rejects_future_version(self, isolated_home): 72 from hermes_cli.model_catalog import _validate_manifest 73 m = _valid_manifest() 74 m["version"] = 999 75 assert _validate_manifest(m) is False 76 77 def test_rejects_missing_providers(self, isolated_home): 78 from hermes_cli.model_catalog import _validate_manifest 79 m = _valid_manifest() 80 del m["providers"] 81 assert _validate_manifest(m) is False 82 83 def test_rejects_malformed_model_entry(self, isolated_home): 84 from hermes_cli.model_catalog import _validate_manifest 85 m = _valid_manifest() 86 m["providers"]["openrouter"]["models"][0] = {"id": ""} # empty id 87 assert _validate_manifest(m) is False 88 89 def test_rejects_non_string_model_id(self, isolated_home): 90 from hermes_cli.model_catalog import _validate_manifest 91 m = _valid_manifest() 92 m["providers"]["openrouter"]["models"][0] = {"id": 42} 93 assert _validate_manifest(m) is False 94 95 96 class TestFetchSuccess: 97 def test_fetch_and_cache_writes_disk(self, isolated_home): 98 from hermes_cli import model_catalog 99 manifest = _valid_manifest() 100 with patch.object( 101 model_catalog, "_fetch_manifest", return_value=manifest 102 ) as fetch: 103 result = model_catalog.get_catalog(force_refresh=True) 104 105 assert result == manifest 106 assert fetch.called 107 108 cache_file = model_catalog._cache_path() 109 assert cache_file.exists() 110 with open(cache_file) as fh: 111 assert json.load(fh) == manifest 112 113 def test_second_call_uses_in_process_cache(self, isolated_home): 114 from hermes_cli import model_catalog 115 manifest = _valid_manifest() 116 with patch.object( 117 model_catalog, "_fetch_manifest", return_value=manifest 118 ) as fetch: 119 model_catalog.get_catalog(force_refresh=True) 120 model_catalog.get_catalog() # should not hit network again 121 assert fetch.call_count == 1 122 123 def test_force_refresh_always_refetches(self, isolated_home): 124 from hermes_cli import model_catalog 125 manifest = _valid_manifest() 126 with patch.object( 127 model_catalog, "_fetch_manifest", return_value=manifest 128 ) as fetch: 129 model_catalog.get_catalog(force_refresh=True) 130 model_catalog.get_catalog(force_refresh=True) 131 assert fetch.call_count == 2 132 133 134 class TestFetchFailure: 135 def test_network_failure_returns_empty_when_no_cache(self, isolated_home): 136 from hermes_cli import model_catalog 137 with patch.object(model_catalog, "_fetch_manifest", return_value=None): 138 result = model_catalog.get_catalog(force_refresh=True) 139 assert result == {} 140 141 def test_network_failure_falls_back_to_disk_cache(self, isolated_home): 142 from hermes_cli import model_catalog 143 # Prime disk cache with a fresh copy. 144 manifest = _valid_manifest() 145 with patch.object(model_catalog, "_fetch_manifest", return_value=manifest): 146 model_catalog.get_catalog(force_refresh=True) 147 148 # Now wipe in-process cache and simulate network failure on refetch. 149 model_catalog.reset_cache() 150 with patch.object(model_catalog, "_fetch_manifest", return_value=None): 151 result = model_catalog.get_catalog(force_refresh=True) 152 153 assert result == manifest 154 155 def test_fetch_failure_falls_back_to_stale_cache(self, isolated_home): 156 from hermes_cli import model_catalog 157 manifest = _valid_manifest() 158 # Write stale cache directly (mtime in the past). 159 cache = model_catalog._cache_path() 160 cache.parent.mkdir(parents=True, exist_ok=True) 161 with open(cache, "w") as fh: 162 json.dump(manifest, fh) 163 old = time.time() - 30 * 24 * 3600 # 30 days ago 164 import os as _os 165 _os.utime(cache, (old, old)) 166 167 with patch.object(model_catalog, "_fetch_manifest", return_value=None): 168 result = model_catalog.get_catalog() 169 170 # Stale cache is better than nothing. 171 assert result == manifest 172 173 174 class TestCuratedAccessors: 175 def test_openrouter_returns_tuples(self, isolated_home): 176 from hermes_cli import model_catalog 177 with patch.object( 178 model_catalog, "_fetch_manifest", return_value=_valid_manifest() 179 ): 180 result = model_catalog.get_curated_openrouter_models() 181 assert result == [ 182 ("anthropic/claude-opus-4.7", "recommended"), 183 ("openai/gpt-5.4", ""), 184 ("openrouter/elephant-alpha", "free"), 185 ] 186 187 def test_nous_returns_ids(self, isolated_home): 188 from hermes_cli import model_catalog 189 with patch.object( 190 model_catalog, "_fetch_manifest", return_value=_valid_manifest() 191 ): 192 result = model_catalog.get_curated_nous_models() 193 assert result == ["anthropic/claude-opus-4.7", "moonshotai/kimi-k2.6"] 194 195 def test_openrouter_returns_none_when_catalog_empty(self, isolated_home): 196 from hermes_cli import model_catalog 197 with patch.object(model_catalog, "_fetch_manifest", return_value=None): 198 assert model_catalog.get_curated_openrouter_models() is None 199 200 def test_nous_returns_none_when_catalog_empty(self, isolated_home): 201 from hermes_cli import model_catalog 202 with patch.object(model_catalog, "_fetch_manifest", return_value=None): 203 assert model_catalog.get_curated_nous_models() is None 204 205 206 class TestDisabled: 207 def test_disabled_config_short_circuits(self, isolated_home): 208 from hermes_cli import model_catalog 209 with patch.object( 210 model_catalog, 211 "_load_catalog_config", 212 return_value={ 213 "enabled": False, 214 "url": "http://ignored", 215 "ttl_hours": 24.0, 216 "providers": {}, 217 }, 218 ): 219 with patch.object(model_catalog, "_fetch_manifest") as fetch: 220 result = model_catalog.get_catalog() 221 assert result == {} 222 fetch.assert_not_called() 223 224 225 class TestProviderOverride: 226 def test_override_url_takes_precedence(self, isolated_home): 227 from hermes_cli import model_catalog 228 229 override_payload = { 230 "version": 1, 231 "providers": { 232 "openrouter": { 233 "models": [ 234 {"id": "override/model", "description": "custom"}, 235 ] 236 } 237 }, 238 } 239 240 def fake_fetch(url, timeout): 241 if "override" in url: 242 return override_payload 243 return _valid_manifest() 244 245 with patch.object( 246 model_catalog, 247 "_load_catalog_config", 248 return_value={ 249 "enabled": True, 250 "url": "http://master", 251 "ttl_hours": 24.0, 252 "providers": {"openrouter": {"url": "http://override"}}, 253 }, 254 ): 255 with patch.object(model_catalog, "_fetch_manifest", side_effect=fake_fetch): 256 result = model_catalog.get_curated_openrouter_models() 257 258 assert result == [("override/model", "custom")] 259 260 261 class TestIntegrationWithModelsModule: 262 """Exercise the fallback paths via the real callers in hermes_cli.models.""" 263 264 def test_curated_nous_ids_falls_back_to_hardcoded_on_empty_catalog( 265 self, isolated_home 266 ): 267 from hermes_cli import model_catalog 268 from hermes_cli.models import get_curated_nous_model_ids, _PROVIDER_MODELS 269 270 with patch.object(model_catalog, "_fetch_manifest", return_value=None): 271 result = get_curated_nous_model_ids() 272 273 assert result == list(_PROVIDER_MODELS["nous"]) 274 275 def test_curated_nous_ids_prefers_manifest(self, isolated_home): 276 from hermes_cli import model_catalog 277 from hermes_cli.models import get_curated_nous_model_ids 278 279 with patch.object( 280 model_catalog, "_fetch_manifest", return_value=_valid_manifest() 281 ): 282 result = get_curated_nous_model_ids() 283 284 assert result == ["anthropic/claude-opus-4.7", "moonshotai/kimi-k2.6"]