litellm/tests/rust-python-harness/strategies/e2e_parity/test_runner.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

48 lines
1.6 KiB
Python

from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from functools import partial
from pathlib import Path
from types import SimpleNamespace
from typing import Final
from unittest.mock import Mock, call
from pytest import MonkeyPatch
from ...shared.reporting.models import Coverage, HarnessCase, RunStatus
from ...shared.reporting.strategy import ModuleCaseSpec
from . import runner as e2e_runner
from .runner import E2ECheck, run_e2e_cases
def test_runs_checks_inside_suite_context(monkeypatch: MonkeyPatch, tmp_path: Path) -> None:
lifecycle: Final = Mock()
@contextmanager
def parity_checks() -> Generator[tuple[E2ECheck, ...]]:
lifecycle("entered")
try:
yield (E2ECheck("check", partial(lifecycle, "checked")),)
finally:
lifecycle("exited")
module: Final = SimpleNamespace(parity_checks=parity_checks)
def import_module(_name: str, _package: str | None = None) -> SimpleNamespace:
return module
monkeypatch.setattr(e2e_runner.importlib, "import_module", import_module)
case: Final = HarnessCase(
strategy_id="e2e_parity",
strategy_label="End-to-end parity",
sdk_function="ocr",
spec=ModuleCaseSpec(coverage=Coverage.PARTIAL, module="example"),
surface="sdk",
)
code, run = run_e2e_cases((case,), tmp_path, lambda _: None)
assert code == 0, run.failures
assert run.results[case.key].status is RunStatus.PASSED
assert lifecycle.call_args_list == [call("entered"), call("checked"), call("exited")]