scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 123

test(ocr): bound fixture strategy domains

Browse source
This commit is contained in:
Yujong Lee 2026-09-01 15:35:14 -07:00
parent c41b740d48
commit 077d6e5e69
7 changed files with 656 additions and 255 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

115
tests/test_litellm/ocr/fixtures/azure.py
View file

@ -4,7 +4,7 @@ from collections.abc import Mapping
from typing import Final, Literal, cast
from hypothesis import strategies as st
from hypothesis.strategies import DrawFn, SearchStrategy
from hypothesis.strategies import SearchStrategy
from pydantic import Field, field_validator
from tests.route_parity.fixtures.recording import ProviderSpec
@ -13,15 +13,17 @@ from tests.test_litellm.ocr.fixtures.common import (
OcrFixtureClient,
OcrRecordingTarget,
invoke_with_api_key,
parameter_strategy,
pdf_document,
public_document_strategy,
sampled_list_strategy,
sampled_scalar_strategy,
)
from tests.test_litellm.ocr.fixtures.mistral import (
MISTRAL_MODEL,
MistralCompatibleOcrSdkInput,
MistralOcrSdkInput,
mistral_input_strategy,
required_mistral_inputs,
)
AzureMistralModel = Literal["azure_ai/mistral-document-ai-2512",]
@ -63,42 +65,80 @@ class AzureDocumentIntelligenceOcrSdkInput(OcrSdkInputBase):
def _as_azure_mistral(case_input: MistralOcrSdkInput, model: AzureMistralModel) -> AzureMistralOcrSdkInput:
values: Final = case_input.model_dump(mode="python", exclude={"boundary", "model", "custom_llm_provider"})
values: Final = case_input.model_dump(
mode="python",
exclude={"boundary", "model", "custom_llm_provider"},
exclude_unset=True,
)
return AzureMistralOcrSdkInput.model_validate({**values, "model": model})
def _required_document_intelligence_inputs() -> tuple[AzureDocumentIntelligenceOcrSdkInput, ...]:
document: Final = pdf_document()
cases: Final[tuple[dict[str, object], ...]] = (
{},
{"pages": [0, 1]},
{"features": ["languages"]},
{"req_format": "litellm"},
)
return tuple(
AzureDocumentIntelligenceOcrSdkInput.model_validate({"model": model, "document": document, **case})
for model in AZURE_DOCUMENT_INTELLIGENCE_MODELS
for case in cases
)
_AZURE_DOCUMENT_INTELLIGENCE_CANONICAL_MODEL: Final[AzureDocumentIntelligenceModel] = (
"azure_ai/doc-intelligence/prebuilt-layout"
)
_AZURE_DOCUMENT_INTELLIGENCE_DOCUMENT_MODEL: Final[AzureDocumentIntelligenceModel] = (
"azure_ai/doc-intelligence/prebuilt-document"
)
@st.composite
def azure_document_intelligence_input_strategy(draw: DrawFn) -> AzureDocumentIntelligenceOcrSdkInput:
optional_params: Final = draw(
st.fixed_dictionaries(
{},
optional={
"pages": st.sampled_from(([0], [0, 1], "1-2")),
"features": st.sampled_from((["languages"], ["keyValuePairs"], "languages,keyValuePairs")),
},
)
)
def _document_intelligence_input(
model: AzureDocumentIntelligenceModel,
document: OcrDocument,
optional_params: dict[str, object] | None = None,
) -> AzureDocumentIntelligenceOcrSdkInput:
return AzureDocumentIntelligenceOcrSdkInput.model_validate(
{
"model": draw(st.sampled_from(AZURE_DOCUMENT_INTELLIGENCE_MODELS)),
"document": draw(public_document_strategy()),
**optional_params,
}
{"model": model, "document": document, **(optional_params or {})}
)
def azure_document_intelligence_input_strategy() -> SearchStrategy[AzureDocumentIntelligenceOcrSdkInput]:
document: Final = pdf_document()
pages: Final = st.one_of(
parameter_strategy("pages", sampled_list_strategy(((0,), (0, 1)))),
parameter_strategy("pages", sampled_scalar_strategy(("1", "1,2", "1-2"))),
)
common_features: Final = parameter_strategy(
"features",
st.one_of(
sampled_list_strategy(
(("languages",), ("ocrHighResolution",), ("barcodes",), ("formulas",), ("styleFont",))
),
sampled_scalar_strategy(("languages,styleFont",)),
),
)
return st.one_of(
st.sampled_from(AZURE_DOCUMENT_INTELLIGENCE_MODELS).map(
lambda model: _document_intelligence_input(model, document)
),
public_document_strategy().map(
lambda selected_document: _document_intelligence_input(
_AZURE_DOCUMENT_INTELLIGENCE_CANONICAL_MODEL, selected_document
)
),
pages.map(
lambda optional_params: _document_intelligence_input(
_AZURE_DOCUMENT_INTELLIGENCE_CANONICAL_MODEL, document, optional_params
)
),
common_features.map(
lambda optional_params: _document_intelligence_input(
_AZURE_DOCUMENT_INTELLIGENCE_CANONICAL_MODEL, document, optional_params
)
),
st.just(
_document_intelligence_input(
_AZURE_DOCUMENT_INTELLIGENCE_DOCUMENT_MODEL,
document,
{"features": ["keyValuePairs"]},
)
),
st.just(
_document_intelligence_input(
_AZURE_DOCUMENT_INTELLIGENCE_CANONICAL_MODEL,
document,
{"req_format": "litellm"},
)
),
)
@ -116,20 +156,12 @@ def azure_mistral_recording_targets(
strategy=cast(
SearchStrategy[OcrSdkInputBase],
st.sampled_from(AZURE_MISTRAL_MODELS).flatmap(
lambda model: mistral_input_strategy(MISTRAL_MODEL).map(
lambda model: mistral_input_strategy(MISTRAL_MODEL, feature_level="2512").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 model in AZURE_MISTRAL_MODELS
for case_input in required_mistral_inputs(MISTRAL_MODEL)
),
),
),
)
@ -147,6 +179,5 @@ def azure_document_intelligence_recording_targets(
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()),
),
)

