test(zai): cover parallel-probe contracts + restore candidate-model loop

Rebase fold: the original PR predates ZAI_ENDPOINTS growing per-endpoint
probe_models lists; the parallel worker now preserves that candidate-model
fallback loop (was: scalar model). Tests (both mutation-checked):
- candidate-model fallback within one endpoint worker
- ZAI_ENDPOINTS priority order wins over completion order
- all-fail returns None
This commit is contained in:
kshitijk4poor 2026-08-03 17:00:41 +05:30 committed by kshitij
parent 9891f4b63f
commit 9e99a335a7
1 changed files with 70 additions and 0 deletions

View File

@ -643,6 +643,76 @@ class TestZaiEndpointAutoDetect:
assert creds["api_key"] == ""
class TestZaiParallelProbe:
"""detect_zai_endpoint probes endpoints in parallel workers.
Contract under test: (1) each endpoint worker preserves the per-endpoint
candidate-model fallback loop, (2) when several endpoints succeed the
winner is chosen by ZAI_ENDPOINTS priority order (not completion order).
"""
def _mock_post(self, ok):
"""Return an httpx.post replacement; `ok` maps (base_url, model) -> bool."""
import httpx as _httpx
def _post(url, headers=None, json=None, timeout=None):
base = url.rsplit("/chat/completions", 1)[0]
code = 200 if ok.get((base, json["model"])) else 401
request = _httpx.Request("POST", url)
return _httpx.Response(code, request=request, json={})
return _post
def test_candidate_model_fallback_within_endpoint(self, monkeypatch):
"""A worker must try its endpoint's later candidate models when the
first ones fail the fallback the scalar-model version dropped."""
from hermes_cli.auth import ZAI_ENDPOINTS, detect_zai_endpoint
coding_global = next(ep for ep in ZAI_ENDPOINTS if ep[0] == "coding-global")
base = coding_global[1]
last_model = coding_global[2][-1]
# Only the LAST candidate model of coding-global succeeds.
monkeypatch.setattr(
"hermes_cli.auth.httpx.post",
self._mock_post({(base, last_model): True}),
)
result = detect_zai_endpoint("test-key", timeout=1.0)
assert result is not None
assert result["id"] == "coding-global"
assert result["model"] == last_model
def test_priority_order_wins_over_completion_order(self, monkeypatch):
"""When multiple endpoints accept the key, the FIRST in
ZAI_ENDPOINTS order must win, even if another finishes earlier."""
import time as _time
from hermes_cli.auth import ZAI_ENDPOINTS, detect_zai_endpoint
first = ZAI_ENDPOINTS[0]
last = ZAI_ENDPOINTS[-1]
ok = {
(first[1], first[2][0]): True,
(last[1], last[2][0]): True,
}
inner = self._mock_post(ok)
def _slow_first(url, headers=None, json=None, timeout=None):
if url.startswith(first[1]):
_time.sleep(0.15) # first-priority endpoint finishes LAST
return inner(url, headers=headers, json=json, timeout=timeout)
monkeypatch.setattr("hermes_cli.auth.httpx.post", _slow_first)
result = detect_zai_endpoint("test-key", timeout=1.0)
assert result is not None
assert result["id"] == first[0]
def test_all_fail_returns_none(self, monkeypatch):
from hermes_cli.auth import detect_zai_endpoint
monkeypatch.setattr("hermes_cli.auth.httpx.post", self._mock_post({}))
assert detect_zai_endpoint("bad-key", timeout=1.0) is None
# =============================================================================
# Kimi / Moonshot model list isolation tests
# =============================================================================