ci: gate changed tests against a mutated cost map

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

.github/workflows/test-cost-map-independence.yml

@@ -0,0 +1,86 @@
+name: "Cost map independence"
+
+on: # zizmor: ignore[dangerous-triggers] runs the PR head's code on a read-only token, same as test-linting.yml
+  pull_request:
+    branches:
+      - main
+      - litellm_internal_staging
+      - litellm_oss_staging
+      - "litellm_**"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.sha }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+env:
+  UV_PYTHON: "3.12"
+  UV_CACHE_DIR: "${{ github.workspace }}/.uv-cache"
+  LITELLM_LOCAL_MODEL_COST_MAP: "True"
+
+jobs:
+  run:
+    name: Run cost map mutation gate
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    steps:
+      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4.3.0
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+          fetch-depth: 1
+          clean: true
+          persist-credentials: false
+
+      - name: Fetch gate base (merge-base with target branch)
+        env:
+          GH_TOKEN: ${{ github.token }}
+          BASE_SHA: ${{ github.event.pull_request.base.sha }}
+          HEAD_SHA: ${{ github.event.pull_request.head.sha }}
+        run: |
+          retry() { "$@" || { sleep 15; "$@"; } || { sleep 30; "$@"; }; }
+          MERGE_BASE=$(retry gh api "repos/${{ github.repository }}/compare/${BASE_SHA}...${HEAD_SHA}?per_page=1" --jq '.merge_base_commit.sha')
+          test -n "$MERGE_BASE"
+          retry git fetch --no-tags --depth=1 origin "$MERGE_BASE"
+          echo "GATE_BASE_SHA=$MERGE_BASE" >> "$GITHUB_ENV"
+
+      - name: Set up Python
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
+        with:
+          python-version: ${{ env.UV_PYTHON }}
+
+      - name: Set up uv
+        uses: ./.github/actions/setup-uv-with-retries
+        with:
+          version: "0.10.9"
+
+      - name: Cache uv dependencies
+        uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
+        with:
+          path: ${{ env.UV_CACHE_DIR }}
+          key: ${{ runner.os }}-uv-downloads-py${{ env.UV_PYTHON }}-${{ hashFiles('uv.lock') }}
+          restore-keys: |
+            ${{ runner.os }}-uv-downloads-py${{ env.UV_PYTHON }}-
+
+      - name: Cache the Rust build
+        uses: ./.github/actions/cache-cargo-build
+
+      - name: Install dependencies
+        timeout-minutes: 8
+        run: |
+          .github/scripts/uv_sync_with_retries.sh --frozen --group ci --group proxy-dev --extra google --extra proxy --extra semantic-router --extra saml
+          uv run --no-sync python -c 'import os, sys; print(sys.version); assert f"{sys.version_info.major}.{sys.version_info.minor}" == os.environ["UV_PYTHON"]'
+
+      - name: Cache Prisma binaries
+        uses: ./.github/actions/cache-prisma-binaries
+
+      - name: Generate Prisma client
+        run: |
+          uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma
+
+      - name: Run cost map mutation gate
+        timeout-minutes: 20
+        run: |
+          uv run --no-sync python scripts/cost_map_mutation_gate.py --base "$GATE_BASE_SHA"

CLAUDE.md

@@ -25,7 +25,7 @@ Same thing for bug fixes. The tests should make it so that this specific bug can
 
 Never test structure of code only function of it
 
-A test must only fail when litellm code changes. Never pin facts we don't own (a vendor's price, a third party's field, an upstream default, today's date) as literals or as "X must be absent"; assert the invariant our code guarantees instead, e.g. two rows agree, a value is within range, a field is derived from another. If an outside fact is truly load-bearing, cite its source and date next to the assertion so a reader can tell stale from broken
+A test must only fail when litellm code changes. Never pin facts we don't own (a vendor's price, a third party's field, an upstream default, today's date) as literals or as "X must be absent"; assert the invariant our code guarantees instead, e.g. two rows agree, a value is within range, a field is derived from another. If an outside fact is truly load-bearing, cite its source and date next to the assertion so a reader can tell stale from broken. `uv run python scripts/cost_map_mutation_gate.py --base origin/main` runs your changed test files against a cost map with every price, limit and deprecation date rewritten, which is what the provider sync does; CI runs the same gate
 
 `tests/test_litellm/` mirrors `litellm/` in a parallel path (see `tests/test_litellm/readme.md`). Name tests `test_<filename>.py`, but always match the existing test file in the directory you touch — many provider dirs use longer descriptive names (e.g. `test_anthropic_chat_transformation.py`) to avoid ambiguity across sibling folders. For bug fixes, extend the existing mapped test file rather than creating a new one. Only create a new test file for a new feature (provider, endpoint, or transformation module) that has no mapped test yet, following that directory's naming convention (or `test_<filename>.py` if you're the first test there). One focused regression test beats many shallow ones
 