24
tests/test_litellm/ocr/fixtures/common.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
import base64
from dataclasses import dataclass, field
from pathlib import Path
from typing import Final, Protocol
from typing import Final, Protocol, TypeVar
from urllib.parse import quote
from hypothesis import strategies as st
@ -19,6 +19,7 @@ from tests.test_litellm.ocr.fixtures.base import (
)
OcrRecordingTarget = RecordingTarget[OcrSdkInputBase]
ValueT = TypeVar("ValueT")
class OcrFixtureClient(Protocol):
@ -57,6 +58,27 @@ def public_document_strategy() -> SearchStrategy[ImageUrlDocument | DocumentUrlD
return st.sampled_from((image_document("invoice 123", 24), pdf_document()))
def sampled_scalar_strategy(values: tuple[ValueT, ...]) -> SearchStrategy[ValueT]:
return st.sampled_from(values)
def sampled_list_strategy(values: tuple[tuple[ValueT, ...], ...]) -> SearchStrategy[list[ValueT]]:
return st.sampled_from(values).map(list)
def sampled_parameter_group_strategy(
values: tuple[tuple[tuple[str, object], ...], ...],
) -> SearchStrategy[dict[str, object]]:
return st.sampled_from(values).map(dict)
def parameter_strategy(name: str, values: SearchStrategy[ValueT]) -> SearchStrategy[dict[str, object]]:
def as_parameter(value: ValueT) -> dict[str, object]:
return {name: value}
return values.map(as_parameter)
def annotation_format(name: str) -> JsonSchemaResponseFormat:
return JsonSchemaResponseFormat(
type="json_schema",

137
tests/test_litellm/ocr/fixtures/mistral.py
View file

@ -20,7 +20,11 @@ from tests.test_litellm.ocr.fixtures.common import (
annotation_format,
image_document,
invoke_with_api_key,
parameter_strategy,
public_document_strategy,
sampled_list_strategy,
sampled_parameter_group_strategy,
sampled_scalar_strategy,
)
MistralModel = Literal[
@ -87,66 +91,83 @@ class MistralOcrSdkInput(MistralCompatibleOcrSdkInput):
MISTRAL_MODEL: Final[MistralModel] = "mistral/mistral-ocr-latest"
_VALUE_TEXT: Final = st.just("case-1")
MistralFeatureLevel = Literal["2505", "2512", "4"]
_MISTRAL_4_MODELS: Final = frozenset(
{
"mistral/mistral-ocr-4",
"mistral/mistral-ocr-4-0",
"mistral/mistral-ocr-4-1",
"mistral/mistral-ocr-latest",
}
)
_MISTRAL_2512_MODELS: Final = frozenset(
{*_MISTRAL_4_MODELS, "mistral/mistral-ocr-2512", "mistral/mistral-ocr-3", "mistral/mistral-ocr-3-0"}
)
def _feature_level(model: str) -> MistralFeatureLevel:
if model in _MISTRAL_4_MODELS:
return "4"
if model in _MISTRAL_2512_MODELS:
return "2512"
return "2505"
def mistral_optional_params_strategy(feature_level: MistralFeatureLevel) -> SearchStrategy[dict[str, object]]:
annotation: Final = annotation_format("document_title")
common: Final[tuple[SearchStrategy[dict[str, object]], ...]] = (
parameter_strategy("pages", sampled_list_strategy(((0,), (0, 1)))),
parameter_strategy("include_image_base64", sampled_scalar_strategy((False, True))),
parameter_strategy("image_limit", sampled_scalar_strategy((1,))),
parameter_strategy("image_min_size", sampled_scalar_strategy((300,))),
parameter_strategy(
"bbox_annotation_format",
sampled_scalar_strategy((annotation_format("bounding_boxes"),)),
),
parameter_strategy("document_annotation_format", sampled_scalar_strategy((annotation,))),
sampled_parameter_group_strategy(
(
(
("document_annotation_format", annotation),
("document_annotation_prompt", "Extract the visible title"),
),
)
),
parameter_strategy("confidence_scores_granularity", sampled_scalar_strategy(("page", "word"))),
parameter_strategy("id", sampled_scalar_strategy(("case-1",))),
)
feature_2512: Final[tuple[SearchStrategy[dict[str, object]], ...]] = (
parameter_strategy("extract_header", sampled_scalar_strategy((False, True))),
parameter_strategy("extract_footer", sampled_scalar_strategy((False, True))),
parameter_strategy("table_format", sampled_scalar_strategy(("markdown", "html"))),
)
feature_4: Final[tuple[SearchStrategy[dict[str, object]], ...]] = (
parameter_strategy("include_blocks", sampled_scalar_strategy((False, True))),
sampled_parameter_group_strategy(((("include_blocks", True), ("confidence_scores_granularity", "block")),)),
)
return st.one_of(
*common,
*(feature_2512 if feature_level in {"2512", "4"} else ()),
*(feature_4 if feature_level == "4" else ()),
)
@st.composite
def mistral_input_strategy(draw: DrawFn, model: str) -> MistralOcrSdkInput:
document: Final = draw(public_document_strategy())
annotation: Final = draw(
st.sampled_from(
(
{},
{"document_annotation_format": annotation_format("document_title")},
{
"document_annotation_format": annotation_format("prompted_document_title"),
"document_annotation_prompt": "Extract the visible title",
},
)
def mistral_input_strategy(
draw: DrawFn,
model: str,
feature_level: MistralFeatureLevel | None = None,
) -> MistralOcrSdkInput:
canonical_document: Final = image_document("invoice 123", 24)
values: Final = draw(
st.one_of(
public_document_strategy().map(lambda document: {"document": document}),
mistral_optional_params_strategy(feature_level or _feature_level(model)).map(
lambda optional_params: {"document": canonical_document, **optional_params}
),
)
)
optional_params: Final = draw(
st.fixed_dictionaries(
{},
optional={
"pages": st.just([0]),
"include_image_base64": st.booleans(),
"image_limit": st.just(1),
"image_min_size": st.just(300),
"bbox_annotation_format": st.just(annotation_format("bounding_boxes")),
"extract_header": st.booleans(),
"extract_footer": st.booleans(),
"table_format": st.just("markdown"),
"confidence_scores_granularity": st.just("page"),
"include_blocks": st.booleans(),
"id": _VALUE_TEXT,
},
)
)
return MistralOcrSdkInput.model_validate({"model": model, "document": document, **annotation, **optional_params})
def required_mistral_inputs(model: str) -> tuple[MistralOcrSdkInput, ...]:
document: Final = image_document("invoice 123", 24)
annotation: Final = annotation_format("document_title")
bbox_annotation: Final = annotation_format("bounding_boxes")
cases: Final[tuple[dict[str, object], ...]] = (
{},
{"pages": [0]},
{"include_image_base64": True},
{"image_limit": 1},
{"image_min_size": 300},
{"bbox_annotation_format": bbox_annotation},
{"document_annotation_format": annotation},
{"document_annotation_format": annotation, "document_annotation_prompt": "Extract the visible title"},
{"extract_header": True},
{"extract_footer": True},
{"table_format": "markdown"},
{"confidence_scores_granularity": "page"},
{"include_blocks": False},
{"id": "case-1"},
)
return tuple(MistralOcrSdkInput.model_validate({"model": model, "document": document, **case}) for case in cases)
return MistralOcrSdkInput.model_validate({"model": model, **values})
def mistral_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]:
@ -161,12 +182,8 @@ def mistral_recording_targets(environ: Mapping[str, str], client: OcrFixtureClie
provider_spec=ProviderSpec(upstream_base=upstream_base),
strategy=cast(
SearchStrategy[OcrSdkInputBase],
st.sampled_from(MISTRAL_MODELS).flatmap(mistral_input_strategy),
sampled_scalar_strategy(MISTRAL_MODELS).flatmap(mistral_input_strategy),
),
invocation=invoke_with_api_key(client, api_key),
required_inputs=cast(
tuple[OcrSdkInputBase, ...],
tuple(case_input for model in MISTRAL_MODELS for case_input in required_mistral_inputs(model)),
),
),
)

211
tests/test_litellm/ocr/fixtures/reducto.py
View file

@ -6,7 +6,7 @@ from collections.abc import Mapping
from typing import Annotated, Final, Literal, cast
from hypothesis import strategies as st
from hypothesis.strategies import DrawFn, SearchStrategy
from hypothesis.strategies import SearchStrategy
from pydantic import Field, field_validator, model_validator
from typing_extensions import Self
@ -18,6 +18,9 @@ from tests.test_litellm.ocr.fixtures.common import (
OcrRecordingTarget,
fixture_pdf_data_uri,
invoke_with_api_key,
parameter_strategy,
sampled_list_strategy,
sampled_scalar_strategy,
)
@ -65,6 +68,7 @@ ReductoDocument = Annotated[
]
ReductoTableOutputFormat = Literal["html", "json", "md", "jsonbbox", "dynamic", "csv"]
ReductoReturnImage = Literal["figure", "table", "page"]
ReductoFormattingInclude = Literal[
"change_tracking",
"highlight",
@ -87,6 +91,19 @@ ReductoBlockType = Literal[
"Comment",
"Signature",
]
_REDUCTO_FILTER_BLOCK_GROUPS: Final[tuple[tuple[ReductoBlockType, ...], ...]] = (
(),
("Header",),
("Header", "Footer", "Page Number"),
("Figure", "Table", "Key Value"),
)
_REDUCTO_RETURN_IMAGE_GROUPS: Final[tuple[tuple[ReductoReturnImage, ...], ...]] = (
(),
("figure",),
("table",),
("page",),
("figure", "table", "page"),
)
class ReductoFormatting(FixtureModel):
@ -164,7 +181,7 @@ class ReductoSettings(FixtureModel):
force_url_result: bool = False
force_file_extension: str | None = None
return_ocr_data: bool = False
return_images: list[Literal["figure", "table", "page"]] = Field(default_factory=list)
return_images: list[ReductoReturnImage] = Field(default_factory=list)
embed_pdf_metadata: bool = False
embed_pdf_metadata_dpi: int = Field(default=100, ge=50, le=250)
persist_results: bool = False
@ -176,9 +193,7 @@ class ReductoSettings(FixtureModel):
@field_validator("return_images")
@classmethod
def validate_unique_images(
cls, value: list[Literal["figure", "table", "page"]]
) -> list[Literal["figure", "table", "page"]]:
def validate_unique_images(cls, value: list[ReductoReturnImage]) -> list[ReductoReturnImage]:
if len(value) != len(set(value)):
raise ValueError("settings.return_images entries must be unique")
return value
@ -218,94 +233,121 @@ _REDUCTO_API_BASE: Final = "https://platform.reducto.ai"
def _formatting_strategy() -> SearchStrategy[ReductoFormatting]:
return st.builds(
ReductoFormatting,
add_page_markers=st.booleans(),
table_output_format=st.sampled_from(("html", "json", "md", "jsonbbox", "dynamic", "csv")),
merge_tables=st.booleans(),
include=st.sampled_from(
(
[],
["hyperlinks"],
["change_tracking", "highlight", "comments"],
["signatures", "ignore_watermarks"],
)
values: Final = st.one_of(
parameter_strategy(
"table_output_format",
sampled_scalar_strategy(("dynamic", "html", "md", "json", "csv", "jsonbbox")),
),
parameter_strategy("add_page_markers", sampled_scalar_strategy((False, True))),
parameter_strategy("merge_tables", sampled_scalar_strategy((False, True))),
parameter_strategy(
"include",
sampled_list_strategy(
(
(),
("hyperlinks",),
("change_tracking", "highlight", "comments"),
("signatures", "ignore_watermarks"),
)
),
),
)
return values.map(ReductoFormatting.model_validate)
def _chunking_strategy() -> SearchStrategy[ReductoChunking]:
return st.one_of(
st.builds(
ReductoChunking,
chunk_mode=st.sampled_from(("section", "page", "disabled", "block", "page_sections")),
chunk_size=st.just(None),
chunk_overlap=st.just(0),
st.sampled_from(("disabled", "section", "page", "block", "page_sections")).map(
lambda mode: ReductoChunking(chunk_mode=mode)
),
st.builds(
ReductoChunking,
chunk_mode=st.just("variable"),
chunk_size=st.sampled_from((250, 1000, 1500)),
chunk_overlap=st.sampled_from((0, 32, 128)),
st.just(ReductoChunking(chunk_mode="variable")),
sampled_scalar_strategy((250, 1000, 1500)).map(
lambda size: ReductoChunking(chunk_mode="variable", chunk_size=size)
),
sampled_scalar_strategy((32, 128)).map(
lambda overlap: ReductoChunking(chunk_mode="variable", chunk_size=1000, chunk_overlap=overlap)
),
)
def _retrieval_strategy() -> SearchStrategy[ReductoRetrieval]:
return st.builds(
ReductoRetrieval,
chunking=_chunking_strategy(),
filter_blocks=st.sampled_from(
(
[],
["Header"],
["Header", "Footer", "Page Number"],
["Figure", "Table", "Key Value"],
filter_blocks: Final[SearchStrategy[list[ReductoBlockType]]] = sampled_list_strategy(_REDUCTO_FILTER_BLOCK_GROUPS)
return st.one_of(
_chunking_strategy().map(lambda chunking: ReductoRetrieval(chunking=chunking)),
filter_blocks.map(lambda selected_blocks: ReductoRetrieval(filter_blocks=selected_blocks)),
sampled_scalar_strategy((False, True)).map(
lambda optimized: ReductoRetrieval(
chunking=ReductoChunking(chunk_mode="variable"),
embedding_optimized=optimized,
)
),
embedding_optimized=st.booleans(),
)
def _settings_strategy() -> SearchStrategy[ReductoSettings]:
return st.builds(
ReductoSettings,
ocr_system=st.sampled_from(("standard", "legacy")),
extraction_mode=st.sampled_from(("ocr", "hybrid")),
force_url_result=st.just(False),
return_ocr_data=st.booleans(),
return_images=st.sampled_from(([], ["figure"], ["table"], ["page"], ["figure", "table", "page"])),
embed_pdf_metadata=st.booleans(),
embed_pdf_metadata_dpi=st.sampled_from((50, 100, 250)),
persist_results=st.just(False),
timeout=st.sampled_from((None, 300.0, 900.0)),
page_range=st.sampled_from((None, [1], [1, 2], ["Sheet1"])),
return_images: Final[SearchStrategy[list[ReductoReturnImage]]] = sampled_list_strategy(_REDUCTO_RETURN_IMAGE_GROUPS)
page_ranges: Final = st.one_of(
st.just(ReductoPageRange(start=1, end=1)),
st.just(ReductoPageRange(start=1, end=3)),
sampled_list_strategy(
(
(
ReductoPageRange(start=1, end=2),
ReductoPageRange(start=4, end=5),
),
)
),
)
return st.one_of(
st.sampled_from(("standard", "legacy")).map(lambda value: ReductoSettings(ocr_system=value)),
st.sampled_from(("hybrid", "ocr")).map(lambda value: ReductoSettings(extraction_mode=value)),
st.just(ReductoSettings(force_url_result=True)),
st.just(ReductoSettings(return_ocr_data=True)),
return_images.map(lambda selected_images: ReductoSettings(return_images=selected_images)),
sampled_scalar_strategy((50, 100, 250)).map(
lambda dpi: ReductoSettings(embed_pdf_metadata=True, embed_pdf_metadata_dpi=dpi)
),
sampled_scalar_strategy((300.0, 900.0)).map(lambda timeout: ReductoSettings(timeout=timeout)),
page_ranges.map(lambda page_range: ReductoSettings(page_range=page_range)),
)
@st.composite
def reducto_v3_input_strategy(
draw: DrawFn, document: ReductoDocumentUrlDocument | None = None
) -> ReductoParseV3SdkInput:
model, custom_llm_provider = draw(st.sampled_from((("reducto/parse-v3", None), ("parse-v3", "reducto"))))
options: Final = draw(
st.fixed_dictionaries(
{},
optional={
"formatting": _formatting_strategy(),
"retrieval": _retrieval_strategy(),
"settings": _settings_strategy(),
},
)
document: ReductoDocumentUrlDocument | None = None,
) -> SearchStrategy[ReductoParseV3SdkInput]:
selected_document: Final = document or ReductoDocumentUrlDocument(
type="document_url", document_url="reducto://fixture-document.pdf"
)
return ReductoParseV3SdkInput.model_validate(
{
"model": model,
"custom_llm_provider": custom_llm_provider,
"document": document
or ReductoDocumentUrlDocument(type="document_url", document_url="reducto://fixture-document.pdf"),
**options,
}
return st.one_of(
st.just(ReductoParseV3SdkInput(model="reducto/parse-v3", document=selected_document)),
st.just(
ReductoParseV3SdkInput(
model="parse-v3",
custom_llm_provider="reducto",
document=selected_document,
)
),
_formatting_strategy().map(
lambda formatting: ReductoParseV3SdkInput(
model="reducto/parse-v3",
document=selected_document,
formatting=formatting,
)
),
_retrieval_strategy().map(
lambda retrieval: ReductoParseV3SdkInput(
model="reducto/parse-v3",
document=selected_document,
retrieval=retrieval,
)
),
_settings_strategy().map(
lambda settings: ReductoParseV3SdkInput(
model="reducto/parse-v3",
document=selected_document,
settings=settings,
)
),
)
@ -324,27 +366,6 @@ def reducto_legacy_input_strategy(
)
def _required_v3_inputs(document: ReductoDocumentUrlDocument) -> tuple[ReductoParseV3SdkInput, ...]:
return (
ReductoParseV3SdkInput(model="reducto/parse-v3", document=document),
ReductoParseV3SdkInput(
model="reducto/parse-v3",
document=document,
formatting=ReductoFormatting(table_output_format="md"),
),
ReductoParseV3SdkInput(
model="reducto/parse-v3",
document=document,
retrieval=ReductoRetrieval(chunking=ReductoChunking(chunk_mode="page")),
),
ReductoParseV3SdkInput(
model="reducto/parse-v3",
document=document,
settings=ReductoSettings(return_ocr_data=True),
),
)
def reducto_recording_targets(environ: Mapping[str, str], client: OcrFixtureClient) -> tuple[OcrRecordingTarget, ...]:
api_key: Final = environ.get("REDUCTO_API_KEY")
if not api_key:
@ -358,19 +379,11 @@ def reducto_recording_targets(environ: Mapping[str, str], client: OcrFixtureClie
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={}),
),
),
),
)

36
tests/test_litellm/ocr/fixtures/vertex.py
View file

@ -12,16 +12,15 @@ from tests.test_litellm.ocr.fixtures.base import OcrDocument, OcrSdkInputBase
from tests.test_litellm.ocr.fixtures.common import (
OcrFixtureClient,
OcrRecordingTarget,
image_document,
invoke_with_api_key,
public_document_strategy,
sampled_scalar_strategy,
)
from tests.test_litellm.ocr.fixtures.mistral import (
MISTRAL_MODEL,
MistralCompatibleOcrSdkInput,
MistralOcrSdkInput,
mistral_input_strategy,
required_mistral_inputs,
)
VertexMistralModel = Literal["vertex_ai/mistral-ocr-2505"]
@ -54,29 +53,21 @@ def _as_vertex_mistral(
location: str,
model: VertexMistralModel,
) -> VertexMistralOcrSdkInput:
values: Final = case_input.model_dump(mode="python", exclude={"boundary", "model", "custom_llm_provider"})
values: Final = case_input.model_dump(
mode="python",
exclude={"boundary", "model", "custom_llm_provider"},
exclude_unset=True,
)
return VertexMistralOcrSdkInput.model_validate(
{**values, "model": model, "vertex_project": project, "vertex_location": location}
)
def _required_deepseek_inputs(project: str, location: str) -> tuple[VertexDeepSeekOcrSdkInput, ...]:
cases: Final = (
VertexDeepSeekOcrSdkInput(
model=model,
document=image_document("invoice 123", 24),
vertex_project=project,
vertex_location=location,
)
for model in VERTEX_DEEPSEEK_MODELS
)
return tuple(cases)
@st.composite
def vertex_deepseek_input_strategy(draw: DrawFn, project: str, location: str) -> VertexDeepSeekOcrSdkInput:
return VertexDeepSeekOcrSdkInput.model_validate(
{
"model": draw(sampled_scalar_strategy(VERTEX_DEEPSEEK_MODELS)),
"document": draw(public_document_strategy()),
"vertex_project": project,
"vertex_location": location,
@ -102,25 +93,16 @@ def vertex_recording_targets(environ: Mapping[str, str], client: OcrFixtureClien
_as_vertex_mistral,
project=st.just(project),
location=st.just(location),
model=st.sampled_from(VERTEX_MISTRAL_MODELS),
case_input=mistral_input_strategy(MISTRAL_MODEL),
model=sampled_scalar_strategy(VERTEX_MISTRAL_MODELS),
case_input=mistral_input_strategy(MISTRAL_MODEL, feature_level="2505"),
),
),
invocation=invocation,
required_inputs=cast(
tuple[OcrSdkInputBase, ...],
tuple(
_as_vertex_mistral(case_input, project, location, model)
for model in VERTEX_MISTRAL_MODELS
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)),
),
)

287
tests/test_litellm/ocr/test_fixture_models.py
View file

@ -3,10 +3,12 @@ from __future__ import annotations
from collections.abc import Callable
from datetime import date
from pathlib import Path
from typing import Final, cast
from typing import Final, TypeVar, cast
import pytest
from hypothesis import given, settings
from hypothesis import find, given, settings
from hypothesis import strategies as st
from hypothesis.strategies import DataObject, SearchStrategy
from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError
from litellm.llms.azure_ai.ocr.document_intelligence.transformation import AzureDocumentIntelligenceOCRConfig
@ -71,6 +73,58 @@ ACTIVE_OCR_MODELS: Final = frozenset(
*REDUCTO_LEGACY_MODELS,
)
)
_MISTRAL_2512_OR_NEWER: Final = frozenset(
{
"mistral/mistral-ocr-2512",
"mistral/mistral-ocr-3",
"mistral/mistral-ocr-3-0",
"mistral/mistral-ocr-4",
"mistral/mistral-ocr-4-0",
"mistral/mistral-ocr-4-1",
"mistral/mistral-ocr-latest",
}
)
_MISTRAL_4_OR_NEWER: Final = frozenset(
{
"mistral/mistral-ocr-4",
"mistral/mistral-ocr-4-0",
"mistral/mistral-ocr-4-1",
"mistral/mistral-ocr-latest",
}
)
_MISTRAL_OPTION_GROUPS: Final = frozenset(
{
frozenset[str](),
*(
frozenset({field})
for field in (
"pages",
"include_image_base64",
"image_limit",
"image_min_size",
"bbox_annotation_format",
"document_annotation_format",
"extract_header",
"extract_footer",
"table_format",
"confidence_scores_granularity",
"include_blocks",
"id",
)
),
frozenset({"document_annotation_format", "document_annotation_prompt"}),
frozenset({"include_blocks", "confidence_scores_granularity"}),
}
)
_FIND_SETTINGS: Final = settings(max_examples=2_000, deadline=None, derandomize=True, database=None)
_FixtureInputT = TypeVar("_FixtureInputT")
def _find_fixture(
strategy: SearchStrategy[_FixtureInputT],
predicate: Callable[[_FixtureInputT], bool],
) -> _FixtureInputT:
return find(strategy, predicate, settings=_FIND_SETTINGS)
class _ModelRegistryEntry(BaseModel):
@ -377,16 +431,190 @@ def test_reducto_nested_constraints() -> None:
ReductoFormatting(include=["hyperlinks", "hyperlinks"])
@settings(max_examples=50, deadline=None)
@given(sdk_input=mistral_input_strategy("mistral/mistral-ocr-4-1"))
def test_mistral_strategy_only_generates_valid_sdk_inputs(sdk_input: MistralOcrSdkInput) -> None:
@settings(max_examples=100, deadline=None)
@given(model=st.sampled_from(MISTRAL_MODELS), data=st.data())
def test_mistral_strategy_only_generates_bounded_valid_sdk_inputs(model: str, data: DataObject) -> None:
sdk_input: Final = data.draw(mistral_input_strategy(model))
assert MistralOcrSdkInput.model_validate(sdk_input.canonical_input()) == sdk_input
optional_fields: Final = frozenset(sdk_input.model_fields_set) - {"model", "document"}
assert optional_fields in _MISTRAL_OPTION_GROUPS
if sdk_input.pages is not None:
assert sdk_input.pages in ([0], [0, 1])
if sdk_input.image_limit is not None:
assert sdk_input.image_limit == 1
if sdk_input.image_min_size is not None:
assert sdk_input.image_min_size == 300
if sdk_input.table_format is not None:
assert sdk_input.table_format in {"markdown", "html"}
if sdk_input.confidence_scores_granularity is not None:
assert sdk_input.confidence_scores_granularity in {"page", "word", "block"}
if sdk_input.confidence_scores_granularity == "block":
assert sdk_input.include_blocks is True
if model not in _MISTRAL_2512_OR_NEWER:
assert optional_fields.isdisjoint({"extract_header", "extract_footer", "table_format"})
if model not in _MISTRAL_4_OR_NEWER:
assert "include_blocks" not in optional_fields
@pytest.mark.parametrize(
("field", "value"),
(
("pages", [0]),
("pages", [0, 1]),
("include_image_base64", False),
("include_image_base64", True),
("image_limit", 1),
("image_min_size", 300),
("extract_header", False),
("extract_header", True),
("extract_footer", False),
("extract_footer", True),
("table_format", "markdown"),
("table_format", "html"),
("confidence_scores_granularity", "page"),
("confidence_scores_granularity", "word"),
("confidence_scores_granularity", "block"),
("include_blocks", False),
("include_blocks", True),
("id", "case-1"),
),
)
def test_mistral_strategy_reaches_every_finite_scalar_value(field: str, value: object) -> None:
sdk_input: Final = _find_fixture(
mistral_input_strategy("mistral/mistral-ocr-4-1"),
lambda candidate: field in candidate.model_fields_set and getattr(candidate, field) == value,
)
assert getattr(sdk_input, field) == value
@settings(max_examples=50, deadline=None)
@given(sdk_input=reducto_v3_input_strategy())
def test_reducto_v3_strategy_only_generates_valid_sdk_inputs(sdk_input: ReductoParseV3SdkInput) -> None:
def test_reducto_v3_strategy_only_generates_bounded_valid_sdk_inputs(sdk_input: ReductoParseV3SdkInput) -> None:
assert ReductoParseV3SdkInput.model_validate(sdk_input.canonical_input()) == sdk_input
option_groups: Final = frozenset(sdk_input.model_fields_set) & {"formatting", "retrieval", "settings"}
assert len(option_groups) <= 1
if "formatting" in option_groups:
assert len(sdk_input.formatting.model_fields_set) == 1
assert sdk_input.formatting.table_output_format in {"dynamic", "html", "md", "json", "csv", "jsonbbox"}
assert tuple(sdk_input.formatting.include) in {
(),
("hyperlinks",),
("change_tracking", "highlight", "comments"),
("signatures", "ignore_watermarks"),
}
if "retrieval" in option_groups:
retrieval_fields: Final = frozenset(sdk_input.retrieval.model_fields_set)
assert retrieval_fields in {
frozenset({"chunking"}),
frozenset({"filter_blocks"}),
frozenset({"chunking", "embedding_optimized"}),
}
chunking: Final = sdk_input.retrieval.chunking
if chunking.chunk_size is not None or chunking.chunk_overlap != 0:
assert chunking.chunk_mode == "variable"
if "embedding_optimized" in retrieval_fields:
assert chunking.chunk_mode == "variable"
if "settings" in option_groups:
settings_fields: Final = frozenset(sdk_input.settings.model_fields_set)
assert settings_fields in {
frozenset({"ocr_system"}),
frozenset({"extraction_mode"}),
frozenset({"force_url_result"}),
frozenset({"return_ocr_data"}),
frozenset({"return_images"}),
frozenset({"embed_pdf_metadata", "embed_pdf_metadata_dpi"}),
frozenset({"timeout"}),
frozenset({"page_range"}),
}
assert "persist_results" not in settings_fields
if "embed_pdf_metadata_dpi" in settings_fields:
assert sdk_input.settings.embed_pdf_metadata is True
assert sdk_input.settings.embed_pdf_metadata_dpi in {50, 100, 250}
if sdk_input.settings.page_range is not None:
ranges: Final = (
sdk_input.settings.page_range
if isinstance(sdk_input.settings.page_range, list)
else [sdk_input.settings.page_range]
)
assert all(isinstance(page_range, ReductoPageRange) for page_range in ranges)
@pytest.mark.parametrize("table_format", ("dynamic", "html", "md", "json", "csv", "jsonbbox"))
def test_reducto_v3_strategy_reaches_every_table_format(table_format: str) -> None:
sdk_input: Final = _find_fixture(
reducto_v3_input_strategy(),
lambda candidate: (
"formatting" in candidate.model_fields_set
and "table_output_format" in candidate.formatting.model_fields_set
and candidate.formatting.table_output_format == table_format
),
)
assert sdk_input.formatting.table_output_format == table_format
@pytest.mark.parametrize("chunk_mode", ("variable", "section", "page", "disabled", "block", "page_sections"))
def test_reducto_v3_strategy_reaches_every_chunk_mode(chunk_mode: str) -> None:
sdk_input: Final = _find_fixture(
reducto_v3_input_strategy(),
lambda candidate: (
"retrieval" in candidate.model_fields_set
and "chunking" in candidate.retrieval.model_fields_set
and candidate.retrieval.chunking.chunk_mode == chunk_mode
),
)
assert sdk_input.retrieval.chunking.chunk_mode == chunk_mode
@pytest.mark.parametrize("chunk_size", (250, 1000, 1500))
def test_reducto_v3_strategy_reaches_every_chunk_size(chunk_size: int) -> None:
sdk_input: Final = _find_fixture(
reducto_v3_input_strategy(),
lambda candidate: candidate.retrieval.chunking.chunk_size == chunk_size,
)
assert sdk_input.retrieval.chunking.chunk_mode == "variable"
assert sdk_input.retrieval.chunking.chunk_size == chunk_size
@pytest.mark.parametrize("dpi", (50, 100, 250))
def test_reducto_v3_strategy_reaches_every_metadata_dpi(dpi: int) -> None:
sdk_input: Final = _find_fixture(
reducto_v3_input_strategy(),
lambda candidate: (
"settings" in candidate.model_fields_set
and "embed_pdf_metadata_dpi" in candidate.settings.model_fields_set
and candidate.settings.embed_pdf_metadata_dpi == dpi
),
)
assert sdk_input.settings.embed_pdf_metadata is True
assert sdk_input.settings.embed_pdf_metadata_dpi == dpi
@pytest.mark.parametrize(
"page_range",
(
{"start": 1, "end": 1},
{"start": 1, "end": 3},
[{"start": 1, "end": 2}, {"start": 4, "end": 5}],
),
)
def test_reducto_v3_strategy_reaches_every_page_range_shape(page_range: object) -> None:
sdk_input: Final = _find_fixture(
reducto_v3_input_strategy(),
lambda candidate: (
cast(
dict[str, object],
candidate.settings.model_dump(mode="json", exclude_unset=True),
).get("page_range")
== page_range
),
)
assert sdk_input.settings.model_dump(mode="json", exclude_unset=True)["page_range"] == page_range
@settings(max_examples=10, deadline=None)
@ -402,6 +630,53 @@ def test_azure_document_intelligence_strategy_only_generates_litellm_inputs(
) -> None:
assert sdk_input.req_format == "litellm"
assert "boundary" not in sdk_input.as_sdk_kwargs()
optional_fields: Final = frozenset(sdk_input.model_fields_set) - {"model", "document"}
assert optional_fields in {
frozenset[str](),
frozenset({"pages"}),
frozenset({"features"}),
frozenset({"req_format"}),
}
if sdk_input.pages is not None:
assert sdk_input.pages in ([0], [0, 1], "1", "1,2", "1-2")
if isinstance(sdk_input.features, list):
assert tuple(sdk_input.features) in {
("languages",),
("ocrHighResolution",),
("barcodes",),
("formulas",),
("styleFont",),
("keyValuePairs",),
}
if isinstance(sdk_input.features, str):
assert sdk_input.features == "languages,styleFont"
@pytest.mark.parametrize(
("field", "value"),
(
("pages", [0]),
("pages", [0, 1]),
("pages", "1"),
("pages", "1,2"),
("pages", "1-2"),
("features", ["languages"]),
("features", ["ocrHighResolution"]),
("features", ["barcodes"]),
("features", ["formulas"]),
("features", ["styleFont"]),
("features", ["keyValuePairs"]),
("features", "languages,styleFont"),
("req_format", "litellm"),
),
)
def test_azure_document_intelligence_strategy_reaches_every_finite_value(field: str, value: object) -> None:
sdk_input: Final = _find_fixture(
azure_document_intelligence_input_strategy(),
lambda candidate: field in candidate.model_fields_set and getattr(candidate, field) == value,
)
assert getattr(sdk_input, field) == value
@settings(max_examples=30, deadline=None)

101
tests/test_litellm/ocr/test_record_fixtures.py
View file

@ -1,11 +1,14 @@
from __future__ import annotations
import queue
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import Final
import pytest
from hypothesis import find, settings
from hypothesis.strategies import SearchStrategy
from tests.route_parity.fixtures.inputs import generate_case_inputs
from tests.route_parity.fixtures.pipeline import parse_recording_args
@ -54,6 +57,9 @@ _MISTRAL_PARAMS: Final = frozenset(
"id",
}
)
_MISTRAL_2512_PARAMS: Final = _MISTRAL_PARAMS - {"include_blocks"}
_MISTRAL_2505_PARAMS: Final = _MISTRAL_2512_PARAMS - {"extract_header", "extract_footer", "table_format"}
_FIND_SETTINGS: Final = settings(max_examples=2_000, deadline=None, derandomize=True, database=None)
def _model(case_input: OcrSdkInputBase) -> str:
@ -62,6 +68,13 @@ def _model(case_input: OcrSdkInputBase) -> str:
return model
def _find_input(
strategy: SearchStrategy[OcrSdkInputBase],
predicate: Callable[[OcrSdkInputBase], bool],
) -> OcrSdkInputBase:
return find(strategy, predicate, settings=_FIND_SETTINGS)
def test_parse_args_has_no_model_selection() -> None:
args: Final = parse_recording_args(["--examples", "2", "--concurrency", "3", "--fixture-dir", "/tmp/ocr"])
@ -123,7 +136,16 @@ def test_azure_mistral_discovery_enumerates_registered_models() -> None:
}
target: Final = discover_targets(environ, _UNUSED_OCR_CLIENT)[0]
assert {_model(case_input) for case_input in target.required_inputs} == set(AZURE_MISTRAL_MODELS)
for model in AZURE_MISTRAL_MODELS:
assert (
_model(
_find_input(
target.strategy,
lambda case_input, expected_model=model: _model(case_input) == expected_model,
)
)
== model
)
@pytest.mark.parametrize(
@ -152,14 +174,6 @@ def test_mistral_target_uses_canonical_model_and_normalized_base(
case_inputs: Final = generate_case_inputs(target.strategy, examples=1)
assert len(case_inputs) == 1
assert case_inputs[0].canonical_input()["model"] in MISTRAL_MODELS
assert len(target.required_inputs) == 14 * len(MISTRAL_MODELS)
covered_params: Final = {
key
for case_input in target.required_inputs
for key in case_input.as_sdk_kwargs()
if key not in {"model", "document", "custom_llm_provider"}
}
assert covered_params == _MISTRAL_PARAMS
def test_mistral_target_invocation_forwards_discovered_credentials() -> None:
@ -177,7 +191,7 @@ def test_mistral_target_invocation_forwards_discovered_credentials() -> None:
assert kwargs["model"] in MISTRAL_MODELS
def test_every_target_covers_every_supported_param_for_every_model() -> None:
def test_every_target_strategy_reaches_every_model_and_coverage_param() -> None:
targets: Final = discover_targets(
{
"MISTRAL_API_KEY": "mistral-secret",
@ -193,12 +207,12 @@ def test_every_target_covers_every_supported_param_for_every_model() -> None:
)
expected: Final[dict[str, tuple[tuple[str, ...], frozenset[str]]]] = {
"mistral-ocr": (MISTRAL_MODELS, _MISTRAL_PARAMS),
"azure-mistral": (AZURE_MISTRAL_MODELS, _MISTRAL_PARAMS),
"azure-mistral": (AZURE_MISTRAL_MODELS, _MISTRAL_2512_PARAMS),
"azure-document-intelligence": (
AZURE_DOCUMENT_INTELLIGENCE_MODELS,
frozenset({"pages", "features", "req_format"}),
),
"vertex-mistral": (VERTEX_MISTRAL_MODELS, _MISTRAL_PARAMS),
"vertex-mistral": (VERTEX_MISTRAL_MODELS, _MISTRAL_2505_PARAMS),
"vertex-deepseek": (VERTEX_DEEPSEEK_MODELS, frozenset[str]()),
"reducto-v3": (REDUCTO_V3_MODELS, frozenset({"formatting", "retrieval", "settings"})),
"reducto-legacy": (REDUCTO_LEGACY_MODELS, frozenset({"enhance"})),
@ -206,13 +220,60 @@ def test_every_target_covers_every_supported_param_for_every_model() -> None:
for target in targets:
expected_models, expected_params = expected[target.name]
assert {_model(case_input) for case_input in target.required_inputs} == set(expected_models)
for model in expected_models:
covered = frozenset(
key
for case_input in target.required_inputs
if _model(case_input) == model
for key in case_input.as_sdk_kwargs()
if key not in {"model", "document", "custom_llm_provider", "vertex_project", "vertex_location"}
assert (
_model(
_find_input(
target.strategy,
lambda case_input, expected_model=model: _model(case_input) == expected_model,
)
)
== model
)
assert covered == expected_params
for param in expected_params:
assert (
param
in _find_input(
target.strategy,
lambda case_input, expected_param=param: expected_param in case_input.as_sdk_kwargs(),
).as_sdk_kwargs()
)
def test_ocr_targets_have_no_hardcoded_required_inputs() -> None:
targets: Final = discover_targets(
{
"MISTRAL_API_KEY": "mistral-secret",
"REDUCTO_API_KEY": "reducto-secret",
"AZURE_AI_API_KEY": "azure-secret",
"AZURE_AI_API_BASE": "https://azure.example",
"AZURE_DOCUMENT_INTELLIGENCE_API_KEY": "document-secret",
"AZURE_DOCUMENT_INTELLIGENCE_ENDPOINT": "https://document.example",
"VERTEX_AI_API_KEY": "vertex-secret",
"VERTEXAI_PROJECT": "project-1",
},
_UNUSED_OCR_CLIENT,
)
assert all(target.required_inputs == () for target in targets)
def test_mistral_adapters_preserve_omitted_optional_params() -> None:
targets: Final = discover_targets(
{
"AZURE_AI_API_KEY": "azure-secret",
"AZURE_AI_API_BASE": "https://azure.example",
"VERTEX_AI_API_KEY": "vertex-secret",
"VERTEXAI_PROJECT": "project-1",
},
_UNUSED_OCR_CLIENT,
)
baselines: Final = tuple(
_find_input(
target.strategy,
lambda case_input: _MISTRAL_PARAMS.isdisjoint(case_input.as_sdk_kwargs()),
)
for target in targets
)
assert all(_MISTRAL_PARAMS.isdisjoint(baseline.as_sdk_kwargs()) for baseline in baselines)