mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-01 07:03:02 +00:00
test(e2e): share the CLI version rule, env lockdown and rate limiter in tests/e2e/cli_harness
Moves the npm release rule (now taking the package name), the env allowlist plus fresh HOME, and the per-upstream rate limiter out of claude_code so the Codex suite can reuse them. Claude Code behavior is unchanged, its CI, cron and Dockerfile paths now call the shared resolver with @anthropic-ai/claude-code
This commit is contained in:
ryan-crabbe-berri
parent
e6b4774ac6
commit
7f83ad7028
21 changed files with 473 additions and 706 deletions
4
.github/workflows/compat-matrix-image.yml
vendored
4
.github/workflows/compat-matrix-image.yml
vendored
|
|
@ -4,7 +4,7 @@ on:
|
|||
pull_request:
|
||||
paths:
|
||||
- tests/e2e/claude_code/cron_vm/**
|
||||
- tests/e2e/claude_code/pr_gate_version_resolver.py
|
||||
- tests/e2e/cli_harness/npm_version_resolver.py
|
||||
- .github/workflows/compat-matrix-image.yml
|
||||
workflow_dispatch:
|
||||
|
||||
|
|
@ -36,7 +36,7 @@ jobs:
|
|||
whoami
|
||||
gh --version
|
||||
uv --version
|
||||
version="$(uv run --no-project --python 3.12 python /opt/litellm/tests/e2e/claude_code/pr_gate_version_resolver.py)"
|
||||
version="$(uv run --no-project --python 3.12 python /opt/litellm/tests/e2e/cli_harness/npm_version_resolver.py @anthropic-ai/claude-code)"
|
||||
/opt/litellm/tests/e2e/claude_code/cron_vm/install_claude_code.sh "${version}" /tmp/claude-cli
|
||||
/tmp/claude-cli/claude --version
|
||||
'
|
||||
|
|
|
|||
2
.github/workflows/test-linting.yml
vendored
2
.github/workflows/test-linting.yml
vendored
|
|
@ -208,7 +208,7 @@ jobs:
|
|||
exit 0
|
||||
fi
|
||||
retry() { "$@" || { sleep 15; "$@"; } || { sleep 30; "$@"; }; }
|
||||
CLAUDE_VERSION="$(retry uv run --no-sync python tests/e2e/claude_code/pr_gate_version_resolver.py)"
|
||||
CLAUDE_VERSION="$(retry uv run --no-sync python tests/e2e/cli_harness/npm_version_resolver.py @anthropic-ai/claude-code)"
|
||||
tests/e2e/claude_code/cron_vm/install_claude_code.sh "$CLAUDE_VERSION" "$RUNNER_TEMP/claude-cli"
|
||||
PATH="$RUNNER_TEMP/claude-cli:$PATH" uv run --no-sync pytest -q tests/e2e_harness
|
||||
uv run --no-sync pytest --tb=short -q --noconftest -o addopts= -o "pythonpath=tests/e2e tests/e2e_harness" -p no:rerunfailures \
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ raw HTTP client imports (requests, urllib.request, httpx, aiohttp, http.client)
|
|||
banned in suite code, and tests/e2e_harness, which tests that harness, is held to the
|
||||
same ban. Importing requests' exception types for catching is fine anywhere; a small
|
||||
allowlist grandfathers the files that legitimately make raw calls (the transport
|
||||
itself, the root conftest liveness probe, the claude_code version resolver's constant
|
||||
itself, the root conftest liveness probe, the cli_harness npm version resolver's constant
|
||||
registry URL fetch, and the mcp OAuth client, whose httpx client is the object the
|
||||
official mcp SDK's streamable_http_client requires and so cannot go through the sync
|
||||
requests transport). Referenced by tests/e2e/AGENTS.md."""
|
||||
|
|
@ -24,7 +24,7 @@ BANNED_MODULES = ("requests", "urllib.request", "http.client", "httpx", "aiohttp
|
|||
ALLOWED_RAW_CLIENT_FILES = {
|
||||
"e2e/e2e_http.py": ("requests",),
|
||||
"e2e/conftest.py": ("requests",),
|
||||
"e2e/claude_code/pr_gate_version_resolver.py": ("urllib.request",),
|
||||
"e2e/cli_harness/npm_version_resolver.py": ("urllib.request",),
|
||||
}
|
||||
|
||||
EXCEPTION_ONLY_NAMES = frozenset({"RequestException", "ConnectionError", "Timeout", "HTTPError"})
|
||||
|
|
|
|||
|
|
@ -49,6 +49,7 @@ Each subdirectory under `tests/e2e/` is one suite, scoped to an endpoint family
|
|||
- `secret_manager/` - the gateway's `key_management_system` against a real secret manager: deployment keys resolved from it (`os.environ/<name>` where the name exists only in the manager) and virtual keys written to and deleted from it. The tests are backend-agnostic and each backend is its own lane, because the setting is global to the proxy: `E2E_SECRET_MANAGER=<system>` opts in and picks the backend from `secret_backends.BACKENDS`, the proxy is booted from `gateway/secret_manager_<system>_ci_config.yml` against the live manager, and the tests reach that manager through the backend's `SecretStore` (`secret_store_<system>.py`). A test needing something not every backend does carries `requires_capability(...)` and is deselected on lanes that lack it. `secret_manager/backend.sh up <system>` runs a backend in Docker and writes the proxy's and the tests' env. Marked `secret_manager`, deselected unless `E2E_SECRET_MANAGER` is set, and kept out of the per-PR selector. Backends today: `hashicorp_vault` and `cyberark` (CyberArk Conjur, which cannot delete, so the delete test is Vault-only)
|
||||
- `gateway/` - proxy configuration only (`litellm-config.yml`). No tests
|
||||
- `claude_code/` - the Claude Code compatibility matrix: drives the real `claude` CLI (and HTTP probes) against a proxy for each feature x provider cell, reporting tagged-union outcomes via the `compat_result` fixture. Ships its own driver/builder/publisher, covered by `tests/e2e_harness/claude_code/`. The HTTP probes ride the shared transport (`ProxyClient.count_tokens` / `ProxyClient.messages`). The CLI-driving path stays bespoke
|
||||
- `cli_harness/` - no tests, only the plumbing every suite that drives a real coding-agent CLI shares: `npm_version_resolver.py` picks the CLI release to install (the newest npm release at least three days old, run by CI as `python tests/e2e/cli_harness/npm_version_resolver.py <npm package>`, stdlib only), `lockdown.py` gives the CLI subprocess an allowlisted env and a fresh empty HOME so it never sees the runner's credentials or dotfiles, and `rate_limiter.py` is the cross-process token bucket per upstream that keeps parallel CLI runs under one request rate. Covered by `tests/e2e_harness/cli_harness/`. Code only one CLI suite needs stays in that suite
|
||||
- `ui/` - the Admin UI browser suite: Playwright in TypeScript, driving the dashboard served by a live proxy on port 4000 (seeded postgres + mock LLM upstream, see its `run_e2e.sh`). It is a self-contained npm package with its own lockfile and does not use the Python harness, pytest markers, or the shared transport. The Python rules in this file (typed models, `Result` unions, basedpyright zero-error gate) do not apply inside it. Its only Python file, `fixtures/mock_llm_server/server.py`, is excluded from the e2e basedpyright gate via the root `pyrightconfig.json`. Its specs follow the dashboard's rule against testing what shadcn and Base UI already guarantee, in `ui/litellm-dashboard/AGENTS.md`
|
||||
|
||||
## MCP suite: real Datadog only
|
||||
|
|
|
|||
|
|
@ -16,7 +16,6 @@ from __future__ import annotations
|
|||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
|
@ -25,14 +24,10 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
|
|||
from dataclasses import dataclass, field
|
||||
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, Union
|
||||
|
||||
from cli_harness.lockdown import isolated_home, locked_down_env
|
||||
from cli_harness.rate_limiter import RateBucket, RateLimiter, get_default_limiter
|
||||
from e2e_metadata import step
|
||||
|
||||
from claude_code.rate_limiter import (
|
||||
RateLimiter,
|
||||
get_default_limiter,
|
||||
infer_provider,
|
||||
)
|
||||
|
||||
CLAUDE_CLI_DEFAULT = "claude"
|
||||
# 120s is fine for a single isolated CLI call against an unloaded
|
||||
# upstream, but the matrix run launches up to 75 concurrent calls and
|
||||
|
|
@ -77,61 +72,35 @@ Anthropic / Bedrock / Vertex) to reset. Both knobs are env-tunable so
|
|||
a matrix run can trade wall time for resilience without code edits;
|
||||
retries=0 disables the behavior entirely."""
|
||||
|
||||
# Env vars the `claude` Node CLI legitimately needs to function:
|
||||
# locating its own binary + node, basic locale/terminal plumbing.
|
||||
# Deliberately excludes every credential-bearing var that the
|
||||
# surrounding CI job sets for the proxy (ANTHROPIC_API_KEY, AWS_*,
|
||||
# AZURE_*, VERTEXAI_CREDENTIALS, GITHUB_TOKEN, OPENAI_API_KEY,
|
||||
# DATABASE_URL, ...). The CLI talks to the proxy via the explicit
|
||||
# ANTHROPIC_BASE_URL/ANTHROPIC_AUTH_TOKEN we set below — it has no
|
||||
# business reading the proxy's upstream credentials, and a
|
||||
# compromised CLI release shouldn't be able to exfiltrate them out
|
||||
# of the CI environment.
|
||||
#
|
||||
# `HOME` is intentionally NOT in this list: see `_make_isolated_home`
|
||||
# below. We give the CLI a fresh empty per-invocation HOME so a
|
||||
# compromised claude package or a model-directed `Read` tool call
|
||||
# can't reach files like `~/.config/gh/hosts.yml`, `~/.ssh/id_rsa`,
|
||||
# or `~/.bash_history` on the cron VM (and on the CircleCI executor
|
||||
# the same isolation prevents accidentally exposing checkout-adjacent
|
||||
# files even though the runner home is ephemeral there).
|
||||
_CLI_ENV_ALLOWLIST: tuple = (
|
||||
"PATH",
|
||||
"USER",
|
||||
"LOGNAME",
|
||||
"SHELL",
|
||||
"TERM",
|
||||
"TMPDIR",
|
||||
"LANG",
|
||||
"LC_ALL",
|
||||
"LC_CTYPE",
|
||||
"NODE_PATH",
|
||||
"NVM_DIR",
|
||||
"NVM_BIN",
|
||||
)
|
||||
def infer_provider(model: str) -> RateBucket:
|
||||
"""Map a model alias to its compat-matrix rate bucket.
|
||||
|
||||
|
||||
def _make_isolated_home() -> str:
|
||||
"""Create a fresh empty HOME directory for a single `claude` subprocess.
|
||||
|
||||
The CLI needs *a* writable HOME (it caches per-session state under
|
||||
`$HOME/.claude/projects/<sha>/`), but it has no legitimate need
|
||||
for the *user's* HOME. Handing it the real one means a compromised
|
||||
`@anthropic-ai/claude-code` release, or a model-directed `Read`
|
||||
tool call during a PDF/vision cell, can read host files like
|
||||
`~/.config/gh/hosts.yml` (GitHub CLI host token), `~/.ssh/`,
|
||||
`~/.bash_history`, or any other dotfile under the runtime user's
|
||||
home. On the cron VM the runtime user is a real interactive
|
||||
account (`mateo`) with a populated home directory, so this is a
|
||||
real exfiltration surface.
|
||||
|
||||
The directory is created under `tempfile.gettempdir()` (which is
|
||||
`/tmp` on Linux; under systemd's `PrivateTmp=true` that's a
|
||||
per-service tmpfs that the service user can't otherwise reach).
|
||||
Caller is responsible for `shutil.rmtree`-ing it after the
|
||||
subprocess exits.
|
||||
The matrix column layout is fixed; aliases registered in the proxy
|
||||
encode the provider via a suffix (`-bedrock-converse`,
|
||||
`-bedrock-invoke`, `-azure`, `-vertex`, `-openai`, `-azure-openai`,
|
||||
`-bedrock-mantle`) or its absence (Anthropic). Order matters:
|
||||
`-azure-openai` also ends with `-openai`, and the bedrock suffixes
|
||||
all contain `bedrock`, so the more-specific suffixes are tested
|
||||
first.
|
||||
"""
|
||||
return tempfile.mkdtemp(prefix="claude-cli-home-")
|
||||
if not model:
|
||||
raise ValueError("model must be a non-empty string")
|
||||
lower = model.lower()
|
||||
if lower.endswith("-azure-openai"):
|
||||
return RateBucket.AZURE_OPENAI
|
||||
if lower.endswith("-openai"):
|
||||
return RateBucket.OPENAI
|
||||
if lower.endswith("-bedrock-mantle"):
|
||||
return RateBucket.BEDROCK_MANTLE
|
||||
if lower.endswith("-bedrock-converse"):
|
||||
return RateBucket.BEDROCK_CONVERSE
|
||||
if lower.endswith("-bedrock-invoke"):
|
||||
return RateBucket.BEDROCK_INVOKE
|
||||
if lower.endswith("-azure"):
|
||||
return RateBucket.AZURE
|
||||
if lower.endswith("-vertex"):
|
||||
return RateBucket.VERTEX_AI
|
||||
return RateBucket.ANTHROPIC
|
||||
|
||||
|
||||
_FIXED_CLI_USER_ID = "0" * 64
|
||||
|
|
@ -176,7 +145,7 @@ def _stable_cli_state() -> Tuple[str, str]:
|
|||
Pinning both makes the bodies repeatable across builds.
|
||||
|
||||
This narrows what survives rather than widening it: HOME stays fresh and
|
||||
empty per invocation, so the isolation `_make_isolated_home` describes is
|
||||
empty per invocation, so the isolation `cli_harness.lockdown` describes is
|
||||
unchanged, and the CLI's own state no longer outlives the pod either. The
|
||||
working directory is deliberately not the checkout, so a model-directed
|
||||
`Read` sees an empty directory instead of the repository.
|
||||
|
|
@ -291,38 +260,26 @@ def run_claude(
|
|||
cmd.append("--")
|
||||
cmd.append(prompt)
|
||||
|
||||
# Build a minimal env for the CLI subprocess: only the allowlisted
|
||||
# process-runtime vars from os.environ, plus the explicit proxy
|
||||
# creds, plus any caller-supplied overrides. See _CLI_ENV_ALLOWLIST
|
||||
# above for the security rationale.
|
||||
env: Dict[str, str] = {
|
||||
key: os.environ[key] for key in _CLI_ENV_ALLOWLIST if key in os.environ
|
||||
}
|
||||
env["ANTHROPIC_BASE_URL"] = base_url
|
||||
env["ANTHROPIC_AUTH_TOKEN"] = api_key
|
||||
env["DISABLE_AUTOUPDATER"] = "1"
|
||||
# Hand the CLI a fresh empty HOME so a compromised claude package
|
||||
# or a model-directed Read tool call can't see the runtime user's
|
||||
# real dotfiles. Created here, removed in the `finally` below
|
||||
# regardless of how the subprocess exits.
|
||||
isolated_home = _make_isolated_home()
|
||||
env["HOME"] = isolated_home
|
||||
config_dir, workspace = _stable_cli_state()
|
||||
env["CLAUDE_CONFIG_DIR"] = config_dir
|
||||
if extra_env:
|
||||
env.update(extra_env)
|
||||
|
||||
# Throttle by provider *before* launching the CLI. Doing this here
|
||||
# (rather than per-test) means every code path that lands on
|
||||
# `run_claude` is rate-limited automatically — including
|
||||
# `run_claude_models_parallel`, which is the hot path during the
|
||||
# full matrix run.
|
||||
limiter = rate_limiter if rate_limiter is not None else get_default_limiter()
|
||||
provider = infer_provider(model)
|
||||
limiter.acquire(provider)
|
||||
|
||||
run_fn = runner or subprocess.run
|
||||
try:
|
||||
with isolated_home("claude-cli-home-") as home:
|
||||
config_dir, workspace = _stable_cli_state()
|
||||
env = locked_down_env(
|
||||
home=home,
|
||||
overrides={
|
||||
"ANTHROPIC_BASE_URL": base_url,
|
||||
"ANTHROPIC_AUTH_TOKEN": api_key,
|
||||
"DISABLE_AUTOUPDATER": "1",
|
||||
"CLAUDE_CONFIG_DIR": config_dir,
|
||||
**(extra_env or {}),
|
||||
},
|
||||
)
|
||||
limiter.acquire(infer_provider(model))
|
||||
try:
|
||||
completed = run_fn(
|
||||
cmd,
|
||||
|
|
@ -340,15 +297,6 @@ def run_claude(
|
|||
) from exc
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise ClaudeCLIError(f"claude CLI timed out after {timeout}s") from exc
|
||||
finally:
|
||||
# Best-effort cleanup. If the subprocess wrote a `.claude/`
|
||||
# session dir under the isolated HOME, we remove it here so
|
||||
# parallel matrix runs don't accumulate per-call tmpdirs.
|
||||
# `ignore_errors=True` because rmtree races with any
|
||||
# not-yet-reaped child (SIGTERM'd `claude` on a host-side
|
||||
# timeout) are benign — the next matrix run starts from a
|
||||
# fresh tmpdir anyway.
|
||||
shutil.rmtree(isolated_home, ignore_errors=True)
|
||||
|
||||
events = _parse_stream_json(completed.stdout or "")
|
||||
text = _extract_assistant_text(events)
|
||||
|
|
|
|||
|
|
@ -52,7 +52,7 @@ than as a GitHub Action or on a dedicated VM. Trade-offs:
|
|||
the stable venv doesn't install) is never loaded. The tag's own
|
||||
`tests/e2e/` is deliberately not used.
|
||||
3. **Resolves and installs the Claude Code CLI under test**:
|
||||
`pr_gate_version_resolver.py` (run on the venv, the image has no
|
||||
`tests/e2e/cli_harness/npm_version_resolver.py` (run on the venv, the image has no
|
||||
Python of its own) picks the newest `@anthropic-ai/claude-code` npm
|
||||
release published at least 3 days ago, the same buffer the PR gate
|
||||
uses, unless `CLAUDE_CODE_VERSION` pins one, and
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@
|
|||
# Releases API.
|
||||
# 2. Update a long-lived worktree at $WORKTREE to that tag and `uv sync` it.
|
||||
# 3. Resolve the Claude Code CLI version under test (the newest npm
|
||||
# release at least 3 days old, via pr_gate_version_resolver.py, or
|
||||
# release at least 3 days old, via cli_harness/npm_version_resolver.py, or
|
||||
# $CLAUDE_CODE_VERSION when set) and download that release's
|
||||
# linux-x64 binary into the run's scratch dir, checksum-verified
|
||||
# against the vendor's release manifest (install_claude_code.sh).
|
||||
|
|
@ -329,7 +329,7 @@ PROXY_CONFIG="${WORKTREE}/tests/e2e/claude_code/test_config.yaml"
|
|||
if [[ -z "${CLAUDE_CODE_VERSION}" ]]; then
|
||||
CLAUDE_CODE_VERSION="$(
|
||||
cd "${WORKTREE}" \
|
||||
&& "${WORKTREE_UV}" run --no-sync python "${POPULATOR_DIR}/../pr_gate_version_resolver.py"
|
||||
&& "${WORKTREE_UV}" run --no-sync python "${POPULATOR_DIR}/../../cli_harness/npm_version_resolver.py" @anthropic-ai/claude-code
|
||||
)" || die "could not resolve the Claude Code version to test"
|
||||
log "resolved claude code: ${CLAUDE_CODE_VERSION}"
|
||||
else
|
||||
|
|
|
|||
|
|
@ -31,8 +31,7 @@ from __future__ import annotations
|
|||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from cli_harness.rate_limiter import RateLimiter, get_default_limiter
|
||||
from e2e_http import (
|
||||
NetworkError,
|
||||
RateLimitedError,
|
||||
|
|
@ -59,12 +58,9 @@ from models import (
|
|||
JsonSchemaProperty,
|
||||
ToolInputSchema,
|
||||
)
|
||||
from pydantic import BaseModel
|
||||
|
||||
from claude_code.rate_limiter import (
|
||||
RateLimiter,
|
||||
get_default_limiter,
|
||||
infer_provider,
|
||||
)
|
||||
from claude_code.cli_driver import infer_provider
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from proxy_client import ProxyClient
|
||||
|
|
|
|||
|
|
@ -1,155 +0,0 @@
|
|||
"""Claude Code PR-Gate Version Resolver.
|
||||
|
||||
Resolves the `@anthropic-ai/claude-code` npm version that the PR-gate CI
|
||||
job installs. Selects the newest version (by publish timestamp) whose
|
||||
publish timestamp is at least 3 days old. The 3-day window is a security
|
||||
review buffer — see PRD #26476, "Version resolvers".
|
||||
|
||||
Two surfaces:
|
||||
|
||||
- ``resolve_pr_gate_version(...)`` — the importable function. Accepts
|
||||
pre-fetched npm metadata (for unit tests) or a custom ``fetcher``
|
||||
callable. The default fetcher hits the public npm registry.
|
||||
- ``python -m claude_code.pr_gate_version_resolver`` — prints the
|
||||
resolved version string to stdout, suitable for piping into a shell
|
||||
``$(...)`` substitution inside the CircleCI job.
|
||||
|
||||
The CLI form is what CircleCI runs at job start; engineers reading the
|
||||
job log can see the selected version on a single line above the
|
||||
``npm install -g`` step (acceptance criterion: "the selected Claude
|
||||
Code version is logged in the CI output").
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import urllib.request
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Callable, Mapping, Optional
|
||||
|
||||
PACKAGE_NAME = "@anthropic-ai/claude-code"
|
||||
NPM_REGISTRY_URL = "https://registry.npmjs.org/{package}"
|
||||
DEFAULT_MIN_AGE = timedelta(days=3)
|
||||
DEFAULT_FETCH_TIMEOUT_SECONDS = 30
|
||||
|
||||
# npm's `time` map mixes per-version timestamps with these meta keys.
|
||||
_TIME_META_KEYS = frozenset({"created", "modified"})
|
||||
|
||||
|
||||
class NoEligibleVersionError(RuntimeError):
|
||||
"""Raised when no version in the npm metadata satisfies the min-age cutoff."""
|
||||
|
||||
|
||||
def _parse_npm_timestamp(value: str) -> datetime:
|
||||
"""Parse the ISO-8601 timestamps npm emits (always UTC, may use ``Z``)."""
|
||||
if value.endswith("Z"):
|
||||
value = value[:-1] + "+00:00"
|
||||
parsed = datetime.fromisoformat(value)
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||||
return parsed
|
||||
|
||||
|
||||
def _default_fetcher(package_name: str) -> dict:
|
||||
"""Fetch the npm packument for ``package_name`` over HTTPS.
|
||||
|
||||
Uses urllib (stdlib) so this module has no extra dependencies in the
|
||||
CI environment. Returns the raw JSON dict.
|
||||
"""
|
||||
# urllib.parse.quote would encode the leading '@' / '/' which the
|
||||
# npm registry expects literally; do a minimal hand-roll instead.
|
||||
url = NPM_REGISTRY_URL.format(package=package_name.replace("/", "%2F"))
|
||||
req = urllib.request.Request(url, headers={"Accept": "application/json"})
|
||||
with urllib.request.urlopen( # noqa: S310 — registry URL is constant
|
||||
req, timeout=DEFAULT_FETCH_TIMEOUT_SECONDS
|
||||
) as response:
|
||||
body = response.read().decode("utf-8")
|
||||
return json.loads(body)
|
||||
|
||||
|
||||
def resolve_pr_gate_version(
|
||||
*,
|
||||
metadata: Optional[Mapping] = None,
|
||||
fetcher: Optional[Callable[[str], Mapping]] = None,
|
||||
as_of: Optional[datetime] = None,
|
||||
min_age: timedelta = DEFAULT_MIN_AGE,
|
||||
package_name: str = PACKAGE_NAME,
|
||||
) -> str:
|
||||
"""Return the newest npm version of ``package_name`` published >= ``min_age`` ago.
|
||||
|
||||
"Newest" means newest by **publish time**, not semver string order —
|
||||
if a patch lands on an older major after a newer release, the
|
||||
patched line is the eligible one. A version npm has unpublished keeps
|
||||
its ``time`` entry but drops out of ``versions``, so only versions
|
||||
still present in ``versions`` are candidates.
|
||||
|
||||
Args:
|
||||
metadata: Pre-fetched npm packument (skips the HTTP call). Useful
|
||||
for unit tests.
|
||||
fetcher: Callable taking a package name and returning the
|
||||
packument. Defaults to a stdlib HTTPS fetcher.
|
||||
as_of: The clock used to decide whether a version is "old
|
||||
enough". Defaults to ``datetime.now(timezone.utc)``.
|
||||
min_age: Minimum publish age. Defaults to 3 days.
|
||||
package_name: Defaults to ``@anthropic-ai/claude-code``.
|
||||
|
||||
Raises:
|
||||
NoEligibleVersionError: when no version in the registry meets
|
||||
the age cutoff.
|
||||
"""
|
||||
if metadata is None:
|
||||
fetch = fetcher or _default_fetcher
|
||||
metadata = fetch(package_name)
|
||||
|
||||
times = metadata.get("time") or {}
|
||||
versions = metadata.get("versions") or {}
|
||||
if as_of is None:
|
||||
as_of = datetime.now(timezone.utc)
|
||||
cutoff = as_of - min_age
|
||||
|
||||
eligible: list[tuple[datetime, str]] = []
|
||||
for version, raw_ts in times.items():
|
||||
if version in _TIME_META_KEYS:
|
||||
continue
|
||||
if version not in versions:
|
||||
continue
|
||||
if not isinstance(raw_ts, str):
|
||||
continue
|
||||
if "-" in version:
|
||||
continue
|
||||
published = _parse_npm_timestamp(raw_ts)
|
||||
if published <= cutoff:
|
||||
eligible.append((published, version))
|
||||
|
||||
if not eligible:
|
||||
raise NoEligibleVersionError(
|
||||
f"no version of {package_name} is at least {min_age} old "
|
||||
f"as of {as_of.isoformat()}"
|
||||
)
|
||||
|
||||
eligible.sort(key=lambda pair: pair[0], reverse=True)
|
||||
return eligible[0][1]
|
||||
|
||||
|
||||
def _main(argv: list[str]) -> int:
|
||||
"""Print the resolved version to stdout. Exit code 0 on success.
|
||||
|
||||
Stderr carries the human-readable announcement so the version can be
|
||||
captured cleanly with ``$(python -m ...)`` in shell.
|
||||
"""
|
||||
try:
|
||||
version = resolve_pr_gate_version()
|
||||
except Exception as exc: # noqa: BLE001 — CLI surface, want everything
|
||||
print(f"pr_gate_version_resolver: {exc}", file=sys.stderr) # noqa: T201
|
||||
return 1
|
||||
print( # noqa: T201
|
||||
f"pr_gate_version_resolver: selected {PACKAGE_NAME}@{version}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
print(version) # noqa: T201
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(_main(sys.argv[1:]))
|
||||
|
|
@ -1,378 +0,0 @@
|
|||
"""Cross-process token-bucket rate limiter for the Claude Code compat suite.
|
||||
|
||||
The compat matrix runs 75 live `claude` CLI invocations (25 cells × 3
|
||||
Claude tiers per cell). When pytest-xdist fans these out across worker
|
||||
processes, each worker would maintain its own in-memory rate limiter
|
||||
and the *aggregate* request rate hitting any one upstream provider
|
||||
would be `workers × per-worker rate` — exactly the situation that
|
||||
trips Anthropic / Azure / Bedrock / Vertex 429s in the middle of a
|
||||
matrix run and silently flips green cells red.
|
||||
|
||||
The fix is a **shared, cross-process** token bucket per provider,
|
||||
backed by a small JSON state file and an OS-level `flock`. Each
|
||||
`run_claude` invocation acquires one token (sleeping if the bucket is
|
||||
empty) before launching the CLI; refills happen lazily based on wall
|
||||
time, so workers can be killed and restarted without losing or
|
||||
double-spending budget.
|
||||
|
||||
Configuration is driven entirely by environment variables so a
|
||||
binary-search workflow can shift per-provider rates without code
|
||||
edits:
|
||||
|
||||
LITELLM_COMPAT_RATE_ANTHROPIC (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_AZURE (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_VERTEX_AI (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_BEDROCK_CONVERSE (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_BEDROCK_INVOKE (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_OPENAI (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_AZURE_OPENAI (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_BEDROCK_MANTLE (req/s, default 5.0)
|
||||
LITELLM_COMPAT_RATE_BURST (per-bucket burst override;
|
||||
default = rate)
|
||||
LITELLM_COMPAT_RATE_STATE_DIR (state file directory;
|
||||
default = $TMPDIR/litellm-claude-compat-ratelimit)
|
||||
|
||||
A rate of 0 (or any non-positive value) disables throttling for that
|
||||
provider — useful when you trust the upstream to handle the burst or
|
||||
when running unit-test-shaped workloads that never actually hit the
|
||||
network.
|
||||
|
||||
The provider id is inferred from the model id by `infer_provider`,
|
||||
mirroring the matrix's column layout (`anthropic`, `azure`,
|
||||
`vertex_ai`, `bedrock_converse`, `bedrock_invoke`, `openai`,
|
||||
`azure_openai`, `bedrock_mantle`). The `vertex_ai_gpt` matrix column
|
||||
has no bucket: its cells are static not_applicable stubs that never
|
||||
reach the network.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Dict, Iterator, Mapping, Optional
|
||||
|
||||
# `fcntl` is POSIX-only; the suite is Linux/macOS only, so we don't
|
||||
# attempt a Windows fallback. Importing at module load fails fast on
|
||||
# the (currently non-existent) Windows runner so we don't silently
|
||||
# degrade to no-locking behavior.
|
||||
import fcntl
|
||||
|
||||
PROVIDER_ANTHROPIC = "anthropic"
|
||||
PROVIDER_AZURE = "azure"
|
||||
PROVIDER_VERTEX_AI = "vertex_ai"
|
||||
PROVIDER_BEDROCK_CONVERSE = "bedrock_converse"
|
||||
PROVIDER_BEDROCK_INVOKE = "bedrock_invoke"
|
||||
PROVIDER_OPENAI = "openai"
|
||||
PROVIDER_AZURE_OPENAI = "azure_openai"
|
||||
PROVIDER_BEDROCK_MANTLE = "bedrock_mantle"
|
||||
|
||||
ALL_PROVIDERS = (
|
||||
PROVIDER_ANTHROPIC,
|
||||
PROVIDER_AZURE,
|
||||
PROVIDER_VERTEX_AI,
|
||||
PROVIDER_BEDROCK_CONVERSE,
|
||||
PROVIDER_BEDROCK_INVOKE,
|
||||
PROVIDER_OPENAI,
|
||||
PROVIDER_AZURE_OPENAI,
|
||||
PROVIDER_BEDROCK_MANTLE,
|
||||
)
|
||||
|
||||
DEFAULT_RATE = 5.0 # req/s per provider, conservative starting point
|
||||
RATE_ENV_PREFIX = "LITELLM_COMPAT_RATE_"
|
||||
BURST_ENV = "LITELLM_COMPAT_RATE_BURST"
|
||||
STATE_DIR_ENV = "LITELLM_COMPAT_RATE_STATE_DIR"
|
||||
DEFAULT_STATE_DIR_NAME = "litellm-claude-compat-ratelimit"
|
||||
|
||||
|
||||
def infer_provider(model: str) -> str:
|
||||
"""Map a model alias to its compat-matrix provider id.
|
||||
|
||||
The matrix column layout is fixed; aliases registered in the proxy
|
||||
encode the provider via a suffix (`-bedrock-converse`,
|
||||
`-bedrock-invoke`, `-azure`, `-vertex`, `-openai`, `-azure-openai`,
|
||||
`-bedrock-mantle`) or its absence (Anthropic). Order matters:
|
||||
`-azure-openai` also ends with `-openai`, and the bedrock suffixes
|
||||
all contain `bedrock`, so the more-specific suffixes are tested
|
||||
first.
|
||||
"""
|
||||
if not model:
|
||||
raise ValueError("model must be a non-empty string")
|
||||
lower = model.lower()
|
||||
if lower.endswith("-azure-openai"):
|
||||
return PROVIDER_AZURE_OPENAI
|
||||
if lower.endswith("-openai"):
|
||||
return PROVIDER_OPENAI
|
||||
if lower.endswith("-bedrock-mantle"):
|
||||
return PROVIDER_BEDROCK_MANTLE
|
||||
if lower.endswith("-bedrock-converse"):
|
||||
return PROVIDER_BEDROCK_CONVERSE
|
||||
if lower.endswith("-bedrock-invoke"):
|
||||
return PROVIDER_BEDROCK_INVOKE
|
||||
if lower.endswith("-azure"):
|
||||
return PROVIDER_AZURE
|
||||
if lower.endswith("-vertex"):
|
||||
return PROVIDER_VERTEX_AI
|
||||
return PROVIDER_ANTHROPIC
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProviderConfig:
|
||||
"""Static config snapshot for one provider's bucket.
|
||||
|
||||
Captured up front (rather than re-read per acquire) so the limiter's
|
||||
behavior in a single process is stable even if env vars are mutated
|
||||
mid-run. A fresh `RateLimiter` picks up env changes on construction.
|
||||
"""
|
||||
|
||||
rate_per_sec: float
|
||||
burst: float
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
return self.rate_per_sec > 0 and self.burst > 0
|
||||
|
||||
|
||||
def load_config(
|
||||
env: Optional[Mapping[str, str]] = None,
|
||||
) -> Dict[str, ProviderConfig]:
|
||||
"""Build a {provider: ProviderConfig} from env, applying defaults.
|
||||
|
||||
Parsing failures fall back to the default rate rather than
|
||||
crashing the test session — a typo in `LITELLM_COMPAT_RATE_AZURE`
|
||||
should not silently disable throttling, but it also shouldn't
|
||||
abort 75 live tests with a `ValueError` ten minutes in.
|
||||
"""
|
||||
src = env if env is not None else os.environ
|
||||
|
||||
def _as_float(value: Optional[str], default: float) -> float:
|
||||
if value is None or value == "":
|
||||
return default
|
||||
try:
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
burst_override = _as_float(src.get(BURST_ENV), -1.0)
|
||||
|
||||
out: Dict[str, ProviderConfig] = {}
|
||||
for provider in ALL_PROVIDERS:
|
||||
env_key = RATE_ENV_PREFIX + provider.upper()
|
||||
rate = _as_float(src.get(env_key), DEFAULT_RATE)
|
||||
burst = burst_override if burst_override > 0 else max(rate, 1.0)
|
||||
out[provider] = ProviderConfig(rate_per_sec=rate, burst=burst)
|
||||
return out
|
||||
|
||||
|
||||
def _state_dir(env: Optional[Mapping[str, str]] = None) -> Path:
|
||||
"""Resolve the directory holding per-provider state files.
|
||||
|
||||
A user-supplied `LITELLM_COMPAT_RATE_STATE_DIR` wins for tests and
|
||||
container environments where `$TMPDIR` may be ephemeral or shared
|
||||
in surprising ways. The directory is created lazily with
|
||||
`parents=True, exist_ok=True` so first-run setup needs no
|
||||
fixture wiring.
|
||||
"""
|
||||
src = env if env is not None else os.environ
|
||||
explicit = src.get(STATE_DIR_ENV)
|
||||
if explicit:
|
||||
return Path(explicit)
|
||||
return Path(tempfile.gettempdir()) / DEFAULT_STATE_DIR_NAME
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
"""Cross-process token bucket per provider.
|
||||
|
||||
Each `acquire(provider)` call:
|
||||
1. Opens (creating if needed) `<state_dir>/<provider>.json`.
|
||||
2. Holds an exclusive `flock` while reading + updating the
|
||||
{tokens, last_refill} state.
|
||||
3. Refills tokens based on `now - last_refill`, capped at burst.
|
||||
4. If tokens >= 1, subtracts one and returns immediately.
|
||||
5. Otherwise, computes the wall-time delay needed to earn a
|
||||
single token at the configured rate, releases the lock, and
|
||||
sleeps. After sleeping it retries — staying under the lock
|
||||
while sleeping would serialize all workers behind whichever
|
||||
one held it longest.
|
||||
|
||||
This bounds the *aggregate* req/s seen by the upstream, regardless
|
||||
of how many xdist workers, threads, or processes are concurrently
|
||||
running tests against the same provider.
|
||||
|
||||
A `_clock` / `_sleep` injection seam keeps the unit tests fast and
|
||||
deterministic; production callers should never override either.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config: Optional[Mapping[str, ProviderConfig]] = None,
|
||||
state_dir: Optional[Path] = None,
|
||||
clock: Optional[callable] = None,
|
||||
sleep: Optional[callable] = None,
|
||||
) -> None:
|
||||
self._config = dict(config) if config is not None else load_config()
|
||||
self._state_dir = Path(state_dir) if state_dir is not None else _state_dir()
|
||||
self._clock = clock or time.monotonic
|
||||
self._sleep = sleep or time.sleep
|
||||
# `_state_dir.mkdir` once on construction is fine; concurrent
|
||||
# workers all racing to create the same directory is benign.
|
||||
self._state_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def acquire(self, provider: str) -> float:
|
||||
"""Block until one token is available for `provider`.
|
||||
|
||||
Returns the cumulative wall-time spent waiting (0.0 when the
|
||||
bucket had budget and we returned immediately). Callers can
|
||||
log this to attribute slow cells to throttling vs. upstream
|
||||
latency — same role `DriverResult.duration_ms` plays for the
|
||||
actual CLI invocation.
|
||||
"""
|
||||
cfg = self._config.get(provider)
|
||||
if cfg is None or not cfg.enabled:
|
||||
return 0.0
|
||||
|
||||
path = self._state_path(provider)
|
||||
total_waited = 0.0
|
||||
while True:
|
||||
now = self._clock()
|
||||
sleep_for = self._try_consume(path, cfg, now)
|
||||
if sleep_for <= 0:
|
||||
return total_waited
|
||||
self._sleep(sleep_for)
|
||||
total_waited += sleep_for
|
||||
|
||||
def _state_path(self, provider: str) -> Path:
|
||||
return self._state_dir / f"{provider}.json"
|
||||
|
||||
def _try_consume(self, path: Path, cfg: ProviderConfig, now: float) -> float:
|
||||
"""Atomically refill and try to take one token.
|
||||
|
||||
Returns 0.0 if a token was consumed, or a positive sleep
|
||||
duration (seconds) if the caller must wait before retrying.
|
||||
|
||||
We hold an exclusive `flock` only across the read-modify-write
|
||||
of the JSON state — never across a `sleep` — so workers don't
|
||||
serialize while one of them is parked.
|
||||
"""
|
||||
# `os.open` + `os.O_CREAT | os.O_RDWR` gives us a fd we can
|
||||
# both lock and read/write through. Opening with `"a+"` then
|
||||
# seeking is equivalent but uglier; this version is closer to
|
||||
# the canonical flock recipe.
|
||||
fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600)
|
||||
try:
|
||||
fcntl.flock(fd, fcntl.LOCK_EX)
|
||||
try:
|
||||
tokens, last_refill = self._read_state(fd, cfg, now)
|
||||
tokens, last_refill = self._refill(tokens, last_refill, now, cfg)
|
||||
if tokens >= 1.0:
|
||||
tokens -= 1.0
|
||||
self._write_state(fd, tokens, last_refill)
|
||||
return 0.0
|
||||
# Not enough budget. Persist the refilled state so a
|
||||
# subsequent caller doesn't have to redo the math, then
|
||||
# release the lock and tell the caller how long to
|
||||
# sleep before retrying.
|
||||
self._write_state(fd, tokens, last_refill)
|
||||
deficit = 1.0 - tokens
|
||||
# `deficit / rate` seconds will earn exactly enough
|
||||
# for one token. Add a tiny safety margin so we don't
|
||||
# wake up nanoseconds early and spin.
|
||||
return (deficit / cfg.rate_per_sec) + 1e-3
|
||||
finally:
|
||||
fcntl.flock(fd, fcntl.LOCK_UN)
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
@staticmethod
|
||||
def _read_state(fd: int, cfg: ProviderConfig, now: float) -> tuple:
|
||||
"""Read {tokens, last_refill} from `fd`, defaulting to a full
|
||||
bucket on a missing/empty/corrupt file.
|
||||
|
||||
New files start full so the very first request never waits;
|
||||
corrupt files are treated like new files because the
|
||||
alternative — refusing to run — is worse than briefly
|
||||
over-spending one bucket's worth of budget.
|
||||
"""
|
||||
os.lseek(fd, 0, os.SEEK_SET)
|
||||
raw = os.read(fd, 4096).decode("utf-8")
|
||||
if not raw.strip():
|
||||
return cfg.burst, now
|
||||
try:
|
||||
obj = json.loads(raw)
|
||||
tokens = float(obj.get("tokens", cfg.burst))
|
||||
last_refill = float(obj.get("last_refill", now))
|
||||
return tokens, last_refill
|
||||
except (ValueError, TypeError):
|
||||
return cfg.burst, now
|
||||
|
||||
@staticmethod
|
||||
def _refill(
|
||||
tokens: float, last_refill: float, now: float, cfg: ProviderConfig
|
||||
) -> tuple:
|
||||
"""Apply elapsed time to the bucket, capped at burst.
|
||||
|
||||
Negative elapsed (clock went backward, e.g. across host
|
||||
sleep/resume or a manually-tweaked monotonic mock) is clamped
|
||||
to zero so we never *remove* tokens.
|
||||
"""
|
||||
elapsed = max(0.0, now - last_refill)
|
||||
tokens = min(cfg.burst, tokens + elapsed * cfg.rate_per_sec)
|
||||
return tokens, now
|
||||
|
||||
@staticmethod
|
||||
def _write_state(fd: int, tokens: float, last_refill: float) -> None:
|
||||
payload = json.dumps({"tokens": tokens, "last_refill": last_refill}).encode(
|
||||
"utf-8"
|
||||
)
|
||||
os.lseek(fd, 0, os.SEEK_SET)
|
||||
os.ftruncate(fd, 0)
|
||||
os.write(fd, payload)
|
||||
|
||||
|
||||
# A single process-wide instance is all we need: provider-keyed state
|
||||
# is stored in files, so multiple `RateLimiter` instances would just
|
||||
# duplicate the in-process bookkeeping. We expose a getter rather than
|
||||
# the instance directly so unit tests can install a custom limiter
|
||||
# scoped to a tmp directory without monkeypatching globals.
|
||||
_default: Optional[RateLimiter] = None
|
||||
|
||||
|
||||
def get_default_limiter() -> RateLimiter:
|
||||
global _default
|
||||
if _default is None:
|
||||
_default = RateLimiter()
|
||||
return _default
|
||||
|
||||
|
||||
def reset_default_limiter() -> None:
|
||||
"""Drop the cached default limiter; the next `get_default_limiter`
|
||||
call rebuilds it from the current environment.
|
||||
|
||||
Useful between unit tests that patch env vars: without this they'd
|
||||
keep reading the stale config snapshot from the first call.
|
||||
"""
|
||||
global _default
|
||||
_default = None
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def use_limiter(limiter: RateLimiter) -> Iterator[RateLimiter]:
|
||||
"""Temporarily install `limiter` as the process default.
|
||||
|
||||
The driver's `run_claude` calls `get_default_limiter()`; tests that
|
||||
want a controlled tmp-dir-backed limiter use this contextmanager
|
||||
to swap one in without touching env vars or the on-disk default
|
||||
state.
|
||||
"""
|
||||
global _default
|
||||
previous = _default
|
||||
_default = limiter
|
||||
try:
|
||||
yield limiter
|
||||
finally:
|
||||
_default = previous
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
#!/usr/bin/env bash
|
||||
# Run the Claude Code compat matrix end-to-end against a live LiteLLM
|
||||
# proxy, with per-provider rate limits applied via the cross-process
|
||||
# token bucket in `tests/e2e/claude_code/rate_limiter.py`.
|
||||
# token bucket in `tests/e2e/cli_harness/rate_limiter.py`.
|
||||
#
|
||||
# Designed for binary-searching the ideal X / Y / Z req/s per provider:
|
||||
# 1. Pick an initial rate (e.g. 5/s for everyone).
|
||||
|
|
|
|||
0
tests/e2e/cli_harness/__init__.py
Normal file
0
tests/e2e/cli_harness/__init__.py
Normal file
49
tests/e2e/cli_harness/lockdown.py
Normal file
49
tests/e2e/cli_harness/lockdown.py
Normal file
|
|
@ -0,0 +1,49 @@
|
|||
"""What a coding-agent CLI under test may see of the host.
|
||||
|
||||
The CI job holds the proxy's upstream credentials (provider keys, AWS, Azure, the database URL, GitHub tokens) in its
|
||||
environment, and the cron VM runs as a real user with a populated home. A compromised CLI release, or a model told to
|
||||
read a file, must reach neither, so the CLI gets only the allowlisted runtime variables plus what its suite passes in,
|
||||
and a fresh empty HOME that is deleted when it exits.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
from collections.abc import Generator, Mapping
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
CLI_ENV_ALLOWLIST: Final = (
|
||||
"PATH",
|
||||
"USER",
|
||||
"LOGNAME",
|
||||
"SHELL",
|
||||
"TERM",
|
||||
"TMPDIR",
|
||||
"LANG",
|
||||
"LC_ALL",
|
||||
"LC_CTYPE",
|
||||
"NODE_PATH",
|
||||
"NVM_DIR",
|
||||
"NVM_BIN",
|
||||
)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def isolated_home(prefix: str) -> Generator[Path]:
|
||||
home: Final = Path(tempfile.mkdtemp(prefix=prefix))
|
||||
try:
|
||||
yield home
|
||||
finally:
|
||||
shutil.rmtree(home, ignore_errors=True)
|
||||
|
||||
|
||||
def locked_down_env(
|
||||
*, home: Path, overrides: Mapping[str, str], source: Mapping[str, str] = os.environ
|
||||
) -> Mapping[str, str]:
|
||||
allowed: Final = {name: source[name] for name in CLI_ENV_ALLOWLIST if name in source}
|
||||
return MappingProxyType({**allowed, "HOME": str(home), **overrides})
|
||||
91
tests/e2e/cli_harness/npm_version_resolver.py
Normal file
91
tests/e2e/cli_harness/npm_version_resolver.py
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
"""The npm release of a coding-agent CLI that CI installs: the newest one published at least three days ago.
|
||||
|
||||
The three days are a security review buffer (PRD #26476, "Version resolvers"), so a release that gets pulled or
|
||||
patched within days never drives a run. Newest means by publish time, not semver order.
|
||||
|
||||
`python tests/e2e/cli_harness/npm_version_resolver.py <npm package>` prints the version on stdout for a shell
|
||||
`$(...)`. It is stdlib only because the cron image runs it without the project's dependencies.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import BinaryIO, Final, cast
|
||||
|
||||
MIN_AGE: Final = timedelta(days=3)
|
||||
_REGISTRY_URL: Final = "https://registry.npmjs.org/{package}"
|
||||
_FETCH_TIMEOUT_SECONDS: Final = 30
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Packument:
|
||||
published_at: Mapping[str, datetime]
|
||||
versions: frozenset[str]
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, raw: object) -> Packument:
|
||||
document: Final = _as_mapping(raw)
|
||||
times: Final = _as_mapping(document.get("time"))
|
||||
return cls(
|
||||
published_at={version: _parse_utc(stamp) for version, stamp in _string_values(times)},
|
||||
versions=frozenset(_as_mapping(document.get("versions"))),
|
||||
)
|
||||
|
||||
|
||||
def newest_version_at_least(packument: Packument, *, min_age: timedelta, as_of: datetime) -> str | None:
|
||||
cutoff: Final = as_of - min_age
|
||||
eligible: Final = (
|
||||
(published, version)
|
||||
for version, published in packument.published_at.items()
|
||||
if version in packument.versions and "-" not in version and published <= cutoff
|
||||
)
|
||||
newest: Final = max(eligible, default=None)
|
||||
return None if newest is None else newest[1]
|
||||
|
||||
|
||||
def fetch_packument(package: str) -> Packument:
|
||||
url: Final = _REGISTRY_URL.format(package=urllib.parse.quote(package, safe="@"))
|
||||
request: Final = urllib.request.Request(url, headers={"Accept": "application/json"})
|
||||
with cast(BinaryIO, urllib.request.urlopen(request, timeout=_FETCH_TIMEOUT_SECONDS)) as response:
|
||||
return Packument.from_json(cast(object, json.loads(response.read())))
|
||||
|
||||
|
||||
def _as_mapping(raw: object) -> Mapping[str, object]:
|
||||
if not isinstance(raw, dict):
|
||||
return {}
|
||||
return {str(key): value for key, value in cast(dict[object, object], raw).items()}
|
||||
|
||||
|
||||
def _string_values(raw: Mapping[str, object]) -> tuple[tuple[str, str], ...]:
|
||||
return tuple((key, value) for key, value in raw.items() if isinstance(value, str))
|
||||
|
||||
|
||||
def _parse_utc(stamp: str) -> datetime:
|
||||
parsed: Final = datetime.fromisoformat(stamp.replace("Z", "+00:00"))
|
||||
return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _main(argv: list[str]) -> int:
|
||||
if len(argv) != 1:
|
||||
print("usage: npm_version_resolver.py <npm package>", file=sys.stderr) # noqa: T201 # CLI surface
|
||||
return 2
|
||||
package: Final = argv[0]
|
||||
version: Final = newest_version_at_least(
|
||||
fetch_packument(package), min_age=MIN_AGE, as_of=datetime.now(timezone.utc)
|
||||
)
|
||||
if version is None:
|
||||
print(f"npm_version_resolver: no release of {package} is at least {MIN_AGE} old", file=sys.stderr) # noqa: T201 # CLI surface
|
||||
return 1
|
||||
print(f"npm_version_resolver: selected {package}@{version}", file=sys.stderr) # noqa: T201 # CLI surface
|
||||
print(version) # noqa: T201 # CLI surface
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(_main(sys.argv[1:]))
|
||||
151
tests/e2e/cli_harness/rate_limiter.py
Normal file
151
tests/e2e/cli_harness/rate_limiter.py
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
"""A token bucket per upstream, shared by every process on the host, so parallel CLI suites and xdist workers
|
||||
together stay under one request rate per provider instead of one rate each.
|
||||
|
||||
Each bucket is a small JSON file read and written under an exclusive `flock`, in `LITELLM_COMPAT_RATE_STATE_DIR`
|
||||
(default: a fixed folder under `$TMPDIR`). `LITELLM_COMPAT_RATE_<BUCKET>` sets a bucket's req/s (default 5, 0 or less
|
||||
turns it off) and `LITELLM_COMPAT_RATE_BURST` overrides every bucket's burst (default: its rate).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
from collections.abc import Callable, Iterator, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from enum import StrEnum
|
||||
from functools import cache
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, ValidationError
|
||||
|
||||
DEFAULT_RATE: Final = 5.0
|
||||
RATE_ENV_PREFIX: Final = "LITELLM_COMPAT_RATE_"
|
||||
BURST_ENV: Final = "LITELLM_COMPAT_RATE_BURST"
|
||||
STATE_DIR_ENV: Final = "LITELLM_COMPAT_RATE_STATE_DIR"
|
||||
DEFAULT_STATE_DIR_NAME: Final = "litellm-claude-compat-ratelimit"
|
||||
_WAKE_MARGIN_SECONDS: Final = 1e-3
|
||||
|
||||
|
||||
class RateBucket(StrEnum):
|
||||
ANTHROPIC = "anthropic"
|
||||
AZURE = "azure"
|
||||
VERTEX_AI = "vertex_ai"
|
||||
BEDROCK_CONVERSE = "bedrock_converse"
|
||||
BEDROCK_INVOKE = "bedrock_invoke"
|
||||
OPENAI = "openai"
|
||||
AZURE_OPENAI = "azure_openai"
|
||||
BEDROCK_MANTLE = "bedrock_mantle"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ProviderConfig:
|
||||
rate_per_sec: float
|
||||
burst: float
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
return self.rate_per_sec > 0 and self.burst > 0
|
||||
|
||||
|
||||
class _BucketState(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
tokens: float
|
||||
last_refill: float
|
||||
|
||||
|
||||
def load_config(env: Mapping[str, str] = os.environ) -> Mapping[RateBucket, ProviderConfig]:
|
||||
burst_override: Final = _as_float(env.get(BURST_ENV), -1.0)
|
||||
return MappingProxyType(
|
||||
{
|
||||
bucket: _bucket_config(_as_float(env.get(_rate_env(bucket)), DEFAULT_RATE), burst_override)
|
||||
for bucket in RateBucket
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def default_state_dir(env: Mapping[str, str] = os.environ) -> Path:
|
||||
explicit: Final = env.get(STATE_DIR_ENV)
|
||||
return Path(explicit) if explicit else Path(tempfile.gettempdir()) / DEFAULT_STATE_DIR_NAME
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RateLimiter:
|
||||
config: Mapping[RateBucket, ProviderConfig] = field(default_factory=load_config)
|
||||
state_dir: Path = field(default_factory=default_state_dir)
|
||||
clock: Callable[[], float] = time.monotonic
|
||||
sleep: Callable[[float], None] = time.sleep
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.state_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def acquire(self, bucket: RateBucket) -> float:
|
||||
cfg: Final = self.config.get(bucket)
|
||||
if cfg is None or not cfg.enabled:
|
||||
return 0.0
|
||||
return sum(self._waits(self.state_dir / f"{bucket.value}.json", cfg))
|
||||
|
||||
def _waits(self, path: Path, cfg: ProviderConfig) -> Iterator[float]:
|
||||
while (wait := _try_consume(path, cfg, self.clock())) > 0:
|
||||
self.sleep(wait)
|
||||
yield wait
|
||||
|
||||
|
||||
@cache
|
||||
def get_default_limiter() -> RateLimiter:
|
||||
return RateLimiter()
|
||||
|
||||
|
||||
def _rate_env(bucket: RateBucket) -> str:
|
||||
return RATE_ENV_PREFIX + bucket.value.upper()
|
||||
|
||||
|
||||
def _bucket_config(rate: float, burst_override: float) -> ProviderConfig:
|
||||
return ProviderConfig(rate_per_sec=rate, burst=burst_override if burst_override > 0 else max(rate, 1.0))
|
||||
|
||||
|
||||
def _as_float(value: str | None, default: float) -> float:
|
||||
if not value:
|
||||
return default
|
||||
try:
|
||||
return float(value)
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
def _try_consume(path: Path, cfg: ProviderConfig, now: float) -> float:
|
||||
fd: Final = os.open(path, os.O_CREAT | os.O_RDWR, 0o600)
|
||||
try:
|
||||
fcntl.flock(fd, fcntl.LOCK_EX)
|
||||
state: Final = _refilled(_read_state(fd, cfg, now), now, cfg)
|
||||
if state.tokens >= 1.0:
|
||||
_write_state(fd, _BucketState(tokens=state.tokens - 1.0, last_refill=state.last_refill))
|
||||
return 0.0
|
||||
_write_state(fd, state)
|
||||
return (1.0 - state.tokens) / cfg.rate_per_sec + _WAKE_MARGIN_SECONDS
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
|
||||
def _read_state(fd: int, cfg: ProviderConfig, now: float) -> _BucketState:
|
||||
os.lseek(fd, 0, os.SEEK_SET)
|
||||
raw: Final = os.read(fd, 4096)
|
||||
try:
|
||||
return _BucketState.model_validate_json(raw)
|
||||
except ValidationError:
|
||||
return _BucketState(tokens=cfg.burst, last_refill=now)
|
||||
|
||||
|
||||
def _refilled(state: _BucketState, now: float, cfg: ProviderConfig) -> _BucketState:
|
||||
elapsed: Final = max(0.0, now - state.last_refill)
|
||||
return _BucketState(tokens=min(cfg.burst, state.tokens + elapsed * cfg.rate_per_sec), last_refill=now)
|
||||
|
||||
|
||||
def _write_state(fd: int, state: _BucketState) -> None:
|
||||
os.lseek(fd, 0, os.SEEK_SET)
|
||||
os.ftruncate(fd, 0)
|
||||
os.write(fd, state.model_dump_json().encode("utf-8"))
|
||||
|
|
@ -2,7 +2,7 @@
|
|||
|
||||
Tests of the harness under `tests/e2e/` (the transport, the clients, the fixture bundle and replay edge, the stack lock, the IdP launcher, the coverage collector, the JUnit properties and the Claude Code driver), not of the product. They live outside `tests/e2e/` because the Buildkite e2e run copies that folder into the runner image and runs every test in it, so a harness test in there counts as a product test in the nightly numbers. Nothing here needs a proxy, provider keys or the network
|
||||
|
||||
The layout mirrors `tests/e2e/`: `test_e2e_http.py` covers `tests/e2e/e2e_http.py`, `logging/test_datadog_reader.py` covers `tests/e2e/logging/datadog_reader.py`, and `claude_code/` covers the driver, builder, probe and version resolver. Put a new harness test under the folder that mirrors the suite folder whose module it covers
|
||||
The layout mirrors `tests/e2e/`: `test_e2e_http.py` covers `tests/e2e/e2e_http.py`, `logging/test_datadog_reader.py` covers `tests/e2e/logging/datadog_reader.py`, `claude_code/` covers the driver, builder and probe, and `cli_harness/` covers the version resolver and the CLI env lockdown. Put a new harness test under the folder that mirrors the suite folder whose module it covers
|
||||
|
||||
Run them from the repo root. `e2e_config` reads `LITELLM_MASTER_KEY` at import and any value will do, the CI lane sets a dummy:
|
||||
|
||||
|
|
|
|||
|
|
@ -1,60 +0,0 @@
|
|||
"""Unit tests for the Claude Code PR-gate version resolver.
|
||||
|
||||
Markerless harness tests: they feed the resolver a hand-built packument and a
|
||||
fixed clock, so they run without a proxy, never reach the npm registry, and
|
||||
carry no `e2e` marker.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Final, Mapping
|
||||
|
||||
import pytest
|
||||
|
||||
from claude_code.pr_gate_version_resolver import NoEligibleVersionError, resolve_pr_gate_version
|
||||
|
||||
NOW: Final = datetime(2026, 4, 25, 12, 0, tzinfo=timezone.utc)
|
||||
INSIDE_THE_2_1_88_WINDOW: Final = datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _packument(times: Mapping[str, str], unpublished: frozenset[str] = frozenset()) -> dict[str, object]:
|
||||
return {
|
||||
"name": "@anthropic-ai/claude-code",
|
||||
"time": {"created": "2024-01-01T00:00:00.000Z", "modified": "2026-04-25T00:00:00.000Z", **times},
|
||||
"versions": {version: {"version": version} for version in times if version not in unpublished},
|
||||
}
|
||||
|
||||
|
||||
def test_skips_a_version_npm_has_unpublished() -> None:
|
||||
metadata: Final = _packument(
|
||||
{
|
||||
"2.1.87": "2026-03-28T20:00:00.000Z",
|
||||
"2.1.88": "2026-03-30T22:36:48.424Z",
|
||||
"2.1.89": "2026-03-31T23:32:40.000Z",
|
||||
},
|
||||
unpublished=frozenset({"2.1.88"}),
|
||||
)
|
||||
assert resolve_pr_gate_version(metadata=metadata, as_of=INSIDE_THE_2_1_88_WINDOW) == "2.1.87"
|
||||
|
||||
|
||||
def test_raises_when_the_only_old_enough_version_is_unpublished() -> None:
|
||||
metadata: Final = _packument(
|
||||
{"2.1.88": "2026-03-30T22:36:48.424Z", "2.1.89": "2026-03-31T23:32:40.000Z"},
|
||||
unpublished=frozenset({"2.1.88"}),
|
||||
)
|
||||
with pytest.raises(NoEligibleVersionError):
|
||||
resolve_pr_gate_version(metadata=metadata, as_of=INSIDE_THE_2_1_88_WINDOW)
|
||||
|
||||
|
||||
def test_picks_the_newest_published_version_at_least_min_age_old() -> None:
|
||||
metadata: Final = _packument(
|
||||
{
|
||||
"2.1.118": "2026-04-15T10:00:00.000Z",
|
||||
"2.1.119": "2026-04-21T10:00:00.000Z",
|
||||
"2.2.0-rc.1": "2026-04-22T10:00:00.000Z",
|
||||
"2.1.120": "2026-04-23T10:00:00.000Z",
|
||||
"2.1.121": "2026-04-25T11:00:00.000Z",
|
||||
}
|
||||
)
|
||||
assert resolve_pr_gate_version(metadata=metadata, as_of=NOW) == "2.1.119"
|
||||
|
|
@ -34,7 +34,7 @@ from typing import List, Tuple
|
|||
import pytest
|
||||
|
||||
from claude_code.cli_driver import _FIXED_CLI_USER_ID, _seed_cli_identity, _stable_cli_state, run_claude
|
||||
from claude_code.rate_limiter import RateLimiter
|
||||
from cli_harness.rate_limiter import RateLimiter
|
||||
|
||||
pytestmark = pytest.mark.cli_determinism
|
||||
|
||||
|
|
|
|||
32
tests/e2e_harness/cli_harness/test_lockdown.py
Normal file
32
tests/e2e_harness/cli_harness/test_lockdown.py
Normal file
|
|
@ -0,0 +1,32 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from cli_harness.lockdown import isolated_home, locked_down_env
|
||||
|
||||
RUNNER_ENV: Final = {
|
||||
"PATH": "/usr/bin",
|
||||
"LANG": "C.UTF-8",
|
||||
"HOME": "/home/runner",
|
||||
"OPENAI_API_KEY": "sk-runner-secret",
|
||||
"AWS_SECRET_ACCESS_KEY": "runner-aws-secret",
|
||||
"DATABASE_URL": "postgresql://runner",
|
||||
}
|
||||
|
||||
|
||||
def test_cli_sees_runtime_vars_and_its_overrides_but_no_runner_credentials(tmp_path: Path) -> None:
|
||||
env: Final = locked_down_env(home=tmp_path, overrides={"OPENAI_BASE_URL": "http://gateway"}, source=RUNNER_ENV)
|
||||
assert dict(env) == {
|
||||
"PATH": "/usr/bin",
|
||||
"LANG": "C.UTF-8",
|
||||
"HOME": str(tmp_path),
|
||||
"OPENAI_BASE_URL": "http://gateway",
|
||||
}
|
||||
|
||||
|
||||
def test_isolated_home_starts_empty_and_is_gone_after_the_cli_exits() -> None:
|
||||
with isolated_home("cli-harness-test-") as home:
|
||||
assert home.is_dir() and not any(home.iterdir())
|
||||
(home / ".config").mkdir()
|
||||
assert not home.exists()
|
||||
63
tests/e2e_harness/cli_harness/test_npm_version_resolver.py
Normal file
63
tests/e2e_harness/cli_harness/test_npm_version_resolver.py
Normal file
|
|
@ -0,0 +1,63 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Final
|
||||
|
||||
from cli_harness.npm_version_resolver import MIN_AGE, Packument, newest_version_at_least
|
||||
|
||||
NOW: Final = datetime(2026, 4, 25, 12, 0, tzinfo=timezone.utc)
|
||||
INSIDE_THE_2_1_88_WINDOW: Final = datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _packument(times: Mapping[str, str], unpublished: frozenset[str] = frozenset()) -> Packument:
|
||||
return Packument.from_json(
|
||||
{
|
||||
"name": "@scope/some-cli",
|
||||
"time": {"created": "2024-01-01T00:00:00.000Z", "modified": "2026-04-25T00:00:00.000Z", **times},
|
||||
"versions": {version: {"version": version} for version in times if version not in unpublished},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def test_skips_a_version_npm_has_unpublished() -> None:
|
||||
packument: Final = _packument(
|
||||
{
|
||||
"2.1.87": "2026-03-28T20:00:00.000Z",
|
||||
"2.1.88": "2026-03-30T22:36:48.424Z",
|
||||
"2.1.89": "2026-03-31T23:32:40.000Z",
|
||||
},
|
||||
unpublished=frozenset({"2.1.88"}),
|
||||
)
|
||||
assert newest_version_at_least(packument, min_age=MIN_AGE, as_of=INSIDE_THE_2_1_88_WINDOW) == "2.1.87"
|
||||
|
||||
|
||||
def test_finds_nothing_when_the_only_old_enough_version_is_unpublished() -> None:
|
||||
packument: Final = _packument(
|
||||
{"2.1.88": "2026-03-30T22:36:48.424Z", "2.1.89": "2026-03-31T23:32:40.000Z"},
|
||||
unpublished=frozenset({"2.1.88"}),
|
||||
)
|
||||
assert newest_version_at_least(packument, min_age=MIN_AGE, as_of=INSIDE_THE_2_1_88_WINDOW) is None
|
||||
|
||||
|
||||
def test_picks_the_newest_stable_version_at_least_min_age_old() -> None:
|
||||
packument: Final = _packument(
|
||||
{
|
||||
"2.1.118": "2026-04-15T10:00:00.000Z",
|
||||
"2.1.119": "2026-04-21T10:00:00.000Z",
|
||||
"2.2.0-rc.1": "2026-04-22T10:00:00.000Z",
|
||||
"2.1.120": "2026-04-23T10:00:00.000Z",
|
||||
"2.1.121": "2026-04-25T11:00:00.000Z",
|
||||
}
|
||||
)
|
||||
assert newest_version_at_least(packument, min_age=MIN_AGE, as_of=NOW) == "2.1.119"
|
||||
|
||||
|
||||
def test_newest_means_latest_published_not_highest_semver() -> None:
|
||||
packument: Final = _packument(
|
||||
{
|
||||
"3.0.0": "2026-04-01T10:00:00.000Z",
|
||||
"2.9.9": "2026-04-10T10:00:00.000Z",
|
||||
}
|
||||
)
|
||||
assert newest_version_at_least(packument, min_age=timedelta(days=3), as_of=NOW) == "2.9.9"
|
||||
29
tests/e2e_harness/cli_harness/test_rate_limiter.py
Normal file
29
tests/e2e_harness/cli_harness/test_rate_limiter.py
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
from cli_harness.rate_limiter import ProviderConfig, RateBucket, RateLimiter
|
||||
|
||||
|
||||
@dataclass
|
||||
class _FakeTime:
|
||||
now: float = 0.0
|
||||
|
||||
def clock(self) -> float:
|
||||
return self.now
|
||||
|
||||
def sleep(self, seconds: float) -> None:
|
||||
self.now += seconds
|
||||
|
||||
|
||||
def test_two_processes_on_one_host_share_a_single_budget_per_bucket(tmp_path: Path) -> None:
|
||||
fake_time: Final = _FakeTime()
|
||||
config: Final = {RateBucket.OPENAI: ProviderConfig(rate_per_sec=2.0, burst=1.0)}
|
||||
first_process: Final = RateLimiter(config, tmp_path, fake_time.clock, fake_time.sleep)
|
||||
second_process: Final = RateLimiter(config, tmp_path, fake_time.clock, fake_time.sleep)
|
||||
|
||||
assert first_process.acquire(RateBucket.OPENAI) == 0.0
|
||||
assert 0.5 <= second_process.acquire(RateBucket.OPENAI) < 0.51
|
||||
assert second_process.acquire(RateBucket.BEDROCK_MANTLE) == 0.0
|
||||
Loading…
Add table
Reference in a new issue