scripts/cost_map_mutation_gate.py

@@ -0,0 +1,222 @@
+#!/usr/bin/env python3
+"""Gate: run changed tests/test_litellm files against a mutated cost map.
+
+The provider sync rewrites prices, context limits and deprecation dates in
+model_prices_and_context_window.json whenever a vendor changes them. A test that
+pins any of those values breaks on the next sync even though no litellm code
+changed. This gate applies one combined mutation to every cost-map entry the
+same way the audit did (prices x1.37, deprecation_date set, max_* limits +1000),
+writes both JSON copies, runs the changed test files, and restores the files
+from git afterwards. A red run means a test asserts a vendor fact instead of a
+litellm-owned invariant.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import signal
+import subprocess
+import sys
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from types import FrameType
+from typing import Final, NamedTuple
+
+from pydantic import TypeAdapter
+
+REPO_ROOT: Final = Path(__file__).resolve().parent.parent
+COST_MAP_PATHS: Final = (
+    "model_prices_and_context_window.json",
+    "litellm/model_prices_and_context_window_backup.json",
+)
+TERMINATION_SIGNALS: Final = (signal.SIGTERM, signal.SIGHUP)
+
+PRICE_MULTIPLIER: Final = 1.37
+DEPRECATION_DATE: Final = "2030-01-01"
+LIMIT_BUMP: Final = 1_000
+LIMIT_FIELDS: Final = frozenset({"max_tokens", "max_input_tokens", "max_output_tokens"})
+
+_COST_MAP_ADAPTER: Final = TypeAdapter(dict[str, object])
+_MODEL_ENTRY_ADAPTER: Final = TypeAdapter(dict[str, object])
+_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
+
+
+class _Args(NamedTuple):
+    base: str | None
+    paths: tuple[str, ...]
+    pytest_args: tuple[str, ...]
+
+
+def _exit_on_termination(signum: int, _frame: FrameType | None) -> None:
+    raise SystemExit(128 + signum)
+
+
+def _install_termination_handlers() -> None:
+    for termination in TERMINATION_SIGNALS:
+        if signal.getsignal(termination) == signal.SIG_DFL:
+            signal.signal(termination, _exit_on_termination)
+
+
+def _run(cmd: Sequence[str], cwd: Path = REPO_ROOT) -> str:
+    proc: Final = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
+    if proc.returncode != 0:
+        sys.stderr.write(proc.stderr)
+        raise SystemExit(f"{cmd[0]} exited {proc.returncode}")
+    return proc.stdout
+
+
+def _cost_map_is_dirty() -> bool:
+    status: Final = _run(["git", "status", "--porcelain", "--", *COST_MAP_PATHS])
+    return bool(status.strip())
+
+
+def _changed_test_files(base: str) -> tuple[str, ...]:
+    out: Final = _run(
+        [
+            "git",
+            "diff",
+            "--name-only",
+            "--diff-filter=ACMR",
+            base,
+            "HEAD",
+            "--",
+            ":(glob)tests/test_litellm/**/*.py",
+        ]
+    )
+    return tuple(
+        line
+        for line in out.splitlines()
+        if line.startswith("tests/test_litellm/") and line.endswith(".py") and Path(line).name != "conftest.py"
+    )
+
+
+def _mutate_value(key: str, value: object, scale_numbers: bool = False) -> object:
+    inside_cost: Final = scale_numbers or "cost" in key
+    if isinstance(value, dict):
+        mapping: Final = _MODEL_ENTRY_ADAPTER.validate_python(value)
+        return {k: _mutate_value(k, v, inside_cost) for k, v in mapping.items()}
+    if isinstance(value, list):
+        items: Final = _OBJECT_LIST_ADAPTER.validate_python(value)
+        return [_mutate_value(key, v, inside_cost) for v in items]
+    if inside_cost and isinstance(value, (int, float)) and not isinstance(value, bool):
+        return value * PRICE_MULTIPLIER
+    return value
+
+
+def mutate_entry(entry: Mapping[str, object]) -> dict[str, object]:
+    return {
+        key: (
+            value + LIMIT_BUMP
+            if key in LIMIT_FIELDS and isinstance(value, int) and not isinstance(value, bool)
+            else _mutate_value(key, value)
+        )
+        for key, value in {**entry, "deprecation_date": DEPRECATION_DATE}.items()
+    }
+
+
+def mutate_cost_map(cost_map: Mapping[str, object]) -> dict[str, object]:
+    return {
+        key: (
+            mutate_entry(_MODEL_ENTRY_ADAPTER.validate_python(value))
+            if isinstance(value, dict) and "litellm_provider" in value
+            else value
+        )
+        for key, value in cost_map.items()
+    }
+
+
+def _serialize(cost_map: Mapping[str, object]) -> str:
+    return json.dumps(cost_map, indent=4, ensure_ascii=False) + "\n"
+
+
+def _mutated_text(path: Path) -> str:
+    original: Final = path.read_text()
+    cost_map: Final = _COST_MAP_ADAPTER.validate_python(json.loads(original))
+    return _serialize(mutate_cost_map(cost_map))
+
+
+def _restore_cost_map_files() -> None:
+    subprocess.run(["git", "checkout", "--", *COST_MAP_PATHS], cwd=REPO_ROOT, check=False)
+
+
+def _pytest_command(files: Sequence[str], extra_args: Sequence[str]) -> list[str]:
+    forwarded: Final = tuple(extra_args)
+    workers: Final = (
+        ()
+        if any(arg == "-n" or arg.startswith("-n=") or arg.startswith("-nauto") for arg in forwarded)
+        else ("-n", "4")
+    )
+    return [
+        "uv",
+        "run",
+        "--no-sync",
+        "pytest",
+        *files,
+        "-q",
+        "-p",
+        "no:cacheprovider",
+        "-p",
+        "no:randomly",
+        *workers,
+        *forwarded,
+    ]
+
+
+def _parse_args(argv: Sequence[str]) -> _Args:
+    parser: Final = argparse.ArgumentParser(
+        description="Run changed tests/test_litellm files against a mutated cost map",
+        epilog="extra arguments after -- are passed to pytest",
+    )
+    parser.add_argument("--base", help="git ref to diff against for changed-test selection")
+    parser.add_argument("paths", nargs="*", help="explicit test paths (overrides --base selection)")
+    argv_tuple: Final = tuple(argv)
+    before, after = (
+        (argv_tuple[: argv_tuple.index("--")], argv_tuple[argv_tuple.index("--") + 1 :])
+        if "--" in argv_tuple
+        else (argv_tuple, ())
+    )
+    args: Final = parser.parse_args(before)
+    return _Args(
+        base=args.base,  # pyright: ignore[reportAny]  # argparse Namespace attributes are untyped
+        paths=tuple(args.paths),  # pyright: ignore[reportAny]  # argparse Namespace attributes are untyped
+        pytest_args=tuple(after),
+    )
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    _install_termination_handlers()
+    args: Final = _parse_args(tuple(argv) if argv is not None else tuple(sys.argv[1:]))
+
+    files: Final = args.paths or (_changed_test_files(args.base) if args.base else ())
+    if not files:
+        sys.stdout.write("No tests/test_litellm files selected; nothing to gate.\n")
+        return 0
+    if _cost_map_is_dirty():
+        sys.stderr.write(
+            "Refusing to run: model_prices_and_context_window.json or its litellm/ backup "
+            "has uncommitted changes. Commit or restore them first.\n"
+        )
+        return 2
+
+    mutated_by_path: Final = tuple((REPO_ROOT / path, _mutated_text(REPO_ROOT / path)) for path in COST_MAP_PATHS)
+    for path, text in mutated_by_path:
+        path.write_text(text)
+    try:
+        env: Final = {**os.environ, "LITELLM_LOCAL_MODEL_COST_MAP": "True"}
+        proc: Final = subprocess.run(_pytest_command(files, args.pytest_args), cwd=REPO_ROOT, env=env)
+        if proc.returncode != 0:
+            sys.stderr.write(
+                "\nCost-map mutation gate failed: the failing assertions pin cost-map values "
+                "the provider sync rewrites (prices, limits, deprecation dates). Derive the "
+                "expected value from the entry the code selects (litellm.model_cost / "
+                "get_model_info) or replace the assertion with an invariant our code owns.\n"
+            )
+        return proc.returncode
+    finally:
+        _restore_cost_map_files()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_litellm/test_cost_map_mutation_gate.py

@@ -0,0 +1,121 @@
+"""Unit tests for scripts/cost_map_mutation_gate.py."""
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from types import ModuleType
+from typing import Final
+
+import pytest
+
+ROOT: Final = Path(__file__).resolve().parents[2]
+GATE_PATH: Final = ROOT / "scripts" / "cost_map_mutation_gate.py"
+
+
+def _load() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("cost_map_mutation_gate", GATE_PATH)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules["cost_map_mutation_gate"] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+gate: Final = _load()
+
+
+def _entry() -> dict[str, object]:
+    return {
+        "input_cost_per_token": 1e-06,
+        "output_cost_per_token": 2e-06,
+        "litellm_provider": "openrouter",
+        "mode": "chat",
+        "max_tokens": 4096,
+        "max_input_tokens": 3000,
+        "max_output_tokens": 1000,
+        "search_context_cost_per_query": {"search_context_size_low": 0.01},
+        "tiered": [{"input_cost_per_token": 5e-06}],
+        "supports_vision": True,
+    }
+
+
+BASE_MAP: Final = {
+    "sample_spec": {"input_cost_per_token": "USD per prompt token"},
+    "fallback_generalizations": {"rules": [{"name": "r", "pattern": "^x"}]},
+    "openrouter/a": _entry(),
+}
+
+
+def test_mutation_scales_cost_fields_including_nested() -> None:
+    mutated: Final = gate.mutate_cost_map(BASE_MAP)
+    entry: Final = mutated["openrouter/a"]
+    assert entry["input_cost_per_token"] == pytest.approx(1e-06 * 1.37)
+    assert entry["output_cost_per_token"] == pytest.approx(2e-06 * 1.37)
+    assert entry["search_context_cost_per_query"]["search_context_size_low"] == pytest.approx(0.01 * 1.37)
+    assert entry["tiered"][0]["input_cost_per_token"] == pytest.approx(5e-06 * 1.37)
+
+
+def test_mutation_adds_deprecation_date_and_bumps_limits() -> None:
+    mutated: Final = gate.mutate_cost_map(BASE_MAP)
+    entry: Final = mutated["openrouter/a"]
+    assert entry["deprecation_date"] == "2030-01-01"
+    assert entry["max_tokens"] == 4096 + 1000
+    assert entry["max_input_tokens"] == 3000 + 1000
+    assert entry["max_output_tokens"] == 1000 + 1000
+    assert entry["supports_vision"] is True
+    assert entry["mode"] == "chat"
+
+
+def test_mutation_leaves_non_model_root_keys_untouched() -> None:
+    mutated: Final = gate.mutate_cost_map(BASE_MAP)
+    assert mutated["sample_spec"] == BASE_MAP["sample_spec"]
+    assert mutated["fallback_generalizations"] == BASE_MAP["fallback_generalizations"]
+
+
+def test_mutation_preserves_key_order() -> None:
+    assert tuple(gate.mutate_cost_map(BASE_MAP)) == tuple(BASE_MAP)
+
+
+def test_changed_test_files_filters_conftest(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        gate,
+        "_run",
+        lambda cmd, cwd=gate.REPO_ROOT: (
+            "tests/test_litellm/test_a.py\n"
+            "tests/test_litellm/conftest.py\n"
+            "tests/test_litellm/llms/conftest.py\n"
+            "tests/test_litellm/llms/test_b.py\n"
+            "litellm/utils.py\n"
+        ),
+    )
+    assert gate._changed_test_files("BASE") == (
+        "tests/test_litellm/test_a.py",
+        "tests/test_litellm/llms/test_b.py",
+    )
+
+
+def test_dirty_cost_map_refuses(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
+    monkeypatch.setattr(gate, "_run", lambda cmd, cwd=gate.REPO_ROOT: " M model_prices_and_context_window.json\n")
+    assert gate.main(["tests/test_litellm/test_a.py"]) == 2
+    assert "Refusing to run" in capsys.readouterr().err
+
+
+def test_no_files_selected_exits_zero(capsys: pytest.CaptureFixture[str]) -> None:
+    assert gate.main([]) == 0
+    assert "nothing to gate" in capsys.readouterr().out
+
+
+def test_pytest_command_adds_workers_only_when_absent() -> None:
+    without_n: Final = gate._pytest_command(("a.py",), ())
+    assert "-n" in without_n and without_n[without_n.index("-n") + 1] == "4"
+    with_n: Final = gate._pytest_command(("a.py",), ("-n", "8"))
+    assert list(with_n).count("-n") == 1 and with_n[with_n.index("-n") + 1] == "8"
+
+
+def test_serialized_mutation_round_trips() -> None:
+    text: Final = gate._serialize(gate.mutate_cost_map(BASE_MAP))
+    parsed: Final = json.loads(text)
+    assert parsed["openrouter/a"]["deprecation_date"] == "2030-01-01"
+    assert parsed["sample_spec"] == BASE_MAP["sample_spec"]
+    assert text.endswith("\n")