litellm/tests/rust-python-harness/shared/reporting/strategy.py
yujonglee ee08c36fc0
refactor(tests): restructure rust python harness around strategy definitions (#39628)
* wip

* refactor(tests): move sdk function tracing into rust python harness

* dead code

* fix: handle harness keyboard interrupts

* refactor(tests): deduplicate rust python harness helpers

* fix(harness): expose validated strategy choices

* wip

* refactor(harness): let strategies own parity reports

* docs(harness): update strategy structure

* refactor(harness): localize strategy report views

* wip

* fix(harness): satisfy mapping runner type checks

* fix(harness): clarify trace parity output

* wip

* fix(harness): clarify unit mapping report

* fix(harness): finalize trace parity contracts

* refactor(harness): structure parity contracts

* feat: derive unit test mapping from traces

* feat(harness): map rstest test families

* feat(ocr): port Azure document intelligence tests

* feat(harness): enforce complete unit mappings

* feat(ocr): add reducto core transforms

* feat(harness): classify host-only unit tests

* fix(ocr): complete Rust provider plumbing

* fix(harness): reuse OCR parity workers
2026-09-03 21:15:01 -07:00

92 lines
2.6 KiB
Python

from __future__ import annotations
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Annotated, Literal, Protocol, TypeAlias
from pydantic import BaseModel, ConfigDict, StringConstraints
from .models import CaseDisposition, Coverage, HarnessCase, HarnessRun, SdkFunction, Surface
from .rendering import StrategyRenderer
UpdateCallback: TypeAlias = Callable[[HarnessRun], None]
NonBlankString: TypeAlias = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)]
class SuiteCaseSpec(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
disposition: Literal[CaseDisposition.RUNNABLE] = CaseDisposition.RUNNABLE
coverage: Coverage
suite: NonBlankString
note: str = ""
class ModuleCaseSpec(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
disposition: Literal[CaseDisposition.RUNNABLE] = CaseDisposition.RUNNABLE
coverage: Coverage
module: NonBlankString
note: str = ""
class NotImplementedCaseSpec(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
disposition: Literal[CaseDisposition.NOT_IMPLEMENTED] = CaseDisposition.NOT_IMPLEMENTED
coverage: None = None
reason: NonBlankString
class SkippedCaseSpec(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
disposition: Literal[CaseDisposition.SKIPPED] = CaseDisposition.SKIPPED
coverage: None = None
reason: NonBlankString
RunnableCaseSpec: TypeAlias = SuiteCaseSpec | ModuleCaseSpec
UnavailableCaseSpec: TypeAlias = NotImplementedCaseSpec | SkippedCaseSpec
CaseSpec: TypeAlias = RunnableCaseSpec | UnavailableCaseSpec
@dataclass(frozen=True, slots=True)
class CaseDefinition:
sdk_function: SdkFunction
spec: CaseSpec
surface: Surface | None = None
@dataclass(frozen=True, slots=True)
class RunnerArgumentDefinition:
option: str
help: str
metavar: str = "ARG"
class StrategyRunner(Protocol):
def __call__(
self,
cases: Sequence[HarnessCase],
repo_root: Path,
on_update: UpdateCallback,
runner_args: Sequence[str] = (),
) -> tuple[int, HarnessRun]: ...
@dataclass(frozen=True, slots=True)
class StrategyDefinition:
id: str
order: int
label: str
description: str
directory: Path
runnable_spec: type[SuiteCaseSpec] | type[ModuleCaseSpec]
cases: tuple[CaseDefinition, ...]
run: StrategyRunner
render: StrategyRenderer
surfaces: tuple[Surface, ...] = ()
runner_argument: RunnerArgumentDefinition | None = None