Merge PR #7165 into stage/7165-custom-allowlist

This commit is contained in:
nesquena-hermes committed 2026-09-23 10:22:41 +00:00
commit dc554cbd1c
5 files changed
+656 -193

No files matched your search

+8 -1
View File
@@ -291,6 +291,13 @@ class TestLiveModelsCustomProviderFallback:
"model": "gpt-5.5",
"models": {"gpt-5.5-mini": {}},
"base_url": "https://right.codes/codex/v1",
# dict-shaped ``models`` is per-model metadata (Agent setup
# flow) and only narrows the live catalog when discovery is
# explicitly off — the new #7165 contract. This test still
# wants provider-scoping (no sibling leak), so pin the
# catalog via ``discover_models: false`` instead of relying
# on the removed dict-as-allowlist behavior.
"discover_models": False,
},
{
"name": "infini-ai",
@@ -308,7 +315,7 @@ class TestLiveModelsCustomProviderFallback:
resp = self._call_live_models(monkeypatch, cfg, "custom:rightcode-codex")
assert resp["provider"] == "custom:rightcode-codex"
assert [m["id"] for m in resp["models"]] == ["gpt-5.5", "gpt-5.5-mini"]
assert [m["id"] for m in resp["models"]] == ["gpt-5.5-mini"]
def test_bare_custom_fallback_ignores_named_custom_provider_models(self, monkeypatch):
"""Bare custom only represents unnamed custom entries, not named siblings."""
+390 -185
View File
@@ -1,193 +1,398 @@
"""Regression test for #3718 -- /api/models/live skips probe for custom providers.
"""Handler-level regression tests for /api/models/live custom-provider probing.
When a custom provider (``custom_providers`` entry in config.yaml) has a
``model:`` field but no ``models:`` allowlist, the live endpoint previously
populated ``ids`` from the config entry and then skipped the live ``/v1/models``
probe (guarded by ``if not ids``). The fix collects config-specified model IDs
in a separate ``_config_ids`` list so the live fetch always runs, and merges
config entries as a fallback after the fetch.
These tests invoke the real ``routes._handle_live_models`` and inspect the
emitted ``models`` payload. Earlier revisions of this file re-implemented the
filtering loop locally, which meant they passed unchanged even with the
production filter deleted (a vacuous oracle). Every case here must fail if the
allowlist gating in ``api/routes.py`` regresses.
Covered shapes:
* native ``models`` list → live catalog filtered
* JSON-array-string ``models`` → live catalog filtered (#7120 class)
* Python-literal-string ``models`` → live catalog filtered
* singular ``model`` only, no ``models`` → live catalog NOT gated
* malformed serialized ``models`` → safe, treated as scalar name
* allowlisted model absent from live catalog → appended
* live probe failure → falls back to config ids
* dict-shaped ``models`` (Agent setup) → live catalog NOT gated (per-model
metadata, not an allowlist)
* dict-shaped ``models`` + ``discover_models:
false`` → pinned to dict keys (explicit opt-out)
* ``models_discovered: true`` catalog → live catalog NOT gated (discovery
metadata is not an allowlist)
* discovered + ``discover_models: false`` → still pinned (explicit opt-out)
"""
import io
import json
import pathlib
import unittest
from unittest import mock
import sys
import types
from urllib.parse import urlparse
REPO = pathlib.Path(__file__).parent.parent
ROUTES_PY = REPO / "api" / "routes.py"
import pytest
class TestLiveModelsCustomProviderProbe(unittest.TestCase):
"""Live endpoint must probe the upstream /v1/models even when config has entries."""
def test_custom_provider_with_model_field_probes_upstream(self):
"""When a custom provider has model: in config, live fetch must still run.
Before the fix, _custom_provider_model_ids() populated ids=["assistant"],
and the `if not ids` guard prevented the upstream /v1/models probe from
running. The endpoint returned only ["assistant"] instead of the full
upstream catalog.
We test the fix by simulating the logic directly: config provides
model IDs, the upstream probe returns a different set, and the result
must contain BOTH the live models and the config fallback.
"""
# Simulate config-specified model IDs
_config_ids = ["assistant"]
# Simulate upstream /v1/models response
live_models = [
"assistant",
"assistant-pro",
"assistant-zdr",
"gpt-5.5:pt",
"claude-sonnet-4.6:pt",
]
# Simulate the merge logic from the fix:
# if ids: merge config entries not in live set
# else: fall back to config-only list
ids = list(live_models) # live fetch succeeded
_live_set = set(ids)
for _cid in _config_ids:
if _cid not in _live_set:
ids.append(_cid)
# "assistant" from config should not duplicate the live result
self.assertEqual(ids, live_models)
# All live models must be present
for m in live_models:
self.assertIn(m, ids)
def test_custom_provider_falls_back_to_config_when_probe_fails(self):
"""When the upstream probe fails, config entries must be used as fallback."""
_config_ids = ["assistant", "custom-only-model"]
# Simulate failed live fetch: ids stays empty
ids = []
# Fallback logic from the fix
if not ids:
ids = list(_config_ids)
self.assertEqual(ids, ["assistant", "custom-only-model"])
def test_config_only_model_appended_when_not_in_live_results(self):
"""Config-specified models not in the live results must be appended."""
_config_ids = ["assistant", "local-finetune"]
live_models = ["assistant", "assistant-pro", "gpt-5.5:pt"]
ids = list(live_models)
_live_set = set(ids)
for _cid in _config_ids:
if _cid not in _live_set:
ids.append(_cid)
# "assistant" is in both, "local-finetune" is config-only
self.assertIn("local-finetune", ids)
self.assertIn("assistant-pro", ids)
self.assertEqual(ids.count("assistant"), 1, "assistant must not be duplicated")
def test_live_fetch_guard_no_longer_checks_ids(self):
"""The live fetch block must not be guarded by `if not ids`.
This is a structural regression guard: the fix replaced
`if not ids and (provider == "custom" ...)` with
`if provider == "custom" ...` so the probe always runs regardless
of whether config entries populated ids.
"""
source = ROUTES_PY.read_text(encoding="utf-8")
# Find the "Always try live fetch" comment (added by the fix)
marker = "Always try live fetch for custom providers"
self.assertIn(marker, source, (
"routes.py must contain the 'Always try live fetch' comment (#3718)"
))
# Extract the block between the marker and the next major section
marker_pos = source.find(marker)
next_section = source.find("OpenAI-compat live fetch fallback", marker_pos)
self.assertNotEqual(next_section, -1, "could not find next section marker")
block = source[marker_pos:next_section]
# The old `if not ids and` guard must NOT appear in this block
self.assertNotIn("if not ids and", block, (
"Live fetch for custom providers must not be guarded by 'if not ids' (#3718)"
))
def test_mocked_live_fetch_returns_full_catalog_plus_config_entry(self):
"""Integration test: mock urlopen and exercise the real fetch/parse/merge path.
This test patches urllib.request.urlopen to return a fake /v1/models
response, then runs the same fetch+merge logic that the live handler
uses, verifying that config entries are merged and live models are
included.
"""
fake_models = {"data": [
{"id": "assistant"},
{"id": "assistant-pro"},
{"id": "gpt-5.5:pt"},
]}
fake_body = json.dumps(fake_models).encode("utf-8")
fake_resp = mock.MagicMock()
fake_resp.read.return_value = fake_body
fake_resp.__enter__ = mock.MagicMock(return_value=fake_resp)
fake_resp.__exit__ = mock.MagicMock(return_value=False)
_config_ids = ["assistant", "local-only"]
with mock.patch("urllib.request.urlopen", return_value=fake_resp):
# Replicate the fetch+parse logic from routes.py
import urllib.request
ids = []
_req = urllib.request.Request(
"http://localhost:4000/v1/models",
headers={"Authorization": "Bearer test-key"},
)
with urllib.request.urlopen(_req, timeout=5) as _resp:
_body = json.loads(_resp.read())
if isinstance(_body, dict):
_data = _body.get("data", [])
if isinstance(_data, list):
ids = [m.get("id", "") for m in _data if m.get("id")]
# Apply the merge logic from the fix
if ids:
_live_set = set(ids)
for _cid in _config_ids:
if _cid not in _live_set:
ids.append(_cid)
else:
ids = list(_config_ids)
# Live models must all be present
self.assertIn("assistant", ids)
self.assertIn("assistant-pro", ids)
self.assertIn("gpt-5.5:pt", ids)
# Config-only entry must be appended
self.assertIn("local-only", ids)
# No duplicates
self.assertEqual(ids.count("assistant"), 1)
def test_timeout_uses_config_constant(self):
"""The live fetch must use CUSTOM_MODELS_ENDPOINT_TIMEOUT_SECONDS, not a hardcoded value."""
source = ROUTES_PY.read_text(encoding="utf-8")
# Find the custom-provider live fetch block
marker = "Always try live fetch for custom providers"
marker_pos = source.find(marker)
next_section = source.find("OpenAI-compat live fetch fallback", marker_pos)
block = source[marker_pos:next_section]
# Must use the constant, not a bare number
self.assertIn("CUSTOM_MODELS_ENDPOINT_TIMEOUT_SECONDS", block, (
"Live fetch timeout must use CUSTOM_MODELS_ENDPOINT_TIMEOUT_SECONDS "
"instead of a hardcoded value (#3718 review feedback)"
))
self.assertNotIn("timeout=8", block, (
"Live fetch must not use hardcoded timeout=8 (#3718 review feedback)"
))
def _install_provider_model_ids(monkeypatch, fn):
"""Stub hermes_cli.models.provider_model_ids (returns [] for custom:*)."""
hermes_cli = types.ModuleType("hermes_cli")
hermes_cli.__path__ = []
models = types.ModuleType("hermes_cli.models")
models.provider_model_ids = fn
monkeypatch.setitem(sys.modules, "hermes_cli", hermes_cli)
monkeypatch.setitem(sys.modules, "hermes_cli.models", models)
if __name__ == "__main__":
unittest.main()
class _FakeResponse(io.BytesIO):
def __enter__(self):
return self
def __exit__(self, *_exc):
self.close()
return False
def _install_live_catalog(monkeypatch, routes, catalog, *, fail=False):
"""Stub the upstream /v1/models probe used by the custom-provider branch."""
import urllib.request
requested: list[str] = []
def _fake_urlopen(req, timeout=None):
requested.append(getattr(req, "full_url", str(req)))
if fail:
raise OSError("probe failed")
body = {"data": [{"id": mid} for mid in catalog]}
return _FakeResponse(json.dumps(body).encode())
monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
return requested
def _run_live_models(monkeypatch, routes, custom_provider, *, catalog, fail=False):
"""Invoke the real handler for provider=custom:test-gateway."""
import api.config as config
import api.profiles as profiles
routes._clear_live_models_cache()
monkeypatch.setattr(
routes, "j", lambda _handler, payload, status=200, extra_headers=None: payload
)
monkeypatch.setattr(config, "_resolve_provider_alias", lambda provider: provider)
monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default")
config_data = {
"model": {"provider": "custom:test-gateway"},
"custom_providers": [custom_provider],
}
monkeypatch.setattr(config, "get_config", lambda: config_data)
monkeypatch.setattr(routes, "cfg", config_data, raising=False)
# provider_model_ids() returns [] for custom:* → drives the custom branch.
_install_provider_model_ids(monkeypatch, lambda provider: [])
requested = _install_live_catalog(monkeypatch, routes, catalog, fail=fail)
parsed = urlparse("/api/models/live?provider=custom:test-gateway")
payload = routes._handle_live_models(object(), parsed)
return payload, requested
def _ids(payload):
return [m["id"] for m in payload.get("models", [])]
_LIVE_CATALOG = [
"chat-a",
"chat-b",
"image-gen-1", # non-chat noise the picker must not surface
"audio-tts-1",
]
_BASE_PROVIDER = {
"name": "Test Gateway",
"base_url": "https://gateway.example/v1",
"api_key": "sk-test",
}
@pytest.mark.parametrize(
"models_value",
[
pytest.param(["chat-a", "chat-b"], id="native-list"),
pytest.param('["chat-a","chat-b"]', id="json-array-string"),
pytest.param("['chat-a', 'chat-b']", id="python-literal-string"),
pytest.param(
[{"id": "chat-a"}, {"model": "chat-b"}], id="list-of-dicts"
),
],
)
def test_allowlist_filters_live_catalog(monkeypatch, models_value):
"""An explicit plural allowlist must trim the live catalog to itself.
The JSON-array-string and Python-literal cases are the regression: those
shapes are what ``hermes config set`` / JSON-mode editor saves persist, and
a dict/list-only parser drops them to ``[]`` → read as "no allowlist" → the
full upstream catalog floods the picker.
"""
import api.routes as routes
provider = dict(_BASE_PROVIDER, models=models_value)
payload, requested = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG
)
assert requested, "live /v1/models probe was never attempted"
assert _ids(payload) == ["chat-a", "chat-b"]
assert "image-gen-1" not in _ids(payload)
assert "audio-tts-1" not in _ids(payload)
def test_dict_models_without_discover_false_does_not_gate(monkeypatch):
"""A dict-shaped ``models`` is per-model metadata — NOT an allowlist (#7165).
Hermes Agent's setup flow and ``hermes_cli/model_switch.py::_save_custom_provider``
persist per-model metadata as a mapping, e.g. ``{chat-a: {context_length: 128000}}``.
Treating its keys as an allowlist would collapse the live picker to the saved
default while the CLI live-probe shows the full catalog. Without an explicit
``discover_models: false`` opt-out the dict must NOT gate — full live catalog wins.
"""
import api.routes as routes
provider = dict(
_BASE_PROVIDER,
models={"chat-a": {"context_length": 128000}, "chat-b": {"context_length": 32000}},
)
payload, requested = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG
)
assert requested, "live /v1/models probe was never attempted"
assert _ids(payload) == _LIVE_CATALOG, (
"dict-shaped models is per-model metadata, not an allowlist — the "
"full live catalog must be surfaced"
)
def test_dict_models_with_discover_false_pins_to_keys(monkeypatch):
"""``discover_models: false`` opts a dict-shaped ``models`` into an allowlist.
The explicit opt-out (bool ``False`` or the string forms ``"false"``/``"no"``/``"0"``)
means the dict keys are a hand-pinned allowlist and must filter the live catalog.
"""
import api.routes as routes
for discover in (False, "false", "no", "0"):
provider = dict(
_BASE_PROVIDER,
models={"chat-a": {"context_length": 128000}, "chat-b": {"context_length": 32000}},
discover_models=discover,
)
payload, requested = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG
)
assert requested, "live /v1/models probe was never attempted"
assert _ids(payload) == ["chat-a", "chat-b"], (
f"discover_models={discover!r} must pin the dict keys as the allowlist"
)
assert "image-gen-1" not in _ids(payload)
assert "audio-tts-1" not in _ids(payload)
def test_singular_model_only_does_not_gate_live_catalog(monkeypatch):
"""``model:`` is default metadata, not an allowlist — keep full discovery."""
import api.routes as routes
provider = dict(_BASE_PROVIDER, model="chat-a")
payload, _ = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG
)
assert _ids(payload) == _LIVE_CATALOG
def test_malformed_serialized_models_is_safe(monkeypatch):
"""A broken JSON-array string must not crash or silently drop the gate.
``_parse_config_string_list()`` treats an undecodable value as one literal
name, so the allowlist stays non-empty and the filter still runs.
"""
import api.routes as routes
provider = dict(_BASE_PROVIDER, models='["chat-a", "chat-b"') # missing ]
payload, _ = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG
)
# Treated as a single (unmatched) name → appended, catalog not flooded.
assert "image-gen-1" not in _ids(payload)
assert "audio-tts-1" not in _ids(payload)
def test_allowlisted_model_missing_from_live_is_appended(monkeypatch):
"""A declared model the upstream didn't return must still be selectable."""
import api.routes as routes
provider = dict(_BASE_PROVIDER, models=["chat-a", "chat-offline"])
payload, _ = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG
)
assert _ids(payload) == ["chat-a", "chat-offline"]
def test_contradictory_singular_model_outside_plural_allowlist_is_ignored(monkeypatch):
"""A singular default outside the plural allowlist must not leak into the picker.
Paired test: whether the live probe succeeds or fails, only the explicit
plural allowlist is exposed.
"""
import api.routes as routes
provider = dict(_BASE_PROVIDER, model="chat-default", models=["chat-a"])
# 1. Successful live fetch -> only plural allowlist
payload_success, _ = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG, fail=False
)
assert _ids(payload_success) == ["chat-a"]
# 2. Failed live fetch -> still only plural allowlist (not chat-default)
payload_fail, _ = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG, fail=True
)
assert _ids(payload_fail) == ["chat-a"]
def test_probe_failure_without_allowlist_falls_back_to_singular_model(monkeypatch):
"""When no plural allowlist is configured and probe fails, singular model is used."""
import api.routes as routes
provider = dict(_BASE_PROVIDER, model="chat-default")
payload, _ = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG, fail=True
)
assert _ids(payload) == ["chat-default"]
@pytest.mark.parametrize(
"models_value",
[
pytest.param([], id="native-empty-list"),
pytest.param("[]", id="serialized-empty-list"),
pytest.param({}, id="empty-mapping"),
pytest.param(" ", id="blank-scalar"),
],
)
def test_empty_allowlist_is_treated_as_not_configured(monkeypatch, models_value):
"""An empty ``models`` value must NOT be read as "allow nothing".
Pins a deliberate semantic decision. Gating the filter on *declared-ness*
rather than on a non-empty allowlist would make ``models: []`` emit an
EMPTY picker, leaving the provider unselectable — a harder failure than
surfacing a few extra models, and unrecoverable from the UI because
``custom_providers`` has no WebUI write path (hand-edited config.yaml only).
There is also no deny-all use case: a provider the user wants hidden is
removed from ``custom_providers`` outright.
If someone later "fixes" the empty case into a deny-all gate, this test
fails and forces the trade-off to be re-argued rather than silently shipped.
"""
import api.routes as routes
provider = dict(_BASE_PROVIDER, models=models_value)
payload, _ = _run_live_models(
monkeypatch, routes, provider, catalog=_LIVE_CATALOG
)
assert _ids(payload) == _LIVE_CATALOG, (
"empty allowlist must fall through to full live discovery, "
"never collapse the picker to zero models"
)
assert _ids(payload), "picker must never be emptied by an empty allowlist"
def test_models_discovered_catalog_is_not_an_allowlist(monkeypatch):
"""``models_discovered: true`` marks ``models:`` as discovery metadata (#7165).
Hermes persists *discovery results* back into config as ``models: {...}``
plus ``models_discovered: true``. That mapping is a snapshot of what the
gateway exposed at discovery time — NOT a hand-curated allowlist. Treating
it as one permanently pins the live catalog to the first-discovery set, so
a model the user later pulls into LM Studio / Ollama is silently dropped
from ``/api/models/live`` (the exact regression the gate reproduced: merge
base returned ``saved-a``, ``saved-b`` AND ``new-live-c``; the pinned head
dropped ``new-live-c``).
"""
import api.routes as routes
provider = dict(
_BASE_PROVIDER,
models={
"saved-a": {"context_length": 128000},
"saved-b": {"context_length": 32000},
},
models_discovered=True,
)
payload, requested = _run_live_models(
monkeypatch,
routes,
provider,
catalog=["saved-a", "saved-b", "new-live-c"],
)
assert requested, "live /v1/models probe was never attempted"
assert _ids(payload) == ["saved-a", "saved-b", "new-live-c"], (
"a discovered catalog must not gate the live probe — models that "
"appear upstream after discovery (new-live-c) must reach the picker"
)
def test_discovered_catalog_with_discover_models_false_still_pins(monkeypatch):
"""``discover_models: false`` re-pins a discovered catalog (control).
The explicit opt-out wins over ``models_discovered: true`` — same rule as
the ``/api/models`` catalog path (`api.config._provider_discover_allowed`).
Filtering to the configured set must survive the discovered-catalog fix.
"""
import api.routes as routes
provider = dict(
_BASE_PROVIDER,
models={
"saved-a": {"context_length": 128000},
"saved-b": {"context_length": 32000},
},
models_discovered=True,
discover_models=False,
)
payload, _ = _run_live_models(
monkeypatch,
routes,
provider,
catalog=["saved-a", "saved-b", "new-live-c"],
)
assert _ids(payload) == ["saved-a", "saved-b"]
def test_discovered_catalog_probe_failure_falls_back_to_saved_models(monkeypatch):
"""Discovered + failed probe → saved models, not an empty picker (control).
Releasing the allowlist must not empty the fallback: when the live probe
fails, the previously discovered ``models:`` mapping is still the best
answer and must be served from ``_config_ids``.
"""
import api.routes as routes
provider = dict(
_BASE_PROVIDER,
models={
"saved-a": {"context_length": 128000},
"saved-b": {"context_length": 32000},
},
models_discovered=True,
)
payload, _ = _run_live_models(
monkeypatch, routes, provider, catalog=[], fail=True
)
assert _ids(payload) == ["saved-a", "saved-b"]