From 65cc43728d1a6af2604776e929b8e42955fbe312 Mon Sep 17 00:00:00 2001 From: Yujong Lee Date: Tue, 1 Sep 2026 14:06:18 -0700 Subject: [PATCH] refactor(tests): clarify fixture recording pipeline --- tests/route_parity/README.md | 19 ++ tests/route_parity/fixture_generator.py | 103 -------- tests/route_parity/fixtures/__init__.py | 1 + tests/route_parity/fixtures/inputs.py | 22 ++ tests/route_parity/fixtures/pipeline.py | 221 ++++++++++++++++++ .../recording.py} | 171 +------------- tests/route_parity/fixtures/store.py | 143 ++++++++++++ tests/route_parity/fixtures/test_inputs.py | 20 ++ tests/route_parity/fixtures/test_pipeline.py | 189 +++++++++++++++ .../test_recording.py} | 140 +++++++---- tests/route_parity/json_file_cache.py | 51 ---- tests/route_parity/replay.py | 4 +- tests/route_parity/test_fixture_generator.py | 80 ------- tests/test_litellm/ocr/conftest.py | 4 +- tests/test_litellm/ocr/fixtures/azure.py | 82 +++---- tests/test_litellm/ocr/fixtures/common.py | 4 +- tests/test_litellm/ocr/fixtures/generate.py | 99 -------- tests/test_litellm/ocr/fixtures/mistral.py | 35 ++- tests/test_litellm/ocr/fixtures/record.py | 72 ++++++ tests/test_litellm/ocr/fixtures/reducto.py | 59 +++-- tests/test_litellm/ocr/fixtures/vertex.py | 75 +++--- tests/test_litellm/ocr/test_fixture_models.py | 11 +- ...te_fixtures.py => test_record_fixtures.py} | 18 +- tests/test_litellm/ocr/test_sdk_parity.py | 2 +- 24 files changed, 931 insertions(+), 694 deletions(-) delete mode 100644 tests/route_parity/fixture_generator.py create mode 100644 tests/route_parity/fixtures/__init__.py create mode 100644 tests/route_parity/fixtures/inputs.py create mode 100644 tests/route_parity/fixtures/pipeline.py rename tests/route_parity/{fixture_recorder.py => fixtures/recording.py} (55%) create mode 100644 tests/route_parity/fixtures/store.py create mode 100644 tests/route_parity/fixtures/test_inputs.py create mode 100644 tests/route_parity/fixtures/test_pipeline.py rename tests/route_parity/{test_fixture_recorder.py => fixtures/test_recording.py} (70%) delete mode 100644 tests/route_parity/json_file_cache.py delete mode 100644 tests/route_parity/test_fixture_generator.py delete mode 100644 tests/test_litellm/ocr/fixtures/generate.py create mode 100644 tests/test_litellm/ocr/fixtures/record.py rename tests/test_litellm/ocr/{test_generate_fixtures.py => test_record_fixtures.py} (91%) diff --git a/tests/route_parity/README.md b/tests/route_parity/README.md index 812f71d3b95..2bba0be66cd 100644 --- a/tests/route_parity/README.md +++ b/tests/route_parity/README.md @@ -34,6 +34,25 @@ - Provider responses are replayed unchanged, so the parity test does not fuzz or validate provider behavior - Because Hypothesis does not run the parity assertion directly, parity failures are not automatically shrunk +## Recording fixtures + +The recording command runs four explicit stages: + +1. Generate deterministic SDK inputs for every configured OCR target +2. Build target-scoped, deduplicated recording jobs +3. Record upstream responses through one globally bounded worker pool +4. Persist each fixture and report whether it was recorded, cached, or failed + +Run it with: + +```shell +uv run python -m tests.test_litellm.ocr.fixtures.record --examples 4 --concurrency 4 +``` + +`--concurrency` caps all provider calls across all targets. Each completed job is reported immediately, and the final +summary reports recorded, cached, and failed totals. Independent jobs finish after a failure, then the command exits +nonzero if any job failed + ## OCR input boundaries OCR strategies generate only public `litellm.ocr()` and `litellm.aocr()` inputs. Every case contains the normalized diff --git a/tests/route_parity/fixture_generator.py b/tests/route_parity/fixture_generator.py deleted file mode 100644 index 4054a888490..00000000000 --- a/tests/route_parity/fixture_generator.py +++ /dev/null @@ -1,103 +0,0 @@ -from __future__ import annotations - -import argparse -import logging -from collections.abc import Mapping, Sequence -from dataclasses import dataclass -from pathlib import Path -from typing import Final, Generic, Protocol, TypeVar, cast - -from hypothesis.strategies import SearchStrategy -from pydantic import BaseModel - -from tests.route_parity.fixture_models import SdkInputBase -from tests.route_parity.fixture_recorder import ProviderSpec, generate_case_inputs, record_cases - -LOGGER: Final = logging.getLogger(__name__) -InputT = TypeVar("InputT", bound=SdkInputBase) -InputT_contra = TypeVar("InputT_contra", bound=SdkInputBase, contravariant=True) -DependencyT_contra = TypeVar("DependencyT_contra", contravariant=True) -CaseT = TypeVar("CaseT", bound=BaseModel) - - -@dataclass(frozen=True, slots=True) -class GeneratorArgs: - concurrency: int - examples: int - fixture_dir: Path | None - - -@dataclass(frozen=True, slots=True) -class FixtureTarget(Generic[InputT]): - name: str - provider_spec: ProviderSpec - strategy: SearchStrategy[InputT] - invocation: FixtureInvocation[InputT] - required_inputs: tuple[InputT, ...] = () - - -class FixtureInvocation(Protocol[InputT_contra]): - def execute(self, provider_url: str, case_input: InputT_contra) -> None: ... - - -class FixtureSource(Protocol[InputT, DependencyT_contra]): - def targets( - self, - environ: Mapping[str, str], - dependency: DependencyT_contra, - /, - ) -> tuple[FixtureTarget[InputT], ...]: ... - - -def discover_fixture_targets( - sources: tuple[FixtureSource[InputT, DependencyT_contra], ...], - environ: Mapping[str, str], - dependency: DependencyT_contra, -) -> tuple[FixtureTarget[InputT], ...]: - return tuple(target for source in sources for target in source.targets(environ, dependency)) - - -def generate_target_fixtures( - target: FixtureTarget[InputT], - root: Path, - examples: int, - concurrency: int, - case_type: type[CaseT], -) -> None: - generated_inputs: Final = generate_case_inputs(target.strategy, examples) - case_inputs: Final = (*target.required_inputs, *generated_inputs) - results: Final = record_cases( - target.provider_spec, - root / target.name, - case_inputs, - target.invocation.execute, - case_type, - concurrency, - ) - for result in results: - LOGGER.info( - "%s %s", - "cached" if result.cache_hit else "recorded", - target.name, - ) - - -def require_targets( - targets: tuple[FixtureTarget[InputT], ...], error_message: str -) -> tuple[FixtureTarget[InputT], ...]: - if targets: - return targets - raise SystemExit(error_message) - - -def parse_generator_args(argv: Sequence[str] | None = None) -> GeneratorArgs: - parser: Final = argparse.ArgumentParser() - parser.add_argument("--concurrency", type=int, default=4) - parser.add_argument("--examples", type=int, default=4) - parser.add_argument("--fixture-dir", type=Path) - namespace: Final = parser.parse_args(argv) - return GeneratorArgs( - concurrency=cast(int, namespace.concurrency), - examples=cast(int, namespace.examples), - fixture_dir=cast(Path | None, namespace.fixture_dir), - ) diff --git a/tests/route_parity/fixtures/__init__.py b/tests/route_parity/fixtures/__init__.py new file mode 100644 index 00000000000..9d48db4f9f8 --- /dev/null +++ b/tests/route_parity/fixtures/__init__.py @@ -0,0 +1 @@ +from __future__ import annotations diff --git a/tests/route_parity/fixtures/inputs.py b/tests/route_parity/fixtures/inputs.py new file mode 100644 index 00000000000..bbca19d0104 --- /dev/null +++ b/tests/route_parity/fixtures/inputs.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +import queue +from typing import Final, TypeVar + +from hypothesis import given, settings +from hypothesis.strategies import SearchStrategy + +InputT = TypeVar("InputT") + + +def generate_case_inputs(strategy: SearchStrategy[InputT], examples: int) -> tuple[InputT, ...]: + generated: Final[queue.SimpleQueue[InputT | None]] = queue.SimpleQueue() + + @settings(max_examples=examples, deadline=None, derandomize=True) + @given(case_input=strategy) + def generate_case(case_input: InputT) -> None: + generated.put(case_input) + + generate_case() + generated.put(None) + return tuple(iter(generated.get, None)) diff --git a/tests/route_parity/fixtures/pipeline.py b/tests/route_parity/fixtures/pipeline.py new file mode 100644 index 00000000000..17a2791593c --- /dev/null +++ b/tests/route_parity/fixtures/pipeline.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +import argparse +import logging +from collections.abc import Sequence +from concurrent.futures import Future, ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path +from types import MappingProxyType +from typing import Final, Generic, Literal, Protocol, TypeVar, cast + +from hypothesis.strategies import SearchStrategy +from pydantic import BaseModel + +from tests.route_parity.fixtures.inputs import generate_case_inputs +from tests.route_parity.fixtures.recording import ProviderSpec, record_upstream_responses +from tests.route_parity.fixtures.store import ( + FixtureInput, + canonical_json, + fixture_cache_key, + fixture_id, + fixture_path, + load_fixture, + save_fixture, +) + +LOGGER: Final = logging.getLogger(__name__) +InputT = TypeVar("InputT", bound=FixtureInput) +InputT_contra = TypeVar("InputT_contra", bound=FixtureInput, contravariant=True) +CaseT = TypeVar("CaseT", bound=BaseModel) + + +@dataclass(frozen=True, slots=True) +class RecordingArgs: + concurrency: int + examples: int + fixture_dir: Path | None + + +class RecordingInvocation(Protocol[InputT_contra]): + def execute(self, provider_url: str, case_input: InputT_contra) -> None: ... + + +@dataclass(frozen=True, slots=True) +class RecordingTarget(Generic[InputT]): + name: str + provider_spec: ProviderSpec + strategy: SearchStrategy[InputT] + invocation: RecordingInvocation[InputT] = field(repr=False) + required_inputs: tuple[InputT, ...] = () + + +@dataclass(frozen=True, slots=True) +class RecordingJob(Generic[InputT]): + target_name: str + directory: Path + provider_spec: ProviderSpec + case_input: InputT + invocation: RecordingInvocation[InputT] = field(repr=False) + + @property + def case_id(self) -> str: + return fixture_id(self.case_input, self.target_name) + + +@dataclass(frozen=True, slots=True) +class RecordedFixture: + target_name: str + case_id: str + path: Path + kind: Literal["recorded"] = field(default="recorded", init=False) + + +@dataclass(frozen=True, slots=True) +class CachedFixture: + target_name: str + case_id: str + path: Path + kind: Literal["cached"] = field(default="cached", init=False) + + +@dataclass(frozen=True, slots=True) +class FailedFixture: + target_name: str + case_id: str + error: Exception = field(repr=False) + kind: Literal["failed"] = field(default="failed", init=False) + + +RecordingOutcome = RecordedFixture | CachedFixture | FailedFixture + + +@dataclass(frozen=True, slots=True) +class RecordingSummary: + recorded: tuple[RecordedFixture, ...] + cached: tuple[CachedFixture, ...] + failed: tuple[FailedFixture, ...] + + @property + def exit_code(self) -> int: + return 1 if self.failed else 0 + + +def _unique_inputs(target: RecordingTarget[InputT], examples: int) -> tuple[InputT, ...]: + generated_inputs: Final = generate_case_inputs(target.strategy, examples) + case_inputs: Final = (*target.required_inputs, *generated_inputs) + return tuple({canonical_json(fixture_cache_key(case_input)): case_input for case_input in case_inputs}.values()) + + +def build_recording_jobs( + targets: tuple[RecordingTarget[InputT], ...], + root: Path, + examples: int, +) -> tuple[RecordingJob[InputT], ...]: + if examples < 1: + raise ValueError("examples must be at least 1") + return tuple( + RecordingJob( + target_name=target.name, + directory=root / target.name, + provider_spec=target.provider_spec, + case_input=case_input, + invocation=target.invocation, + ) + for target in targets + for case_input in _unique_inputs(target, examples) + ) + + +def _record_job(job: RecordingJob[InputT], case_type: type[CaseT]) -> RecordedFixture | CachedFixture: + cached: Final = load_fixture(job.directory, job.case_input, case_type) + if cached is not None: + return CachedFixture( + target_name=job.target_name, + case_id=job.case_id, + path=fixture_path(job.directory, job.case_input), + ) + provider_responses: Final = record_upstream_responses( + job.provider_spec, + job.case_input, + job.invocation.execute, + ) + case: Final = case_type.model_validate({"litellm_input": job.case_input, "provider_responses": provider_responses}) + path: Final = save_fixture(job.directory, job.case_input, case) + return RecordedFixture(target_name=job.target_name, case_id=job.case_id, path=path) + + +def _completed_outcome( + completed: int, + total: int, + job: RecordingJob[InputT], + future: Future[RecordedFixture | CachedFixture], +) -> RecordingOutcome: + try: + outcome: Final = future.result() + except Exception as error: + failed: Final = FailedFixture(target_name=job.target_name, case_id=job.case_id, error=error) + LOGGER.error( + "[%d/%d] failed %s %s: %s", + completed, + total, + failed.target_name, + failed.case_id, + type(error).__name__, + ) + return failed + LOGGER.info("[%d/%d] %s %s %s", completed, total, outcome.kind, outcome.target_name, outcome.case_id) + return outcome + + +def record_fixtures( + targets: tuple[RecordingTarget[InputT], ...], + root: Path, + examples: int, + concurrency: int, + case_type: type[CaseT], +) -> RecordingSummary: + if concurrency < 1: + raise ValueError("concurrency must be at least 1") + jobs: Final = build_recording_jobs(targets, root, examples) + total: Final = len(jobs) + LOGGER.info("Recording %d fixtures across %d targets with concurrency %d", total, len(targets), concurrency) + with ThreadPoolExecutor(max_workers=concurrency) as executor: + future_jobs: Final = MappingProxyType({executor.submit(_record_job, job, case_type): job for job in jobs}) + outcomes: Final = tuple( + _completed_outcome(completed, total, future_jobs[future], future) + for completed, future in enumerate(as_completed(future_jobs), start=1) + ) + summary: Final = RecordingSummary( + recorded=tuple(outcome for outcome in outcomes if isinstance(outcome, RecordedFixture)), + cached=tuple(outcome for outcome in outcomes if isinstance(outcome, CachedFixture)), + failed=tuple(outcome for outcome in outcomes if isinstance(outcome, FailedFixture)), + ) + LOGGER.info( + "Finished %d fixtures: %d recorded, %d cached, %d failed", + total, + len(summary.recorded), + len(summary.cached), + len(summary.failed), + ) + return summary + + +def _positive_int(value: str) -> int: + parsed: Final = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be at least 1") + return parsed + + +def parse_recording_args(argv: Sequence[str] | None = None) -> RecordingArgs: + parser: Final = argparse.ArgumentParser() + parser.add_argument("--concurrency", type=_positive_int, default=4) + parser.add_argument("--examples", type=_positive_int, default=4) + parser.add_argument("--fixture-dir", type=Path) + namespace: Final = parser.parse_args(argv) + return RecordingArgs( + concurrency=cast(int, namespace.concurrency), + examples=cast(int, namespace.examples), + fixture_dir=cast(Path | None, namespace.fixture_dir), + ) diff --git a/tests/route_parity/fixture_recorder.py b/tests/route_parity/fixtures/recording.py similarity index 55% rename from tests/route_parity/fixture_recorder.py rename to tests/route_parity/fixtures/recording.py index 36a5f41d292..ea0ee77c215 100644 --- a/tests/route_parity/fixture_recorder.py +++ b/tests/route_parity/fixtures/recording.py @@ -1,26 +1,16 @@ from __future__ import annotations -import hashlib -import os import queue import threading from collections.abc import Callable, Generator, Iterable -from concurrent.futures import ThreadPoolExecutor from contextlib import contextmanager from dataclasses import dataclass -from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -from pathlib import Path -from typing import Final, Generic, Protocol, TypeVar, cast +from typing import Final, TypeVar, cast from urllib.parse import urlsplit, urlunsplit import httpx -import pytest -from hypothesis import given, settings -from hypothesis.strategies import SearchStrategy -from pydantic import AwareDatetime, BaseModel, ConfigDict, ValidationError -from tests.route_parity.json_file_cache import JsonFileCache, canonical_json from tests.route_parity.recorded_http import ( HttpHeader, RecordedHttpResponse, @@ -29,7 +19,6 @@ from tests.route_parity.recorded_http import ( RecordedStreamChunk, ) -FIXTURE_SCHEMA_VERSION: Final = 1 _PARITY_PROVIDER_HOST: Final = "parity-provider.invalid" _HOP_BY_HOP_HEADERS: Final = frozenset( @@ -46,12 +35,7 @@ _HOP_BY_HOP_HEADERS: Final = frozenset( ) -class FixtureInput(Protocol): - def canonical_input(self) -> dict[str, object]: ... - - -InputT = TypeVar("InputT", bound=FixtureInput) -CaseT = TypeVar("CaseT", bound=BaseModel) +InputT = TypeVar("InputT") @dataclass(frozen=True, slots=True) @@ -59,20 +43,6 @@ class ProviderSpec: upstream_base: str -@dataclass(frozen=True, slots=True) -class RecorderResult(Generic[CaseT]): - case: CaseT - cache_hit: bool - - -class FixtureEnvelope(BaseModel): - model_config = ConfigDict(frozen=True, extra="forbid") - - schema_version: int - recorded_at: AwareDatetime - case: dict[str, object] - - def _excluded_headers(headers: tuple[tuple[str, str], ...]) -> frozenset[str]: connection_values: Final = tuple(value for name, value in headers if name.lower() == "connection") connection_headers: Final = frozenset( @@ -100,7 +70,7 @@ def _normalized_response_header(name: str, value: str) -> str: return urlunsplit(("http", _PARITY_PROVIDER_HOST, parsed.path, parsed.query, parsed.fragment)) -def _local_response_header(name: str, value: str, provider_url: str) -> str: +def local_response_header(name: str, value: str, provider_url: str) -> str: if name.lower() not in {"location", "operation-location"}: return value parsed: Final = urlsplit(value) @@ -193,7 +163,7 @@ class _RecordingHandler(BaseHTTPRequestHandler): provider: Final = self.server assert isinstance(provider, _RecordingProvider) for header in headers: - self.send_header(header.name, _local_response_header(header.name, header.value, provider.url)) + self.send_header(header.name, local_response_header(header.name, header.value, provider.url)) self.send_header("transfer-encoding", "chunked") self.end_headers() chunks: Final = tuple(self._relay_chunks(upstream.iter_bytes())) @@ -219,7 +189,7 @@ class _RecordingHandler(BaseHTTPRequestHandler): provider: Final = self.server assert isinstance(provider, _RecordingProvider) for header in headers: - self.send_header(header.name, _local_response_header(header.name, header.value, provider.url)) + self.send_header(header.name, local_response_header(header.name, header.value, provider.url)) self.send_header("content-length", str(len(body))) self.end_headers() self.wfile.write(body) @@ -241,23 +211,6 @@ def _recording_provider(spec: ProviderSpec) -> Generator[_RecordingProvider]: thread.join(timeout=5) -def generate_case_inputs(strategy: SearchStrategy[InputT], examples: int) -> tuple[InputT, ...]: - generated: Final[queue.SimpleQueue[InputT | None]] = queue.SimpleQueue() - - @settings(max_examples=examples, deadline=None, derandomize=True) - @given(case_input=strategy) - def generate_case(case_input: InputT) -> None: - generated.put(case_input) - - generate_case() - generated.put(None) - return tuple(iter(generated.get, None)) - - -def fixture_cache_key(case_input: FixtureInput) -> dict[str, object]: - return case_input.canonical_input() - - def _invoke_and_take_responses( recorder: _RecordingProvider, case_input: InputT, @@ -273,118 +226,10 @@ def _invoke_and_take_responses( return recorder.take_responses() -def _load_fixture(raw_fixture: dict[str, object], path: Path, case_type: type[CaseT]) -> CaseT: - schema_version: Final = raw_fixture.get("schema_version") - if schema_version != FIXTURE_SCHEMA_VERSION: - raise ValueError( - f"fixture {path} has schema_version {schema_version!r}, expected {FIXTURE_SCHEMA_VERSION}; " - "delete it and regenerate the fixture bundle" - ) - try: - envelope: Final = FixtureEnvelope.model_validate(raw_fixture) - return case_type.model_validate(envelope.case) - except ValidationError as error: - raise ValueError(f"invalid parity fixture {path} ({len(error.errors())} validation errors)") from error - - -def record_case( +def record_upstream_responses( spec: ProviderSpec, - root: Path, case_input: InputT, sdk_call: Callable[[str, InputT], object], - case_type: type[CaseT], -) -> RecorderResult[CaseT]: - cache: Final = JsonFileCache(root) - cache_key: Final = fixture_cache_key(case_input) - cached: Final = cache.get(cache_key) - if cached is not None: - return RecorderResult(case=_load_fixture(cached, cache.path_for(cache_key), case_type), cache_hit=True) - +) -> tuple[RecordedResponse, ...]: with _recording_provider(spec) as recorder: - upstream_responses: Final = _invoke_and_take_responses(recorder, case_input, sdk_call) - - case: Final = case_type.model_validate({"litellm_input": case_input, "provider_responses": upstream_responses}) - envelope: Final = FixtureEnvelope( - schema_version=FIXTURE_SCHEMA_VERSION, - recorded_at=datetime.now(timezone.utc), - case=cast(dict[str, object], case.model_dump(mode="json", exclude_unset=True)), - ) - cache.put(cache_key, cast(dict[str, object], envelope.model_dump(mode="json", exclude_unset=True))) - return RecorderResult(case=case, cache_hit=False) - - -def record_cases( - spec: ProviderSpec, - root: Path, - case_inputs: tuple[InputT, ...], - sdk_call: Callable[[str, InputT], object], - case_type: type[CaseT], - max_concurrency: int, -) -> tuple[RecorderResult[CaseT], ...]: - if max_concurrency < 1: - raise ValueError("max_concurrency must be at least 1") - unique_inputs: Final = tuple( - {canonical_json(fixture_cache_key(case_input)): case_input for case_input in case_inputs}.values() - ) - with ThreadPoolExecutor(max_workers=max_concurrency) as executor: - futures: Final = tuple( - executor.submit(record_case, spec, root, case_input, sdk_call, case_type) for case_input in unique_inputs - ) - return tuple(future.result() for future in futures) - - -def fixture_directory(configured: Path | None, env_value: str | None, default: Path) -> Path: - return (configured or Path(env_value or default)).expanduser() - - -def recorded_fixtures(directory: Path, case_type: type[CaseT]) -> tuple[CaseT, ...]: - cache: Final = JsonFileCache(directory) - return tuple(_load_fixture(raw_fixture, path, case_type) for path, raw_fixture in cache.values_with_paths()) - - -def fixture_id(case_input: FixtureInput, prefix: str) -> str: - input_json: Final = canonical_json(case_input.canonical_input()) - digest: Final = hashlib.sha256(input_json.encode("utf-8")).hexdigest()[:8] - return f"{prefix}-{digest}" - - -def parametrize_recorded_fixtures( - metafunc: pytest.Metafunc, - *, - fixture_name: str, - case_type: type[CaseT], - env_var: str, - default_directory: Path, - regeneration_command: str, - id_builder: Callable[[CaseT], str], -) -> None: - if fixture_name not in metafunc.fixturenames: - return - configured: Final = os.environ.get(env_var) - if configured == "": - raise pytest.UsageError(f"{env_var} is set but empty") - directory: Final = Path(configured).expanduser() if configured is not None else default_directory - try: - fixtures: Final = recorded_fixtures(directory, case_type) - except (ValidationError, ValueError) as error: - raise pytest.UsageError( - f"Invalid parity fixture bundle at {directory}. " - "Each fixture must use the current versioned envelope. " - f"Record fresh fixtures in an empty directory with: `{regeneration_command}`. " - f"Validation details: {error}" - ) from error - if fixtures: - metafunc.parametrize(fixture_name, fixtures, ids=tuple(id_builder(fixture) for fixture in fixtures)) - return - if configured is not None: - raise pytest.UsageError(f"no recorded fixtures in {directory}") - metafunc.parametrize( - fixture_name, - ( - pytest.param( - None, - marks=pytest.mark.skip(reason=f"no recorded fixtures in {directory}"), - id="no-recorded-fixtures", - ), - ), - ) + return _invoke_and_take_responses(recorder, case_input, sdk_call) diff --git a/tests/route_parity/fixtures/store.py b/tests/route_parity/fixtures/store.py new file mode 100644 index 00000000000..af6e7417c65 --- /dev/null +++ b/tests/route_parity/fixtures/store.py @@ -0,0 +1,143 @@ +from __future__ import annotations + +import hashlib +import json +import os +from collections.abc import Callable, Mapping +from datetime import datetime, timezone +from pathlib import Path +from typing import Final, Protocol, TypeVar, cast + +import pytest +from pydantic import AwareDatetime, BaseModel, ConfigDict, TypeAdapter, ValidationError + +FIXTURE_SCHEMA_VERSION: Final = 1 +JSON_OBJECT: Final = TypeAdapter(dict[str, object]) + + +class FixtureInput(Protocol): + def canonical_input(self) -> dict[str, object]: ... + + +CaseT = TypeVar("CaseT", bound=BaseModel) + + +class FixtureEnvelope(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + schema_version: int + recorded_at: AwareDatetime + case: dict[str, object] + + +def canonical_json(value: Mapping[str, object]) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def fixture_cache_key(case_input: FixtureInput) -> dict[str, object]: + return case_input.canonical_input() + + +def fixture_path(directory: Path, case_input: FixtureInput) -> Path: + input_json: Final = canonical_json(fixture_cache_key(case_input)) + digest: Final = hashlib.sha256(input_json.encode("utf-8")).hexdigest() + return directory / f"{digest}.json" + + +def load_fixture(directory: Path, case_input: FixtureInput, case_type: type[CaseT]) -> CaseT | None: + path: Final = fixture_path(directory, case_input) + if not path.is_file(): + return None + return _load_fixture(JSON_OBJECT.validate_json(path.read_text(encoding="utf-8")), path, case_type) + + +def save_fixture(directory: Path, case_input: FixtureInput, case: BaseModel) -> Path: + directory.mkdir(parents=True, exist_ok=True) + path: Final = fixture_path(directory, case_input) + temporary_path: Final = path.with_suffix(".tmp") + envelope: Final = FixtureEnvelope( + schema_version=FIXTURE_SCHEMA_VERSION, + recorded_at=datetime.now(timezone.utc), + case=cast(dict[str, object], case.model_dump(mode="json", exclude_unset=True)), + ) + serialized: Final = ( + json.dumps(envelope.model_dump(mode="json", exclude_unset=True), indent=2, sort_keys=True) + "\n" + ) + temporary_path.write_text(serialized, encoding="utf-8") + temporary_path.replace(path) + return path + + +def _load_fixture(raw_fixture: dict[str, object], path: Path, case_type: type[CaseT]) -> CaseT: + schema_version: Final = raw_fixture.get("schema_version") + if schema_version != FIXTURE_SCHEMA_VERSION: + raise ValueError( + f"fixture {path} has schema_version {schema_version!r}, expected {FIXTURE_SCHEMA_VERSION}; " + "delete it and regenerate the fixture bundle" + ) + try: + envelope: Final = FixtureEnvelope.model_validate(raw_fixture) + return case_type.model_validate(envelope.case) + except ValidationError as error: + raise ValueError(f"invalid parity fixture {path} ({len(error.errors())} validation errors)") from error + + +def recorded_fixtures(directory: Path, case_type: type[CaseT]) -> tuple[CaseT, ...]: + if not directory.is_dir(): + return () + paths: Final = tuple(sorted(directory.rglob("*.json"))) + return tuple( + _load_fixture(JSON_OBJECT.validate_json(path.read_text(encoding="utf-8")), path, case_type) for path in paths + ) + + +def fixture_directory(configured: Path | None, env_value: str | None, default: Path) -> Path: + return (configured or Path(env_value or default)).expanduser() + + +def fixture_id(case_input: FixtureInput, prefix: str) -> str: + input_json: Final = canonical_json(case_input.canonical_input()) + digest: Final = hashlib.sha256(input_json.encode("utf-8")).hexdigest()[:8] + return f"{prefix}-{digest}" + + +def parametrize_recorded_fixtures( + metafunc: pytest.Metafunc, + *, + fixture_name: str, + case_type: type[CaseT], + env_var: str, + default_directory: Path, + regeneration_command: str, + id_builder: Callable[[CaseT], str], +) -> None: + if fixture_name not in metafunc.fixturenames: + return + configured: Final = os.environ.get(env_var) + if configured == "": + raise pytest.UsageError(f"{env_var} is set but empty") + directory: Final = Path(configured).expanduser() if configured is not None else default_directory + try: + fixtures: Final = recorded_fixtures(directory, case_type) + except (ValidationError, ValueError) as error: + raise pytest.UsageError( + f"Invalid parity fixture bundle at {directory}. " + "Each fixture must use the current versioned envelope. " + f"Record fresh fixtures in an empty directory with: `{regeneration_command}`. " + f"Validation details: {error}" + ) from error + if fixtures: + metafunc.parametrize(fixture_name, fixtures, ids=tuple(id_builder(fixture) for fixture in fixtures)) + return + if configured is not None: + raise pytest.UsageError(f"no recorded fixtures in {directory}") + metafunc.parametrize( + fixture_name, + ( + pytest.param( + None, + marks=pytest.mark.skip(reason=f"no recorded fixtures in {directory}"), + id="no-recorded-fixtures", + ), + ), + ) diff --git a/tests/route_parity/fixtures/test_inputs.py b/tests/route_parity/fixtures/test_inputs.py new file mode 100644 index 00000000000..76081771b31 --- /dev/null +++ b/tests/route_parity/fixtures/test_inputs.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from typing import Final + +from hypothesis import strategies as st +from pydantic import BaseModel, ConfigDict + +from tests.route_parity.fixtures.inputs import generate_case_inputs + + +class _Input(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + identifier: str + + +def test_generate_case_inputs_is_deterministic() -> None: + strategy: Final = st.builds(_Input, identifier=st.integers().map(str)) + + assert generate_case_inputs(strategy, examples=4) == generate_case_inputs(strategy, examples=4) diff --git a/tests/route_parity/fixtures/test_pipeline.py b/tests/route_parity/fixtures/test_pipeline.py new file mode 100644 index 00000000000..52f54f332d5 --- /dev/null +++ b/tests/route_parity/fixtures/test_pipeline.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import logging +import threading +from collections.abc import Generator +from contextlib import contextmanager +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Final, Literal + +import httpx +import pytest +from hypothesis import strategies as st +from pydantic import BaseModel, ConfigDict + +from tests.route_parity.fixtures.pipeline import ( + RecordingInvocation, + RecordingTarget, + build_recording_jobs, + record_fixtures, +) +from tests.route_parity.fixtures.recording import ProviderSpec +from tests.route_parity.fixtures.store import fixture_path +from tests.route_parity.recorded_http import RecordedResponse + + +class _FixtureInput(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + identifier: str + + def canonical_input(self) -> dict[str, object]: + return {"identifier": self.identifier} + + +class _ParityCase(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + litellm_input: _FixtureInput + provider_responses: tuple[RecordedResponse, ...] + + +class _Upstream(ThreadingHTTPServer): + daemon_threads = True + + def __init__(self) -> None: + super().__init__(("127.0.0.1", 0), _UpstreamHandler) + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.server_address[1]}" + + +class _UpstreamHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def do_POST(self) -> None: + length: Final = int(self.headers.get("content-length") or "0") + self.rfile.read(length) + body: Final = b"{}" + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: object) -> None: + return + + +@contextmanager +def _upstream() -> Generator[_Upstream]: + server: Final = _Upstream() + thread: Final = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +@dataclass(frozen=True, slots=True) +class _OrderedInvocation: + order: Literal["slow", "fast"] + slow_started: threading.Event + fast_finished: threading.Event + + def execute(self, provider_url: str, case_input: _FixtureInput) -> None: + if self.order == "slow": + self.slow_started.set() + if not self.fast_finished.wait(timeout=2): + raise TimeoutError("fast recording did not finish") + else: + if not self.slow_started.wait(timeout=2): + raise TimeoutError("slow recording did not start") + response: Final = httpx.post(f"{provider_url}/record", json={"id": case_input.identifier}, timeout=5) + response.raise_for_status() + if self.order == "fast": + self.fast_finished.set() + + +@dataclass(frozen=True, slots=True) +class _Invocation: + def execute(self, provider_url: str, case_input: _FixtureInput) -> None: + response: Final = httpx.post(f"{provider_url}/record", json={"id": case_input.identifier}, timeout=5) + response.raise_for_status() + + +def _target( + name: str, + upstream_url: str, + case_input: _FixtureInput, + invocation: RecordingInvocation[_FixtureInput], +) -> RecordingTarget[_FixtureInput]: + return RecordingTarget( + name=name, + provider_spec=ProviderSpec(upstream_base=upstream_url), + strategy=st.just(case_input), + invocation=invocation, + required_inputs=(case_input,), + ) + + +def test_build_jobs_keeps_required_inputs_before_generated_inputs_and_deduplicates(tmp_path: Path) -> None: + required: Final = _FixtureInput(identifier="required") + generated: Final = _FixtureInput(identifier="generated") + target: Final = RecordingTarget( + name="ordered", + provider_spec=ProviderSpec(upstream_base="https://provider.invalid"), + strategy=st.just(generated), + invocation=_Invocation(), + required_inputs=(required, required), + ) + + jobs: Final = build_recording_jobs((target,), tmp_path, examples=1) + + assert tuple(job.case_input.identifier for job in jobs) == ("required", "generated") + + +def test_progress_follows_completion_order(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + slow_started: Final = threading.Event() + fast_finished: Final = threading.Event() + with _upstream() as upstream: + targets: Final = ( + _target( + "slow", + upstream.url, + _FixtureInput(identifier="slow"), + _OrderedInvocation("slow", slow_started, fast_finished), + ), + _target( + "fast", + upstream.url, + _FixtureInput(identifier="fast"), + _OrderedInvocation("fast", slow_started, fast_finished), + ), + ) + with caplog.at_level(logging.INFO, logger="tests.route_parity.fixtures.pipeline"): + summary: Final = record_fixtures(targets, tmp_path, 1, 2, _ParityCase) + + progress: Final = tuple(record.message for record in caplog.records if record.message.startswith("[")) + assert len(summary.recorded) == 2 + assert summary.exit_code == 0 + assert "recorded fast" in progress[0] + assert "recorded slow" in progress[1] + assert caplog.records[0].message == "Recording 2 fixtures across 2 targets with concurrency 2" + assert caplog.records[-1].message == "Finished 2 fixtures: 2 recorded, 0 cached, 0 failed" + + +def test_failure_does_not_stop_independent_recordings(tmp_path: Path) -> None: + stale_input: Final = _FixtureInput(identifier="stale") + stale_directory: Final = tmp_path / "stale" + stale_directory.mkdir() + fixture_path(stale_directory, stale_input).write_text('{"schema_version": 0}\n', encoding="utf-8") + with _upstream() as upstream: + targets: Final = ( + _target("stale", upstream.url, stale_input, _Invocation()), + _target("valid", upstream.url, _FixtureInput(identifier="valid"), _Invocation()), + ) + summary: Final = record_fixtures(targets, tmp_path, 1, 2, _ParityCase) + + assert len(summary.recorded) == 1 + assert summary.recorded[0].target_name == "valid" + assert len(summary.failed) == 1 + assert summary.failed[0].target_name == "stale" + assert summary.exit_code == 1 diff --git a/tests/route_parity/test_fixture_recorder.py b/tests/route_parity/fixtures/test_recording.py similarity index 70% rename from tests/route_parity/test_fixture_recorder.py rename to tests/route_parity/fixtures/test_recording.py index 3dde6b5cad7..d27b14ee1de 100644 --- a/tests/route_parity/test_fixture_recorder.py +++ b/tests/route_parity/fixtures/test_recording.py @@ -3,6 +3,7 @@ from __future__ import annotations import threading from collections.abc import Callable, Generator from contextlib import contextmanager +from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Final @@ -12,16 +13,13 @@ import pytest from hypothesis import strategies as st from pydantic import BaseModel, ConfigDict -from tests.route_parity.fixture_recorder import ( +from tests.route_parity.fixtures.pipeline import RecordingTarget, record_fixtures +from tests.route_parity.fixtures.recording import ProviderSpec, record_upstream_responses +from tests.route_parity.fixtures.store import ( FIXTURE_SCHEMA_VERSION, - ProviderSpec, - fixture_cache_key, - generate_case_inputs, - record_case, - record_cases, + fixture_path, recorded_fixtures, ) -from tests.route_parity.json_file_cache import JsonFileCache from tests.route_parity.recorded_http import ( HttpHeader, RecordedHttpStreamResponse, @@ -53,6 +51,14 @@ class _ParityCase(BaseModel): provider_responses: tuple[RecordedResponse, ...] +@dataclass(frozen=True, slots=True) +class _Invocation: + sdk_call: Callable[[str, _FixtureInput], object] + + def execute(self, provider_url: str, case_input: _FixtureInput) -> None: + self.sdk_call(provider_url, case_input) + + class _ControlledUpstream(ThreadingHTTPServer): daemon_threads = True @@ -195,54 +201,97 @@ def _polling_sdk_call(api_base: str, case_input: _FixtureInput) -> object: return completed -def test_generate_case_inputs_is_deterministic() -> None: - strategy: Final = st.builds(_FixtureInput, identifier=st.integers().map(str)) - - assert generate_case_inputs(strategy, examples=4) == generate_case_inputs(strategy, examples=4) - - -def test_record_cases_deduplicates_and_limits_concurrency(tmp_path: Path) -> None: - case_inputs: Final = (_case("one"), _case("two"), _case("one"), _case("three")) +def test_recording_deduplicates_per_target_and_caps_global_concurrency(tmp_path: Path) -> None: + shared_input: Final = _case("shared") with _controlled_upstream() as upstream: spec: Final = ProviderSpec(upstream_base=upstream.url) - results: Final = record_cases(spec, tmp_path, case_inputs, _sdk_call, _ParityCase, max_concurrency=2) + targets: Final = ( + RecordingTarget( + name="first", + provider_spec=spec, + strategy=st.just(shared_input), + invocation=_Invocation(_sdk_call), + required_inputs=(shared_input, shared_input), + ), + RecordingTarget( + name="second", + provider_spec=spec, + strategy=st.just(shared_input), + invocation=_Invocation(_sdk_call), + required_inputs=(shared_input,), + ), + ) + summary: Final = record_fixtures(targets, tmp_path, examples=1, concurrency=2, case_type=_ParityCase) - assert len(results) == 3 - assert upstream.request_count == 3 + assert len(summary.recorded) == 2 + assert {result.target_name for result in summary.recorded} == {"first", "second"} + assert summary.cached == () + assert summary.failed == () + assert upstream.request_count == 2 assert upstream.max_active_requests == 2 - assert len(recorded_fixtures(tmp_path, _ParityCase)) == 3 - for fixture_path in tmp_path.glob("*.json"): - contents = fixture_path.read_text(encoding="utf-8") + assert len(recorded_fixtures(tmp_path, _ParityCase)) == 2 + for path in tmp_path.rglob("*.json"): + contents = path.read_text(encoding="utf-8") assert f'"schema_version": {FIXTURE_SCHEMA_VERSION}' in contents assert '"recorded_at":' in contents -def test_record_case_rejects_stale_fixture_before_provider_call(tmp_path: Path) -> None: +def test_pipeline_rejects_stale_fixture_before_provider_call(tmp_path: Path) -> None: case_input: Final = _case("stale") - cache: Final = JsonFileCache(tmp_path) - fixture_path: Final = cache.put(fixture_cache_key(case_input), {"schema_version": 0}) + directory: Final = tmp_path / "stale-target" + directory.mkdir() + path: Final = fixture_path(directory, case_input) + path.write_text('{"schema_version": 0}\n', encoding="utf-8") + target: Final = RecordingTarget( + name="stale-target", + provider_spec=ProviderSpec(upstream_base="http://127.0.0.1:1"), + strategy=st.just(case_input), + invocation=_Invocation(_sdk_call), + ) - with pytest.raises(ValueError, match=f"{fixture_path} has schema_version 0, expected {FIXTURE_SCHEMA_VERSION}"): - record_case( - ProviderSpec(upstream_base="http://127.0.0.1:1"), - tmp_path, - case_input, - _sdk_call, - _ParityCase, - ) + summary: Final = record_fixtures( + (target,), + tmp_path, + examples=1, + concurrency=1, + case_type=_ParityCase, + ) + + assert summary.recorded == () + assert summary.cached == () + assert len(summary.failed) == 1 + assert str(summary.failed[0].error) == ( + f"fixture {path} has schema_version 0, expected {FIXTURE_SCHEMA_VERSION}; " + "delete it and regenerate the fixture bundle" + ) -def test_streaming_response_records_and_replays_chunks(tmp_path: Path) -> None: +def test_cached_fixture_is_reported_without_provider_call(tmp_path: Path) -> None: + case_input: Final = _case("cached") with _controlled_upstream() as upstream: - result: Final = record_case( + target: Final = RecordingTarget( + name="cached-target", + provider_spec=ProviderSpec(upstream_base=upstream.url), + strategy=st.just(case_input), + invocation=_Invocation(_sdk_call), + ) + first: Final = record_fixtures((target,), tmp_path, 1, 1, _ParityCase) + second: Final = record_fixtures((target,), tmp_path, 1, 1, _ParityCase) + + assert len(first.recorded) == 1 + assert len(second.cached) == 1 + assert upstream.request_count == 1 + + +def test_streaming_response_records_and_replays_chunks() -> None: + with _controlled_upstream() as upstream: + responses: Final = record_upstream_responses( ProviderSpec(upstream_base=upstream.url), - tmp_path, _case("stream"), _stream_sdk_call, - _ParityCase, ) - response: Final = result.case.provider_responses[0] + response: Final = responses[0] assert isinstance(response, RecordedHttpStreamResponse) assert tuple(chunk.data_bytes() for chunk in response.chunks) == _SSE_CHUNKS assert isinstance(response.model_dump(mode="json")["chunks"], list) @@ -256,17 +305,15 @@ def test_streaming_response_records_and_replays_chunks(tmp_path: Path) -> None: assert replayed_chunks == _SSE_CHUNKS -def test_non_successful_provider_response_is_recorded(tmp_path: Path) -> None: +def test_non_successful_provider_response_is_recorded() -> None: with _controlled_upstream() as upstream: - result: Final = record_case( + responses: Final = record_upstream_responses( ProviderSpec(upstream_base=upstream.url), - tmp_path, _case("provider-error"), _error_sdk_call, - _ParityCase, ) - response: Final = result.case.provider_responses[0] + response: Final = responses[0] assert response.status_code == 429 @@ -285,21 +332,18 @@ def test_stream_response_model_rejects_buffered_body() -> None: @pytest.mark.parametrize("sdk_call", (_multi_sdk_call, _polling_sdk_call)) def test_multiple_provider_calls_record_and_replay_in_order( - tmp_path: Path, sdk_call: Callable[[str, _FixtureInput], object], ) -> None: with _controlled_upstream() as upstream: - result: Final = record_case( + responses: Final = record_upstream_responses( ProviderSpec(upstream_base=upstream.url), - tmp_path, _case(sdk_call.__name__), sdk_call, - _ParityCase, ) - assert len(result.case.provider_responses) == 2 + assert len(responses) == 2 with replay_server() as provider: - for response in result.case.provider_responses: + for response in responses: provider.enqueue_response(response) sdk_call(provider.url, _case(sdk_call.__name__)) requests: Final = provider.take_requests(2) diff --git a/tests/route_parity/json_file_cache.py b/tests/route_parity/json_file_cache.py deleted file mode 100644 index dfc8b6331f4..00000000000 --- a/tests/route_parity/json_file_cache.py +++ /dev/null @@ -1,51 +0,0 @@ -from __future__ import annotations - -import hashlib -import json -from collections.abc import Mapping -from dataclasses import dataclass -from pathlib import Path -from typing import Final - -from pydantic import TypeAdapter - -JSON_OBJECT: Final = TypeAdapter(dict[str, object]) - - -def canonical_json(value: Mapping[str, object]) -> str: - return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) - - -@dataclass(frozen=True, slots=True) -class JsonFileCache: - root: Path - - def path_for(self, key: Mapping[str, object]) -> Path: - digest: Final = hashlib.sha256(canonical_json(key).encode("utf-8")).hexdigest() - return self.root / f"{digest}.json" - - def get(self, key: Mapping[str, object]) -> dict[str, object] | None: - path: Final = self.path_for(key) - if not path.is_file(): - return None - return JSON_OBJECT.validate_json(path.read_text(encoding="utf-8")) - - def put(self, key: Mapping[str, object], value: Mapping[str, object]) -> Path: - self.root.mkdir(parents=True, exist_ok=True) - path: Final = self.path_for(key) - temporary_path: Final = path.with_suffix(".tmp") - temporary_path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") - temporary_path.replace(path) - return path - - def values(self) -> tuple[dict[str, object], ...]: - if not self.root.is_dir(): - return () - paths: Final = tuple(sorted(self.root.rglob("*.json"))) - return tuple(JSON_OBJECT.validate_json(path.read_text(encoding="utf-8")) for path in paths) - - def values_with_paths(self) -> tuple[tuple[Path, dict[str, object]], ...]: - if not self.root.is_dir(): - return () - paths: Final = tuple(sorted(self.root.rglob("*.json"))) - return tuple((path, JSON_OBJECT.validate_json(path.read_text(encoding="utf-8"))) for path in paths) diff --git a/tests/route_parity/replay.py b/tests/route_parity/replay.py index 6cfa7ad76a3..9a185ead100 100644 --- a/tests/route_parity/replay.py +++ b/tests/route_parity/replay.py @@ -10,7 +10,7 @@ from typing import Final from pydantic import JsonValue, TypeAdapter -from tests.route_parity.fixture_recorder import _local_response_header +from tests.route_parity.fixtures.recording import local_response_header from tests.route_parity.models import CapturedRequest from tests.route_parity.recorded_http import RecordedHttpResponse, RecordedHttpStreamResponse, RecordedResponse @@ -102,7 +102,7 @@ class _ReplayHandler(BaseHTTPRequestHandler): self.send_response_only(response.status_code) for header in response.headers: if header.name.lower() not in EXCLUDED_RESPONSE_HEADERS: - self.send_header(header.name, _local_response_header(header.name, header.value, provider.url)) + self.send_header(header.name, local_response_header(header.name, header.value, provider.url)) if isinstance(response, RecordedHttpResponse): response_body: Final = response.body_bytes() self.send_header("content-length", str(len(response_body))) diff --git a/tests/route_parity/test_fixture_generator.py b/tests/route_parity/test_fixture_generator.py deleted file mode 100644 index 1843729a939..00000000000 --- a/tests/route_parity/test_fixture_generator.py +++ /dev/null @@ -1,80 +0,0 @@ -from __future__ import annotations - -import queue -from collections.abc import Mapping -from dataclasses import dataclass, field -from typing import Final - -from hypothesis import strategies as st - -from tests.route_parity.fixture_generator import FixtureSource, FixtureTarget, discover_fixture_targets -from tests.route_parity.fixture_models import SdkInputBase -from tests.route_parity.fixture_recorder import ProviderSpec - - -class ExampleSdkInput(SdkInputBase): - model: str - - -@dataclass(frozen=True, slots=True) -class ExampleInvocation: - calls: queue.SimpleQueue[dict[str, object]] - api_key: str = field(repr=False) - - def execute(self, provider_url: str, case_input: ExampleSdkInput) -> None: - self.calls.put( - { - "api_base": provider_url, - "api_key": self.api_key, - **case_input.as_sdk_kwargs(), - } - ) - - -@dataclass(frozen=True, slots=True) -class ExampleProvider: - name: str - key_name: str - - def targets( - self, - environ: Mapping[str, str], - calls: queue.SimpleQueue[dict[str, object]], - ) -> tuple[FixtureTarget[ExampleSdkInput], ...]: - api_key: Final = environ.get(self.key_name) - if not api_key: - return () - - case_input: Final = ExampleSdkInput(model=f"{self.name}/model") - return ( - FixtureTarget( - name=self.name, - provider_spec=ProviderSpec(upstream_base=f"https://{self.name}.example"), - strategy=st.just(case_input), - invocation=ExampleInvocation(calls=calls, api_key=api_key), - required_inputs=(case_input,), - ), - ) - - -def test_discover_fixture_targets_flattens_configured_providers_and_injects_sdk_call() -> None: - calls: Final[queue.SimpleQueue[dict[str, object]]] = queue.SimpleQueue() - - providers: Final[tuple[FixtureSource[ExampleSdkInput, queue.SimpleQueue[dict[str, object]]], ...]] = ( - ExampleProvider(name="first", key_name="FIRST_KEY"), - ExampleProvider(name="skipped", key_name="SKIPPED_KEY"), - ExampleProvider(name="second", key_name="SECOND_KEY"), - ) - targets: Final = discover_fixture_targets( - providers, - {"FIRST_KEY": "first-secret", "SECOND_KEY": "second-secret"}, - calls, - ) - - assert tuple(target.name for target in targets) == ("first", "second") - targets[1].invocation.execute("http://127.0.0.1:1234", targets[1].required_inputs[0]) - assert calls.get_nowait() == { - "api_base": "http://127.0.0.1:1234", - "api_key": "second-secret", - "model": "second/model", - } diff --git a/tests/test_litellm/ocr/conftest.py b/tests/test_litellm/ocr/conftest.py index 6018c5e1dda..d99c0de7cf9 100644 --- a/tests/test_litellm/ocr/conftest.py +++ b/tests/test_litellm/ocr/conftest.py @@ -5,7 +5,7 @@ from typing import Final import pytest -from tests.route_parity.fixture_recorder import fixture_id, parametrize_recorded_fixtures +from tests.route_parity.fixtures.store import fixture_id, parametrize_recorded_fixtures from tests.test_litellm.ocr.fixtures.models import OcrParityCase FIXTURE_DIR_ENV: Final = "LITELLM_OCR_FIXTURE_DIR" @@ -27,7 +27,7 @@ def pytest_generate_tests(metafunc: pytest.Metafunc) -> None: env_var=FIXTURE_DIR_ENV, default_directory=default_directory, regeneration_command=( - f"uv run python -m tests.test_litellm.ocr.fixtures.generate --fixture-dir {default_directory}" + f"uv run python -m tests.test_litellm.ocr.fixtures.record --fixture-dir {default_directory}" ), id_builder=_fixture_id, ) diff --git a/tests/test_litellm/ocr/fixtures/azure.py b/tests/test_litellm/ocr/fixtures/azure.py index 119f2a5fca8..229cdc7205d 100644 --- a/tests/test_litellm/ocr/fixtures/azure.py +++ b/tests/test_litellm/ocr/fixtures/azure.py @@ -6,10 +6,10 @@ from typing import Final, cast from hypothesis import strategies as st from hypothesis.strategies import DrawFn, SearchStrategy -from tests.route_parity.fixture_recorder import ProviderSpec +from tests.route_parity.fixtures.recording import ProviderSpec from tests.test_litellm.ocr.fixtures.common import ( OcrFixtureClient, - OcrFixtureTarget, + OcrRecordingTarget, invoke_with_api_key, pdf_document, public_document_strategy, @@ -71,45 +71,45 @@ def azure_document_intelligence_input_strategy(draw: DrawFn) -> AzureDocumentInt ) -class AzureMistralFixtureSource: - def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]: - api_key: Final = environ.get("AZURE_AI_API_KEY") - upstream_base: Final = environ.get("AZURE_AI_API_BASE") - configured_model: Final = environ.get("AZURE_AI_OCR_MODEL") - if not api_key or not upstream_base or not configured_model: - return () - model: Final = configured_model if configured_model.startswith("azure_ai/") else f"azure_ai/{configured_model}" - return ( - OcrFixtureTarget( - name="azure-mistral", - provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), - strategy=cast( - SearchStrategy[OcrSdkInputBase], - mistral_input_strategy(MISTRAL_MODEL).map(lambda case_input: _as_azure_mistral(case_input, model)), - ), - invocation=invoke_with_api_key(client, api_key), - required_inputs=cast( - tuple[OcrSdkInputBase, ...], - tuple( - _as_azure_mistral(case_input, model) for case_input in required_mistral_inputs(MISTRAL_MODEL) - ), - ), +def azure_mistral_recording_targets( + environ: Mapping[str, str], client: OcrFixtureClient +) -> tuple[OcrRecordingTarget, ...]: + api_key: Final = environ.get("AZURE_AI_API_KEY") + upstream_base: Final = environ.get("AZURE_AI_API_BASE") + configured_model: Final = environ.get("AZURE_AI_OCR_MODEL") + if not api_key or not upstream_base or not configured_model: + return () + model: Final = configured_model if configured_model.startswith("azure_ai/") else f"azure_ai/{configured_model}" + return ( + OcrRecordingTarget( + name="azure-mistral", + provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), + strategy=cast( + SearchStrategy[OcrSdkInputBase], + mistral_input_strategy(MISTRAL_MODEL).map(lambda case_input: _as_azure_mistral(case_input, model)), ), - ) + invocation=invoke_with_api_key(client, api_key), + required_inputs=cast( + tuple[OcrSdkInputBase, ...], + tuple(_as_azure_mistral(case_input, model) for case_input in required_mistral_inputs(MISTRAL_MODEL)), + ), + ), + ) -class AzureDocumentIntelligenceFixtureSource: - def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]: - api_key: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_API_KEY") - upstream_base: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT") - if not api_key or not upstream_base: - return () - return ( - OcrFixtureTarget( - name="azure-document-intelligence", - provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), - strategy=cast(SearchStrategy[OcrSdkInputBase], azure_document_intelligence_input_strategy()), - invocation=invoke_with_api_key(client, api_key), - required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_document_intelligence_inputs()), - ), - ) +def azure_document_intelligence_recording_targets( + environ: Mapping[str, str], client: OcrFixtureClient +) -> tuple[OcrRecordingTarget, ...]: + api_key: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_API_KEY") + upstream_base: Final = environ.get("AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT") + if not api_key or not upstream_base: + return () + return ( + OcrRecordingTarget( + name="azure-document-intelligence", + provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), + strategy=cast(SearchStrategy[OcrSdkInputBase], azure_document_intelligence_input_strategy()), + invocation=invoke_with_api_key(client, api_key), + required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_document_intelligence_inputs()), + ), + ) diff --git a/tests/test_litellm/ocr/fixtures/common.py b/tests/test_litellm/ocr/fixtures/common.py index 74e4aa915ee..a60712eff57 100644 --- a/tests/test_litellm/ocr/fixtures/common.py +++ b/tests/test_litellm/ocr/fixtures/common.py @@ -9,7 +9,7 @@ from urllib.parse import quote from hypothesis import strategies as st from hypothesis.strategies import SearchStrategy -from tests.route_parity.fixture_generator import FixtureTarget +from tests.route_parity.fixtures.pipeline import RecordingTarget from tests.test_litellm.ocr.fixtures.models import ( JsonSchemaDefinition, JsonSchemaResponseFormat, @@ -18,7 +18,7 @@ from tests.test_litellm.ocr.fixtures.models import ( OcrSdkInputBase, ) -OcrFixtureTarget = FixtureTarget[OcrSdkInputBase] +OcrRecordingTarget = RecordingTarget[OcrSdkInputBase] class OcrFixtureClient(Protocol): diff --git a/tests/test_litellm/ocr/fixtures/generate.py b/tests/test_litellm/ocr/fixtures/generate.py deleted file mode 100644 index 4de560b4064..00000000000 --- a/tests/test_litellm/ocr/fixtures/generate.py +++ /dev/null @@ -1,99 +0,0 @@ -from __future__ import annotations - -import logging -import os -from collections.abc import Mapping -from pathlib import Path -from typing import Final, cast - -from dotenv import load_dotenv - -import litellm -from litellm.rust_bridge.ocr import use_litellm_rust -from tests.route_parity.fixture_generator import ( - FixtureSource, - discover_fixture_targets, - generate_target_fixtures, - parse_generator_args, -) -from tests.route_parity.fixture_generator import require_targets as require_fixture_targets -from tests.route_parity.fixture_recorder import fixture_directory -from tests.test_litellm.ocr.fixtures.azure import ( - AzureDocumentIntelligenceFixtureSource, - AzureMistralFixtureSource, - azure_document_intelligence_input_strategy, -) -from tests.test_litellm.ocr.fixtures.common import OcrFixtureClient, OcrFixtureTarget, OcrSdkCall -from tests.test_litellm.ocr.fixtures.mistral import ( - MistralFixtureSource, - mistral_input_strategy, -) -from tests.test_litellm.ocr.fixtures.models import OcrParityCase, OcrSdkInputBase -from tests.test_litellm.ocr.fixtures.reducto import ( - ReductoFixtureSource, - reducto_legacy_input_strategy, - reducto_v3_input_strategy, -) -from tests.test_litellm.ocr.fixtures.vertex import ( - VertexFixtureSource, - vertex_deepseek_input_strategy, -) - -__all__ = ( - "azure_document_intelligence_input_strategy", - "mistral_input_strategy", - "reducto_legacy_input_strategy", - "reducto_v3_input_strategy", - "vertex_deepseek_input_strategy", -) - -FIXTURE_DIR_ENV: Final = "LITELLM_OCR_FIXTURE_DIR" -OCR_FIXTURE_SOURCES: Final[tuple[FixtureSource[OcrSdkInputBase, OcrFixtureClient], ...]] = ( - MistralFixtureSource(), - AzureMistralFixtureSource(), - AzureDocumentIntelligenceFixtureSource(), - VertexFixtureSource(), - ReductoFixtureSource(), -) - - -class LiteLLMOcrFixtureClient: - def __init__(self, sdk_call: OcrSdkCall) -> None: - self.sdk_call: Final = sdk_call - - def execute(self, api_base: str, api_key: str, case_input: OcrSdkInputBase) -> None: - self.sdk_call(api_base=api_base, api_key=api_key, **case_input.as_sdk_kwargs()) - - -def discover_targets( - environ: Mapping[str, str], - client: OcrFixtureClient, -) -> tuple[OcrFixtureTarget, ...]: - return discover_fixture_targets(OCR_FIXTURE_SOURCES, environ, client) - - -def require_targets(targets: tuple[OcrFixtureTarget, ...]) -> tuple[OcrFixtureTarget, ...]: - return require_fixture_targets( - targets, - "No OCR fixture providers are configured. Set a supported provider API key and endpoint", - ) - - -def main() -> None: - logging.basicConfig(level=logging.INFO, format="%(message)s") - load_dotenv() - args: Final = parse_generator_args() - client: Final = LiteLLMOcrFixtureClient(cast(OcrSdkCall, litellm.ocr)) - targets: Final = require_targets(discover_targets(os.environ, client)) - root: Final = fixture_directory( - args.fixture_dir, - os.environ.get(FIXTURE_DIR_ENV), - Path(__file__).with_name("data"), - ) - use_litellm_rust(False, ocr=None, aocr=None) - for target in targets: - generate_target_fixtures(target, root, args.examples, args.concurrency, OcrParityCase) - - -if __name__ == "__main__": - main() diff --git a/tests/test_litellm/ocr/fixtures/mistral.py b/tests/test_litellm/ocr/fixtures/mistral.py index fdf08330ac1..4136c0585be 100644 --- a/tests/test_litellm/ocr/fixtures/mistral.py +++ b/tests/test_litellm/ocr/fixtures/mistral.py @@ -6,10 +6,10 @@ from typing import Final, cast from hypothesis import strategies as st from hypothesis.strategies import DrawFn, SearchStrategy -from tests.route_parity.fixture_recorder import ProviderSpec +from tests.route_parity.fixtures.recording import ProviderSpec from tests.test_litellm.ocr.fixtures.common import ( OcrFixtureClient, - OcrFixtureTarget, + OcrRecordingTarget, annotation_format, image_document, invoke_with_api_key, @@ -80,19 +80,18 @@ def required_mistral_inputs(model: str) -> tuple[MistralOcrSdkInput, ...]: return tuple(MistralOcrSdkInput.model_validate({"model": model, "document": document, **case}) for case in cases) -class MistralFixtureSource: - def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]: - api_key: Final = environ.get("MISTRAL_API_KEY") - if not api_key: - return () - configured: Final = environ.get("MISTRAL_API_BASE", "https://api.mistral.ai").rstrip("/") - upstream_base: Final = configured.removesuffix("/v1") - return ( - OcrFixtureTarget( - name="mistral-ocr", - provider_spec=ProviderSpec(upstream_base=upstream_base), - strategy=cast(SearchStrategy[OcrSdkInputBase], mistral_input_strategy(MISTRAL_MODEL)), - invocation=invoke_with_api_key(client, api_key), - required_inputs=cast(tuple[OcrSdkInputBase, ...], required_mistral_inputs(MISTRAL_MODEL)), - ), - ) +def mistral_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]: + api_key: Final = environ.get("MISTRAL_API_KEY") + if not api_key: + return () + configured: Final = environ.get("MISTRAL_API_BASE", "https://api.mistral.ai").rstrip("/") + upstream_base: Final = configured.removesuffix("/v1") + return ( + OcrRecordingTarget( + name="mistral-ocr", + provider_spec=ProviderSpec(upstream_base=upstream_base), + strategy=cast(SearchStrategy[OcrSdkInputBase], mistral_input_strategy(MISTRAL_MODEL)), + invocation=invoke_with_api_key(client, api_key), + required_inputs=cast(tuple[OcrSdkInputBase, ...], required_mistral_inputs(MISTRAL_MODEL)), + ), + ) diff --git a/tests/test_litellm/ocr/fixtures/record.py b/tests/test_litellm/ocr/fixtures/record.py new file mode 100644 index 00000000000..f4b2e4b14ee --- /dev/null +++ b/tests/test_litellm/ocr/fixtures/record.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import logging +import os +from collections.abc import Mapping +from pathlib import Path +from typing import Final, cast + +from dotenv import load_dotenv + +import litellm +from litellm.rust_bridge.ocr import use_litellm_rust +from tests.route_parity.fixtures.pipeline import parse_recording_args, record_fixtures +from tests.route_parity.fixtures.store import fixture_directory +from tests.test_litellm.ocr.fixtures.azure import ( + azure_document_intelligence_recording_targets, + azure_mistral_recording_targets, +) +from tests.test_litellm.ocr.fixtures.common import OcrFixtureClient, OcrRecordingTarget, OcrSdkCall +from tests.test_litellm.ocr.fixtures.mistral import mistral_recording_targets +from tests.test_litellm.ocr.fixtures.models import OcrParityCase, OcrSdkInputBase +from tests.test_litellm.ocr.fixtures.reducto import reducto_recording_targets +from tests.test_litellm.ocr.fixtures.vertex import vertex_recording_targets + +FIXTURE_DIR_ENV: Final = "LITELLM_OCR_FIXTURE_DIR" + + +class LiteLLMOcrFixtureClient: + def __init__(self, sdk_call: OcrSdkCall) -> None: + self.sdk_call: Final = sdk_call + + def execute(self, api_base: str, api_key: str, case_input: OcrSdkInputBase) -> None: + self.sdk_call(api_base=api_base, api_key=api_key, **case_input.as_sdk_kwargs()) + + +def discover_targets( + environ: Mapping[str, str], + client: OcrFixtureClient, +) -> tuple[OcrRecordingTarget, ...]: + return ( + *mistral_recording_targets(environ, client), + *azure_mistral_recording_targets(environ, client), + *azure_document_intelligence_recording_targets(environ, client), + *vertex_recording_targets(environ, client), + *reducto_recording_targets(environ, client), + ) + + +def require_targets(targets: tuple[OcrRecordingTarget, ...]) -> tuple[OcrRecordingTarget, ...]: + if targets: + return targets + raise SystemExit("No OCR fixture providers are configured. Set a supported provider API key and endpoint") + + +def main() -> int: + logging.basicConfig(level=logging.INFO, format="%(message)s") + load_dotenv() + args: Final = parse_recording_args() + client: Final = LiteLLMOcrFixtureClient(cast(OcrSdkCall, litellm.ocr)) + targets: Final = require_targets(discover_targets(os.environ, client)) + root: Final = fixture_directory( + args.fixture_dir, + os.environ.get(FIXTURE_DIR_ENV), + Path(__file__).with_name("data"), + ) + use_litellm_rust(False, ocr=None, aocr=None) + summary: Final = record_fixtures(targets, root, args.examples, args.concurrency, OcrParityCase) + return summary.exit_code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_litellm/ocr/fixtures/reducto.py b/tests/test_litellm/ocr/fixtures/reducto.py index ceb44460349..0c05f7bbd44 100644 --- a/tests/test_litellm/ocr/fixtures/reducto.py +++ b/tests/test_litellm/ocr/fixtures/reducto.py @@ -6,10 +6,10 @@ from typing import Final, cast from hypothesis import strategies as st from hypothesis.strategies import DrawFn, SearchStrategy -from tests.route_parity.fixture_recorder import ProviderSpec +from tests.route_parity.fixtures.recording import ProviderSpec from tests.test_litellm.ocr.fixtures.common import ( OcrFixtureClient, - OcrFixtureTarget, + OcrRecordingTarget, fixture_pdf_data_uri, invoke_with_api_key, ) @@ -155,33 +155,32 @@ def _required_v3_inputs(document: ReductoDocumentUrlDocument) -> tuple[ReductoPa ) -class ReductoFixtureSource: - def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]: - api_key: Final = environ.get("REDUCTO_API_KEY") - if not api_key: - return () - upstream_base: Final = environ.get("REDUCTO_API_BASE", _REDUCTO_API_BASE).rstrip("/") - document: Final = ReductoDocumentUrlDocument(type="document_url", document_url=fixture_pdf_data_uri()) - invocation: Final = invoke_with_api_key(client, api_key) - return ( - OcrFixtureTarget( - name="reducto-v3", - provider_spec=ProviderSpec(upstream_base=upstream_base), - strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_v3_input_strategy(document)), - invocation=invocation, - required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_v3_inputs(document)), - ), - OcrFixtureTarget( - name="reducto-legacy", - provider_spec=ProviderSpec(upstream_base=upstream_base), - strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_legacy_input_strategy(document)), - invocation=invocation, - required_inputs=cast( - tuple[OcrSdkInputBase, ...], - ( - ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document), - ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document, enhance={}), - ), +def reducto_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]: + api_key: Final = environ.get("REDUCTO_API_KEY") + if not api_key: + return () + upstream_base: Final = environ.get("REDUCTO_API_BASE", _REDUCTO_API_BASE).rstrip("/") + document: Final = ReductoDocumentUrlDocument(type="document_url", document_url=fixture_pdf_data_uri()) + invocation: Final = invoke_with_api_key(client, api_key) + return ( + OcrRecordingTarget( + name="reducto-v3", + provider_spec=ProviderSpec(upstream_base=upstream_base), + strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_v3_input_strategy(document)), + invocation=invocation, + required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_v3_inputs(document)), + ), + OcrRecordingTarget( + name="reducto-legacy", + provider_spec=ProviderSpec(upstream_base=upstream_base), + strategy=cast(SearchStrategy[OcrSdkInputBase], reducto_legacy_input_strategy(document)), + invocation=invocation, + required_inputs=cast( + tuple[OcrSdkInputBase, ...], + ( + ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document), + ReductoParseLegacySdkInput(model="reducto/parse-legacy", document=document, enhance={}), ), ), - ) + ), + ) diff --git a/tests/test_litellm/ocr/fixtures/vertex.py b/tests/test_litellm/ocr/fixtures/vertex.py index 1fb36c7c665..46a274c76b3 100644 --- a/tests/test_litellm/ocr/fixtures/vertex.py +++ b/tests/test_litellm/ocr/fixtures/vertex.py @@ -6,10 +6,10 @@ from typing import Final, cast from hypothesis import strategies as st from hypothesis.strategies import DrawFn, SearchStrategy -from tests.route_parity.fixture_recorder import ProviderSpec +from tests.route_parity.fixtures.recording import ProviderSpec from tests.test_litellm.ocr.fixtures.common import ( OcrFixtureClient, - OcrFixtureTarget, + OcrRecordingTarget, image_document, invoke_with_api_key, public_document_strategy, @@ -53,42 +53,41 @@ def vertex_deepseek_input_strategy(draw: DrawFn, project: str, location: str) -> ) -class VertexFixtureSource: - def targets(self, environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrFixtureTarget, ...]: - api_key: Final = environ.get("VERTEX_AI_API_KEY") - project: Final = environ.get("VERTEXAI_PROJECT") or environ.get("VERTEX_PROJECT") - location: Final = environ.get("VERTEXAI_LOCATION") or environ.get("VERTEX_LOCATION") or "us-central1" - if not api_key or not project: - return () - upstream_base: Final = environ.get("VERTEX_AI_API_BASE") or f"https://{location}-aiplatform.googleapis.com" - invocation: Final = invoke_with_api_key(client, api_key) - return ( - OcrFixtureTarget( - name="vertex-mistral", - provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), - strategy=cast( - SearchStrategy[OcrSdkInputBase], - st.builds( - _as_vertex_mistral, - case_input=mistral_input_strategy(MISTRAL_MODEL), - project=st.just(project), - location=st.just(location), - ), - ), - invocation=invocation, - required_inputs=cast( - tuple[OcrSdkInputBase, ...], - tuple( - _as_vertex_mistral(case_input, project, location) - for case_input in required_mistral_inputs(MISTRAL_MODEL) - ), +def vertex_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]: + api_key: Final = environ.get("VERTEX_AI_API_KEY") + project: Final = environ.get("VERTEXAI_PROJECT") or environ.get("VERTEX_PROJECT") + location: Final = environ.get("VERTEXAI_LOCATION") or environ.get("VERTEX_LOCATION") or "us-central1" + if not api_key or not project: + return () + upstream_base: Final = environ.get("VERTEX_AI_API_BASE") or f"https://{location}-aiplatform.googleapis.com" + invocation: Final = invoke_with_api_key(client, api_key) + return ( + OcrRecordingTarget( + name="vertex-mistral", + provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), + strategy=cast( + SearchStrategy[OcrSdkInputBase], + st.builds( + _as_vertex_mistral, + case_input=mistral_input_strategy(MISTRAL_MODEL), + project=st.just(project), + location=st.just(location), ), ), - OcrFixtureTarget( - name="vertex-deepseek", - provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), - strategy=cast(SearchStrategy[OcrSdkInputBase], vertex_deepseek_input_strategy(project, location)), - invocation=invocation, - required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_deepseek_inputs(project, location)), + invocation=invocation, + required_inputs=cast( + tuple[OcrSdkInputBase, ...], + tuple( + _as_vertex_mistral(case_input, project, location) + for case_input in required_mistral_inputs(MISTRAL_MODEL) + ), ), - ) + ), + OcrRecordingTarget( + name="vertex-deepseek", + provider_spec=ProviderSpec(upstream_base=upstream_base.rstrip("/")), + strategy=cast(SearchStrategy[OcrSdkInputBase], vertex_deepseek_input_strategy(project, location)), + invocation=invocation, + required_inputs=cast(tuple[OcrSdkInputBase, ...], _required_deepseek_inputs(project, location)), + ), + ) diff --git a/tests/test_litellm/ocr/test_fixture_models.py b/tests/test_litellm/ocr/test_fixture_models.py index 26264a97b6d..e8fc6d1a358 100644 --- a/tests/test_litellm/ocr/test_fixture_models.py +++ b/tests/test_litellm/ocr/test_fixture_models.py @@ -11,13 +11,8 @@ from litellm.llms.base_llm.ocr.transformation import BaseOCRConfig from litellm.llms.mistral.ocr.transformation import MistralOCRConfig from litellm.llms.reducto.ocr.transformation import ReductoParseLegacyConfig, ReductoParseV3Config from litellm.llms.vertex_ai.ocr.deepseek_transformation import VertexAIDeepSeekOCRConfig -from tests.test_litellm.ocr.fixtures.generate import ( - azure_document_intelligence_input_strategy, - mistral_input_strategy, - reducto_legacy_input_strategy, - reducto_v3_input_strategy, - vertex_deepseek_input_strategy, -) +from tests.test_litellm.ocr.fixtures.azure import azure_document_intelligence_input_strategy +from tests.test_litellm.ocr.fixtures.mistral import mistral_input_strategy from tests.test_litellm.ocr.fixtures.models import ( AzureDocumentIntelligenceOcrSdkInput, AzureMistralOcrSdkInput, @@ -39,6 +34,8 @@ from tests.test_litellm.ocr.fixtures.models import ( VertexDeepSeekOcrSdkInput, VertexMistralOcrSdkInput, ) +from tests.test_litellm.ocr.fixtures.reducto import reducto_legacy_input_strategy, reducto_v3_input_strategy +from tests.test_litellm.ocr.fixtures.vertex import vertex_deepseek_input_strategy COMMON_FIELDS: Final = frozenset( {"boundary", "model", "document", "custom_llm_provider", "vertex_project", "vertex_location"} diff --git a/tests/test_litellm/ocr/test_generate_fixtures.py b/tests/test_litellm/ocr/test_record_fixtures.py similarity index 91% rename from tests/test_litellm/ocr/test_generate_fixtures.py rename to tests/test_litellm/ocr/test_record_fixtures.py index 1004a69178c..dfd66ac1248 100644 --- a/tests/test_litellm/ocr/test_generate_fixtures.py +++ b/tests/test_litellm/ocr/test_record_fixtures.py @@ -7,13 +7,13 @@ from typing import Final import pytest -from tests.route_parity.fixture_recorder import generate_case_inputs -from tests.test_litellm.ocr.fixtures.generate import ( +from tests.route_parity.fixtures.inputs import generate_case_inputs +from tests.route_parity.fixtures.pipeline import parse_recording_args +from tests.test_litellm.ocr.fixtures.models import OcrSdkInputBase +from tests.test_litellm.ocr.fixtures.record import ( discover_targets, - parse_generator_args, require_targets, ) -from tests.test_litellm.ocr.fixtures.models import OcrSdkInputBase class _UnusedOcrClient: @@ -33,13 +33,13 @@ _UNUSED_OCR_CLIENT: Final = _UnusedOcrClient() def test_parse_args_has_no_model_selection() -> None: - args: Final = parse_generator_args(["--examples", "2", "--concurrency", "3", "--fixture-dir", "/tmp/ocr"]) + args: Final = parse_recording_args(["--examples", "2", "--concurrency", "3", "--fixture-dir", "/tmp/ocr"]) assert args.examples == 2 assert args.concurrency == 3 assert args.fixture_dir == Path("/tmp/ocr") with pytest.raises(SystemExit): - parse_generator_args(["--model", "mistral/mistral-ocr-latest"]) + parse_recording_args(["--model", "mistral/mistral-ocr-latest"]) @pytest.mark.parametrize( @@ -94,9 +94,9 @@ def test_azure_mistral_discovery_requires_and_normalizes_deployment_model() -> N } assert discover_targets(incomplete, _UNUSED_OCR_CLIENT) == () - target: Final = discover_targets({**incomplete, "AZURE_AI_OCR_MODEL": "mistral-ocr-deployment"}, _UNUSED_OCR_CLIENT)[ - 0 - ] + target: Final = discover_targets( + {**incomplete, "AZURE_AI_OCR_MODEL": "mistral-ocr-deployment"}, _UNUSED_OCR_CLIENT + )[0] assert target.required_inputs[0].canonical_input()["model"] == "azure_ai/mistral-ocr-deployment" diff --git a/tests/test_litellm/ocr/test_sdk_parity.py b/tests/test_litellm/ocr/test_sdk_parity.py index 1c6539da40a..4352a85fbe1 100644 --- a/tests/test_litellm/ocr/test_sdk_parity.py +++ b/tests/test_litellm/ocr/test_sdk_parity.py @@ -18,7 +18,7 @@ from litellm.rust_bridge import get_native_bridge from litellm.rust_bridge import ocr as rust_ocr_bridge from litellm.rust_bridge.ocr import RustAocr, RustOcr from tests.route_parity.compare import assert_model_parity, assert_parity, assert_request_parity -from tests.route_parity.fixture_recorder import recorded_fixtures +from tests.route_parity.fixtures.store import recorded_fixtures from tests.route_parity.inprocess import run_in_process from tests.route_parity.models import ( SDKCommand,