mirror of
https://github.com/nesquena/hermes-webui.git
synced 2026-07-21 15:10:30 +00:00
fix(#3928): return a real static catalog when /api/models times out
This commit is contained in:
Rod Boev
+468
-11
@@ -3045,17 +3045,118 @@ def _invoke_models_rebuild(builder):
|
||||
return builder()
|
||||
|
||||
|
||||
def _minimal_static_models_catalog() -> dict:
|
||||
"""Return a network-free /api/models catalog derived from config + auth.
|
||||
def _configured_model_badges_from_static_catalog(
|
||||
groups: list[dict],
|
||||
*,
|
||||
active_provider: str | None,
|
||||
default_model: str,
|
||||
) -> dict[str, dict[str, str]]:
|
||||
configured_entries: list[dict[str, str]] = []
|
||||
if active_provider and default_model:
|
||||
configured_entries.append(
|
||||
{
|
||||
"provider": active_provider,
|
||||
"model": default_model,
|
||||
"role": "primary",
|
||||
"label": "Primary",
|
||||
}
|
||||
)
|
||||
|
||||
Used as the fast fallback when a foreground caller must NOT pay the live
|
||||
provider probe: server-initiated wakeup turns (Option Z) and the
|
||||
bounded-rebuild timeout path. It is enough for
|
||||
``_resolve_compatible_session_model_state`` (which only needs
|
||||
``default_model`` / ``active_provider`` plus the persisted session model)
|
||||
and keeps the picker non-empty. Intentionally NOT written to the 24h
|
||||
cache so a subsequent human ``/api/models`` still triggers a real rebuild.
|
||||
"""
|
||||
fallback_cfg = cfg.get("fallback_providers", []) if isinstance(cfg, dict) else []
|
||||
if isinstance(fallback_cfg, list):
|
||||
for idx, entry in enumerate(fallback_cfg, start=1):
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
provider = _resolve_provider_alias(entry.get("provider"))
|
||||
model = str(entry.get("model") or "").strip()
|
||||
if not provider or not model:
|
||||
continue
|
||||
configured_entries.append(
|
||||
{
|
||||
"provider": provider,
|
||||
"model": model,
|
||||
"role": "fallback",
|
||||
"label": f"Fallback {idx}",
|
||||
}
|
||||
)
|
||||
|
||||
option_ids = [
|
||||
m.get("id", "")
|
||||
for g in groups
|
||||
for m in g.get("models", [])
|
||||
if m.get("id")
|
||||
]
|
||||
option_lookup = {str(opt_id): str(opt_id) for opt_id in option_ids}
|
||||
option_provider_lookup = {
|
||||
str(m.get("id")): str(g.get("provider_id") or "")
|
||||
for g in groups
|
||||
for m in g.get("models", [])
|
||||
if m.get("id")
|
||||
}
|
||||
|
||||
def _norm_model_id(model_id: str) -> str:
|
||||
s = str(model_id or "").strip().lower()
|
||||
if s.startswith("@") and ":" in s:
|
||||
parts = s.split(":")
|
||||
s = parts[-1] or s
|
||||
if "://" not in s and "/" in s:
|
||||
stripped = s.split("/", 1)[1]
|
||||
s = stripped or s
|
||||
return s.replace("-", ".")
|
||||
|
||||
norm_lookup: dict[str, list[str]] = {}
|
||||
for opt_id in option_ids:
|
||||
norm_lookup.setdefault(_norm_model_id(opt_id), []).append(opt_id)
|
||||
|
||||
badges: dict[str, dict[str, str]] = {}
|
||||
for entry in configured_entries:
|
||||
provider = entry["provider"]
|
||||
model = entry["model"]
|
||||
raw_candidates = []
|
||||
for candidate in (model, f"{provider}/{model}", f"@{provider}:{model}"):
|
||||
if candidate and candidate not in raw_candidates:
|
||||
raw_candidates.append(candidate)
|
||||
|
||||
match_id = None
|
||||
for candidate in raw_candidates:
|
||||
if (
|
||||
candidate in option_lookup
|
||||
and option_provider_lookup.get(candidate) == provider
|
||||
):
|
||||
match_id = option_lookup[candidate]
|
||||
break
|
||||
if match_id is None:
|
||||
for candidate in raw_candidates:
|
||||
normalized = _norm_model_id(candidate)
|
||||
matches = norm_lookup.get(normalized, [])
|
||||
if not matches:
|
||||
continue
|
||||
provider_match = next(
|
||||
(m for m in matches if option_provider_lookup.get(m) == provider),
|
||||
None,
|
||||
)
|
||||
match_id = provider_match or matches[0]
|
||||
if match_id:
|
||||
break
|
||||
|
||||
badge_payload = {
|
||||
"role": entry["role"],
|
||||
"label": entry["label"],
|
||||
"provider": provider,
|
||||
}
|
||||
for candidate in raw_candidates:
|
||||
candidate_provider = option_provider_lookup.get(candidate)
|
||||
if candidate_provider and candidate_provider != provider:
|
||||
continue
|
||||
badges[candidate] = badge_payload
|
||||
if match_id:
|
||||
badges[match_id] = badge_payload
|
||||
|
||||
return badges
|
||||
|
||||
|
||||
def _minimal_static_models_catalog() -> dict:
|
||||
"""Return the emergency one-model fallback for /api/models."""
|
||||
try:
|
||||
active_provider = None
|
||||
cfg_base_url = ""
|
||||
@@ -3112,6 +3213,362 @@ def _minimal_static_models_catalog() -> dict:
|
||||
"groups": [],
|
||||
}
|
||||
|
||||
|
||||
def _static_models_catalog_without_live_probes() -> dict:
|
||||
"""Return a network-free /api/models catalog from local config/auth only."""
|
||||
try:
|
||||
from api.providers import _provider_has_key
|
||||
|
||||
active_provider = None
|
||||
cfg_base_url = ""
|
||||
model_cfg = cfg.get("model", {}) if isinstance(cfg, dict) else {}
|
||||
if isinstance(model_cfg, dict):
|
||||
active_provider = model_cfg.get("provider")
|
||||
cfg_base_url = model_cfg.get("base_url", "") or ""
|
||||
if active_provider:
|
||||
try:
|
||||
active_provider = _resolve_configured_provider_id(
|
||||
active_provider,
|
||||
cfg,
|
||||
base_url=cfg_base_url,
|
||||
)
|
||||
except Exception:
|
||||
active_provider = str(active_provider or "").strip() or None
|
||||
|
||||
auth_store: dict = {}
|
||||
try:
|
||||
auth_store_path = _get_auth_store_path()
|
||||
if auth_store_path.exists():
|
||||
auth_store = json.loads(auth_store_path.read_text(encoding="utf-8"))
|
||||
if not active_provider:
|
||||
active_provider = (
|
||||
_resolve_configured_provider_id(
|
||||
auth_store.get("active_provider"),
|
||||
cfg,
|
||||
base_url=cfg_base_url,
|
||||
)
|
||||
or None
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Failed to load auth store for static models catalog", exc_info=True)
|
||||
|
||||
default_model = get_effective_default_model(cfg)
|
||||
detected_providers: set[str] = set()
|
||||
configured_model_ids: dict[str, list[str]] = {}
|
||||
named_custom_groups: dict[str, dict[str, object]] = {}
|
||||
custom_group_models: list[dict] = []
|
||||
canonical_to_raw_provider_key: dict[str, str] = {}
|
||||
providers_cfg = _get_providers_cfg()
|
||||
|
||||
def _append_model_id(provider_id: str | None, model_id: object) -> None:
|
||||
pid = _canonicalise_provider_id(provider_id)
|
||||
mid = str(model_id or "").strip()
|
||||
if not pid or not mid:
|
||||
return
|
||||
configured_model_ids.setdefault(pid, [])
|
||||
if mid not in configured_model_ids[pid]:
|
||||
configured_model_ids[pid].append(mid)
|
||||
|
||||
if active_provider:
|
||||
detected_providers.add(active_provider)
|
||||
_append_model_id(active_provider, default_model)
|
||||
|
||||
try:
|
||||
_pool = auth_store.get("credential_pool", {}) if isinstance(auth_store, dict) else {}
|
||||
if isinstance(_pool, dict):
|
||||
for _pid, _entries in _pool.items():
|
||||
if not isinstance(_entries, list) or not _entries:
|
||||
continue
|
||||
if any(
|
||||
isinstance(_entry, dict)
|
||||
and not _is_ambient_gh_cli_entry(
|
||||
str(_entry.get("source", "") or ""),
|
||||
str(_entry.get("label", "") or ""),
|
||||
str(_entry.get("key_source", "") or ""),
|
||||
)
|
||||
for _entry in _entries
|
||||
):
|
||||
detected_providers.add(_resolve_provider_alias(str(_pid)))
|
||||
except Exception:
|
||||
logger.debug("Failed to inspect auth-store credential pool", exc_info=True)
|
||||
|
||||
if isinstance(providers_cfg, dict):
|
||||
for provider_key, provider_cfg in providers_cfg.items():
|
||||
canonical = _canonicalise_provider_id(provider_key)
|
||||
if not canonical:
|
||||
continue
|
||||
is_known_provider = (
|
||||
canonical in _PROVIDER_MODELS
|
||||
or canonical in _PROVIDER_DISPLAY
|
||||
or _is_plugin_model_provider(canonical)
|
||||
)
|
||||
is_provider_config = isinstance(provider_cfg, dict)
|
||||
if not (is_known_provider or is_provider_config):
|
||||
continue
|
||||
canonical_to_raw_provider_key.setdefault(canonical, provider_key)
|
||||
if isinstance(provider_cfg, dict):
|
||||
has_local_signal = any(
|
||||
str(provider_cfg.get(key) or "").strip()
|
||||
for key in ("api_key", "key_env", "base_url")
|
||||
)
|
||||
provider_models = provider_cfg.get("models")
|
||||
if isinstance(provider_models, dict):
|
||||
for model_id in provider_models:
|
||||
_append_model_id(canonical, model_id)
|
||||
has_local_signal = True
|
||||
elif isinstance(provider_models, list):
|
||||
for item in provider_models:
|
||||
if isinstance(item, dict):
|
||||
_append_model_id(
|
||||
canonical,
|
||||
item.get("id") or item.get("model") or item.get("name"),
|
||||
)
|
||||
else:
|
||||
_append_model_id(canonical, item)
|
||||
has_local_signal = True
|
||||
if has_local_signal:
|
||||
detected_providers.add(canonical)
|
||||
|
||||
for provider_id in set(_PROVIDER_MODELS) | set(_PROVIDER_DISPLAY):
|
||||
canonical = _canonicalise_provider_id(provider_id)
|
||||
if canonical and _provider_has_key(canonical):
|
||||
detected_providers.add(canonical)
|
||||
|
||||
fallback_cfg = cfg.get("fallback_providers", []) if isinstance(cfg, dict) else []
|
||||
if isinstance(fallback_cfg, list):
|
||||
for entry in fallback_cfg:
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
provider = _resolve_provider_alias(entry.get("provider"))
|
||||
if provider:
|
||||
detected_providers.add(provider)
|
||||
_append_model_id(provider, entry.get("model"))
|
||||
|
||||
for entry in _custom_provider_entries(cfg):
|
||||
provider_name = str(entry.get("name") or "").strip()
|
||||
provider_slug = _custom_provider_slug_from_name(provider_name) or "custom"
|
||||
if provider_slug != "custom":
|
||||
named_custom_groups.setdefault(
|
||||
provider_slug,
|
||||
{"name": provider_name, "models": []},
|
||||
)
|
||||
detected_providers.add(provider_slug)
|
||||
|
||||
configured_ids: list[str] = []
|
||||
model_id = str(entry.get("model") or "").strip()
|
||||
if model_id:
|
||||
configured_ids.append(model_id)
|
||||
raw_models = entry.get("models")
|
||||
if isinstance(raw_models, dict):
|
||||
for key in raw_models:
|
||||
if isinstance(key, str) and key.strip() and key.strip() not in configured_ids:
|
||||
configured_ids.append(key.strip())
|
||||
elif isinstance(raw_models, list):
|
||||
for item in raw_models:
|
||||
if isinstance(item, dict):
|
||||
candidate = str(
|
||||
item.get("id") or item.get("model") or item.get("name") or ""
|
||||
).strip()
|
||||
else:
|
||||
candidate = str(item or "").strip()
|
||||
if candidate and candidate not in configured_ids:
|
||||
configured_ids.append(candidate)
|
||||
|
||||
for configured_id in configured_ids:
|
||||
label = _get_label_for_model(configured_id, [])
|
||||
if provider_slug == "custom":
|
||||
custom_group_models.append({"id": configured_id, "label": label})
|
||||
else:
|
||||
named_custom_groups[provider_slug]["models"].append(
|
||||
{"id": configured_id, "label": label}
|
||||
)
|
||||
_append_model_id(provider_slug, configured_id)
|
||||
|
||||
if cfg_base_url:
|
||||
detected_providers.add(
|
||||
_named_custom_provider_slug_for_base_url(cfg_base_url, cfg)
|
||||
or active_provider
|
||||
or "custom"
|
||||
)
|
||||
|
||||
if detected_providers:
|
||||
detected_providers = {
|
||||
_canonicalise_provider_id(provider_id) or provider_id
|
||||
for provider_id in detected_providers
|
||||
if provider_id
|
||||
}
|
||||
|
||||
groups: list[dict] = []
|
||||
for pid in sorted(detected_providers):
|
||||
if pid.startswith("custom:"):
|
||||
custom_group = named_custom_groups.get(pid, {})
|
||||
group_models = copy.deepcopy(custom_group.get("models", []))
|
||||
if group_models or pid == active_provider:
|
||||
groups.append(
|
||||
{
|
||||
"provider": custom_group.get("name") or pid.replace("custom:", ""),
|
||||
"provider_id": pid,
|
||||
"models": _apply_provider_prefix(
|
||||
group_models,
|
||||
pid,
|
||||
active_provider,
|
||||
),
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
if pid == "custom":
|
||||
group_models = copy.deepcopy(custom_group_models)
|
||||
for model_id in configured_model_ids.get(pid, []):
|
||||
if not any(m.get("id") == model_id for m in group_models):
|
||||
group_models.append(
|
||||
{"id": model_id, "label": _get_label_for_model(model_id, [])}
|
||||
)
|
||||
if group_models or cfg_base_url or pid == active_provider:
|
||||
groups.append(
|
||||
{
|
||||
"provider": _PROVIDER_DISPLAY.get(pid, "Custom"),
|
||||
"provider_id": pid,
|
||||
"models": _apply_provider_prefix(
|
||||
group_models,
|
||||
pid,
|
||||
active_provider,
|
||||
),
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
provider_name = _PROVIDER_DISPLAY.get(pid, pid.replace("-", " ").title())
|
||||
raw_key = canonical_to_raw_provider_key.get(pid, pid)
|
||||
provider_cfg = _get_provider_cfg(raw_key)
|
||||
raw_models = []
|
||||
if isinstance(provider_cfg, dict) and "models" in provider_cfg:
|
||||
cfg_models = provider_cfg["models"]
|
||||
if isinstance(cfg_models, dict):
|
||||
raw_models = [{"id": key, "label": key} for key in cfg_models.keys()]
|
||||
elif isinstance(cfg_models, list):
|
||||
raw_models = [
|
||||
{
|
||||
"id": item["id"] if isinstance(item, dict) else item,
|
||||
"label": (
|
||||
item.get("label", item["id"])
|
||||
if isinstance(item, dict)
|
||||
else item
|
||||
),
|
||||
}
|
||||
for item in cfg_models
|
||||
]
|
||||
if not raw_models:
|
||||
raw_models = copy.deepcopy(_PROVIDER_MODELS.get(pid, []))
|
||||
for model_id in configured_model_ids.get(pid, []):
|
||||
if model_id and not any(m.get("id") == model_id for m in raw_models):
|
||||
raw_models.append(
|
||||
{"id": model_id, "label": _get_label_for_model(model_id, groups)}
|
||||
)
|
||||
if raw_models:
|
||||
groups.append(
|
||||
{
|
||||
"provider": provider_name,
|
||||
"provider_id": pid,
|
||||
"models": _apply_provider_prefix(raw_models, pid, active_provider),
|
||||
}
|
||||
)
|
||||
|
||||
if default_model:
|
||||
all_model_ids = {
|
||||
str(model.get("id") or "")
|
||||
for group in groups
|
||||
for model in group.get("models", [])
|
||||
}
|
||||
if default_model not in all_model_ids and f"@{active_provider}:{default_model}" not in all_model_ids:
|
||||
label = _get_label_for_model(default_model, groups)
|
||||
target_group = next(
|
||||
(group for group in groups if group.get("provider_id") == active_provider),
|
||||
None,
|
||||
)
|
||||
if target_group is not None:
|
||||
target_group.setdefault("models", []).insert(0, {"id": default_model, "label": label})
|
||||
elif groups:
|
||||
groups.append(
|
||||
{
|
||||
"provider": "Default",
|
||||
"provider_id": active_provider or "default",
|
||||
"models": [{"id": default_model, "label": label}],
|
||||
}
|
||||
)
|
||||
|
||||
_deduplicate_model_ids(groups)
|
||||
groups = [
|
||||
group
|
||||
for group in groups
|
||||
if group.get("models") or str(group.get("provider_id") or "").startswith("custom:")
|
||||
]
|
||||
|
||||
providers_with_keys: set[str] = set()
|
||||
try:
|
||||
_pool = auth_store.get("credential_pool", {}) if isinstance(auth_store, dict) else {}
|
||||
if isinstance(_pool, dict):
|
||||
for _pid in _pool:
|
||||
_canonical = _canonicalise_provider_id(_pid)
|
||||
if _canonical:
|
||||
providers_with_keys.add(_canonical)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
for _pk, _pv in providers_cfg.items():
|
||||
if isinstance(_pv, dict) and (
|
||||
_pv.get("api_key")
|
||||
or _pv.get("key_env")
|
||||
or _pv.get("base_url")
|
||||
):
|
||||
_canonical = _canonicalise_provider_id(_pk)
|
||||
if _canonical:
|
||||
providers_with_keys.add(_canonical)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _group_sort_key(group: dict) -> tuple[int, str]:
|
||||
provider_id = str(group.get("provider_id") or "")
|
||||
if provider_id == active_provider:
|
||||
return (0, provider_id)
|
||||
if provider_id.startswith("custom:"):
|
||||
return (1, provider_id)
|
||||
if provider_id in providers_with_keys:
|
||||
return (2, provider_id)
|
||||
return (3, provider_id)
|
||||
|
||||
groups.sort(key=_group_sort_key)
|
||||
|
||||
model_aliases: dict[str, str] = {}
|
||||
try:
|
||||
raw_aliases = cfg.get("model", {}).get("aliases", {})
|
||||
if isinstance(raw_aliases, dict):
|
||||
model_aliases = {
|
||||
str(k).strip(): str(v).strip()
|
||||
for k, v in raw_aliases.items()
|
||||
if k and v
|
||||
}
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not groups and default_model:
|
||||
return copy.deepcopy(_minimal_static_models_catalog())
|
||||
|
||||
return {
|
||||
"active_provider": active_provider,
|
||||
"default_model": default_model,
|
||||
"configured_model_badges": _configured_model_badges_from_static_catalog(
|
||||
groups,
|
||||
active_provider=active_provider,
|
||||
default_model=default_model,
|
||||
),
|
||||
"groups": groups,
|
||||
"aliases": model_aliases,
|
||||
}
|
||||
except Exception:
|
||||
logger.debug("static models catalog build failed", exc_info=True)
|
||||
return copy.deepcopy(_minimal_static_models_catalog())
|
||||
|
||||
# Cache for credential pool results -- calling load_pool() per-provider per-server
|
||||
# session is expensive (~10s for zai due to endpoint probing). The credential pool
|
||||
# only changes when the user adds/removes credentials, which is rare; a 24h TTL
|
||||
@@ -5457,7 +5914,7 @@ def get_available_models(*, prefer_cache: bool = False) -> dict:
|
||||
# PR #2971: the previous ``if disk_groups is not None`` branch
|
||||
# here was dead code. Fall back directly to the static minimal
|
||||
# catalog (no second disk read).
|
||||
return copy.deepcopy(_minimal_static_models_catalog())
|
||||
return copy.deepcopy(_static_models_catalog_without_live_probes())
|
||||
|
||||
|
||||
# ── Static file path ─────────────────────────────────────────────────────────
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
"""Regression coverage for the /api/models rebuild-budget fallback."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
import api.config as cfg
|
||||
import api.profiles as profiles
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolate_models_catalog_state(monkeypatch, tmp_path):
|
||||
config_path = tmp_path / "config.yaml"
|
||||
config_path.write_text("model: {}\n", encoding="utf-8")
|
||||
auth_store_path = tmp_path / "auth.json"
|
||||
auth_store_path.write_text("{}", encoding="utf-8")
|
||||
hermes_home = tmp_path / "hermes-home"
|
||||
hermes_home.mkdir()
|
||||
(hermes_home / ".env").write_text("", encoding="utf-8")
|
||||
|
||||
monkeypatch.setattr(cfg, "_get_config_path", lambda: config_path)
|
||||
monkeypatch.setattr(cfg, "_cfg_path", config_path, raising=False)
|
||||
monkeypatch.setattr(cfg, "_cfg_mtime", config_path.stat().st_mtime, raising=False)
|
||||
monkeypatch.setattr(cfg, "_cfg_has_in_memory_overrides", lambda: True)
|
||||
monkeypatch.setattr(cfg, "_get_auth_store_path", lambda: auth_store_path)
|
||||
monkeypatch.setattr(cfg, "_load_models_cache_from_disk", lambda: None)
|
||||
monkeypatch.setattr(cfg, "_save_models_cache_to_disk", lambda *_a, **_k: None)
|
||||
monkeypatch.setattr(cfg, "_delete_models_cache_on_disk", lambda: None)
|
||||
monkeypatch.setattr(
|
||||
cfg,
|
||||
"_models_cache_source_fingerprint",
|
||||
lambda: "unit-test-fingerprint",
|
||||
)
|
||||
monkeypatch.setattr(cfg, "_available_models_cache", None, raising=False)
|
||||
monkeypatch.setattr(cfg, "_available_models_cache_ts", 0.0, raising=False)
|
||||
monkeypatch.setattr(
|
||||
cfg,
|
||||
"_available_models_cache_source_fingerprint",
|
||||
None,
|
||||
raising=False,
|
||||
)
|
||||
monkeypatch.setattr(cfg, "_cache_build_in_progress", False, raising=False)
|
||||
monkeypatch.setattr(cfg, "cfg", {}, raising=False)
|
||||
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: hermes_home)
|
||||
monkeypatch.setattr(cfg.os, "getenv", lambda key, default=None: default or "")
|
||||
|
||||
return {
|
||||
"auth_store_path": auth_store_path,
|
||||
}
|
||||
|
||||
|
||||
def _configure_local_sources(
|
||||
monkeypatch,
|
||||
auth_store_path,
|
||||
*,
|
||||
env_vars: dict[str, str] | None = None,
|
||||
):
|
||||
cfg.cfg = {
|
||||
"model": {
|
||||
"provider": "openai-api",
|
||||
"default": "gpt-5.5",
|
||||
},
|
||||
"fallback_providers": [
|
||||
{"provider": "anthropic", "model": "claude-sonnet-4.6"},
|
||||
],
|
||||
"providers": {
|
||||
"openai_api": {"api_key": "sk-openai"},
|
||||
"anthropic": {"api_key": "sk-anthropic"},
|
||||
},
|
||||
"custom_providers": [
|
||||
{
|
||||
"name": "ZenMux",
|
||||
"models": ["vendor/model-a", {"id": "vendor/model-b"}],
|
||||
}
|
||||
],
|
||||
}
|
||||
auth_store_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"credential_pool": {
|
||||
"openrouter": [
|
||||
{
|
||||
"source": "env:openrouter_api_key",
|
||||
"label": "manual key",
|
||||
"key_source": "env",
|
||||
}
|
||||
],
|
||||
"copilot": [
|
||||
{
|
||||
"source": "gh_cli",
|
||||
"label": "gh auth token",
|
||||
"key_source": "gh auth token",
|
||||
}
|
||||
],
|
||||
}
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
env_lookup = dict(env_vars or {})
|
||||
monkeypatch.setattr(cfg.os, "getenv", lambda key, default=None: env_lookup.get(key, default or ""))
|
||||
|
||||
|
||||
def test_static_catalog_budget_fallback_lists_local_providers(
|
||||
monkeypatch,
|
||||
isolate_models_catalog_state,
|
||||
):
|
||||
_configure_local_sources(
|
||||
monkeypatch,
|
||||
isolate_models_catalog_state["auth_store_path"],
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
cfg,
|
||||
"_read_live_provider_model_ids",
|
||||
lambda _provider_id: (_ for _ in ()).throw(
|
||||
AssertionError("static timeout fallback must stay network-free")
|
||||
),
|
||||
)
|
||||
|
||||
catalog = cfg._static_models_catalog_without_live_probes()
|
||||
provider_ids = [group["provider_id"] for group in catalog["groups"]]
|
||||
|
||||
assert provider_ids[0] == "openai-api"
|
||||
assert "anthropic" in provider_ids
|
||||
assert "custom:zenmux" in provider_ids
|
||||
assert "copilot" not in provider_ids
|
||||
assert not (
|
||||
len(catalog["groups"]) == 1
|
||||
and catalog["groups"][0]["provider"] == "Default"
|
||||
)
|
||||
assert any(
|
||||
badge["role"] == "fallback" and badge["provider"] == "anthropic"
|
||||
for badge in catalog["configured_model_badges"].values()
|
||||
)
|
||||
|
||||
|
||||
def test_budget_exceeded_foreground_uses_richer_static_catalog_and_refreshes_out_of_band(
|
||||
monkeypatch,
|
||||
isolate_models_catalog_state,
|
||||
):
|
||||
_configure_local_sources(
|
||||
monkeypatch,
|
||||
isolate_models_catalog_state["auth_store_path"],
|
||||
)
|
||||
monkeypatch.setattr(cfg, "_LIVE_REBUILD_BUDGET_SECONDS", 0.01, raising=False)
|
||||
|
||||
live_result = {
|
||||
"active_provider": "openrouter",
|
||||
"default_model": "openrouter/google/gemini-2.5-pro",
|
||||
"configured_model_badges": {},
|
||||
"groups": [
|
||||
{
|
||||
"provider": "OpenRouter",
|
||||
"provider_id": "openrouter",
|
||||
"models": [
|
||||
{
|
||||
"id": "openrouter/google/gemini-2.5-pro",
|
||||
"label": "Gemini 2.5 Pro",
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
"aliases": {},
|
||||
}
|
||||
rebuild_calls = {"count": 0}
|
||||
|
||||
def _slow_rebuild(_builder):
|
||||
rebuild_calls["count"] += 1
|
||||
time.sleep(0.05)
|
||||
return copy.deepcopy(live_result)
|
||||
|
||||
monkeypatch.setattr(cfg, "_invoke_models_rebuild", _slow_rebuild)
|
||||
|
||||
expected_fallback = cfg._static_models_catalog_without_live_probes()
|
||||
started = time.monotonic()
|
||||
result = cfg.get_available_models()
|
||||
elapsed = time.monotonic() - started
|
||||
|
||||
assert rebuild_calls["count"] == 1
|
||||
assert elapsed < 0.04
|
||||
assert result == expected_fallback
|
||||
assert not (
|
||||
len(result["groups"]) == 1 and result["groups"][0]["provider"] == "Default"
|
||||
)
|
||||
|
||||
deadline = time.monotonic() + 0.5
|
||||
while time.monotonic() < deadline:
|
||||
if cfg._available_models_cache == live_result:
|
||||
break
|
||||
time.sleep(0.01)
|
||||
|
||||
assert cfg._available_models_cache == live_result
|
||||
assert cfg._cache_build_in_progress is False
|
||||
|
||||
|
||||
def test_default_group_survives_only_as_emergency_last_resort(
|
||||
monkeypatch,
|
||||
isolate_models_catalog_state,
|
||||
):
|
||||
cfg.cfg = {
|
||||
"model": {
|
||||
"provider": "anthropic",
|
||||
"default": "claude-sonnet-4.6",
|
||||
}
|
||||
}
|
||||
monkeypatch.setattr(
|
||||
cfg,
|
||||
"_get_providers_cfg",
|
||||
lambda: (_ for _ in ()).throw(RuntimeError("boom")),
|
||||
)
|
||||
|
||||
catalog = cfg._static_models_catalog_without_live_probes()
|
||||
|
||||
assert catalog["active_provider"] == "anthropic"
|
||||
assert len(catalog["groups"]) == 1
|
||||
assert catalog["groups"][0]["provider"] == "Default"
|
||||
assert catalog["groups"][0]["provider_id"] == "anthropic"
|
||||
assert catalog["groups"][0]["models"][0]["id"] == "claude-sonnet-4.6"
|
||||
Reference in New Issue
Block a user