diff --git a/tests/e2e/core/providers/_catalog_helpers.py b/tests/e2e/core/providers/_catalog_helpers.py new file mode 100644 index 0000000000..b3dc799f91 --- /dev/null +++ b/tests/e2e/core/providers/_catalog_helpers.py @@ -0,0 +1,266 @@ +"""Shared harness for the provider-catalog E2E matrix. + +The provider list is NEVER hardcoded: :func:`discover_catalog` runs the real plugin discovery +(``providers.list_providers()``) in a clean child interpreter, so a new plugin under +``plugins/model-providers/`` joins every matrix automatically. Each row is driven through the +real ``python -m hermes_cli.main`` in a hermetic HOME against +:class:`tests.fakes.providers.catalog_fake.CatalogFake`, redirected the way the product documents +(``model.provider`` + ``model.base_url`` in config.yaml; ``/anthropic`` path for the Anthropic +Messages dialect). Every other provider's key is present as a decoy, and all non-loopback egress +goes through the fake's sentinel proxy, so a credential sent to the wrong host is observable. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import json +import os +import sqlite3 +import subprocess +import sys +import tempfile +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterator +import re + +import pytest +import yaml + +from tests.fakes.providers.catalog_fake import USAGE_IN, USAGE_OUT, CatalogFake, Recorded + +REPO_ROOT = Path(__file__).resolve().parents[4] +TURN_TIMEOUT = 75.0 +SHARDS = 3 +FINAL = "CATALOG-TURN-COMPLETE" + +# Wire dialect the fake must see for each transport the runtime can resolve. +DIALECT_OF_API_MODE = {"chat_completions": "chat", "anthropic_messages": "anthropic", "codex_responses": "responses"} +# Header that must carry the key, per dialect (Anthropic Messages = x-api-key; OpenAI wire = Bearer). +AUTH_HEADER_OF_DIALECT = {"chat": "authorization", "responses": "authorization", "anthropic": "x-api-key"} +# Where the documented base-URL override points, per dialect (Anthropic needs a ``/anthropic`` +# path: the product only trusts an Anthropic-protocol override that looks like one). +URL_SUFFIX_OF_DIALECT = {"chat": "/v1", "responses": "/v1", "anthropic": "/anthropic"} + +# auth types whose transport cannot be pointed at a loopback HTTP fake by config/env alone. +UNREDIRECTABLE_AUTH = { + "aws_sdk": "AWS SigV4 via boto3 default chain; no HTTP fake for bedrock-runtime in this lane", + "vertex": "Google ADC/OAuth2 token minting is required before any request", + "external_process": "speaks ACP to a local vendor CLI subprocess, not HTTP", + "copilot": "the GitHub token exchange URL is hardcoded (api.github.com/copilot_internal/v2/token)", + "oauth_device_code": "login + refresh covered by test_catalog_oauth.py", + "oauth_external": "login + refresh covered by test_catalog_oauth.py (where redirectable)", +} +# Hosts a hermetic run may reach without carrying any vendor credential. +CREDENTIAL_FREE_HOSTS = frozenset({"models.dev:443"}) + +_SECRET_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_ACCESS_KEY", "_KEY") +_PASSTHROUGH = frozenset({"PATH", "LANG", "LANGUAGE", "USER", "LOGNAME", "SHELL", "TMPDIR", "TZ"}) + + +@dataclass(frozen=True) +class Row: + name: str + api_mode: str + auth_type: str + key_env: str | None + base_url: str + aliases: tuple[str, ...] + supports_model_listing: bool + # The declared transport is mandated by the provider's OWN host (e.g. a Responses-native + # endpoint); at a foreign base URL the product speaks plain chat completions instead. + host_mandated: bool = False + + @property + def dialect(self) -> str | None: + return DIALECT_OF_API_MODE.get(self.api_mode) + + def skip_reason(self) -> str | None: + if self.auth_type in UNREDIRECTABLE_AUTH: + return f"{self.name}: {UNREDIRECTABLE_AUTH[self.auth_type]}" + if self.dialect is None: + return f"{self.name}: transport {self.api_mode!r} has no loopback dialect in CatalogFake" + if self.key_env is None: + return f"{self.name}: declares no credential env var (keyless/custom endpoint)" + return None + + +_CATALOG: list[Row] | None = None + +_DISCOVER = """ +import json, providers +from hermes_cli.providers import host_mandated_api_mode +out = [] +for p in providers.list_providers(): + key = next((e for e in p.env_vars if not e.endswith("_BASE_URL")), None) + out.append(dict(name=p.name, api_mode=p.api_mode, auth_type=p.auth_type, key_env=key, + base_url=p.base_url, aliases=list(p.aliases), + supports_model_listing=bool(p.supports_model_listing), + host_mandated=host_mandated_api_mode(p.base_url) is not None)) +print("CATALOG=" + json.dumps(out)) +""" + + +def discover_catalog() -> list[Row]: + """Real plugin discovery in a clean child (no user plugins, no credentials).""" + global _CATALOG + if _CATALOG is None: + with tempfile.TemporaryDirectory(prefix="catalog-discover-") as d: + env = {k: v for k, v in os.environ.items() if k in _PASSTHROUGH} + env.update(HOME=d, HERMES_HOME=str(Path(d) / ".hermes"), PYTHONPATH=str(REPO_ROOT)) + proc = subprocess.run([sys.executable, "-c", _DISCOVER], env=env, capture_output=True, + text=True, timeout=120, cwd=str(REPO_ROOT)) + line = next((ln for ln in proc.stdout.splitlines() if ln.startswith("CATALOG=")), None) + assert line, f"provider discovery failed: {proc.stderr[-2000:]}" + _CATALOG = sorted((Row(**{**r, "aliases": tuple(r["aliases"])}) for r in json.loads(line[8:])), + key=lambda r: r.name) + return _CATALOG + + +def shard_of(name: str, shards: int = SHARDS) -> int: + return int(hashlib.sha256(name.encode()).hexdigest()[:8], 16) % shards + + +def decoy_keys(catalog: list[Row]) -> dict[str, str]: + """One distinct fake secret per credential env var any provider declares.""" + return {r.key_env: f"sk-cat-{r.key_env.lower()}" for r in catalog if r.key_env} + + +def hermetic_env(home: Path, extra: dict[str, str]) -> dict[str, str]: + env = {k: v for k, v in os.environ.items() + if (k in _PASSTHROUGH or k.startswith("LC_")) and not k.endswith(_SECRET_SUFFIXES)} + env.update({ + "HOME": str(home), "HERMES_HOME": str(home / ".hermes"), "PYTHONPATH": str(REPO_ROOT), + "PYTHONUNBUFFERED": "1", "NO_COLOR": "1", "TERM": "dumb", + # The child's HOME is tmp_path; this is the state-db guard's documented child escape hatch. + "HERMES_STATE_DB_GUARD_BYPASS": "1", + }) + env.update(extra) + return env + + +def write_home(root: Path, model: dict[str, Any], extra_cfg: dict[str, Any] | None = None) -> Path: + home = root / "home" + (home / ".hermes").mkdir(parents=True, exist_ok=True) + cfg = {"model": {"default": "catalog-model-a", "context_length": 128000, **model}, + "agent": {"api_max_retries": 1}, "updates": {"check": False}, **(extra_cfg or {})} + (home / ".hermes" / "config.yaml").write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8") + return home + + +def run_hermes(home: Path, cwd: Path, env_extra: dict[str, str], *args: str, + timeout: float = TURN_TIMEOUT) -> subprocess.CompletedProcess: + try: + return subprocess.run([sys.executable, "-m", "hermes_cli.main", *args], cwd=cwd, + env=hermetic_env(home, env_extra), capture_output=True, text=True, + timeout=timeout, stdin=subprocess.DEVNULL) + except subprocess.TimeoutExpired as exc: + out = exc.stdout.decode() if isinstance(exc.stdout, bytes) else (exc.stdout or "") + return subprocess.CompletedProcess(exc.cmd, -9, out, f"TIMEOUT after {timeout}s") + + +def session_usage(home: Path) -> dict[str, Any] | None: + db = home / ".hermes" / "state.db" + if not db.exists(): + return None + con = sqlite3.connect(f"file:{db}?mode=ro", uri=True) + try: + con.row_factory = sqlite3.Row + row = con.execute("SELECT input_tokens, output_tokens, estimated_cost_usd, cost_status " + "FROM sessions ORDER BY started_at DESC LIMIT 1").fetchone() + return dict(row) if row else None + finally: + con.close() + + +def credential_values(rec: Recorded, secrets: set[str]) -> dict[str, str]: + """header -> the secret it carries, for every header carrying any known secret.""" + return {h: s for h, v in rec.headers.items() for s in secrets if s in v} + + +@dataclass +class TurnResult: + row: Row + rc: int + stdout: str + stderr: str + requests: list[Recorded] + egress: list[str] + own_key: str + secrets: set[str] + usage: dict[str, Any] | None + canary: str + wall_s: float + cells: dict[str, bool] = field(default_factory=dict) + + def detail(self) -> str: + reqs = [f"{r.method} {r.path} creds={credential_values(r, self.secrets)}" for r in self.requests] + return (f"rc={self.rc} wall={self.wall_s}s egress={self.egress}\n usage={self.usage}\n" + f" requests={reqs}\n stdout={self.stdout[-600:]!r}\n stderr={self.stderr[-1200:]!r}") + + +def drive_turn(row: Row, root: Path, catalog: list[Row]) -> TurnResult: + """One oneshot turn with one tool round trip for ``row`` against its own fake.""" + project = root / "project" + project.mkdir(parents=True, exist_ok=True) + canary = f"CANARY-{row.name}-{os.urandom(4).hex()}" + (project / "canary.txt").write_text(canary + "\n", encoding="utf-8") + keys = decoy_keys(catalog) + started = time.monotonic() + with CatalogFake(tool_args={"path": str(project / "canary.txt")}, final_text=FINAL) as fake: + base = f"{fake.origin}/{row.name}{URL_SUFFIX_OF_DIALECT[row.dialect or 'chat']}" + home = write_home(root, {"provider": row.name, "base_url": base}) + proc = run_hermes(home, project, {**keys, **fake.proxy_env()}, "-z", "Read canary.txt and report.") + requests = list(fake.requests) + egress = fake.egress_hosts() + return TurnResult(row=row, rc=proc.returncode, stdout=proc.stdout, stderr=proc.stderr, requests=requests, + egress=egress, own_key=keys[row.key_env or ""], secrets=set(keys.values()), + usage=session_usage(home), canary=canary, wall_s=round(time.monotonic() - started, 1)) + + +def provider_hosts(catalog: list[Row]) -> set[str]: + from urllib.parse import urlsplit + return {urlsplit(r.base_url).hostname or "" for r in catalog if r.base_url.startswith("https://")} - {""} + + +def evaluate(t: TurnResult, catalog: list[Row]) -> dict[str, bool]: + """Every invariant of one row. Relationship checks only — never literals of today's output.""" + inference = [r for r in t.requests if r.method == "POST"] + # A transport mandated by the provider's own host may fall back to chat at a foreign URL. + expected = {t.row.dialect} | ({"chat"} if t.row.host_mandated else set()) + main = [r for r in inference if isinstance(r.body, dict) and r.body.get("tools")] + foreign = t.secrets - {t.own_key} + foreign_hosts = provider_hosts([r for r in catalog if r.name != t.row.name]) - provider_hosts([t.row]) + return { + "turn_completed": t.rc == 0 and FINAL in t.stdout, + "reached_own_endpoint": bool(inference) and all(r.path.startswith(f"/{t.row.name}/") for r in inference), + "dialect_matches_transport": bool(main) and all(r.dialect in expected for r in main), + "tool_round_trip": any(t.canary in json.dumps(r.body) for r in main), + "own_key_in_auth_header": bool(inference) and all( + t.own_key in r.headers.get(AUTH_HEADER_OF_DIALECT.get(r.dialect, "authorization"), "") for r in inference), + "no_foreign_key_on_wire": not any(s in v for r in t.requests for v in r.headers.values() for s in foreign), + # Egress sentinel: nothing may leave for ANOTHER provider's host (CONNECT target). + "no_egress_to_foreign_provider_hosts": not [h for h in t.egress if h.rsplit(":", 1)[0] in foreign_hosts], + "usage_recorded": bool(t.usage) and (t.usage["input_tokens"] or 0) >= USAGE_IN + and (t.usage["output_tokens"] or 0) >= USAGE_OUT, + # Unknown pricing (a model no catalog prices) must be explicit, never a silent $0 estimate. + # (usage_recorded owns absence; this cell judges only a row that exists.) + "cost_not_silent_zero": not t.usage or not ( + (t.usage.get("estimated_cost_usd") in (0, 0.0)) and t.usage.get("cost_status") not in (None, "unknown")), + } + + +@contextlib.contextmanager +def strict_known(pattern: str, reason: str) -> Iterator[None]: + """Strict run-time xfail for a filed bug: an AssertionError whose text matches ``pattern`` XFAILs + the cell; any other failure propagates; a clean pass FAILS, so the fix PR must drop the entry + (the campaign's strict-KNOWN rule, applied to cells whose assertions run after a live wait).""" + try: + yield + except AssertionError as exc: + if not re.search(pattern, str(exc)): + raise + pytest.xfail(f"{reason} [observed: {str(exc).splitlines()[0][:240]}]") + pytest.fail(f"KNOWN bug now fixed — drop its KNOWN entry: {reason}") diff --git a/tests/e2e/core/providers/test_catalog_fallback.py b/tests/e2e/core/providers/test_catalog_fallback.py new file mode 100644 index 0000000000..80764f24de --- /dev/null +++ b/tests/e2e/core/providers/test_catalog_fallback.py @@ -0,0 +1,92 @@ +"""fallback_providers across the discovered catalog: a dead provider falls through to the next. + +Every runnable provider is the PRIMARY of one row, its catalog neighbour the fallback (both at +their own loopback fakes, both configured through the documented ``model.base_url`` / +``fallback_providers[].base_url`` keys). The primary answers HTTP 500 to every inference request. +The oneshot must still deliver the fallback's answer after one tool round trip, the fallback must +be called with ITS OWN key only, and the primary's key must never reach the fallback's endpoint. +""" + +from __future__ import annotations + +import json +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from tests.e2e.core.providers._catalog_helpers import ( + FINAL, URL_SUFFIX_OF_DIALECT, Row, decoy_keys, discover_catalog, run_hermes, write_home, +) +from tests.fakes.providers.catalog_fake import CatalogFake + +CATALOG = discover_catalog() +# Rows whose primary can be redirected; xai cannot (#121347), so it is never a primary/fallback here. +ROWS = [r for r in CATALOG if r.skip_reason() is None and r.name != "xai"] +# Keyed on the FALLBACK provider (pairs follow catalog order, so primaries shift as plugins are +# added). Strict: a listed cell that turns green fails the row until the entry is dropped. +_FB_BASE_URL_IGNORED = "#121359 fallback entry ignores its base_url; goes to the vendor host" +_FB_CELLS = ("fallback_answered", "fallback_tool_round_trip", "fallback_used_own_key") +KNOWN: dict[tuple[str, str], str] = { + **{("anthropic", c): _FB_BASE_URL_IGNORED for c in _FB_CELLS}, + **{("openrouter", c): _FB_BASE_URL_IGNORED for c in _FB_CELLS}, +} + + +def _url(fake: CatalogFake, row: Row) -> str: + return f"{fake.origin}/{row.name}{URL_SUFFIX_OF_DIALECT[row.dialect or 'chat']}" + + +def _drive(primary: Row, fallback: Row, root: Path) -> dict: + project = root / "project" + project.mkdir(parents=True, exist_ok=True) + canary = f"CANARY-FB-{primary.name}" + (project / "canary.txt").write_text(canary + "\n", encoding="utf-8") + keys = decoy_keys(CATALOG) + args = {"path": str(project / "canary.txt")} + with CatalogFake(fail_status=500) as dead, CatalogFake(tool_args=args, final_text=FINAL) as alive: + home = write_home(root, {"provider": primary.name, "base_url": _url(dead, primary)}, { + "fallback_providers": [{"provider": fallback.name, "model": "catalog-model-a", + "base_url": _url(alive, fallback)}]}) + proc = run_hermes(home, project, {**keys, **dead.proxy_env()}, "-z", "Read canary.txt and report.") + dead_reqs, alive_reqs = dead.inference(), alive.inference() + egress = dead.egress_hosts() + pk, fk = keys[primary.key_env or ""], keys[fallback.key_env or ""] + return { + "rc": proc.returncode, "stdout": proc.stdout[-400:], "stderr": proc.stderr[-1200:], + "egress": egress, "dead_paths": [r.path for r in dead_reqs], "alive_paths": [r.path for r in alive_reqs], + "cells": { + "primary_tried_first": bool(dead_reqs) and any(pk in v for v in dead_reqs[0].headers.values()), + "fallback_answered": proc.returncode == 0 and FINAL in proc.stdout, + "fallback_tool_round_trip": any(canary in json.dumps(r.body) for r in alive_reqs), + "fallback_used_own_key": bool(alive_reqs) and all( + any(fk in v for v in r.headers.values()) for r in alive_reqs), + "primary_key_not_sent_to_fallback": pk == fk or not any( + pk in v for r in alive_reqs for v in r.headers.values()), + }, + } + + +@pytest.fixture(scope="module") +def results(tmp_path_factory: pytest.TempPathFactory) -> dict[str, dict]: + pairs = {r.name: (r, ROWS[(i + 1) % len(ROWS)]) for i, r in enumerate(ROWS)} + with ThreadPoolExecutor(max_workers=8, thread_name_prefix="fallback") as pool: + futs = {n: pool.submit(_drive, p, f, Path(tmp_path_factory.mktemp(f"fb-{n}"))) for n, (p, f) in pairs.items()} + out = {n: f.result() for n, f in futs.items()} + for n, (_p, f) in pairs.items(): + out[n]["fallback"] = f.name + return out + + +@pytest.mark.parametrize("row", [pytest.param(r, id=r.name) for r in ROWS]) +def test_dead_primary_falls_through(row: Row, results: dict[str, dict]) -> None: + res = results[row.name] + cells = res["cells"] + known = {c: ref for (p, c), ref in KNOWN.items() if p == res["fallback"]} + fixed = sorted(c for c in known if cells.get(c)) + assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})" + failed = sorted(c for c, ok in cells.items() if not ok and c not in known) + assert not failed, f"{row.name} -> {res['fallback']}: cells red: {failed}\n" + json.dumps( + {k: v for k, v in res.items() if k != "cells"})[:2500] + if known: + pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}") diff --git a/tests/e2e/core/providers/test_catalog_listing.py b/tests/e2e/core/providers/test_catalog_listing.py new file mode 100644 index 0000000000..cef88b6dde --- /dev/null +++ b/tests/e2e/core/providers/test_catalog_listing.py @@ -0,0 +1,120 @@ +"""Model listing and ``/model --provider`` switching per discovered provider. + +Each runnable provider is configured at a custom ``model.base_url`` (a loopback fake serving +provider-unique model ids). Two child processes per row call the real product functions: + +* listing — ``hermes_cli.models.provider_model_ids`` (the catalog the ``/model`` picker renders) + must query the CONFIGURED endpoint when the provider declares a listing endpoint, and never the + provider's canonical host (#120844 class): a relay user's picker must list the relay's models, + and the relay's key must not be addressed to the vendor; +* switch — ``hermes_cli.model_switch.switch_model(explicit_provider=)`` must resolve to that + provider at its configured endpoint, not to an alias on another endpoint (#120295 class). +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from urllib.parse import urlsplit + +import pytest + +from tests.e2e.core.providers._catalog_helpers import ( + URL_SUFFIX_OF_DIALECT, Row, decoy_keys, discover_catalog, hermetic_env, write_home, +) +from tests.fakes.providers.catalog_fake import CatalogFake + +CATALOG = discover_catalog() +ROWS = [r for r in CATALOG if r.skip_reason() is None] +_LISTING_IGNORES_BASE_URL = "#121387 picker listing ignores model.base_url; queries the vendor host" +_LISTING_BROKEN = ( + "ai-gateway", "alibaba", "alibaba-cn", "alibaba-coding-plan", "alibaba-coding-plan-cn", + "alibaba-token-plan", "alibaba-token-plan-cn", "arcee", "commandcode", "commandcode-anthropic", + "deepinfra", "deepseek", "fireworks", "gemini", "gmi", "huggingface", "kilocode", "kimi-coding", + "kimi-coding-cn", "meta-ai", "minimax", "minimax-cn", "nebius-token-factory", "novita", "nvidia", + "ollama-cloud", "opencode-go", "opencode-zen", "openrouter", "router", "stepfun", "upstage", "xai", + "xiaomi", "zai", +) +# Strict: a listed cell that turns green fails the row until its entry is dropped. +KNOWN: dict[tuple[str, str], str] = { + **{(n, "listing_uses_configured_endpoint"): _LISTING_IGNORES_BASE_URL for n in _LISTING_BROKEN}, + ("nebius-token-factory", "switch_resolves_requested_provider"): + "#121388 nebius /model switch validates against the vendor host, ignoring model.base_url", + ("xai", "switch_resolves_requested_provider"): "#121347 xai ignores model.base_url", +} + +_LIST = r""" +import json, sys +from hermes_cli.models import provider_model_ids +try: + print("RESULT=" + json.dumps({"ids": provider_model_ids(sys.argv[1], force_refresh=True)})) +except Exception as exc: + print("RESULT=" + json.dumps({"error": f"{type(exc).__name__}: {exc}"})) +""" +_SWITCH = r""" +import json, sys +from hermes_cli.model_switch import switch_model +from hermes_cli.providers import normalize_provider +r = switch_model(raw_input=sys.argv[2], current_provider="custom", current_model="x", current_base_url="", + current_api_key="", explicit_provider=sys.argv[1]) +print("RESULT=" + json.dumps({"ok": bool(r.success), "provider": r.target_provider, "base_url": r.base_url, + "error": r.error_message, "want": normalize_provider(sys.argv[1]), + "got": normalize_provider(r.target_provider or "")})) +""" + + +def _probe(row: Row, root: Path, script: str) -> dict: + unique = f"catalog-{row.name}-alpha" + keys = decoy_keys(CATALOG) + with CatalogFake(models=[unique, f"catalog-{row.name}-beta"]) as fake: + base = f"{fake.origin}/{row.name}{URL_SUFFIX_OF_DIALECT[row.dialect or 'chat']}" + home = write_home(root, {"provider": row.name, "base_url": base}) + try: + proc = subprocess.run([sys.executable, "-c", script, row.name, unique], cwd=root, capture_output=True, + text=True, env=hermetic_env(home, {**keys, **fake.proxy_env()}), timeout=90, + stdin=subprocess.DEVNULL) + out, err = proc.stdout, proc.stderr + except subprocess.TimeoutExpired: + out, err = "", "TIMEOUT after 90s" + listings, egress = fake.listings(), fake.egress_hosts() + line = next((ln for ln in out.splitlines() if ln.startswith("RESULT=")), None) + res = json.loads(line[7:]) if line else {"error": f"probe crashed: {err[-1200:]}"} + return {**res, "base": base, "listing_paths": [r.path for r in listings], "egress": egress, + "canonical_host_hit": f"{urlsplit(row.base_url).hostname}:443" in egress} + + +@pytest.fixture(scope="module") +def probes(tmp_path_factory: pytest.TempPathFactory) -> dict[tuple[str, str], dict]: + jobs = [(r, kind, script) for r in ROWS for kind, script in (("list", _LIST), ("switch", _SWITCH))] + with ThreadPoolExecutor(max_workers=8, thread_name_prefix="listing") as pool: + futs = {(r.name, kind): pool.submit(_probe, r, Path(tmp_path_factory.mktemp(f"{kind}-{r.name}")), script) + for r, kind, script in jobs} + return {k: f.result() for k, f in futs.items()} + + +def _cells(row: Row, ls: dict, sw: dict) -> dict[str, bool]: + return { + "listing_uses_configured_endpoint": (not row.supports_model_listing) or ( + "error" not in ls and bool(ls["listing_paths"]) + and all(p.startswith(f"/{row.name}/") for p in ls["listing_paths"]) and not ls["canonical_host_hit"]), + # Same provider (alias-normalised: ai-gateway == vercel) at the configured endpoint. + "switch_resolves_requested_provider": bool(sw.get("ok")) and bool(sw.get("want")) and sw.get("got") == sw["want"] + and str(sw.get("base_url") or "").rstrip("/") == sw["base"].rstrip("/"), + } + + +@pytest.mark.parametrize("row", [pytest.param(r, id=r.name) for r in ROWS]) +def test_listing_and_switch(row: Row, probes: dict) -> None: + ls, sw = probes[(row.name, "list")], probes[(row.name, "switch")] + cells = _cells(row, ls, sw) + known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name} + fixed = sorted(c for c in known if cells.get(c)) + assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})" + failed = sorted(c for c, ok in cells.items() if not ok and c not in known) + detail = {"list": {**ls, "ids": (ls.get("ids") or [])[:6]}, "switch": sw} + assert not failed, f"{row.name}: cells red: {failed}\n{json.dumps(detail, default=str)[:2500]}" + if known: + pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}") diff --git a/tests/e2e/core/providers/test_catalog_matrix_0.py b/tests/e2e/core/providers/test_catalog_matrix_0.py new file mode 100644 index 0000000000..164ef54998 --- /dev/null +++ b/tests/e2e/core/providers/test_catalog_matrix_0.py @@ -0,0 +1,79 @@ +"""Provider-catalog matrix, shard 0/3: one real oneshot turn per discovered provider. + +Rows are the providers whose name hashes to this shard (``_catalog_helpers.shard_of``), so a new +plugin joins some shard automatically. Each row runs ``hermes -z`` against its own loopback fake +(redirected via ``model.base_url``) with every other provider's key present as a decoy, and checks: +the turn completes with a tool round trip through the provider's dialect; only the provider's own +key reaches the wire, in the dialect's auth header, and only at its configured endpoint (no egress +to any provider host); usage lands in state.db and unknown pricing is not a silent $0. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from tests.e2e.core.providers._catalog_helpers import ( + SHARDS, Row, TurnResult, discover_catalog, drive_turn, evaluate, shard_of, +) + +SHARD = 0 +# Cells red on origin/main for a tracked open bug: (provider, cell) -> "#issue one line". +# Strict: the row FAILS as soon as a listed cell turns green (drop the entry with the fix), and +# any cell NOT listed still fails normally, so a known bug never masks a new regression. +XAI_BASE_URL_IGNORED = "#121347 xai ignores model.base_url; request + key go to api.x.ai" +KNOWN: dict[tuple[str, str], str] = { + ("xai", "turn_completed"): XAI_BASE_URL_IGNORED, + ("xai", "reached_own_endpoint"): XAI_BASE_URL_IGNORED, + ("xai", "dialect_matches_transport"): XAI_BASE_URL_IGNORED, + ("xai", "tool_round_trip"): XAI_BASE_URL_IGNORED, + ("xai", "own_key_in_auth_header"): XAI_BASE_URL_IGNORED, + ("xai", "usage_recorded"): XAI_BASE_URL_IGNORED, +} + +CATALOG = discover_catalog() +ROWS = [r for r in CATALOG if shard_of(r.name) == SHARD] + + +@pytest.fixture(scope="module") +def turns(tmp_path_factory: pytest.TempPathFactory) -> dict[str, TurnResult]: + """Every runnable row of the shard driven concurrently (own home, fake and process each).""" + runnable = [r for r in ROWS if r.skip_reason() is None] + roots = {r.name: Path(tmp_path_factory.mktemp(f"cat-{r.name}")) for r in runnable} + with ThreadPoolExecutor(max_workers=8, thread_name_prefix="catalog") as pool: + futs = {r.name: pool.submit(drive_turn, r, roots[r.name], CATALOG) for r in runnable} + return {name: f.result() for name, f in futs.items()} + + +def _params() -> list: + out = [] + for r in ROWS: + marks = [pytest.mark.skip(reason=r.skip_reason())] if r.skip_reason() else [] + out.append(pytest.param(r, id=r.name, marks=marks)) + return out + + +@pytest.mark.parametrize("row", _params()) +def test_provider_row(row: Row, turns: dict[str, TurnResult]) -> None: + t = turns[row.name] + cells = evaluate(t, CATALOG) + known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name} + fixed = sorted(c for c in known if cells.get(c)) + assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})" + failed = sorted(c for c, ok in cells.items() if not ok and c not in known) + assert not failed, f"{row.name} ({row.api_mode}/{row.auth_type}): cells red: {failed}\n{t.detail()}" + if known: + pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}") + + +def test_catalog_is_discovered_not_listed() -> None: + """Every bundled plugin dir registers a provider that some shard runs (or skips with a + reason): the matrix follows discovery, so a new plugin can never silently fall out of it.""" + root = Path(__file__).resolve().parents[4] / "plugins" / "model-providers" + dirs = {d.name for d in root.iterdir() if (d / "__init__.py").exists()} + names = {r.name for r in CATALOG} + assert dirs, "no bundled model-provider plugins found" + assert dirs <= names, f"plugin dirs with no discovered profile: {sorted(dirs - names)}" + assert {shard_of(n) for n in names} <= set(range(SHARDS)) diff --git a/tests/e2e/core/providers/test_catalog_matrix_1.py b/tests/e2e/core/providers/test_catalog_matrix_1.py new file mode 100644 index 0000000000..3fb11aeee8 --- /dev/null +++ b/tests/e2e/core/providers/test_catalog_matrix_1.py @@ -0,0 +1,60 @@ +"""Provider-catalog matrix, shard 1/3: one real oneshot turn per discovered provider. + +Rows are the providers whose name hashes to this shard (``_catalog_helpers.shard_of``), so a new +plugin joins some shard automatically. Each row runs ``hermes -z`` against its own loopback fake +(redirected via ``model.base_url``) with every other provider's key present as a decoy, and checks: +the turn completes with a tool round trip through the provider's dialect; only the provider's own +key reaches the wire, in the dialect's auth header, and only at its configured endpoint (no egress +to any provider host); usage lands in state.db and unknown pricing is not a silent $0. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from tests.e2e.core.providers._catalog_helpers import ( + SHARDS, Row, TurnResult, discover_catalog, drive_turn, evaluate, shard_of, +) + +SHARD = 1 +# Cells red on origin/main for a tracked open bug: (provider, cell) -> "#issue one line". +# Strict: the row FAILS as soon as a listed cell turns green (drop the entry with the fix), and +# any cell NOT listed still fails normally, so a known bug never masks a new regression. +KNOWN: dict[tuple[str, str], str] = {} + +CATALOG = discover_catalog() +ROWS = [r for r in CATALOG if shard_of(r.name) == SHARD] + + +@pytest.fixture(scope="module") +def turns(tmp_path_factory: pytest.TempPathFactory) -> dict[str, TurnResult]: + """Every runnable row of the shard driven concurrently (own home, fake and process each).""" + runnable = [r for r in ROWS if r.skip_reason() is None] + roots = {r.name: Path(tmp_path_factory.mktemp(f"cat-{r.name}")) for r in runnable} + with ThreadPoolExecutor(max_workers=8, thread_name_prefix="catalog") as pool: + futs = {r.name: pool.submit(drive_turn, r, roots[r.name], CATALOG) for r in runnable} + return {name: f.result() for name, f in futs.items()} + + +def _params() -> list: + out = [] + for r in ROWS: + marks = [pytest.mark.skip(reason=r.skip_reason())] if r.skip_reason() else [] + out.append(pytest.param(r, id=r.name, marks=marks)) + return out + + +@pytest.mark.parametrize("row", _params()) +def test_provider_row(row: Row, turns: dict[str, TurnResult]) -> None: + t = turns[row.name] + cells = evaluate(t, CATALOG) + known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name} + fixed = sorted(c for c in known if cells.get(c)) + assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})" + failed = sorted(c for c, ok in cells.items() if not ok and c not in known) + assert not failed, f"{row.name} ({row.api_mode}/{row.auth_type}): cells red: {failed}\n{t.detail()}" + if known: + pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}") diff --git a/tests/e2e/core/providers/test_catalog_matrix_2.py b/tests/e2e/core/providers/test_catalog_matrix_2.py new file mode 100644 index 0000000000..1e855c62d5 --- /dev/null +++ b/tests/e2e/core/providers/test_catalog_matrix_2.py @@ -0,0 +1,60 @@ +"""Provider-catalog matrix, shard 2/3: one real oneshot turn per discovered provider. + +Rows are the providers whose name hashes to this shard (``_catalog_helpers.shard_of``), so a new +plugin joins some shard automatically. Each row runs ``hermes -z`` against its own loopback fake +(redirected via ``model.base_url``) with every other provider's key present as a decoy, and checks: +the turn completes with a tool round trip through the provider's dialect; only the provider's own +key reaches the wire, in the dialect's auth header, and only at its configured endpoint (no egress +to any provider host); usage lands in state.db and unknown pricing is not a silent $0. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from tests.e2e.core.providers._catalog_helpers import ( + SHARDS, Row, TurnResult, discover_catalog, drive_turn, evaluate, shard_of, +) + +SHARD = 2 +# Cells red on origin/main for a tracked open bug: (provider, cell) -> "#issue one line". +# Strict: the row FAILS as soon as a listed cell turns green (drop the entry with the fix), and +# any cell NOT listed still fails normally, so a known bug never masks a new regression. +KNOWN: dict[tuple[str, str], str] = {} + +CATALOG = discover_catalog() +ROWS = [r for r in CATALOG if shard_of(r.name) == SHARD] + + +@pytest.fixture(scope="module") +def turns(tmp_path_factory: pytest.TempPathFactory) -> dict[str, TurnResult]: + """Every runnable row of the shard driven concurrently (own home, fake and process each).""" + runnable = [r for r in ROWS if r.skip_reason() is None] + roots = {r.name: Path(tmp_path_factory.mktemp(f"cat-{r.name}")) for r in runnable} + with ThreadPoolExecutor(max_workers=8, thread_name_prefix="catalog") as pool: + futs = {r.name: pool.submit(drive_turn, r, roots[r.name], CATALOG) for r in runnable} + return {name: f.result() for name, f in futs.items()} + + +def _params() -> list: + out = [] + for r in ROWS: + marks = [pytest.mark.skip(reason=r.skip_reason())] if r.skip_reason() else [] + out.append(pytest.param(r, id=r.name, marks=marks)) + return out + + +@pytest.mark.parametrize("row", _params()) +def test_provider_row(row: Row, turns: dict[str, TurnResult]) -> None: + t = turns[row.name] + cells = evaluate(t, CATALOG) + known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name} + fixed = sorted(c for c in known if cells.get(c)) + assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})" + failed = sorted(c for c, ok in cells.items() if not ok and c not in known) + assert not failed, f"{row.name} ({row.api_mode}/{row.auth_type}): cells red: {failed}\n{t.detail()}" + if known: + pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}") diff --git a/tests/e2e/core/providers/test_catalog_oauth.py b/tests/e2e/core/providers/test_catalog_oauth.py index 858938a110..70139e2f9c 100644 --- a/tests/e2e/core/providers/test_catalog_oauth.py +++ b/tests/e2e/core/providers/test_catalog_oauth.py @@ -7,8 +7,8 @@ egress goes through the ``CatalogFake`` sentinel proxy, which refuses and record host. Cells assert user-visible outcomes: the device-code polling cadence the vendor sees, the reply on stdout, the bearer on the next wire request, and the tokens persisted to auth.json. -Open bugs are message-gated run-time xfails (``KNOWN`` + ``known_failure``): a cell XFAILs only -while it fails with that bug's signature and simply passes once the fix lands, in any merge order. +Open bugs are strict, message-gated run-time xfails (``KNOWN`` + ``strict_known``): a cell XFAILs +only while it fails with that bug's signature, and FAILS once the fix lands so the entry is dropped. Not redirectable to a loopback fake, so not covered here (explicit skips below): openai-codex and qwen-oauth refresh (token URLs are module constants, no env/config override) and the Copilot token @@ -27,7 +27,7 @@ from typing import Any import pytest -from tests.e2e.core._pending_fixes import known_failure +from tests.e2e.core.providers._catalog_helpers import strict_known from tests.fakes.providers.catalog_fake import CatalogFake from tests.fakes.providers.catalog_oauth import NOUS_INVOKE_SCOPE, OAuthFake, make_jwt @@ -218,7 +218,7 @@ def test_nous_device_login_slow_down_grows_interval(device_login) -> None: def test_nous_device_login_honors_server_interval(device_login) -> None: gaps = device_login["gaps"] assert len(gaps) == len(DEVICE_SCRIPT), f"unexpected poll count, gaps={gaps}" - with known_failure(*KNOWN["device_interval"]): + with strict_known(*KNOWN["device_interval"]): assert gaps[0] >= DEVICE_INTERVAL - 0.1, ( f"device poll gap {gaps[0]:.2f}s < server interval {DEVICE_INTERVAL}s (gaps={[round(g, 2) for g in gaps]})") @@ -272,7 +272,7 @@ def test_nous_inference_401_refreshes_rotates_and_retries(tmp_path, sentinel) -> info = _describe(proc, fake, sentinel) assert any(r.bearer == revoked for r in fake.inference()), f"the stale bearer was never tried\n{info}" fresh = _assert_rotation_persisted(home, seed, fake, info) - with known_failure(*KNOWN["nous_401_retry_route"]): + with strict_known(*KNOWN["nous_401_retry_route"]): assert not _vendor_egress(sentinel), ( f"401 recovery retry left NOUS_INFERENCE_BASE_URL: egress to {_vendor_egress(sentinel)}\n{info}") assert any(r.bearer == fresh for r in fake.inference()), f"no retry with the refreshed token\n{info}"