test: delete unit-test assertions that pin cost-map prices, limits and deprecation dates

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

@@ -0,0 +1,85 @@
+name: "Cost map independence"
+
+on:
+  pull_request:
+    branches:
+      - main
+      - litellm_internal_staging
+      - litellm_oss_staging
+      - "litellm_**"
+
+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/local_testing/test_completion_cost.py

@@ -5,7 +5,7 @@
 
 import asyncio
 import time
-from typing import Optional
+from typing import Final, Optional
 from unittest.mock import AsyncMock, MagicMock, patch
 import base64
 import pytest
@@ -685,8 +685,10 @@ def test_vertex_ai_claude_completion_cost():
         completion_response=response,
         messages=[{"role": "user", "content": "Hey, how's it going?"}],
     )
-    predicted_cost = input_tokens * 0.000003 + 0.000015 * output_tokens
-    assert cost == predicted_cost
+    model_info: Final = litellm.model_cost["vertex_ai/claude-3-sonnet@20240229"]
+    assert model_info["input_cost_per_token"] > 0
+    assert model_info["output_cost_per_token"] > 0
+    assert cost > 0
 
 
 def test_vertex_ai_embedding_completion_cost(caplog):

tests/test_litellm/batches/test_batch_utils.py

@@ -15,6 +15,7 @@
 """
 
 import json
+from typing import Final
 import logging
 from types import MappingProxyType
 
@@ -1670,8 +1671,10 @@ async def fake_fetch(batch: object, custom_llm_provider: str, litellm_params: di
     )
 
     assert (result.usage.prompt_tokens, result.usage.completion_tokens, result.usage.total_tokens) == (1800, 1000, 2800)
-    # 3e-06 / 1.5e-05 on-demand, halved for batch.
-    assert result.cost == pytest.approx(1800 * 3e-06 / 2 + 1000 * 1.5e-05 / 2)
+    entry: Final = litellm.model_cost["global.anthropic.claude-sonnet-4-6"]
+    assert result.cost == pytest.approx(
+        1800 * entry["input_cost_per_token"] / 2 + 1000 * entry["output_cost_per_token"] / 2
+    )
 
     # The response model alone cannot price a bedrock batch: this is the $0 bug.
     zero_result = await bu._handle_completed_batch(

tests/test_litellm/containers/test_container_transformation.py

@@ -387,4 +387,3 @@ def test_container_create_response_includes_cost(self, monkeypatch):
         ]
 
         assert actual_cost == expected_cost
-        assert actual_cost == 0.03  # OpenAI code interpreter costs $0.03 per session