diff --git a/.circleci/scripts/run_integration.sh b/.circleci/scripts/run_integration.sh index 8194fb94bbc..501bf68b7ca 100644 --- a/.circleci/scripts/run_integration.sh +++ b/.circleci/scripts/run_integration.sh @@ -10,12 +10,8 @@ suite="${1:?integration suite required}" results="test-results/integration-${suite}" mkdir -p "$results" shard_timeout=11m -if [ "$suite" = cost ]; then - shard_timeout=20m -fi integration_identity="$(.venv/bin/python -c 'import uuid; print(uuid.uuid4().hex)')" upstream_pid="" -scripted_provider_pid="" proxy_pid="" peer_pid="" launched_pid="" @@ -27,9 +23,9 @@ cleanup() { original_status=$? trap - EXIT INT TERM sudo .venv/bin/python .circleci/scripts/stop_integration_processes.py \ - "$integration_identity" "$(id -u)" "$proxy_pid" "$peer_pid" "$upstream_pid" "$scripted_provider_pid" \ + "$integration_identity" "$(id -u)" "$proxy_pid" "$peer_pid" "$upstream_pid" \ > "$results/process-cleanup.txt" 2>&1 || original_status=1 - for owned_pid in "$peer_pid" "$proxy_pid" "$upstream_pid" "$scripted_provider_pid"; do + for owned_pid in "$peer_pid" "$proxy_pid" "$upstream_pid"; do if [ -n "$owned_pid" ]; then kill -- "-$owned_pid" 2>/dev/null || true for _ in {1..50}; do @@ -74,7 +70,6 @@ export STORE_MODEL_IN_DB=True AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 export INTEGRATION_PROXY_URL=http://127.0.0.1:4000 export INTEGRATION_PEER_URL="" export INTEGRATION_UPSTREAM_URL=http://127.0.0.1:8190 -export INTEGRATION_SCRIPTED_PROVIDER_URL="" export INTEGRATION_MASTER_KEY="$LITELLM_MASTER_KEY" export LITELLM_UI_PATH="$PWD/litellm/proxy/_experimental/out" if [ "$suite" = browser ]; then @@ -115,18 +110,7 @@ setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN .venv/bin/python -m integration._support.upstream > "$results/upstream.log" 2>&1 & upstream_pid=$! if [ "$suite" = cost ]; then - export INTEGRATION_SCRIPTED_PROVIDER_URL=http://127.0.0.1:8191 - setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \ - .venv/bin/python -m integration._support.scripted_provider --port 8191 \ - > "$results/scripted-provider.log" 2>&1 & - scripted_provider_pid=$! - for _ in {1..90}; do - if curl --noproxy '*' -fsS "$INTEGRATION_SCRIPTED_PROVIDER_URL/health" >/dev/null 2>&1; then - break - fi - sleep 1 - done - curl --noproxy '*' -fsS "$INTEGRATION_SCRIPTED_PROVIDER_URL/health" >/dev/null + export INTEGRATION_WORKERS=8 fi start_proxy() { local port="$1" @@ -134,7 +118,7 @@ start_proxy() { local -a cost_map_env if [ "$suite" = cost ]; then cost_map_env=( - "LITELLM_MODEL_COST_MAP_URL=$INTEGRATION_SCRIPTED_PROVIDER_URL/_cost_map" + "LITELLM_MODEL_COST_MAP_URL=$INTEGRATION_UPSTREAM_URL/_cost_map" "MODEL_COST_MAP_MIN_MODEL_COUNT=1" "MODEL_COST_MAP_MAX_SHRINK_RATIO=0" ) @@ -190,7 +174,7 @@ timeout --signal=TERM --kill-after=20s "$shard_timeout" env -i PATH="$PATH" HOME DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \ INTEGRATION_PROXY_URL="$INTEGRATION_PROXY_URL" INTEGRATION_PEER_URL="$INTEGRATION_PEER_URL" \ INTEGRATION_UPSTREAM_URL="$INTEGRATION_UPSTREAM_URL" \ - INTEGRATION_SCRIPTED_PROVIDER_URL="$INTEGRATION_SCRIPTED_PROVIDER_URL" \ + INTEGRATION_WORKERS="${INTEGRATION_WORKERS:-1}" \ INTEGRATION_MASTER_KEY="$INTEGRATION_MASTER_KEY" LITELLM_MODE=PRODUCTION \ INTEGRATION_SEED="$INTEGRATION_SEED" \ INTEGRATION_ORDER_SEED="$INTEGRATION_ORDER_SEED" \ diff --git a/.circleci/scripts/wait_integration_services.py b/.circleci/scripts/wait_integration_services.py index 462874e8aa6..486e37cba00 100644 --- a/.circleci/scripts/wait_integration_services.py +++ b/.circleci/scripts/wait_integration_services.py @@ -9,7 +9,6 @@ from redis import Redis def main() -> None: primary: Final = os.environ["INTEGRATION_PROXY_URL"] peer: Final = os.environ.get("INTEGRATION_PEER_URL") - scripted_provider: Final = os.environ.get("INTEGRATION_SCRIPTED_PROVIDER_URL") or None proxies: Final = (primary, peer) if peer else (primary,) deadline: Final = time.monotonic() + 90 headers: Final = {"Authorization": f"Bearer {os.environ['INTEGRATION_MASTER_KEY']}"} @@ -20,10 +19,6 @@ def main() -> None: try: ready: Final = ( client.get(f"{os.environ['INTEGRATION_UPSTREAM_URL']}/health").status_code == 200 - and ( - scripted_provider is None - or client.get(f"{scripted_provider}/health").status_code == 200 - ) and all(client.get(f"{url}/health/readiness").status_code == 200 for url in proxies) ) if ready: diff --git a/tests/integration/README.md b/tests/integration/README.md index 814d03a2875..49b413b17c5 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -2,7 +2,7 @@ These tests exercise a running gateway, PostgreSQL and Redis with an owned local upstream. CircleCI owns this suite. Tests are grouped by behavior, with no automatic test retries or fallback to paid provider calls -The `cost` group runs the scripted-provider cost matrix through a dedicated sidecar. The sidecar serves the test-owned cost map over loopback through `LITELLM_MODEL_COST_MAP_URL`; cost goldens are checked into the integration suite and must not be copied into the E2E coverage registry +The `cost` group runs the scripted-wire cost matrix through the shared integration upstream. The upstream serves the test-owned cost map over loopback through `LITELLM_MODEL_COST_MAP_URL`; cost goldens are checked into the integration suite and must not be copied into the E2E coverage registry Use `tests/integration/run.py management`, `accounting`, `database`, `providers`, `extensions`, `sdk` or `cost` to run a selected group. Set `INTEGRATION_PROXY_URL`, `INTEGRATION_UPSTREAM_URL`, `INTEGRATION_MASTER_KEY` and `DATABASE_URL` to an isolated test deployment. The runner selects the new domain directories explicitly; the legacy OCI and sandbox selections remain separate @@ -22,7 +22,7 @@ Fixtures must contain synthetic data only. Keep private incident records and sou Database cases own their temporary schemas, roles, constraints and proxy processes. They prove reader-versus-writer execution with PostgreSQL lock observations, exercise real transaction wait limits and verify rollback after a reached database failure -Accounting cases compare persisted input and output cost components against literal rates, including zero and default prices. Cache state models assert actual upstream calls, response identity and every persisted charge. Generated accounting tests have a 180-second test limit to accommodate the asynchronous spend writer; CircleCI keeps other shards capped at 11 minutes and gives the cost shard 20 minutes +Accounting cases compare persisted input and output cost components against literal rates, including zero and default prices. Cache state models assert actual upstream calls, response identity and every persisted charge. Generated accounting tests have a 180-second test limit to accommodate the asynchronous spend writer; CircleCI keeps the whole shard capped at 11 minutes Provider contracts exercise actual TCP requests with synthetic credentials and local protocol peers. The S3 verifier uses independently implemented equations, a published known-answer vector, a fixed signing clock and deliberately invalid signed requests. Bedrock cases clear ambient AWS credential sources and check the literal model path, loaded role references, STS requests and bearer-only behavior diff --git a/tests/integration/_support/scripted_client.py b/tests/integration/_support/scripted_client.py index 7818488fae0..9502740b1b5 100644 --- a/tests/integration/_support/scripted_client.py +++ b/tests/integration/_support/scripted_client.py @@ -1,4 +1,4 @@ -"""Client for registering scenarios with the integration scripted provider.""" +"""Client for registering scenarios with the integration upstream.""" from __future__ import annotations @@ -7,7 +7,7 @@ from dataclasses import dataclass from typing import Final import httpx -from integration._support.scripted_provider import ( +from integration._support.scripted_wires import ( WIRE_MOUNTS, Scenario, ScenarioDeleted, @@ -15,7 +15,7 @@ from integration._support.scripted_provider import ( Wire, ) -CONTROL_URL: Final = os.environ.get("INTEGRATION_SCRIPTED_PROVIDER_URL", "http://127.0.0.1:8191").rstrip("/") +CONTROL_URL: Final = os.environ.get("INTEGRATION_UPSTREAM_URL", "http://127.0.0.1:8190").rstrip("/") @dataclass(frozen=True, slots=True) @@ -33,7 +33,7 @@ class ScenarioHandle: def register_scenario(scenario: Scenario) -> ScenarioHandle: response: Final = httpx.post( - f"{CONTROL_URL}/_scenarios", + f"{CONTROL_URL}/__scenarios", json=scenario.model_dump(mode="json"), trust_env=False, timeout=15, @@ -49,7 +49,7 @@ def register_scenario(scenario: Scenario) -> ScenarioHandle: def delete_scenario(handle: ScenarioHandle) -> None: response: Final = httpx.delete( - f"{CONTROL_URL}/_scenarios/{handle.scenario_id}", + f"{CONTROL_URL}/__scenarios/{handle.scenario_id}", trust_env=False, timeout=15, ) diff --git a/tests/integration/_support/scripted_provider.py b/tests/integration/_support/scripted_wires.py similarity index 91% rename from tests/integration/_support/scripted_provider.py rename to tests/integration/_support/scripted_wires.py index d5e0fd7e9cf..ae5ed3abd61 100644 --- a/tests/integration/_support/scripted_provider.py +++ b/tests/integration/_support/scripted_wires.py @@ -1,22 +1,17 @@ -"""Scripted provider sidecar for the cost-calculation integration suite. +"""Scripted provider wires for the cost-calculation integration suite. -A standalone process (``python -m integration._support.scripted_provider``) that -pretends to be an LLM provider for the proxy under test. The suite registers a -Scenario over a small control API; the provider wire routes then answer the -proxy's upstream calls with the scripted usage figures, in the exact wire shape +The shared integration upstream registers a Scenario over a small control API; +the provider wire routes answer the proxy's upstream calls with the scripted usage figures, in the exact wire shape the real provider would emit (OpenAI chat completions, OpenAI Responses, Anthropic Messages, Gemini generateContent, or the OpenAI-compatible Together / Fireworks surfaces). Because the usage is scripted, expected spend is literal arithmetic on the test cost map's rates, with no dependency on what a real provider would report. -Layout on one port: +The upstream exposes: -- ``GET /health`` liveness -- ``POST /_scenarios`` register a Scenario JSON, returns its id -- ``DELETE /_scenarios/`` remove it -- ``POST /_oauth/token`` fake Google OAuth token endpoint for the - Vertex service-account credential's refresh call +- ``POST /__scenarios`` register a Scenario JSON, returns its id +- ``DELETE /__scenarios/`` remove it - ``POST ///`` provider wire; mount is one of ``openai``, ``anthropic``, ``gemini``, ``together``, ``fireworks``, ``azure``, ``bedrock``, ``vertex`` and the remainder is whatever path the provider @@ -32,22 +27,18 @@ final stream chunk carries usage or the provider reports none. from __future__ import annotations -import argparse import json import struct -import sys import threading import time import zlib from collections.abc import Mapping from dataclasses import dataclass -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from pathlib import Path from types import MappingProxyType -from typing import Final, Literal, TypeAlias, cast +from typing import Final, Literal, TypeAlias from urllib.parse import unquote, urlsplit -from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, model_validator +from pydantic import BaseModel, ConfigDict, TypeAdapter, model_validator Wire: TypeAlias = Literal[ "openai_chat", @@ -1307,7 +1298,7 @@ def _render( # ---------- registry + request routing ---------- -class _ScenarioStore: +class ScenarioStore: def __init__(self) -> None: self._lock: Final = threading.Lock() self._scenarios: dict[str, Scenario] = {} # mutable-ok: server state, guarded by _lock @@ -1359,55 +1350,9 @@ def _request_model(body: bytes, path_tail: str, scenario: Scenario) -> str: return scenario.model -def handle_request(store: _ScenarioStore, method: str, raw_path: str, body: bytes) -> RenderedResponse: +def render(store: ScenarioStore, method: str, raw_path: str, body: bytes) -> RenderedResponse: path: Final = urlsplit(raw_path).path segments: Final = tuple(segment for segment in path.split("/") if segment) - if method == "GET" and segments == ("health",): - return RenderedResponse(200, "application/json", _json_bytes(_jobj(("status", "ok")))) - if method == "GET" and segments == ("_cost_map",): - return RenderedResponse( - 200, - "application/json", - (Path(__file__).resolve().parents[1] / "cost_calculation" / "cost_map.json").read_bytes(), - ) - if segments and segments[0] == "_oauth": - if method == "POST" and segments == ("_oauth", "token"): - return RenderedResponse( - 200, - "application/json", - _json_bytes( - _jobj( - ("access_token", "scripted-token"), - ("token_type", "Bearer"), - ("expires_in", 3600), - ) - ), - ) - return RenderedResponse( - 404, "application/json", _json_bytes(_jobj(("error", "unknown control route"))) - ) - if segments and segments[0] == "_scenarios": - if method == "POST" and len(segments) == 1: - try: - scenario: Final = Scenario.model_validate_json(body) - except ValidationError as exc: - return RenderedResponse( - 400, "application/json", _json_bytes(_jobj(("error", str(exc)))) - ) - store.put(scenario) - return RenderedResponse( - 200, "application/json", _json_bytes(_jobj(("scenario_id", scenario.scenario_id))) - ) - if method == "DELETE" and len(segments) == 2: - deleted: Final = store.drop(segments[1]) - return RenderedResponse( - 200 if deleted else 404, - "application/json", - _json_bytes(_jobj(("deleted", deleted))), - ) - return RenderedResponse( - 404, "application/json", _json_bytes(_jobj(("error", "unknown control route"))) - ) if len(segments) < 2 or method != "POST": return RenderedResponse( 404, "application/json", _json_bytes(_jobj(("error", f"no route for {method} {path}"))) @@ -1441,42 +1386,3 @@ def handle_request(store: _ScenarioStore, method: str, raw_path: str, body: byte requested_model=_request_model(body, tail, found), path_tail=tail, ) - - -class _ScriptedHandler(BaseHTTPRequestHandler): - store: Final[_ScenarioStore] = _ScenarioStore() - - def _dispatch(self, method: str) -> None: - length: Final = int(self.headers.get("content-length") or 0) - body: Final = self.rfile.read(length) if length else b"" - rendered: Final = handle_request(self.store, method, self.path, body) - self.send_response(rendered.status_code) - self.send_header("content-type", rendered.content_type) - self.send_header("content-length", str(len(rendered.body))) - self.end_headers() - self.wfile.write(rendered.body) - - def do_GET(self) -> None: - self._dispatch("GET") - - def do_POST(self) -> None: - self._dispatch("POST") - - def do_DELETE(self) -> None: - self._dispatch("DELETE") - - - -DEFAULT_PORT: Final = 8191 - - -def serve(port: int = DEFAULT_PORT, bind_host: str = "127.0.0.1") -> None: - server: Final = ThreadingHTTPServer((bind_host, port), _ScriptedHandler) - sys.stderr.write(f"scripted-provider listening on http://{bind_host}:{port}\n") - server.serve_forever() - - -if __name__ == "__main__": - parser: Final = argparse.ArgumentParser() - parser.add_argument("--port", type=int, default=8191) - serve(port=cast(int, parser.parse_args().port)) diff --git a/tests/integration/_support/upstream.py b/tests/integration/_support/upstream.py index 04a6ea02eec..c8e77ad513a 100644 --- a/tests/integration/_support/upstream.py +++ b/tests/integration/_support/upstream.py @@ -1,19 +1,22 @@ from __future__ import annotations import argparse -from dataclasses import dataclass, field from collections import deque +import json +from dataclasses import dataclass, field +from pathlib import Path from queue import SimpleQueue -from typing import Final +from typing import Final, cast import uvicorn -from pydantic import JsonValue, TypeAdapter +from pydantic import JsonValue, TypeAdapter, ValidationError from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import JSONResponse, Response from starlette.routing import Route from _fake_openai_endpoint_server import chat_completions, completions, embeddings, health, moderations +from integration._support.scripted_wires import RenderedResponse, Scenario, ScenarioStore, render JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) INTERNAL_FIELDS: Final = frozenset( @@ -48,6 +51,7 @@ class Observation: class Provider: observations: SimpleQueue[Observation] = field(default_factory=SimpleQueue) scripts: dict[str, deque[int]] = field(default_factory=dict) + scenario_store: ScenarioStore = field(default_factory=ScenarioStore) async def chat(self, request: Request) -> Response: body: Final = JSON_OBJECT.validate_json(await request.body()) @@ -103,16 +107,89 @@ class Provider: } ) + async def register_scenario(self, request: Request) -> Response: + try: + scenario: Final = Scenario.model_validate_json(await request.body()) + except ValidationError as exc: + return self._render( + RenderedResponse(400, "application/json", json.dumps({"error": str(exc)}).encode("utf-8")) + ) + self.scenario_store.put(scenario) + return self._render( + RenderedResponse( + 200, + "application/json", + json.dumps({"scenario_id": scenario.scenario_id}).encode("utf-8"), + ) + ) + + async def delete_scenario(self, request: Request) -> Response: + scenario_id: Final = cast(str, request.path_params["scenario_id"]) + deleted: Final = self.scenario_store.drop(scenario_id) + return self._render( + RenderedResponse( + 200 if deleted else 404, + "application/json", + json.dumps({"deleted": deleted}).encode("utf-8"), + ) + ) + + async def cost_map(self, _request: Request) -> Response: + return self._render( + RenderedResponse( + 200, + "application/json", + (Path(__file__).resolve().parents[1] / "cost_calculation" / "cost_map.json").read_bytes(), + ) + ) + + async def oauth_token(self, _request: Request) -> Response: + return self._render( + RenderedResponse( + 200, + "application/json", + json.dumps( + { + "access_token": "scripted-token", + "token_type": "Bearer", + "expires_in": 3600, + } + ).encode("utf-8"), + ) + ) + + async def scripted(self, request: Request) -> Response: + rendered: Final = render( + self.scenario_store, + request.method, + request.url.path, + await request.body(), + ) + return self._render(rendered) + + @staticmethod + def _render(rendered: RenderedResponse) -> Response: + return Response( + content=rendered.body, + status_code=rendered.status_code, + media_type=rendered.content_type, + ) + def app(self) -> Starlette: return Starlette( routes=[ Route("/health", health), Route("/__observations", self.observed), Route("/__scripts/{model}", self.script, methods=["POST", "DELETE", "GET"]), + Route("/__scenarios", self.register_scenario, methods=["POST"]), + Route("/__scenarios/{scenario_id}", self.delete_scenario, methods=["DELETE"]), + Route("/_cost_map", self.cost_map, methods=["GET"]), + Route("/_oauth/token", self.oauth_token, methods=["POST"]), Route("/v1/chat/completions", self.chat, methods=["POST"]), Route("/v1/completions", completions, methods=["POST"]), Route("/v1/embeddings", embeddings, methods=["POST"]), Route("/v1/moderations", moderations, methods=["POST"]), + Route("/{scenario_id}/{tail:path}", self.scripted, methods=["POST"]), ] ) @@ -121,7 +198,7 @@ def main() -> None: parser: Final = argparse.ArgumentParser() parser.add_argument("--port", type=int, default=8190) arguments: Final = parser.parse_args() - uvicorn.run(Provider().app(), host="127.0.0.1", port=arguments.port, access_log=False) + uvicorn.run(Provider().app(), host="127.0.0.1", port=cast(int, arguments.port), access_log=False) if __name__ == "__main__": diff --git a/tests/integration/cost_calculation/conftest.py b/tests/integration/cost_calculation/conftest.py index ab162725eef..66eb373df33 100644 --- a/tests/integration/cost_calculation/conftest.py +++ b/tests/integration/cost_calculation/conftest.py @@ -122,7 +122,7 @@ def register_scenario_deployment( case: Case, marker: str, ) -> str: - control_url: Final = os.environ["INTEGRATION_SCRIPTED_PROVIDER_URL"].rstrip("/") + control_url: Final = os.environ["INTEGRATION_UPSTREAM_URL"].rstrip("/") sidecar_scenario: Final = case.scenario( scenario_id=f"sc-{marker}", model=model, text=f"scripted answer {marker}" ) diff --git a/tests/integration/cost_calculation/cost_matrix.py b/tests/integration/cost_calculation/cost_matrix.py index 3c47cc16051..8b9e0aa9424 100644 --- a/tests/integration/cost_calculation/cost_matrix.py +++ b/tests/integration/cost_calculation/cost_matrix.py @@ -27,7 +27,7 @@ from types import MappingProxyType from typing import Final, Literal from pydantic import BaseModel, ConfigDict, Field, TypeAdapter -from integration._support.scripted_provider import Scenario, ScriptedOutput, ScriptedToolCall, ScriptedUsage, Wire +from integration._support.scripted_wires import Scenario, ScriptedOutput, ScriptedToolCall, ScriptedUsage, Wire COST_MAP_PATH: Final = Path(__file__).resolve().parent / "cost_map.json" CASES_PATH: Final = Path(__file__).resolve().parent / "cases.json" diff --git a/tests/integration/cost_calculation/test_token_pricing.py b/tests/integration/cost_calculation/test_token_pricing.py index 72510b03423..69e2ac7ca0c 100644 --- a/tests/integration/cost_calculation/test_token_pricing.py +++ b/tests/integration/cost_calculation/test_token_pricing.py @@ -1,4 +1,4 @@ -"""Token pricing coverage for the integration scripted-provider cost shard.""" +"""Token pricing coverage for the integration scripted-wire cost shard.""" from __future__ import annotations @@ -9,7 +9,7 @@ import pytest from pydantic import JsonValue from integration._support.client import JSON_OBJECT, Gateway -from integration._support.scripted_provider import ScriptedUsage, Wire +from integration._support.scripted_wires import ScriptedUsage, Wire from integration.cost_calculation.conftest import ( approx_equal, assert_total_is_sum_of_components,