diff --git a/litellm-rust/crates/ai-gateway/src/bin/trace_parity_gateway.rs b/litellm-rust/crates/ai-gateway/src/bin/trace_parity_gateway.rs index 9036deb9871..e247c650fad 100644 --- a/litellm-rust/crates/ai-gateway/src/bin/trace_parity_gateway.rs +++ b/litellm-rust/crates/ai-gateway/src/bin/trace_parity_gateway.rs @@ -5,6 +5,7 @@ use serde_json::Value; #[derive(Deserialize)] struct Input { + path: String, model_alias: String, provider_model: String, api_base: String, @@ -21,7 +22,8 @@ async fn main() { Ok(input) => input, Err(error) => fail(error), }; - let result = litellm_ai_gateway::trace_parity::traced_messages_request( + let result = litellm_ai_gateway::trace_parity::traced_request( + input.path, input.model_alias, input.provider_model, input.api_base, diff --git a/litellm-rust/crates/ai-gateway/src/trace_parity.rs b/litellm-rust/crates/ai-gateway/src/trace_parity.rs index 00c9b53e691..7540a71fb12 100644 --- a/litellm-rust/crates/ai-gateway/src/trace_parity.rs +++ b/litellm-rust/crates/ai-gateway/src/trace_parity.rs @@ -29,14 +29,15 @@ pub struct TracedGatewayResponse { pub trace: Vec, } -pub async fn traced_messages_request( +pub async fn traced_request( + path: String, model_alias: String, provider_model: String, api_base: String, body: Value, ) -> TracedGatewayResponse { let trace = litellm_core::observability::FunctionTrace::default(); - let result = messages_request(model_alias, provider_model, api_base, body) + let result = request(path, model_alias, provider_model, api_base, body) .with_subscriber(trace.dispatcher()) .await; let events = trace.events(); @@ -54,7 +55,8 @@ pub async fn traced_messages_request( } } -pub async fn messages_request( +pub async fn request( + path: String, model_alias: String, provider_model: String, api_base: String, @@ -75,7 +77,7 @@ pub async fn messages_request( }; let request = Request::builder() .method("POST") - .uri("/v1/messages") + .uri(path) .header(AUTHORIZATION, "Bearer trace-master-key") .header(CONTENT_TYPE, "application/json") .body(Body::from(body.to_string())) diff --git a/tests/rust-python-harness/AGENTS.md b/tests/rust-python-harness/AGENTS.md index 017668d4289..ec973035785 100644 --- a/tests/rust-python-harness/AGENTS.md +++ b/tests/rust-python-harness/AGENTS.md @@ -63,7 +63,7 @@ tests/rust-python-harness/ - Examples: `run e2e_parity --surface sdk --function ocr`, `run unit_tests_parity --function ocr --pytest-arg=-x`, or `run all --function ocr` - `cli/catalog.py` discovers strategies, validates their Python definitions, and orders them; `cli/__init__.py` builds the Click command tree; `cli/commands.py` runs selected cases - `e2e_parity/` compares SDK objects, exceptions, callbacks, and streams, or gateway HTTP responses -- `trace_parity/` compares mapped operations, call counts, and required execution ordering; before running it rebuilds the native bridge with the `trace-parity` feature whenever `litellm-rust` sources are newer than the installed extension (`shared/native_build.py`) +- `trace_parity/` prints filtered Python and Rust execution traces without comparing them; before running it rebuilds the native bridge with the `trace-parity` feature whenever `litellm-rust` sources are newer than the installed extension (`shared/native_build.py`) - E2E and trace strategies load their registered module cases and run surface-specific execution from their folders - `unit_tests_mapping/contracts.py` owns typed harness-side mapping contracts, per-function contracts live below `cases/`, and `mappings.py` exports the registry; live test discovery derives unmapped Python and Rust-only tests without an exhaustive manifest - `unit_tests_mapping/runner.py` validates confirmed mappings against the live Python and Rust inventories and attaches the derived status report diff --git a/tests/rust-python-harness/cli/__init__.py b/tests/rust-python-harness/cli/__init__.py index 13b995825dd..d2bfdc55b19 100644 --- a/tests/rust-python-harness/cli/__init__.py +++ b/tests/rust-python-harness/cli/__init__.py @@ -58,16 +58,29 @@ def _strategy_command(strategy: Strategy) -> click.Command: help=runner_argument.help, ) ) + for runner_option in strategy.definition.runner_options: + name: Final = runner_option.option.removeprefix("--").replace("-", "_") + params.append( + click.Option( + (runner_option.option, name), + type=click.Choice(runner_option.choices), + help=runner_option.help, + ) + ) def run_strategy( sdk_functions: tuple[str, ...], surface: str | None = None, runner_args: tuple[str, ...] = (), + **runner_options: str | None, ) -> int: selected_functions: Final = cast(frozenset[SdkFunction], frozenset(sdk_functions)) selected_surface: Final = cast(Surface | None, surface) cases: Final = select_cases((strategy,), selected_functions, selected_surface) - return run_command((strategy,), cases, runner_args) + option_args: Final = tuple( + f"--{name.replace('_', '-')}={value}" for name, value in runner_options.items() if value is not None + ) + return run_command((strategy,), cases, (*runner_args, *option_args)) return click.Command( strategy.id, diff --git a/tests/rust-python-harness/cli/catalog.py b/tests/rust-python-harness/cli/catalog.py index 03eb032d9c6..073873d0121 100644 --- a/tests/rust-python-harness/cli/catalog.py +++ b/tests/rust-python-harness/cli/catalog.py @@ -21,9 +21,7 @@ def _load_strategy_module(name: str, folder: Path, prefix: str | None) -> Module if prefix is not None: return importlib.import_module(f"{prefix}.{name}") module_name: Final = _synthetic_module_name(folder) - spec: Final = importlib.util.spec_from_file_location( - module_name, folder / "__init__.py" - ) + spec: Final = importlib.util.spec_from_file_location(module_name, folder / "__init__.py") if spec is None or spec.loader is None: raise ValueError(f"{folder}: cannot load strategy package") module: Final = importlib.util.module_from_spec(spec) @@ -59,9 +57,7 @@ def _load_strategy(name: str, folder: Path, prefix: str | None) -> Strategy: if duplicates: raise ValueError(f"{folder}: duplicate strategy cases: {duplicates}") expected: Final = frozenset( - (surface, function) - for surface in (definition.surfaces or (None,)) - for function in SDK_FUNCTIONS + (surface, function) for surface in (definition.surfaces or (None,)) for function in SDK_FUNCTIONS ) actual: Final = frozenset(keys) if actual != expected: @@ -73,8 +69,7 @@ def _load_strategy(name: str, folder: Path, prefix: str | None) -> Strategy: incompatible: Final = tuple( (case.surface, case.sdk_function) for case in definition.cases - if case.spec.disposition is CaseDisposition.RUNNABLE - and not isinstance(case.spec, definition.runnable_spec) + if case.spec.disposition is CaseDisposition.RUNNABLE and not isinstance(case.spec, definition.runnable_spec) ) if incompatible: raise ValueError(f"{folder}: runnable cases do not match {definition.runnable_spec.__name__}: {incompatible}") @@ -102,14 +97,10 @@ def _load_strategy(name: str, folder: Path, prefix: str | None) -> Strategy: def load_catalog(root: Path | None = None) -> tuple[Strategy, ...]: resolved: Final = STRATEGIES_ROOT if root is None else root prefix: Final = _STRATEGIES_PACKAGE.__name__ if resolved == STRATEGIES_ROOT else None - folders: Final = tuple( - info.name for info in pkgutil.iter_modules([str(resolved)]) if info.ispkg - ) + folders: Final = tuple(info.name for info in pkgutil.iter_modules([str(resolved)]) if info.ispkg) if not folders: raise ValueError(f"No strategy packages found below {resolved}") - strategies: Final = tuple( - _load_strategy(name, resolved / name, prefix) for name in sorted(folders) - ) + strategies: Final = tuple(_load_strategy(name, resolved / name, prefix) for name in sorted(folders)) ids: Final = [strategy.id for strategy in strategies] if len(set(ids)) != len(ids): raise ValueError(f"Duplicate strategy id in {resolved}") diff --git a/tests/rust-python-harness/cli/commands.py b/tests/rust-python-harness/cli/commands.py index f94c3277dc2..e51bbb5d966 100644 --- a/tests/rust-python-harness/cli/commands.py +++ b/tests/rust-python-harness/cli/commands.py @@ -21,8 +21,7 @@ def select_cases( case for strategy in strategies for case in strategy.cases - if (not sdk_functions or case.sdk_function in sdk_functions) - and (surface is None or case.surface == surface) + if (not sdk_functions or case.sdk_function in sdk_functions) and (surface is None or case.surface == surface) ) @@ -32,8 +31,7 @@ def run_command( runner_args: Sequence[str] = (), ) -> int: grouped: Final = { - strategy.id: tuple(case for case in cases if case.strategy_id == strategy.id) - for strategy in strategies + strategy.id: tuple(case for case in cases if case.strategy_id == strategy.id) for strategy in strategies } visible: Final = tuple(strategy for strategy in strategies if grouped[strategy.id]) runners: Final = tuple(replace(strategy, cases=grouped[strategy.id]) for strategy in visible) diff --git a/tests/rust-python-harness/cli/test_cli.py b/tests/rust-python-harness/cli/test_cli.py index 226c89843d0..5641aa8a539 100644 --- a/tests/rust-python-harness/cli/test_cli.py +++ b/tests/rust-python-harness/cli/test_cli.py @@ -242,7 +242,7 @@ def _assert_unavailable_cell(strategy: Strategy, case: HarnessCase, section_titl def test_every_unavailable_case_finishes_and_explains_itself() -> None: section_titles: Final = { "e2e_parity": "End-to-end parity outcomes", - "trace_parity": "trace comparisons", + "trace_parity": "traces", "unit_tests_mapping": "Python/Rust unit-test mappings", "unit_tests_parity": "Python backend parity outcomes", "unit_tests_rust": "Native Rust unit-test outcomes", @@ -359,6 +359,25 @@ def test_strategy_command_forwards_repeated_filters_and_runner_arguments( ] +def test_trace_command_forwards_engine_and_scenario(monkeypatch: pytest.MonkeyPatch) -> None: + cli: Final = importlib.import_module("tests.rust-python-harness.cli") + captured: list[tuple[str, ...]] = [] + + def capture_run( + strategies: Sequence[Strategy], + cases: Sequence[HarnessCase], + runner_args: Sequence[str] = (), + ) -> int: + del strategies, cases + captured.append(tuple(runner_args)) + return 0 + + monkeypatch.setattr(cli, "run_command", capture_run) + + assert main(["run", "trace_parity", "--scenario", "async-mistral", "--engine", "python"]) == 0 + assert captured == [("async-mistral", "--engine=python")] + + def test_omitted_surface_selects_every_strategy_surface(monkeypatch: pytest.MonkeyPatch) -> None: cli: Final = importlib.import_module("tests.rust-python-harness.cli") selected: list[str] = [] diff --git a/tests/rust-python-harness/conftest.py b/tests/rust-python-harness/conftest.py index d50d0fa4204..c487b25a9cc 100644 --- a/tests/rust-python-harness/conftest.py +++ b/tests/rust-python-harness/conftest.py @@ -19,9 +19,7 @@ def subprocess_test_environment(monkeypatch: pytest.MonkeyPatch) -> None: def cargo_project(tmp_path: Path) -> Callable[[str, str], Path]: def create(package: str, source: str) -> Path: manifest: Final = tmp_path / "Cargo.toml" - manifest.write_text( - f'[package]\nname = "{package}"\nversion = "0.1.0"\nedition = "2021"\n[workspace]\n' - ) + manifest.write_text(f'[package]\nname = "{package}"\nversion = "0.1.0"\nedition = "2021"\n[workspace]\n') (tmp_path / "src").mkdir() (tmp_path / "src/lib.rs").write_text(source) return manifest diff --git a/tests/rust-python-harness/shared/native_build.py b/tests/rust-python-harness/shared/native_build.py index 8693cf3bac2..b50def98e3a 100644 --- a/tests/rust-python-harness/shared/native_build.py +++ b/tests/rust-python-harness/shared/native_build.py @@ -85,7 +85,8 @@ def trace_bridge_error() -> str | None: def ensure_trace_bridge(repo_root: Path) -> str | None: native_path: Final = _native_module_path() native_mtime: Final = native_path.stat().st_mtime if native_path is not None and native_path.exists() else None - if needs_rebuild(native_mtime, _newest_source_mtime(repo_root)): + rebuild_required: Final = needs_rebuild(native_mtime, _newest_source_mtime(repo_root)) or trace_bridge_error() is not None + if rebuild_required: print(f"Rebuilding native Rust bridge ({BRIDGE_FEATURE} feature)...", flush=True) succeeded: Final output: Final diff --git a/tests/rust-python-harness/shared/parity/fixtures/recording.py b/tests/rust-python-harness/shared/parity/fixtures/recording.py index ee0d4b6de7b..cb6ed968b90 100644 --- a/tests/rust-python-harness/shared/parity/fixtures/recording.py +++ b/tests/rust-python-harness/shared/parity/fixtures/recording.py @@ -191,6 +191,7 @@ class _RecordingHandler(LocalHttpHandler): self.end_headers() self.wfile.write(body) + def _recording_provider(spec: UpstreamEndpoint) -> AbstractContextManager[_RecordingProvider]: return serve_in_thread(_RecordingProvider(spec)) diff --git a/tests/rust-python-harness/shared/parity/fixtures/store.py b/tests/rust-python-harness/shared/parity/fixtures/store.py index 270af2a7625..1145b5d7d27 100644 --- a/tests/rust-python-harness/shared/parity/fixtures/store.py +++ b/tests/rust-python-harness/shared/parity/fixtures/store.py @@ -15,6 +15,8 @@ from .cassette import deserialize_cassette, serialize_cassette from .recording import RecordedInteraction FIXTURE_SCHEMA_VERSION: Final = 1 + + class FixtureInput(Protocol): def canonical_input(self) -> dict[str, object]: ... diff --git a/tests/rust-python-harness/shared/parity/fixtures/test_pipeline.py b/tests/rust-python-harness/shared/parity/fixtures/test_pipeline.py index 4535ba05bf6..b162e4949d2 100644 --- a/tests/rust-python-harness/shared/parity/fixtures/test_pipeline.py +++ b/tests/rust-python-harness/shared/parity/fixtures/test_pipeline.py @@ -45,8 +45,8 @@ class _Upstream(LocalHttpServer): super().__init__(("127.0.0.1", 0), _UpstreamHandler) self.response_status: Final = status -class _UpstreamHandler(LocalHttpHandler): +class _UpstreamHandler(LocalHttpHandler): def do_POST(self) -> None: length: Final = int(self.headers.get("content-length") or "0") self.rfile.read(length) @@ -59,6 +59,7 @@ class _UpstreamHandler(LocalHttpHandler): self.end_headers() self.wfile.write(body) + def _upstream(status: int = 200) -> AbstractContextManager[_Upstream]: return serve_in_thread(_Upstream(status)) diff --git a/tests/rust-python-harness/shared/parity/fixtures/test_recording.py b/tests/rust-python-harness/shared/parity/fixtures/test_recording.py index 6181f18e89a..c1e7c2af8d8 100644 --- a/tests/rust-python-harness/shared/parity/fixtures/test_recording.py +++ b/tests/rust-python-harness/shared/parity/fixtures/test_recording.py @@ -238,6 +238,7 @@ class _ControlledUpstreamHandler(LocalHttpHandler): self.end_headers() self.wfile.write(body) + def _controlled_upstream( stream_chunks: tuple[bytes, ...] = _SSE_CHUNKS, ) -> AbstractContextManager[_ControlledUpstream]: diff --git a/tests/rust-python-harness/shared/reporting/models.py b/tests/rust-python-harness/shared/reporting/models.py index 1ebba6c9793..4a46a6d4cfe 100644 --- a/tests/rust-python-harness/shared/reporting/models.py +++ b/tests/rust-python-harness/shared/reporting/models.py @@ -180,15 +180,11 @@ class HarnessRun: @property def unique_checks(self) -> int: - return len( - {nodeid for result in self.results.values() for nodeid in result.collected} - ) + return len({nodeid for result in self.results.values() for nodeid in result.collected}) @property def completed_checks(self) -> int: - return len( - {nodeid for result in self.results.values() for nodeid in result.completed} - ) + return len({nodeid for result in self.results.values() for nodeid in result.completed}) @classmethod def from_cases(cls, cases: Iterable[HarnessCase]) -> HarnessRun: diff --git a/tests/rust-python-harness/shared/reporting/strategy.py b/tests/rust-python-harness/shared/reporting/strategy.py index 7e76f035e20..d8e9d9e5ba9 100644 --- a/tests/rust-python-harness/shared/reporting/strategy.py +++ b/tests/rust-python-harness/shared/reporting/strategy.py @@ -67,6 +67,13 @@ class RunnerArgumentDefinition: metavar: str = "ARG" +@dataclass(frozen=True, slots=True) +class RunnerOptionDefinition: + option: str + help: str + choices: tuple[str, ...] + + class StrategyRunner(Protocol): def __call__( self, @@ -90,3 +97,4 @@ class StrategyDefinition: render: StrategyRenderer surfaces: tuple[Surface, ...] = () runner_argument: RunnerArgumentDefinition | None = None + runner_options: tuple[RunnerOptionDefinition, ...] = () diff --git a/tests/rust-python-harness/shared/test_native_build.py b/tests/rust-python-harness/shared/test_native_build.py index f3e5aead846..982f787aefa 100644 --- a/tests/rust-python-harness/shared/test_native_build.py +++ b/tests/rust-python-harness/shared/test_native_build.py @@ -89,8 +89,8 @@ def test_ensure_trace_bridge_reports_failed_rebuild(tmp_path: Final, monkeypatch assert "boom" in message -def test_ensure_trace_bridge_flags_missing_trace_feature_without_rebuild( - tmp_path: Final, monkeypatch: pytest.MonkeyPatch +def test_ensure_trace_bridge_rebuilds_when_trace_feature_is_missing( + tmp_path: Final, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] ) -> None: native: Final = tmp_path / "_native.abi3.so" native.write_bytes(b"") @@ -107,10 +107,14 @@ def test_ensure_trace_bridge_flags_missing_trace_feature_without_rebuild( monkeypatch.setattr(native_build, "_native_module_path", lambda: native) monkeypatch.setattr(native_build, "_rebuild", fake_rebuild) - monkeypatch.setattr(native_build, "get_native_bridge", lambda: SimpleNamespace(_trace=None)) + monkeypatch.setattr( + native_build, + "get_native_bridge", + lambda: SimpleNamespace(_trace=object() if state.rebuilt else None), + ) message: Final = native_build.ensure_trace_bridge(tmp_path) - assert message is not None - assert "_trace" in message - assert state.rebuilt is False + assert message is None + assert state.rebuilt is True + assert "Rebuilding native Rust bridge" in capsys.readouterr().out diff --git a/tests/rust-python-harness/shared/tracing/steps.py b/tests/rust-python-harness/shared/tracing/steps.py index 2475f0fdcc5..152f753a536 100644 --- a/tests/rust-python-harness/shared/tracing/steps.py +++ b/tests/rust-python-harness/shared/tracing/steps.py @@ -181,12 +181,7 @@ class TraceDiff: @property def matches(self) -> bool: - return ( - not self.python_only - and not self.rust_only - and not self.missing_mappings - and self.shared_order_matches - ) + return not self.python_only and not self.rust_only and not self.missing_mappings and self.shared_order_matches def _missing_mappings( @@ -257,9 +252,7 @@ def trace_diff( rust_counts: Final = Counter(rust_spans) python_only_counts: Final = python_counts - rust_counts rust_only_counts: Final = rust_counts - python_counts - python_only: Final = tuple( - span for span, count in python_only_counts.items() for _ in range(count) - ) + python_only: Final = tuple(span for span, count in python_only_counts.items() for _ in range(count)) rust_only: Final = tuple(span for span, count in rust_only_counts.items() for _ in range(count)) first_difference: Final = _first_difference(python, rust, mappings, contract) return TraceDiff( diff --git a/tests/rust-python-harness/shared/tracing/test_steps.py b/tests/rust-python-harness/shared/tracing/test_steps.py index 2efc6a3c579..1b2f02a4ed7 100644 --- a/tests/rust-python-harness/shared/tracing/test_steps.py +++ b/tests/rust-python-harness/shared/tracing/test_steps.py @@ -146,9 +146,7 @@ def test_trace_diff_allows_reordered_concurrent_children() -> None: def test_trace_diff_prunes_declared_engine_only_nodes_but_requires_them() -> None: mappings: Final = (MAPPINGS[0], mapping(rust_span="rust_prepare")) python: Final = pipeline_projection("python", (event(0, "module.py:1 entry"),), mappings).steps - rust: Final = pipeline_projection( - "rust", (event(0, "route"), event(1, "rust_prepare", 0)), mappings - ).steps + rust: Final = pipeline_projection("rust", (event(0, "route"), event(1, "rust_prepare", 0)), mappings).steps assert trace_diff(python, rust, mappings).matches assert trace_diff(python, rust[:1], mappings).missing_mappings == ("rust_prepare",) diff --git a/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/fixtures/reducto.py b/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/fixtures/reducto.py index bcca8ac6d42..1b283d75ca5 100644 --- a/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/fixtures/reducto.py +++ b/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/fixtures/reducto.py @@ -264,9 +264,7 @@ def _formatting_strategy() -> SearchStrategy[ReductoFormatting]: ), st.sampled_from((False, True)).map(lambda value: {"add_page_markers": value}), st.sampled_from((False, True)).map(lambda value: {"merge_tables": value}), - st.sampled_from(REDUCTO_FORMATTING_INCLUDE_GROUPS) - .map(list) - .map(lambda value: {"include": value}), + st.sampled_from(REDUCTO_FORMATTING_INCLUDE_GROUPS).map(list).map(lambda value: {"include": value}), ) return values.map(ReductoFormatting.model_validate) diff --git a/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/test_sdk_parity.py b/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/test_sdk_parity.py index 5c4abc78081..f14296fbf06 100644 --- a/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/test_sdk_parity.py +++ b/tests/rust-python-harness/strategies/e2e_parity/sdk/ocr/test_sdk_parity.py @@ -124,10 +124,7 @@ class RecordingCallback(CustomLogger): if isinstance(value, Mapping): if any(not isinstance(map_key, str) for map_key in value): raise TypeError("callback kwarg mappings must use string keys") - return { - map_key: self._normalized_kwargs(map_value, map_key) - for map_key, map_value in value.items() - } + return {map_key: self._normalized_kwargs(map_value, map_key) for map_key, map_value in value.items()} if isinstance(value, (list, tuple)): return [self._normalized_kwargs(item) for item in value] raise TypeError(f"unsupported callback kwarg type: {type(value)}") diff --git a/tests/rust-python-harness/strategies/trace_parity/AGENTS.md b/tests/rust-python-harness/strategies/trace_parity/AGENTS.md index bb7cb8c91d8..5db17974d07 100644 --- a/tests/rust-python-harness/strategies/trace_parity/AGENTS.md +++ b/tests/rust-python-harness/strategies/trace_parity/AGENTS.md @@ -1 +1 @@ -Maps Python profiler frames onto feature-gated Rust span names via an explicit per-case mapping (Rust span name is the identity) and compares steps, order, and nesting of both live traces against a replayed provider response. +Prints filtered Python profiler frames and feature-gated Rust spans from live traces against a replayed provider response. The two traces are independent and are not compared. diff --git a/tests/rust-python-harness/strategies/trace_parity/__init__.py b/tests/rust-python-harness/strategies/trace_parity/__init__.py index ec88b0169fa..710bdaa3d39 100644 --- a/tests/rust-python-harness/strategies/trace_parity/__init__.py +++ b/tests/rust-python-harness/strategies/trace_parity/__init__.py @@ -7,6 +7,7 @@ from ...shared.reporting.strategy import ( ModuleCaseSpec, NotImplementedCaseSpec, RunnerArgumentDefinition, + RunnerOptionDefinition, StrategyDefinition, ) from .reporting import render_trace_results @@ -26,13 +27,17 @@ CASES: Final[tuple[CaseDefinition, ...]] = ( ModuleCaseSpec( coverage=Coverage.PARTIAL, module="tests.rust-python-harness.strategies.trace_parity.sdk.messages.case", - note="Async only until anthropic_messages_handler supports sync calls.", + note="Success paths are async; sync tracing captures the currently unsupported behavior.", ), surface="sdk", ), CaseDefinition( "responses", - NotImplementedCaseSpec(reason="No Responses trace-parity case is registered."), + ModuleCaseSpec( + coverage=Coverage.PARTIAL, + module="tests.rust-python-harness.strategies.trace_parity.sdk.responses.case", + note="Core create paths: native, streaming, provider error, Azure override, and chat bridge.", + ), surface="sdk", ), CaseDefinition( @@ -70,13 +75,17 @@ CASES: Final[tuple[CaseDefinition, ...]] = ( ModuleCaseSpec( coverage=Coverage.PARTIAL, module="tests.rust-python-harness.strategies.trace_parity.gateway.messages.case", - note="Non-streaming success paths only.", + note="Anthropic/Azure provider routes plus a fully consumed downstream streaming path.", ), surface="gateway", ), CaseDefinition( "responses", - NotImplementedCaseSpec(reason="No gateway Responses trace-parity case is registered."), + ModuleCaseSpec( + coverage=Coverage.PARTIAL, + module="tests.rust-python-harness.strategies.trace_parity.gateway.responses.case", + note="Native OpenAI non-streaming and fully consumed downstream streaming paths.", + ), surface="gateway", ), CaseDefinition( @@ -86,7 +95,11 @@ CASES: Final[tuple[CaseDefinition, ...]] = ( ), CaseDefinition( "chat_completions", - NotImplementedCaseSpec(reason="No gateway chat trace-parity case is registered."), + ModuleCaseSpec( + coverage=Coverage.PARTIAL, + module="tests.rust-python-harness.strategies.trace_parity.gateway.chat_completions.case", + note="Anthropic non-streaming and fully consumed downstream streaming paths.", + ), surface="gateway", ), CaseDefinition( @@ -99,8 +112,8 @@ CASES: Final[tuple[CaseDefinition, ...]] = ( STRATEGY: Final = StrategyDefinition( id="trace_parity", order=20, - label="Trace parity", - description="Compare pipeline steps, order, and nesting between Python profiler frames and Rust spans via an explicit mapping.", + label="Traces", + description="Print Python profiler frames and Rust spans for representative pipeline scenarios.", directory=Path(__file__).parent, runnable_spec=ModuleCaseSpec, cases=CASES, @@ -112,4 +125,11 @@ STRATEGY: Final = StrategyDefinition( metavar="NAME", help="run only this named trace scenario; repeat to select more than one", ), + runner_options=( + RunnerOptionDefinition( + option="--engine", + choices=("python", "rust"), + help="show only this engine's trace; omit to print both engines", + ), + ), ) diff --git a/tests/rust-python-harness/strategies/trace_parity/fixtures.py b/tests/rust-python-harness/strategies/trace_parity/fixtures.py new file mode 100644 index 00000000000..84ac525efb6 --- /dev/null +++ b/tests/rust-python-harness/strategies/trace_parity/fixtures.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import base64 +import binascii +import json +import struct +from collections.abc import Iterable, Mapping +from typing import Final + +from ...shared.parity.recorded_http import ( + HttpHeader, + RecordedHttpResponse, + RecordedHttpStreamResponse, + RecordedStreamChunk, +) + +JSON_HEADERS: Final = (HttpHeader(name="content-type", value="application/json"),) +SSE_HEADERS: Final = (HttpHeader(name="content-type", value="text/event-stream"),) +AWS_EVENT_STREAM_HEADERS: Final = (HttpHeader(name="content-type", value="application/vnd.amazon.eventstream"),) + + +def json_response(body: Mapping[str, object] | bytes, *, status: int = 200) -> RecordedHttpResponse: + encoded: Final = body if isinstance(body, bytes) else json.dumps(body).encode() + return RecordedHttpResponse.from_bytes(status, JSON_HEADERS, encoded) + + +def sse_event(event: str, payload: Mapping[str, object]) -> bytes: + return f"event: {event}\ndata: {json.dumps(payload, separators=(',', ':'))}\n\n".encode() + + +def sse_response(events: Iterable[tuple[str, Mapping[str, object]]]) -> RecordedHttpStreamResponse: + return RecordedHttpStreamResponse( + kind="http_stream", + status_code=200, + headers=SSE_HEADERS, + chunks=tuple(RecordedStreamChunk.from_bytes(sse_event(event, payload)) for event, payload in events), + ) + + +def _aws_string_header(name: str, value: str) -> bytes: + name_bytes: Final = name.encode() + value_bytes: Final = value.encode() + return ( + struct.pack("!B", len(name_bytes)) + + name_bytes + + struct.pack("!B", 7) + + struct.pack("!H", len(value_bytes)) + + value_bytes + ) + + +def aws_event_stream_frame(payload: Mapping[str, object]) -> bytes: + event_payload: Final = json.dumps( + {"bytes": base64.b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode()}, + separators=(",", ":"), + ).encode() + headers: Final = ( + _aws_string_header(":event-type", "chunk") + + _aws_string_header(":content-type", "application/json") + + _aws_string_header(":message-type", "event") + ) + total_length: Final = 12 + len(headers) + len(event_payload) + 4 + prelude: Final = struct.pack("!II", total_length, len(headers)) + prelude_crc: Final = binascii.crc32(prelude) & 0xFFFFFFFF + prelude_crc_bytes: Final = struct.pack("!I", prelude_crc) + message_crc: Final = binascii.crc32(prelude_crc_bytes + headers + event_payload, prelude_crc) & 0xFFFFFFFF + return prelude + prelude_crc_bytes + headers + event_payload + struct.pack("!I", message_crc) + + +def aws_event_stream_response( + events: Iterable[Mapping[str, object]], *, corrupt_last_frame: bool = False +) -> RecordedHttpStreamResponse: + frames: Final = [aws_event_stream_frame(event) for event in events] + if corrupt_last_frame: + corrupted: Final = bytearray(frames[-1]) + corrupted[-1] ^= 0xFF + frames[-1] = bytes(corrupted) + return RecordedHttpStreamResponse( + kind="http_stream", + status_code=200, + headers=AWS_EVENT_STREAM_HEADERS, + chunks=(RecordedStreamChunk.from_bytes(b"".join(frames)),), + ) + + +def anthropic_response_body(*, model: str = "claude-sonnet-5") -> dict[str, object]: + return { + "id": "msg_trace", + "type": "message", + "role": "assistant", + "model": model, + "content": [{"type": "text", "text": "hello"}], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 2, "output_tokens": 3}, + } + + +def anthropic_stream_events(*, model: str = "claude-sonnet-5") -> tuple[tuple[str, Mapping[str, object]], ...]: + return ( + ( + "message_start", + { + "type": "message_start", + "message": { + "id": "msg_trace", + "type": "message", + "role": "assistant", + "model": model, + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 2, "output_tokens": 0}, + }, + }, + ), + ( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ), + ( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hello"}}, + ), + ("content_block_stop", {"type": "content_block_stop", "index": 0}), + ( + "message_delta", + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 1}, + }, + ), + ("message_stop", {"type": "message_stop"}), + ) + + +def responses_body(*, model: str = "gpt-5", status: str = "completed") -> dict[str, object]: + return { + "id": "resp_trace", + "object": "response", + "created_at": 1_750_000_000, + "status": status, + "model": model, + "output": [ + { + "type": "message", + "id": "msg_trace", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "hello", "annotations": []}], + } + ], + "usage": {"input_tokens": 2, "output_tokens": 3, "total_tokens": 5}, + } + + +def responses_stream_events(*, model: str = "gpt-5") -> tuple[tuple[str, Mapping[str, object]], ...]: + response: Final = responses_body(model=model) + return ( + ( + "response.created", + {"type": "response.created", "response": {**response, "status": "in_progress", "output": []}}, + ), + ( + "response.output_text.delta", + { + "type": "response.output_text.delta", + "item_id": "msg_trace", + "output_index": 0, + "content_index": 0, + "delta": "hello", + }, + ), + ("response.completed", {"type": "response.completed", "response": response}), + ) diff --git a/tests/rust-python-harness/strategies/trace_parity/gateway/chat_completions/case.py b/tests/rust-python-harness/strategies/trace_parity/gateway/chat_completions/case.py new file mode 100644 index 00000000000..522d2d0259c --- /dev/null +++ b/tests/rust-python-harness/strategies/trace_parity/gateway/chat_completions/case.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from typing import Final + +from .....shared.tracing.steps import Engine, mapping +from ...fixtures import anthropic_response_body, anthropic_stream_events, json_response, sse_response +from ...models import GatewayRouteSpec, RouteFixture, TraceScenario, TraceSuite + +MAPPINGS: Final = ( + mapping(span="python_chat_gateway_route", python_frame=r"proxy_server\.py:\d+ chat_completion$"), + mapping(span="python_gateway_service", python_frame=r"ProxyBaseLLMRequestProcessing\.base_process_llm_request$"), + mapping(rust_span="chat_completions_gateway_route"), + mapping(rust_span="chat_completions"), + mapping(span="python_chat_entrypoint", python_frame=r"main\.py:\d+ a?completion$"), + mapping(span="python_provider_config", python_frame=r"ProviderConfigManager\.get_provider_chat_config$"), + mapping(rust_span="validate_environment", python_frame=r"(? RouteFixture: + return RouteFixture( + kwargs={ + "model_alias": "trace-model", + "provider_model": "anthropic/claude-sonnet-5", + "body": { + "model": "trace-model", + "messages": [{"role": "user", "content": "hello"}], + "max_tokens": 16, + }, + }, + provider_responses=(json_response(anthropic_response_body()),), + ) + + +def _stream_fixture(engine: Engine, base_url: str) -> RouteFixture: + fixture: Final = _fixture(engine, base_url) + return fixture.with_body(stream=True).derive( + provider_responses=(sse_response(anthropic_stream_events()),), + ) + + +TRACE_SUITE: Final = TraceSuite( + route=GatewayRouteSpec("chat_completions"), + scenarios=( + TraceScenario(name="async-anthropic", fixture=_fixture, mappings=MAPPINGS, asynchronous=True), + TraceScenario( + name="async-anthropic-downstream-stream", + fixture=_stream_fixture, + mappings=(*MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, + ), + ), +) diff --git a/tests/rust-python-harness/strategies/trace_parity/gateway/execution.py b/tests/rust-python-harness/strategies/trace_parity/gateway/execution.py index 860e872dd44..88130dab808 100644 --- a/tests/rust-python-harness/strategies/trace_parity/gateway/execution.py +++ b/tests/rust-python-harness/strategies/trace_parity/gateway/execution.py @@ -13,8 +13,8 @@ from ....shared.parity.replay import replay_server from ....shared.tracing.native import TraceResponsePayload, native_trace_events from ....shared.tracing.profiler import FunctionTraceEvent, profile_python from ....shared.tracing.steps import Engine, PipelineProjection, pipeline_projection -from ..models import GatewayRouteSpec, RouteFixture, TraceExecutionFailure, TraceMode, TraceScenario -from ..reporting import TraceComparisonArtifact +from ..models import GatewayRouteSpec, RouteFixture, TraceEngine, TraceExecutionFailure, TraceScenario +from ..reporting import TraceArtifact class _GatewayResponsePayload(BaseModel): @@ -28,7 +28,14 @@ class _GatewayClient(Protocol): def post(self, url: str, *, json: object, headers: dict[str, str]) -> httpx.Response: ... -def _collect_python(fixture: RouteFixture) -> tuple[FunctionTraceEvent, ...]: +_ROUTE_PATHS: Final = { + "messages": "/v1/messages", + "chat_completions": "/v1/chat/completions", + "responses": "/v1/responses", +} + + +def _collect_python(fixture: RouteFixture, route: GatewayRouteSpec) -> tuple[FunctionTraceEvent, ...]: from fastapi.testclient import TestClient import litellm @@ -61,7 +68,7 @@ def _collect_python(fixture: RouteFixture) -> tuple[FunctionTraceEvent, ...]: with profile_python(Path(litellm.__file__).parent, threads=True) as profiler: client: Final = cast(_GatewayClient, TestClient(proxy_server.app)) response: Final = client.post( - "/v1/messages", + _ROUTE_PATHS[route.route], json=fixture.kwargs["body"], headers={"authorization": "Bearer trace-key"}, ) @@ -76,9 +83,10 @@ def _collect_python(fixture: RouteFixture) -> tuple[FunctionTraceEvent, ...]: proxy_server.app.dependency_overrides[user_api_key_auth] = old_override -def _collect_rust(fixture: RouteFixture) -> tuple[FunctionTraceEvent, ...]: +def _collect_rust(fixture: RouteFixture, route: GatewayRouteSpec) -> tuple[FunctionTraceEvent, ...]: payload: Final = json.dumps( { + "path": _ROUTE_PATHS[route.route], "model_alias": fixture.kwargs["model_alias"], "provider_model": fixture.kwargs["provider_model"], "api_base": fixture.kwargs["api_base"], @@ -130,7 +138,9 @@ def _gateway_trace_binary() -> Path: return rust_root / "target" / "debug" / "trace-parity-gateway" -def _collect(scenario: TraceScenario, engine: Engine) -> tuple[FunctionTraceEvent, ...] | TraceExecutionFailure: +def _collect( + route: GatewayRouteSpec, scenario: TraceScenario, engine: Engine +) -> tuple[FunctionTraceEvent, ...] | TraceExecutionFailure: try: with replay_server() as provider: base_fixture: Final = scenario.fixture(engine, provider.url) @@ -140,7 +150,7 @@ def _collect(scenario: TraceScenario, engine: Engine) -> tuple[FunctionTraceEven ) for response in fixture.provider_responses: provider.enqueue_response(response) - events: Final = _collect_python(fixture) if engine == "python" else _collect_rust(fixture) + events: Final = _collect_python(fixture, route) if engine == "python" else _collect_rust(fixture, route) provider.take_requests(len(fixture.provider_responses)) return events except Exception as error: @@ -151,9 +161,8 @@ def _projections( python_events: tuple[FunctionTraceEvent, ...], rust_events: tuple[FunctionTraceEvent, ...], scenario: TraceScenario, - mode: TraceMode, ) -> tuple[PipelineProjection, PipelineProjection, str | None]: - mappings: Final = scenario.mappings_for(mode) + mappings: Final = scenario.mappings try: return ( pipeline_projection("python", python_events, mappings), @@ -164,26 +173,26 @@ def _projections( return PipelineProjection(), PipelineProjection(), f"harness: {error}" -def execute_gateway_trace(route: GatewayRouteSpec, scenario: TraceScenario, mode: TraceMode) -> TraceComparisonArtifact: - mappings: Final = scenario.mappings_for(mode) - python_trace: Final = _collect(scenario, "python") - rust_trace: Final = _collect(scenario, "rust") +def execute_gateway_trace( + route: GatewayRouteSpec, + scenario: TraceScenario, + engine: TraceEngine = "both", +) -> TraceArtifact: + python_trace: Final = _collect(route, scenario, "python") if engine != "rust" else () + rust_trace: Final = _collect(route, scenario, "rust") if engine != "python" else () collection_python_error: Final = None if isinstance(python_trace, tuple) else f"python: {python_trace.message}" rust_error: Final = None if isinstance(rust_trace, tuple) else f"rust: {rust_trace.message}" python_events: Final = python_trace if isinstance(python_trace, tuple) else () rust_events: Final = rust_trace if isinstance(rust_trace, tuple) else () - python, rust, projection_error = _projections(python_events, rust_events, scenario, mode) + python, rust, projection_error = _projections(python_events, rust_events, scenario) python_error: Final = projection_error or collection_python_error - return TraceComparisonArtifact.from_traces( + return TraceArtifact.from_traces( + engine=engine, surface="gateway", sdk_function=route.route, scenario=scenario.name, - mode=mode, - mappings=mappings, - contract=scenario.contract, python=python.steps, rust=rust.steps, - python_unmatched=python.unmatched, python_error=python_error, rust_error=rust_error, ) diff --git a/tests/rust-python-harness/strategies/trace_parity/gateway/messages/case.py b/tests/rust-python-harness/strategies/trace_parity/gateway/messages/case.py index 30f51cee353..ca9c858f6b7 100644 --- a/tests/rust-python-harness/strategies/trace_parity/gateway/messages/case.py +++ b/tests/rust-python-harness/strategies/trace_parity/gateway/messages/case.py @@ -1,10 +1,9 @@ from __future__ import annotations -import json from typing import Final -from .....shared.parity.recorded_http import HttpHeader, RecordedHttpResponse from .....shared.tracing.steps import Engine, mapping +from ...fixtures import anthropic_response_body, anthropic_stream_events, json_response, sse_response from ...models import GatewayRouteSpec, RouteFixture, TraceScenario, TraceSuite @@ -49,24 +48,7 @@ def _fixture(_engine: Engine, provider: str) -> RouteFixture: "max_tokens": 16, }, }, - provider_responses=( - RecordedHttpResponse.from_bytes( - 200, - (HttpHeader(name="content-type", value="application/json"),), - json.dumps( - { - "id": "msg_trace", - "type": "message", - "role": "assistant", - "model": "claude-sonnet-5", - "content": [{"type": "text", "text": "hello"}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 2, "output_tokens": 3}, - } - ).encode(), - ), - ), + provider_responses=(json_response(anthropic_response_body()),), ) @@ -78,6 +60,13 @@ def _azure_fixture(engine: Engine, _base_url: str) -> RouteFixture: return _fixture(engine, "azure_ai") +def _stream_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _anthropic_fixture(engine, _base_url) + return fixture.with_body(stream=True).derive( + provider_responses=(sse_response(anthropic_stream_events()),), + ) + + ANTHROPIC_MAPPINGS: Final = ( *GATEWAY_MAPPINGS, mapping( @@ -100,7 +89,22 @@ AZURE_MAPPINGS: Final = ( TRACE_SUITE: Final = TraceSuite( route=GatewayRouteSpec("messages"), scenarios=( - TraceScenario(name="anthropic", fixture=_anthropic_fixture, mappings=ANTHROPIC_MAPPINGS, modes=("async",)), - TraceScenario(name="azure-ai", fixture=_azure_fixture, mappings=AZURE_MAPPINGS, modes=("async",)), + TraceScenario( + name="async-anthropic", fixture=_anthropic_fixture, mappings=ANTHROPIC_MAPPINGS, asynchronous=True + ), + TraceScenario(name="async-azure-ai", fixture=_azure_fixture, mappings=AZURE_MAPPINGS, asynchronous=True), + TraceScenario( + name="async-anthropic-downstream-stream", + fixture=_stream_fixture, + mappings=( + *ANTHROPIC_MAPPINGS, + mapping(span="python_upstream_stream", python_frame=r"AnthropicMessagesStreamingResponse\.__anext__$"), + mapping( + span="python_downstream_stream", python_frame=r"DataGenerator\.__anext__$|async_data_generator$" + ), + mapping(span="python_stream_callback", python_frame=r"Logging\.async_success_handler$"), + ), + asynchronous=True, + ), ), ) diff --git a/tests/rust-python-harness/strategies/trace_parity/gateway/responses/case.py b/tests/rust-python-harness/strategies/trace_parity/gateway/responses/case.py new file mode 100644 index 00000000000..52c1ebb391f --- /dev/null +++ b/tests/rust-python-harness/strategies/trace_parity/gateway/responses/case.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from typing import Final + +from .....shared.tracing.steps import Engine, mapping +from ...fixtures import json_response, responses_body, responses_stream_events, sse_response +from ...models import GatewayRouteSpec, RouteFixture, TraceScenario, TraceSuite + +MAPPINGS: Final = ( + mapping( + span="python_responses_gateway_route", python_frame=r"response_api_endpoints/endpoints\.py:\d+ responses_api$" + ), + mapping(span="python_gateway_service", python_frame=r"ProxyBaseLLMRequestProcessing\.base_process_llm_request$"), + mapping(rust_span="responses_gateway_route"), + mapping(span="python_responses", python_frame=r"responses/main\.py:\d+ a?responses$"), + mapping(span="python_provider_config", python_frame=r"ProviderConfigManager\.get_provider_responses_api_config$"), + mapping(rust_span="validate_environment", python_frame=r"OpenAIResponsesAPIConfig\.validate_environment$"), + mapping(rust_span="complete_url", python_frame=r"OpenAIResponsesAPIConfig\.get_complete_url$"), + mapping(rust_span="transform_request", python_frame=r"OpenAIResponsesAPIConfig\.transform_responses_api_request$"), + mapping(span="python_logging_pre_call", python_frame=r"Logging\.pre_call$"), + mapping(rust_span="http_request", python_frame=r"AsyncHTTPHandler\.post$|HTTPHandler\.post$"), + mapping(rust_span="transform_response", python_frame=r"OpenAIResponsesAPIConfig\.transform_response_api_response$"), + mapping(span="python_success_callback", python_frame=r"Logging\.async_success_handler$|Logging\.success_handler$"), +) + +STREAM_MAPPINGS: Final = ( + mapping(span="python_stream_iterator", python_frame=r"ResponsesAPIStreamingIterator\.__init__$"), + mapping(span="python_stream_next", python_frame=r"ResponsesAPIStreamingIterator\.__anext__$"), + mapping(span="python_stream_transform", python_frame=r"OpenAIResponsesAPIConfig\.transform_streaming_response$"), + mapping(span="python_downstream_stream", python_frame=r"DataGenerator\.__anext__$|async_data_generator$"), +) + + +def _fixture(_engine: Engine, _base_url: str) -> RouteFixture: + return RouteFixture( + kwargs={ + "model_alias": "trace-model", + "provider_model": "openai/gpt-5", + "body": {"model": "trace-model", "input": "hello"}, + }, + provider_responses=(json_response(responses_body()),), + ) + + +def _stream_fixture(engine: Engine, base_url: str) -> RouteFixture: + fixture: Final = _fixture(engine, base_url) + return fixture.with_body(stream=True).derive( + provider_responses=(sse_response(responses_stream_events()),), + ) + + +TRACE_SUITE: Final = TraceSuite( + route=GatewayRouteSpec("responses"), + scenarios=( + TraceScenario(name="async-openai", fixture=_fixture, mappings=MAPPINGS, asynchronous=True), + TraceScenario( + name="async-openai-downstream-stream", + fixture=_stream_fixture, + mappings=(*MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, + ), + ), +) diff --git a/tests/rust-python-harness/strategies/trace_parity/models.py b/tests/rust-python-harness/strategies/trace_parity/models.py index 04659b25382..078af2b316d 100644 --- a/tests/rust-python-harness/strategies/trace_parity/models.py +++ b/tests/rust-python-harness/strategies/trace_parity/models.py @@ -1,22 +1,45 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass -from typing import Final, Literal, TypeAlias +from typing import Final, Literal, TypeAlias, cast -from ...shared.parity.recorded_http import RecordedHttpResponse +from ...shared.parity.recorded_http import RecordedResponse from ...shared.reporting.models import SdkFunction -from ...shared.tracing.steps import Engine, TraceContract, TraceMapping +from ...shared.tracing.steps import Engine, TraceMapping -TraceMode = Literal["sync", "async"] +TraceEngine = Literal["python", "rust", "both"] TraceFailureSource = Literal["python", "rust", "harness"] @dataclass(frozen=True, slots=True) class RouteFixture: kwargs: dict[str, object] - provider_responses: tuple[RecordedHttpResponse, ...] + provider_responses: tuple[RecordedResponse, ...] expected_failure: bool = False + consume_stream: bool = False + + def derive( + self, + *, + kwargs: Mapping[str, object] | None = None, + provider_responses: tuple[RecordedResponse, ...] | None = None, + expected_failure: bool | None = None, + consume_stream: bool | None = None, + ) -> RouteFixture: + return RouteFixture( + kwargs={**self.kwargs, **(kwargs or {})}, + provider_responses=self.provider_responses if provider_responses is None else provider_responses, + expected_failure=self.expected_failure if expected_failure is None else expected_failure, + consume_stream=self.consume_stream if consume_stream is None else consume_stream, + ) + + def with_body(self, **updates: object) -> RouteFixture: + raw_body: Final = self.kwargs.get("body") + if not isinstance(raw_body, dict): + raise ValueError("route fixture does not contain an object body") + body: Final = cast(dict[str, object], raw_body) + return self.derive(kwargs={"body": {**body, **updates}}) @dataclass(frozen=True, slots=True) @@ -40,14 +63,7 @@ class TraceScenario: name: str fixture: Callable[[Engine, str], RouteFixture] mappings: tuple[TraceMapping, ...] - modes: tuple[TraceMode, ...] = ("sync", "async") - contract: TraceContract = TraceContract() - sync_mappings: tuple[TraceMapping, ...] | None = None - async_mappings: tuple[TraceMapping, ...] | None = None - - def mappings_for(self, mode: TraceMode) -> tuple[TraceMapping, ...]: - selected: Final = self.async_mappings if mode == "async" else self.sync_mappings - return self.mappings if selected is None else selected + asynchronous: bool @dataclass(frozen=True, slots=True) diff --git a/tests/rust-python-harness/strategies/trace_parity/reporting.py b/tests/rust-python-harness/strategies/trace_parity/reporting.py index 9c5bf9e88cd..e7c07ef9c0f 100644 --- a/tests/rust-python-harness/strategies/trace_parity/reporting.py +++ b/tests/rust-python-harness/strategies/trace_parity/reporting.py @@ -1,31 +1,24 @@ from __future__ import annotations import os -import re import sys from collections.abc import Sequence -from typing import Final, Literal +from typing import Final from pydantic import BaseModel, ConfigDict, ValidationError from ...shared.reporting.models import SURFACES, CaseResult, RunStatus, SdkFunction, Surface from ...shared.reporting.rendering import ReportSection from ...shared.reporting.strategy import NotImplementedCaseSpec, SkippedCaseSpec -from ...shared.tracing.steps import ( - PipelineStep, - TraceContract, - TraceDiff, - TraceMapping, - trace_depths, - trace_diff, -) +from ...shared.tracing.steps import PipelineStep, trace_depths +from .models import TraceEngine -TRACE_COMPARISON_ARTIFACT: Final = "trace_comparison" +TRACE_ARTIFACT: Final = "trace" TRACE_PARITY_HINT: Final = ( "rebuild the native bridge with the trace-parity feature, e.g. `uvx maturin develop --features trace-parity`" ) -_COLORS: Final[dict[str, str]] = {"green": "32", "yellow": "33", "red": "31", "cyan": "36"} +_COLORS: Final[dict[str, str]] = {"yellow": "33", "red": "31", "cyan": "36"} _RESET: Final = "\033[0m" @@ -47,26 +40,15 @@ class TraceEventArtifact(BaseModel): return PipelineStep(self.id, self.parent_id, self.span, self.raw) -class TraceMappingArtifact(BaseModel): - model_config = ConfigDict(frozen=True, extra="forbid") - - span: str - python: str | None - rust: str | None - - -class TraceComparisonArtifact(BaseModel): +class TraceArtifact(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") + engine: TraceEngine = "both" surface: Surface sdk_function: SdkFunction scenario: str - mode: Literal["sync", "async"] - mappings: tuple[TraceMappingArtifact, ...] python: tuple[TraceEventArtifact, ...] rust: tuple[TraceEventArtifact, ...] - python_unmatched: int - unordered_children_of: frozenset[str] python_error: str | None = None rust_error: str | None = None @@ -74,41 +56,27 @@ class TraceComparisonArtifact(BaseModel): def from_traces( cls, *, + engine: TraceEngine = "both", surface: Surface, sdk_function: SdkFunction, scenario: str, - mode: Literal["sync", "async"], - mappings: Sequence[TraceMapping], - contract: TraceContract, python: Sequence[PipelineStep], rust: Sequence[PipelineStep], - python_unmatched: int, python_error: str | None = None, rust_error: str | None = None, - ) -> TraceComparisonArtifact: + ) -> TraceArtifact: return cls( + engine=engine, surface=surface, sdk_function=sdk_function, scenario=scenario, - mode=mode, - mappings=tuple( - TraceMappingArtifact( - span=item.span, - python=item.python.pattern if item.python else None, - rust=item.rust, - ) - for item in mappings - ), python=tuple( TraceEventArtifact(id=step.id, parent_id=step.parent_id, span=step.span, raw=step.raw) for step in python ), rust=tuple( - TraceEventArtifact(id=step.id, parent_id=step.parent_id, span=step.span, raw=step.raw) - for step in rust + TraceEventArtifact(id=step.id, parent_id=step.parent_id, span=step.span, raw=step.raw) for step in rust ), - python_unmatched=python_unmatched, - unordered_children_of=contract.unordered_children_of, python_error=python_error, rust_error=rust_error, ) @@ -119,32 +87,9 @@ class TraceComparisonArtifact(BaseModel): def rust_steps(self) -> tuple[PipelineStep, ...]: return tuple(event.step() for event in self.rust) - def diff(self) -> TraceDiff: - return trace_diff( - self.python_steps(), - self.rust_steps(), - tuple( - TraceMapping( - item.span, - re.compile(item.python) if item.python is not None else None, - item.rust, - ) - for item in self.mappings - ), - TraceContract(self.unordered_children_of), - ) - - def exact_match(self) -> bool: - return self.diff().matches - def has_errors(self) -> bool: return self.python_error is not None or self.rust_error is not None - def contract_matches(self) -> bool: - if self.has_errors(): - return False - return self.diff().matches - def _split_raw(raw: str) -> tuple[str, str]: location, separator, name = raw.partition(" ") @@ -153,72 +98,28 @@ def _split_raw(raw: str) -> tuple[str, str]: return raw, "" -def _python_line(index: int, step: PipelineStep, depth: int, exclusive: frozenset[str]) -> str: +def _python_line(index: int, step: PipelineStep, depth: int) -> str: name: Final = _split_raw(step.raw)[0] location: Final = _split_raw(step.raw)[1] suffix: Final = f" ({location})" if location else "" - marker: Final = " [python only]" if step.span in exclusive else "" - return _paint(f"{index} {' ' * depth}{name}{suffix}{marker}", "cyan") + return _paint(f"{index} {' ' * depth}{name}{suffix}", "cyan") -def _python_lines(steps: tuple[PipelineStep, ...], exclusive: frozenset[str]) -> str: +def _python_lines(steps: tuple[PipelineStep, ...]) -> str: depths: Final = trace_depths(steps) - lines: Final = tuple( - _python_line(index, step, depths[step.id], exclusive) for index, step in enumerate(steps, start=1) - ) + lines: Final = tuple(_python_line(index, step, depths[step.id]) for index, step in enumerate(steps, start=1)) return f"{_paint('PYTHON', 'cyan')} ({len(steps)} steps)\n" + ("\n".join(lines) if lines else "(empty)") -def _python_references(steps: tuple[PipelineStep, ...]) -> dict[tuple[str, int], str]: - references: dict[tuple[str, int], str] = {} - occurrences: dict[str, int] = {} - for index, step in enumerate(steps, start=1): - name = _split_raw(step.raw)[0] - occurrence = occurrences.get(step.span, 0) + 1 - occurrences[step.span] = occurrence - references[(step.span, occurrence)] = f"{index} {name}" - return references - - -def _rust_line( - step: PipelineStep, - depth: int, - occurrence: int, - references: dict[tuple[str, int], str], -) -> str: - span: Final = _paint(step.span, "yellow") - key: Final = (step.span, occurrence) - reference: Final = ( - _paint(references[key], "cyan") if key in references else _paint("[rust only]", "yellow") - ) - suffix: Final = f"#{occurrence}" if occurrence > 1 else "" - return f"{' ' * depth}{span}{suffix} -> {reference}" - - -def _rust_lines(steps: tuple[PipelineStep, ...], references: dict[tuple[str, int], str]) -> str: +def _rust_lines(steps: tuple[PipelineStep, ...]) -> str: depths: Final = trace_depths(steps) - occurrences: dict[str, int] = {} - lines: list[str] = [] - for step in steps: - occurrence = occurrences.get(step.span, 0) + 1 - occurrences[step.span] = occurrence - lines.append(_rust_line(step, depths[step.id], occurrence, references)) + lines: Final = tuple( + _paint(f"{index} {' ' * depths[step.id]}{step.span}", "yellow") for index, step in enumerate(steps, 1) + ) return f"{_paint('RUST', 'yellow')} ({len(steps)} steps)\n" + ("\n".join(lines) if lines else "(empty)") -def _state_text(state: str, *, good: bool) -> str: - return _paint(state, "green" if good else "red") - - -def _contract_line(artifact: TraceComparisonArtifact) -> str: - matches: Final = artifact.contract_matches() - status: Final = _state_text("PASS" if matches else "FAIL", good=matches) - if artifact.python_error or artifact.rust_error: - return f"Contract: {status}" - return f"Contract: {status}" - - -def _error_lines(artifact: TraceComparisonArtifact) -> tuple[str, ...]: +def _error_lines(artifact: TraceArtifact) -> tuple[str, ...]: lines: list[str] = [] for engine, error in (("Python", artifact.python_error), ("Rust", artifact.rust_error)): if error is None: @@ -229,68 +130,20 @@ def _error_lines(artifact: TraceComparisonArtifact) -> tuple[str, ...]: return tuple(lines) -def _unseen_mappings( - artifact: TraceComparisonArtifact, - python: tuple[PipelineStep, ...], - rust: tuple[PipelineStep, ...], -) -> tuple[str, ...]: - return artifact.diff().missing_mappings - - -def _comparison_status_lines( - artifact: TraceComparisonArtifact, - python: tuple[PipelineStep, ...], - rust: tuple[PipelineStep, ...], -) -> tuple[str, ...]: - diff: Final = artifact.diff() - exact_match: Final = artifact.exact_match() - if artifact.has_errors(): - return (*_error_lines(artifact), _contract_line(artifact)) - unseen: Final = _unseen_mappings(artifact, python, rust) - unseen_line: Final[tuple[str, ...]] = (f"Unseen mappings: {', '.join(unseen)}",) if unseen else () - drift_lines: Final[tuple[str, ...]] = ( - (_state_text("Same steps, order, and nesting", good=True),) - if exact_match - else ( - _paint(f"Python only: {', '.join(diff.python_only) or 'none'}", "cyan"), - _paint(f"Rust only: {', '.join(diff.rust_only) or 'none'}", "yellow"), - f"First difference: {diff.first_difference or 'none'}", - f"Python frames outside mapping: {artifact.python_unmatched}", - ) - ) - return ( - f"Trace: {_state_text('MATCH' if exact_match else 'DRIFT', good=exact_match)}", - *drift_lines, - *unseen_line, - _contract_line(artifact), - ) - - -def _render_comparison(artifact: TraceComparisonArtifact) -> str: - python: Final = artifact.python_steps() - rust: Final = artifact.rust_steps() - diff: Final = artifact.diff() - python_exclusive: Final = frozenset(item.span for item in artifact.mappings if item.rust is None) - status_lines: Final = _comparison_status_lines(artifact, python, rust) - return "\n\n".join( - ( - _python_lines(python, python_exclusive | frozenset(diff.python_only)), - _rust_lines(rust, _python_references(python)), - "\n".join(status_lines), - ) - ) - - -def _mode(nodeid: str) -> str: - if "[" in nodeid: - return nodeid.rsplit("[", 1)[-1].removesuffix("]") - head, _, tail = nodeid.rpartition(":") - return tail if head else "unknown mode" +def _render_trace(artifact: TraceArtifact) -> str: + traces: tuple[str, ...] + if artifact.engine == "python": + traces = (_python_lines(artifact.python_steps()),) + elif artifact.engine == "rust": + traces = (_rust_lines(artifact.rust_steps()),) + else: + traces = (_python_lines(artifact.python_steps()), _rust_lines(artifact.rust_steps())) + return "\n\n".join((*traces, *_error_lines(artifact))) def _scenario(nodeid: str) -> str: parts: Final = nodeid.split(":") - return parts[-2] if len(parts) >= 5 else "default" + return parts[-1] if len(parts) >= 4 else "default" def _unavailable(status: RunStatus) -> str: @@ -299,20 +152,20 @@ def _unavailable(status: RunStatus) -> str: def _render_artifact(body: str) -> str: try: - artifact: Final = TraceComparisonArtifact.model_validate_json(body) + artifact: Final = TraceArtifact.model_validate_json(body) except ValidationError as error: - return f"Trace comparison artifact is invalid: {error}" - return _render_comparison(artifact) + return f"Trace artifact is invalid: {error}" + return _render_trace(artifact) -def _mode_section(result: CaseResult, nodeid: str, status: RunStatus) -> str: +def _scenario_section(result: CaseResult, nodeid: str, status: RunStatus) -> str: artifacts: Final = tuple( - artifact for artifact in result.artifacts.get(nodeid, ()) if artifact.kind == TRACE_COMPARISON_ARTIFACT + artifact for artifact in result.artifacts.get(nodeid, ()) if artifact.kind == TRACE_ARTIFACT ) body: Final = ( "\n\n".join(_render_artifact(artifact.body) for artifact in artifacts) if artifacts else _unavailable(status) ) - label: Final = f"Scenario: {_scenario(nodeid)} / Mode: {_mode(nodeid)}" + label: Final = f"Scenario: {_scenario(nodeid)}" return f"{label}\n{'-' * len(label)}\n\n{body}" @@ -321,7 +174,7 @@ def _case_block(result: CaseResult) -> str: outcomes: Final = tuple(result.outcomes.items()) or ( (nodeid, RunStatus.NOT_RUN) for nodeid in sorted(result.collected) ) - sections: Final = tuple(_mode_section(result, nodeid, status) for nodeid, status in outcomes) + sections: Final = tuple(_scenario_section(result, nodeid, status) for nodeid, status in outcomes) return "\n\n".join((f"{header}\n{'=' * len(header)}", *sections)) @@ -357,11 +210,11 @@ def _surface_section(surface: Surface, results: Sequence[CaseResult]) -> ReportS *((not_implemented,) if not_implemented else ()), *((skipped,) if skipped else ()), ) - return ReportSection(f"{surface.upper()} trace comparisons", blocks or ("No runnable trace comparisons",)) + return ReportSection(f"{surface.upper()} traces", blocks or ("No runnable traces",)) def render_trace_results(results: Sequence[CaseResult]) -> tuple[ReportSection, ...]: sections: Final = tuple( section for surface in SURFACES if (section := _surface_section(surface, results)) is not None ) - return sections or (ReportSection("Trace comparisons", ("No trace comparisons selected",)),) + return sections or (ReportSection("Traces", ("No traces selected",)),) diff --git a/tests/rust-python-harness/strategies/trace_parity/runner.py b/tests/rust-python-harness/strategies/trace_parity/runner.py index b78a3c7da3f..706e054bf52 100644 --- a/tests/rust-python-harness/strategies/trace_parity/runner.py +++ b/tests/rust-python-harness/strategies/trace_parity/runner.py @@ -4,13 +4,20 @@ import importlib from collections.abc import Sequence from pathlib import Path from time import monotonic -from typing import Final +from typing import Final, cast +from ...shared.native_build import ensure_trace_bridge from ...shared.reporting.models import CaseResult, HarnessCase, HarnessRun, ResultArtifact, RunStatus, Surface from ...shared.reporting.strategy import ModuleCaseSpec, UpdateCallback -from ...shared.native_build import ensure_trace_bridge -from .models import GatewayRouteSpec, RouteSpec, TraceExecutionFailure, TraceMode, TraceScenario, TraceSuite -from .reporting import TRACE_COMPARISON_ARTIFACT, TraceComparisonArtifact +from .models import ( + GatewayRouteSpec, + RouteSpec, + TraceEngine, + TraceExecutionFailure, + TraceScenario, + TraceSuite, +) +from .reporting import TRACE_ARTIFACT, TraceArtifact from .sdk.execution import execute_trace @@ -32,15 +39,13 @@ def validate_trace_suite(suite: TraceSuite, harness_case: HarnessCase) -> str | names: Final = tuple(scenario.name for scenario in suite.scenarios) if not names or len(names) != len(set(names)) or any(not name or ":" in name for name in names): return "scenario names must be non-empty, unique, and colon-free" - invalid_modes: Final = tuple( + invalid_names: Final = tuple( scenario.name for scenario in suite.scenarios - if not scenario.modes - or len(scenario.modes) != len(set(scenario.modes)) - or any(mode not in {"sync", "async"} for mode in scenario.modes) + if not scenario.name.startswith("async-" if scenario.asynchronous else "sync-") ) - if invalid_modes: - return f"scenarios must use non-empty, unique sync/async modes: {', '.join(invalid_modes)}" + if invalid_names: + return f"scenario names must start with sync- or async-: {', '.join(invalid_names)}" surface: Final = harness_case.surface if surface == "sdk" and not isinstance(suite.route, RouteSpec): return "must use RouteSpec for the sdk surface" @@ -57,15 +62,14 @@ def scenario_nodeids( trace_suite: TraceSuite, harness_case: HarnessCase, selected_scenarios: frozenset[str] = frozenset(), -) -> tuple[tuple[TraceScenario, TraceMode, str], ...]: +) -> tuple[tuple[TraceScenario, str], ...]: surface: Final = harness_case.surface if surface is None: return () return tuple( - (scenario, mode, f"trace:{surface}:{harness_case.sdk_function}:{scenario.name}:{mode}") + (scenario, f"trace:{surface}:{harness_case.sdk_function}:{scenario.name}") for scenario in trace_suite.scenarios if not selected_scenarios or scenario.name in selected_scenarios - for mode in scenario.modes ) @@ -77,54 +81,49 @@ def _record_setup_failure(run: HarnessRun, case: HarnessCase, message: str, stag run.failures.append((nodeid, message)) -def run_trace_mode( +def run_trace_scenario( run: HarnessRun, result: CaseResult, trace_suite: TraceSuite, scenario: TraceScenario, - mode: TraceMode, surface: Surface, nodeid: str, on_update: UpdateCallback, + engine: TraceEngine = "both", ) -> None: started_at: Final = monotonic() - comparison: Final = _execute_mode(trace_suite, scenario, mode, surface) + trace: Final = _execute_scenario(trace_suite, scenario, surface, engine) duration: Final = monotonic() - started_at - if isinstance(comparison, TraceExecutionFailure): + if isinstance(trace, TraceExecutionFailure): result.record(nodeid, RunStatus.ERROR, duration) - run.failures.append((nodeid, comparison.message)) + run.failures.append((nodeid, trace.message)) on_update(run) return - artifact: Final = ResultArtifact(TRACE_COMPARISON_ARTIFACT, comparison.model_dump_json()) - if comparison.has_errors(): + artifact: Final = ResultArtifact(TRACE_ARTIFACT, trace.model_dump_json()) + if trace.has_errors(): result.record(nodeid, RunStatus.ERROR, duration, (artifact,)) - run.failures.append( - (nodeid, "\n".join(error for error in (comparison.python_error, comparison.rust_error) if error)) - ) + run.failures.append((nodeid, "\n".join(error for error in (trace.python_error, trace.rust_error) if error))) else: - status: Final = RunStatus.PASSED if comparison.contract_matches() else RunStatus.FAILED - result.record(nodeid, status, duration, (artifact,)) - if status is RunStatus.FAILED: - run.failures.append((nodeid, "trace contract mismatch; see the rendered comparison")) + result.record(nodeid, RunStatus.PASSED, duration, (artifact,)) on_update(run) -def _execute_mode( +def _execute_scenario( trace_suite: TraceSuite, scenario: TraceScenario, - mode: TraceMode, surface: Surface, -) -> TraceComparisonArtifact | TraceExecutionFailure: + engine: TraceEngine, +) -> TraceArtifact | TraceExecutionFailure: route: Final = trace_suite.route if isinstance(route, GatewayRouteSpec): if surface != "gateway": return TraceExecutionFailure("harness", "gateway route cannot run on the sdk surface") from .gateway.execution import execute_gateway_trace - return execute_gateway_trace(route, scenario, mode) + return execute_gateway_trace(route, scenario, engine) if surface != "sdk": return TraceExecutionFailure("harness", "sdk route cannot run on the gateway surface") - return execute_trace(route, scenario, mode, surface) + return execute_trace(route, scenario, surface, engine) def _run_case( @@ -132,6 +131,7 @@ def _run_case( harness_case: HarnessCase, selected_scenarios: frozenset[str], on_update: UpdateCallback, + engine: TraceEngine, ) -> None: result: Final = run.results[harness_case.key] spec: Final = harness_case.spec @@ -146,15 +146,29 @@ def _run_case( on_update(run) return nodeids: Final = scenario_nodeids(trace_suite, harness_case, selected_scenarios) - result.collected.update(nodeid for _, _, nodeid in nodeids) + result.collected.update(nodeid for _, nodeid in nodeids) if not nodeids: result.status = RunStatus.SKIPPED on_update(run) return result.status = RunStatus.RUNNING on_update(run) - for scenario, mode, nodeid in nodeids: - run_trace_mode(run, result, trace_suite, scenario, mode, surface, nodeid, on_update) + for scenario, nodeid in nodeids: + run_trace_scenario(run, result, trace_suite, scenario, surface, nodeid, on_update, engine) + + +def runner_selection(runner_args: Sequence[str]) -> tuple[frozenset[str], TraceEngine]: + engine: TraceEngine = "both" + scenarios: list[str] = [] + for argument in runner_args: + if argument.startswith("--engine="): + value = argument.removeprefix("--engine=") + if value not in {"python", "rust"}: + raise ValueError(f"invalid trace engine: {value}") + engine = cast(TraceEngine, value) + else: + scenarios.append(argument) + return frozenset(scenarios), engine def run_trace_cases( @@ -163,10 +177,10 @@ def run_trace_cases( on_update: UpdateCallback, runner_args: Sequence[str] = (), ) -> tuple[int, HarnessRun]: - selected_scenarios: Final = frozenset(runner_args) + selected_scenarios, engine = runner_selection(runner_args) run: Final = HarnessRun.from_cases(cases) runnable_cases: Final = tuple(case for case in cases if isinstance(case.spec, ModuleCaseSpec)) - bridge_error: Final = ensure_trace_bridge(repo_root) if runnable_cases else None + bridge_error: Final = ensure_trace_bridge(repo_root) if runnable_cases and engine != "python" else None if bridge_error is not None: for harness_case in runnable_cases: _record_setup_failure(run, harness_case, bridge_error, "bridge") @@ -174,7 +188,7 @@ def run_trace_cases( on_update(run) return 1, run for harness_case in cases: - _run_case(run, harness_case, selected_scenarios, on_update) + _run_case(run, harness_case, selected_scenarios, on_update, engine) run.finished_at = monotonic() on_update(run) failed: Final = any( diff --git a/tests/rust-python-harness/strategies/trace_parity/sdk/chat_completions/case.py b/tests/rust-python-harness/strategies/trace_parity/sdk/chat_completions/case.py index 6be5afd60d6..1221f237570 100644 --- a/tests/rust-python-harness/strategies/trace_parity/sdk/chat_completions/case.py +++ b/tests/rust-python-harness/strategies/trace_parity/sdk/chat_completions/case.py @@ -1,10 +1,15 @@ from __future__ import annotations -import json from typing import Final -from .....shared.parity.recorded_http import HttpHeader, RecordedHttpResponse from .....shared.tracing.steps import Engine, mapping +from ...fixtures import ( + anthropic_response_body, + anthropic_stream_events, + aws_event_stream_response, + json_response, + sse_response, +) from ...models import RouteFixture, RouteSpec, TraceScenario, TraceSuite COMMON_MAPPINGS: Final = ( @@ -24,6 +29,22 @@ COMMON_MAPPINGS: Final = ( mapping(rust_span="execute_chat_completions_provider_call"), mapping(rust_span="http_request", python_frame=r"AsyncHTTPHandler\.post$|HTTPHandler\.post$"), mapping(rust_span="transform_response", python_frame=r"(? RouteFixture: - response: Final = json.dumps( - { - "id": "msg_trace", - "type": "message", - "role": "assistant", - "model": "claude-sonnet-5", - "content": [{"type": "text", "text": "hello"}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 2, "output_tokens": 3}, - } - ).encode() return RouteFixture( kwargs={ "model": "anthropic/claude-sonnet-5", "messages": [{"role": "user", "content": "hello"}], **({"optional_params": {"max_tokens": 16}} if engine == "rust" else {"max_tokens": 16}), }, - provider_responses=( - RecordedHttpResponse.from_bytes( - 200, (HttpHeader(name="content-type", value="application/json"),), response - ), - ), + provider_responses=(json_response(anthropic_response_body()),), ) def _bedrock_fixture(engine: Engine, _base_url: str) -> RouteFixture: - response: Final = json.dumps( - { - "output": {"message": {"role": "assistant", "content": [{"text": "hello"}]}}, - "stopReason": "end_turn", - "usage": {"inputTokens": 2, "outputTokens": 3, "totalTokens": 5}, - "metrics": {"latencyMs": 1}, - } - ).encode() + response: Final[dict[str, object]] = { + "output": {"message": {"role": "assistant", "content": [{"text": "hello"}]}}, + "stopReason": "end_turn", + "usage": {"inputTokens": 2, "outputTokens": 3, "totalTokens": 5}, + "metrics": {"latencyMs": 1}, + } credentials: Final = { "aws_access_key_id": "test-access", "aws_secret_access_key": "test-secret", @@ -94,11 +97,60 @@ def _bedrock_fixture(engine: Engine, _base_url: str) -> RouteFixture: else {**credentials, "max_tokens": 16} ), }, + provider_responses=(json_response(response),), + ) + + +def _anthropic_stream_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _anthropic_fixture(engine, _base_url) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(sse_response(anthropic_stream_events()),), + consume_stream=True, + ) + + +def _bedrock_stream_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _bedrock_fixture(engine, _base_url) + events: Final[tuple[dict[str, object], ...]] = ( + {"messageStart": {"role": "assistant"}}, + {"contentBlockStart": {"contentBlockIndex": 0, "start": {}}}, + {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "hello"}}}, + {"contentBlockStop": {"contentBlockIndex": 0}}, + {"messageStop": {"stopReason": "end_turn"}}, + {"metadata": {"usage": {"inputTokens": 2, "outputTokens": 1, "totalTokens": 3}}}, + ) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(aws_event_stream_response(events),), + consume_stream=True, + ) + + +def _provider_error_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _anthropic_fixture(engine, _base_url) + return fixture.derive( provider_responses=( - RecordedHttpResponse.from_bytes( - 200, (HttpHeader(name="content-type", value="application/json"),), response + json_response( + {"type": "error", "error": {"type": "invalid_request_error", "message": "bad request"}}, + status=400, ), ), + expected_failure=True, + ) + + +def _stream_error_fixture(engine: Engine, base_url: str) -> RouteFixture: + fixture: Final = _anthropic_fixture(engine, base_url) + events: Final = ( + anthropic_stream_events()[0], + ("error", {"type": "error", "error": {"type": "overloaded_error", "message": "overloaded"}}), + ) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(sse_response(events),), + expected_failure=True, + consume_stream=True, ) @@ -132,18 +184,64 @@ TRACE_SUITE: Final = TraceSuite( route=SPEC, scenarios=( TraceScenario( - name="anthropic", + name="sync-anthropic", fixture=_anthropic_fixture, - mappings=COMMON_MAPPINGS, - sync_mappings=SYNC_MAPPINGS, - async_mappings=ASYNC_MAPPINGS, + mappings=SYNC_MAPPINGS, + asynchronous=False, ), TraceScenario( - name="bedrock", + name="async-anthropic", + fixture=_anthropic_fixture, + mappings=ASYNC_MAPPINGS, + asynchronous=True, + ), + TraceScenario( + name="sync-anthropic-stream", + fixture=_anthropic_stream_fixture, + mappings=(*SYNC_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=False, + ), + TraceScenario( + name="async-anthropic-stream", + fixture=_anthropic_stream_fixture, + mappings=(*ASYNC_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="async-anthropic-provider-error", + fixture=_provider_error_fixture, + mappings=(*ASYNC_MAPPINGS, *FAILURE_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="async-anthropic-stream-error", + fixture=_stream_error_fixture, + mappings=(*ASYNC_MAPPINGS, *STREAM_MAPPINGS, *FAILURE_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="sync-bedrock", fixture=_bedrock_fixture, - mappings=BEDROCK_COMMON_MAPPINGS, - sync_mappings=BEDROCK_SYNC_MAPPINGS, - async_mappings=BEDROCK_ASYNC_MAPPINGS, + mappings=BEDROCK_SYNC_MAPPINGS, + asynchronous=False, + ), + TraceScenario( + name="async-bedrock", + fixture=_bedrock_fixture, + mappings=BEDROCK_ASYNC_MAPPINGS, + asynchronous=True, + ), + TraceScenario( + name="sync-bedrock-event-stream", + fixture=_bedrock_stream_fixture, + mappings=(*BEDROCK_SYNC_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=False, + ), + TraceScenario( + name="async-bedrock-event-stream", + fixture=_bedrock_stream_fixture, + mappings=(*BEDROCK_ASYNC_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, ), ), ) diff --git a/tests/rust-python-harness/strategies/trace_parity/sdk/execution.py b/tests/rust-python-harness/strategies/trace_parity/sdk/execution.py index eb6c9233565..c249d09fa73 100644 --- a/tests/rust-python-harness/strategies/trace_parity/sdk/execution.py +++ b/tests/rust-python-harness/strategies/trace_parity/sdk/execution.py @@ -1,7 +1,7 @@ from __future__ import annotations import asyncio -from collections.abc import Awaitable +from collections.abc import AsyncIterable, Awaitable, Iterable from dataclasses import dataclass from pathlib import Path from typing import Final, Protocol, cast @@ -11,8 +11,8 @@ from ....shared.reporting.models import Surface from ....shared.tracing.native import TraceResponsePayload, native_trace_events from ....shared.tracing.profiler import FunctionTraceEvent, profile_python from ....shared.tracing.steps import Engine, pipeline_projection -from ..models import RouteFixture, RouteSpec, TraceExecutionFailure, TraceMode, TraceScenario -from ..reporting import TraceComparisonArtifact +from ..models import RouteFixture, RouteSpec, TraceEngine, TraceExecutionFailure, TraceScenario +from ..reporting import TraceArtifact class SdkCall(Protocol): @@ -25,10 +25,20 @@ class _CollectedTrace: error: str | None = None -def _invoke(function: SdkCall, kwargs: dict[str, object], *, asynchronous: bool) -> object: +def _invoke( + function: SdkCall, + kwargs: dict[str, object], + *, + asynchronous: bool, + consume_stream: bool = False, +) -> object: async def invoke_async() -> object: try: - return await cast(Awaitable[object], function(**kwargs)) + response: Final = await cast(Awaitable[object], function(**kwargs)) + if consume_stream and isinstance(response, AsyncIterable): + stream = cast(AsyncIterable[object], response) + return tuple([item async for item in stream]) + return response finally: await asyncio.sleep(0) from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER @@ -38,7 +48,10 @@ def _invoke(function: SdkCall, kwargs: dict[str, object], *, asynchronous: bool) if asynchronous: return asyncio.run(invoke_async()) - return function(**kwargs) + response: Final = function(**kwargs) + if consume_stream and isinstance(response, Iterable): + return tuple(cast(Iterable[object], response)) + return response def _entrypoint(spec: RouteSpec, engine: Engine, *, asynchronous: bool) -> SdkCall | TraceExecutionFailure: @@ -62,14 +75,6 @@ def _entrypoint(spec: RouteSpec, engine: Engine, *, asynchronous: bool) -> SdkCa return cast(SdkCall, getattr(owner, spec.python_entrypoints[int(asynchronous)])) -def _python_invocation_error(function: SdkCall, kwargs: dict[str, object], *, asynchronous: bool) -> str | None: - try: - _invoke(function, kwargs, asynchronous=asynchronous) - except Exception as error: - return f"{type(error).__name__}: {error}" - return None - - def _collect( function: SdkCall, fixture: RouteFixture, @@ -83,8 +88,19 @@ def _collect( return _CollectedTrace(native_trace_events(payload), payload.error) import litellm - with profile_python(Path(litellm.__file__).parent, threads=True) as profiler: - error: Final = _python_invocation_error(function, kwargs, asynchronous=asynchronous) + previous_suppress_debug_info: Final = litellm.suppress_debug_info + try: + if fixture.expected_failure: + litellm.suppress_debug_info = True + with profile_python(Path(litellm.__file__).parent, threads=True) as profiler: + error: str | None + try: + _invoke(function, kwargs, asynchronous=asynchronous, consume_stream=fixture.consume_stream) + error = None + except Exception as caught: + error = f"{type(caught).__name__}: {caught}" + finally: + litellm.suppress_debug_info = previous_suppress_debug_info return _CollectedTrace(tuple(profiler.events), error) @@ -108,6 +124,7 @@ def collect_trace( }, provider_responses=base_fixture.provider_responses, expected_failure=base_fixture.expected_failure, + consume_stream=base_fixture.consume_stream, ) collected: Final = _collect(function, fixture, engine, asynchronous=asynchronous) provider.take_requests(len(fixture.provider_responses)) @@ -129,18 +146,24 @@ def _failure_message(result: tuple[FunctionTraceEvent, ...] | TraceExecutionFail def execute_trace( - route: RouteSpec, scenario: TraceScenario, mode: TraceMode, surface: Surface -) -> TraceComparisonArtifact: - asynchronous: Final = mode == "async" - mappings: Final = scenario.mappings_for(mode) + route: RouteSpec, + scenario: TraceScenario, + surface: Surface, + engine: TraceEngine = "both", +) -> TraceArtifact: + mappings: Final = scenario.mappings scenario_route: Final = RouteSpec( route=route.route, python_entrypoints=route.python_entrypoints, rust_entrypoints=route.rust_entrypoints, fixture=scenario.fixture, ) - python_trace: Final = collect_trace(scenario_route, "python", asynchronous=asynchronous) - rust_trace: Final = collect_trace(scenario_route, "rust", asynchronous=asynchronous) + python_trace: Final = ( + collect_trace(scenario_route, "python", asynchronous=scenario.asynchronous) if engine != "rust" else () + ) + rust_trace: Final = ( + collect_trace(scenario_route, "rust", asynchronous=scenario.asynchronous) if engine != "python" else () + ) python_error: Final = _failure_message(python_trace) rust_error: Final = _failure_message(rust_trace) python_events: Final = python_trace if isinstance(python_trace, tuple) else () @@ -149,28 +172,22 @@ def execute_trace( python: Final = pipeline_projection("python", python_events, mappings) rust: Final = pipeline_projection("rust", rust_events, mappings) except ValueError as error: - return TraceComparisonArtifact.from_traces( + return TraceArtifact.from_traces( + engine=engine, surface=surface, sdk_function=route.route, scenario=scenario.name, - mode=mode, - mappings=mappings, - contract=scenario.contract, python=(), rust=(), - python_unmatched=0, python_error=f"harness: {error}", ) - return TraceComparisonArtifact.from_traces( + return TraceArtifact.from_traces( + engine=engine, surface=surface, sdk_function=route.route, scenario=scenario.name, - mode=mode, - mappings=mappings, - contract=scenario.contract, python=python.steps, rust=rust.steps, - python_unmatched=python.unmatched, python_error=python_error, rust_error=rust_error, ) diff --git a/tests/rust-python-harness/strategies/trace_parity/sdk/messages/case.py b/tests/rust-python-harness/strategies/trace_parity/sdk/messages/case.py index 27079c28cd8..211c454eadf 100644 --- a/tests/rust-python-harness/strategies/trace_parity/sdk/messages/case.py +++ b/tests/rust-python-harness/strategies/trace_parity/sdk/messages/case.py @@ -1,14 +1,26 @@ from __future__ import annotations -import json from typing import Final -from .....shared.parity.recorded_http import HttpHeader, RecordedHttpResponse from .....shared.tracing.steps import Engine, mapping +from ...fixtures import ( + anthropic_response_body, + anthropic_stream_events, + aws_event_stream_response, + json_response, + sse_response, +) from ...models import RouteFixture, RouteSpec, TraceScenario, TraceSuite COMMON_MAPPINGS: Final = ( mapping(rust_span="messages", python_frame=r"anthropic_interface/messages/__init__\.py:\d+ a?create$"), + mapping(span="python_sanitize_empty_content", python_frame=r"strip_empty_content_blocks_from_anthropic_messages$"), + mapping(span="python_sanitize_tool_ids", python_frame=r"sanitize_tool_use_ids_in_anthropic_messages$"), + mapping( + span="python_flatten_web_search", python_frame=r"flatten_unencrypted_web_search_results_in_anthropic_messages$" + ), + mapping(span="python_cache_control", python_frame=r"AnthropicCacheControlHook\.maybe_inject_cache_control$"), + mapping(span="python_pre_request_hooks", python_frame=r"_execute_pre_request_hooks$"), mapping( span="python_messages_provider_config", python_frame=r"ProviderConfigManager\.get_provider_anthropic_messages_config$", @@ -30,10 +42,42 @@ COMMON_MAPPINGS: Final = ( ), mapping(rust_span="http_request", python_frame=r"AsyncHTTPHandler\.post$|HTTPHandler\.post$"), mapping(rust_span="transform_response", python_frame=r"(? RouteFixture: conversation: Final = {"messages": [{"role": "user", "content": "hello"}], "max_tokens": 16} - response: Final = json.dumps( - { - "id": "msg_trace", - "type": "message", - "role": "assistant", - "model": "claude-sonnet-5", - "content": [{"type": "text", "text": "hello"}], - "stop_reason": "end_turn", - "stop_sequence": None, - "usage": {"input_tokens": 2, "output_tokens": 3}, - } - ).encode() return RouteFixture( kwargs={ "model": f"{provider}/claude-sonnet-5", **({"body": {**conversation, "model": "claude-sonnet-5"}} if engine == "rust" else conversation), }, - provider_responses=( - RecordedHttpResponse.from_bytes( - 200, (HttpHeader(name="content-type", value="application/json"),), response - ), - ), + provider_responses=(json_response(anthropic_response_body()),), ) @@ -88,11 +156,170 @@ def _azure_fixture(engine: Engine, _base_url: str) -> RouteFixture: return _fixture(engine, "azure_ai") +def _bedrock_kwargs(engine: Engine) -> dict[str, object]: + conversation: Final = {"messages": [{"role": "user", "content": "hello"}], "max_tokens": 16} + return { + "model": "bedrock/anthropic.claude-3-sonnet-20240229-v1:0", + **( + {"body": {**conversation, "model": "anthropic.claude-3-sonnet-20240229-v1:0"}} + if engine == "rust" + else conversation + ), + "aws_access_key_id": "AKIAIOSFODNN7EXAMPLE", + "aws_secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + "aws_region_name": "us-east-1", + } + + +def _bedrock_fixture(engine: Engine, _base_url: str) -> RouteFixture: + response_fixture: Final = _fixture(engine, "anthropic") + return RouteFixture(kwargs=_bedrock_kwargs(engine), provider_responses=response_fixture.provider_responses) + + +def _bedrock_retry_fixture(engine: Engine, _base_url: str) -> RouteFixture: + success_fixture: Final = _bedrock_fixture(engine, _base_url) + messages: Final = [ + {"role": "user", "content": "hello"}, + { + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "old reasoning", "signature": ""}, + {"type": "text", "text": "partial answer"}, + ], + }, + {"role": "user", "content": "continue"}, + ] + kwargs: Final = { + **_bedrock_kwargs(engine), + **( + {"body": {"messages": messages, "max_tokens": 16, "model": "anthropic.claude-3-sonnet-20240229-v1:0"}} + if engine == "rust" + else {"messages": messages} + ), + } + return success_fixture.derive( + kwargs=kwargs, + provider_responses=( + json_response({"message": "messages.1.content.0: Invalid `signature` in `thinking` block"}, status=400), + *success_fixture.provider_responses, + ), + ) + + +def _mock_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _fixture(engine, "anthropic") + return fixture.derive(kwargs={"mock_response": "hello from mock"}, provider_responses=()) + + +def _provider_error_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _fixture(engine, "anthropic") + return fixture.derive( + provider_responses=( + json_response( + {"type": "error", "error": {"type": "invalid_request_error", "message": "bad request"}}, + status=400, + ), + ), + expected_failure=True, + ) + + +def _sync_unsupported_fixture(engine: Engine, base_url: str) -> RouteFixture: + if engine == "rust": + return _anthropic_fixture(engine, base_url) + fixture: Final = _fixture(engine, "anthropic") + return fixture.derive(provider_responses=(), expected_failure=True) + + +def _stream_fixture_for(engine: Engine, provider: str) -> RouteFixture: + fixture: Final = _fixture(engine, provider) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(sse_response(anthropic_stream_events()),), + consume_stream=True, + ) + + +def _stream_fixture(engine: Engine, _base_url: str) -> RouteFixture: + return _stream_fixture_for(engine, "anthropic") + + +def _azure_stream_fixture(engine: Engine, _base_url: str) -> RouteFixture: + return _stream_fixture_for(engine, "azure_ai") + + +def _bedrock_stream_fixture(engine: Engine, base_url: str) -> RouteFixture: + fixture: Final = _bedrock_fixture(engine, base_url) + events: Final = tuple(payload for _, payload in anthropic_stream_events()) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(aws_event_stream_response(events),), + consume_stream=True, + ) + + +def _bedrock_stream_error_fixture(engine: Engine, base_url: str) -> RouteFixture: + fixture: Final = _bedrock_fixture(engine, base_url) + start: Final = anthropic_stream_events(model="anthropic.claude-3-sonnet-20240229-v1:0")[0][1] + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(aws_event_stream_response((start, {"type": "message_stop"}), corrupt_last_frame=True),), + expected_failure=True, + consume_stream=True, + ) + + SPEC: Final = RouteSpec("messages", ("create", "acreate"), ("messages", "amessages"), _anthropic_fixture) TRACE_SUITE: Final = TraceSuite( route=SPEC, scenarios=( - TraceScenario(name="anthropic", fixture=_anthropic_fixture, mappings=ANTHROPIC_MAPPINGS, modes=("async",)), - TraceScenario(name="azure-ai", fixture=_azure_fixture, mappings=AZURE_MAPPINGS, modes=("async",)), + TraceScenario( + name="async-anthropic", fixture=_anthropic_fixture, mappings=ANTHROPIC_MAPPINGS, asynchronous=True + ), + TraceScenario(name="async-azure-ai", fixture=_azure_fixture, mappings=AZURE_MAPPINGS, asynchronous=True), + TraceScenario(name="async-bedrock", fixture=_bedrock_fixture, mappings=BEDROCK_MAPPINGS, asynchronous=True), + TraceScenario( + name="async-bedrock-invalid-thinking-retry", + fixture=_bedrock_retry_fixture, + mappings=RETRY_MAPPINGS, + asynchronous=True, + ), + TraceScenario(name="async-mock-response", fixture=_mock_fixture, mappings=MOCK_MAPPINGS, asynchronous=True), + TraceScenario( + name="async-anthropic-provider-error", + fixture=_provider_error_fixture, + mappings=ANTHROPIC_FAILURE_MAPPINGS, + asynchronous=True, + ), + TraceScenario( + name="async-anthropic-stream", + fixture=_stream_fixture, + mappings=(*ANTHROPIC_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="async-azure-ai-stream", + fixture=_azure_stream_fixture, + mappings=(*AZURE_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="async-bedrock-event-stream", + fixture=_bedrock_stream_fixture, + mappings=(*BEDROCK_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="async-bedrock-event-stream-error", + fixture=_bedrock_stream_error_fixture, + mappings=(*BEDROCK_MAPPINGS, *STREAM_MAPPINGS, *FAILURE_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="sync-unsupported", + fixture=_sync_unsupported_fixture, + mappings=ANTHROPIC_MAPPINGS, + asynchronous=False, + ), ), ) diff --git a/tests/rust-python-harness/strategies/trace_parity/sdk/ocr/case.py b/tests/rust-python-harness/strategies/trace_parity/sdk/ocr/case.py index 2a4a1b3a152..effd1a0b4f6 100644 --- a/tests/rust-python-harness/strategies/trace_parity/sdk/ocr/case.py +++ b/tests/rust-python-harness/strategies/trace_parity/sdk/ocr/case.py @@ -358,53 +358,88 @@ TRACE_SUITE: Final = TraceSuite( route=SPEC, scenarios=( TraceScenario( - name="mistral", + name="sync-mistral", fixture=_mistral_fixture, - mappings=COMMON_MAPPINGS, - sync_mappings=(*SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), - async_mappings=(*ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + mappings=(*SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=False, ), TraceScenario( - name="mistral-callback-success", + name="async-mistral", + fixture=_mistral_fixture, + mappings=(*ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=True, + ), + TraceScenario( + name="sync-mistral-callback-success", fixture=_mistral_callback_success_fixture, - mappings=COMMON_MAPPINGS, - sync_mappings=CALLBACK_SUCCESS_SYNC_MAPPINGS, - async_mappings=CALLBACK_SUCCESS_ASYNC_MAPPINGS, + mappings=CALLBACK_SUCCESS_SYNC_MAPPINGS, + asynchronous=False, ), TraceScenario( - name="mistral-callback-failure", + name="async-mistral-callback-success", + fixture=_mistral_callback_success_fixture, + mappings=CALLBACK_SUCCESS_ASYNC_MAPPINGS, + asynchronous=True, + ), + TraceScenario( + name="sync-mistral-callback-failure", fixture=_mistral_callback_failure_fixture, - mappings=(*COMMON_MAPPINGS, FAILURE_CALLBACK_MAPPING), - sync_mappings=CALLBACK_FAILURE_SYNC_MAPPINGS, - async_mappings=CALLBACK_FAILURE_ASYNC_MAPPINGS, + mappings=CALLBACK_FAILURE_SYNC_MAPPINGS, + asynchronous=False, ), TraceScenario( - name="azure-ai", + name="async-mistral-callback-failure", + fixture=_mistral_callback_failure_fixture, + mappings=CALLBACK_FAILURE_ASYNC_MAPPINGS, + asynchronous=True, + ), + TraceScenario( + name="sync-azure-ai", fixture=_azure_fixture, - mappings=AZURE_COMMON_MAPPINGS, - sync_mappings=(*AZURE_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), - async_mappings=(*AZURE_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + mappings=(*AZURE_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=False, ), TraceScenario( - name="azure-document-intelligence", + name="async-azure-ai", + fixture=_azure_fixture, + mappings=(*AZURE_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=True, + ), + TraceScenario( + name="sync-azure-document-intelligence", fixture=_azure_document_intelligence_fixture, - mappings=DOCUMENT_INTELLIGENCE_COMMON_MAPPINGS, - sync_mappings=(*DOCUMENT_INTELLIGENCE_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), - async_mappings=(*DOCUMENT_INTELLIGENCE_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + mappings=(*DOCUMENT_INTELLIGENCE_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=False, ), TraceScenario( - name="vertex-ai", + name="async-azure-document-intelligence", + fixture=_azure_document_intelligence_fixture, + mappings=(*DOCUMENT_INTELLIGENCE_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=True, + ), + TraceScenario( + name="sync-vertex-ai", fixture=_vertex_fixture, - mappings=VERTEX_COMMON_MAPPINGS, - sync_mappings=(*VERTEX_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), - async_mappings=(*VERTEX_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + mappings=(*VERTEX_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=False, ), TraceScenario( - name="vertex-deepseek", + name="async-vertex-ai", + fixture=_vertex_fixture, + mappings=(*VERTEX_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=True, + ), + TraceScenario( + name="sync-vertex-deepseek", fixture=_vertex_deepseek_fixture, - mappings=DEEPSEEK_COMMON_MAPPINGS, - sync_mappings=(*DEEPSEEK_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), - async_mappings=(*DEEPSEEK_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + mappings=(*DEEPSEEK_SYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=False, + ), + TraceScenario( + name="async-vertex-deepseek", + fixture=_vertex_deepseek_fixture, + mappings=(*DEEPSEEK_ASYNC_MAPPINGS, IGNORED_SUCCESS_CALLBACK_MAPPING), + asynchronous=True, ), ), ) diff --git a/tests/rust-python-harness/strategies/trace_parity/sdk/responses/case.py b/tests/rust-python-harness/strategies/trace_parity/sdk/responses/case.py new file mode 100644 index 00000000000..68d2eaba9bf --- /dev/null +++ b/tests/rust-python-harness/strategies/trace_parity/sdk/responses/case.py @@ -0,0 +1,216 @@ +from __future__ import annotations + +from typing import Final + +from .....shared.tracing.steps import Engine, mapping +from ...fixtures import ( + anthropic_response_body, + anthropic_stream_events, + json_response, + responses_body, + responses_stream_events, + sse_response, +) +from ...models import RouteFixture, RouteSpec, TraceScenario, TraceSuite + +COMMON_MAPPINGS: Final = ( + mapping(span="python_responses", python_frame=r"responses/main\.py:\d+ a?responses$"), + mapping( + span="python_responses_provider_config", + python_frame=r"ProviderConfigManager\.get_provider_responses_api_config$", + ), + mapping(rust_span="responses_provider_config"), + mapping(rust_span="validate_environment", python_frame=r"validate_environment$"), + mapping(rust_span="complete_url", python_frame=r"get_complete_url$"), + mapping( + rust_span="transform_request", + python_frame=r"(? RouteFixture: + model: Final = "gpt-5" + return RouteFixture( + kwargs={ + "model": f"{provider}/{model}", + "input": "hello", + **({"body": {"model": model, "input": "hello"}} if engine == "rust" else {}), + }, + provider_responses=(json_response(responses_body(model=model)),), + ) + + +def _openai_fixture(engine: Engine, _base_url: str) -> RouteFixture: + return _native_fixture(engine, "openai") + + +def _azure_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _native_fixture(engine, "azure") + return fixture.derive(kwargs={"api_version": "2025-04-01-preview"}) + + +def _openai_stream_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _openai_fixture(engine, _base_url) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(sse_response(responses_stream_events()),), + consume_stream=True, + ) + + +def _provider_error_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _openai_fixture(engine, _base_url) + return fixture.derive( + provider_responses=( + json_response({"error": {"message": "bad request", "type": "invalid_request_error"}}, status=400), + ), + expected_failure=True, + ) + + +def _stream_failed_fixture(engine: Engine, base_url: str) -> RouteFixture: + fixture: Final = _openai_fixture(engine, base_url) + failed_response: Final[dict[str, object]] = { + **responses_body(), + "status": "failed", + "output": [], + "error": {"message": "stream failed", "type": "server_error", "code": "server_error"}, + } + events: Final = ( + ( + "response.created", + {"type": "response.created", "response": {**failed_response, "status": "in_progress", "error": None}}, + ), + ("response.failed", {"type": "response.failed", "response": failed_response}), + ) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(sse_response(events),), + expected_failure=True, + consume_stream=True, + ) + + +def _anthropic_bridge_fixture(engine: Engine, _base_url: str) -> RouteFixture: + return RouteFixture( + kwargs={ + "model": "anthropic/claude-sonnet-5", + "input": "hello", + "max_output_tokens": 16, + **({"body": {"model": "claude-sonnet-5", "input": "hello"}} if engine == "rust" else {}), + }, + provider_responses=(json_response(anthropic_response_body()),), + ) + + +def _anthropic_bridge_stream_fixture(engine: Engine, _base_url: str) -> RouteFixture: + fixture: Final = _anthropic_bridge_fixture(engine, _base_url) + return fixture.derive( + kwargs={"stream": True}, + provider_responses=(sse_response(anthropic_stream_events()),), + consume_stream=True, + ) + + +SPEC: Final = RouteSpec("responses", ("responses", "aresponses"), ("responses", "aresponses"), _openai_fixture) +TRACE_SUITE: Final = TraceSuite( + route=SPEC, + scenarios=( + TraceScenario(name="sync-openai", fixture=_openai_fixture, mappings=COMMON_MAPPINGS, asynchronous=False), + TraceScenario(name="async-openai", fixture=_openai_fixture, mappings=COMMON_MAPPINGS, asynchronous=True), + TraceScenario( + name="sync-openai-stream", + fixture=_openai_stream_fixture, + mappings=(*COMMON_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=False, + ), + TraceScenario( + name="async-openai-stream", + fixture=_openai_stream_fixture, + mappings=(*COMMON_MAPPINGS, *STREAM_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="async-openai-provider-error", + fixture=_provider_error_fixture, + mappings=(*COMMON_MAPPINGS, *FAILURE_MAPPINGS), + asynchronous=True, + ), + TraceScenario( + name="async-openai-stream-failed", + fixture=_stream_failed_fixture, + mappings=(*COMMON_MAPPINGS, *STREAM_MAPPINGS, *FAILURE_MAPPINGS), + asynchronous=True, + ), + TraceScenario(name="async-azure", fixture=_azure_fixture, mappings=AZURE_MAPPINGS, asynchronous=True), + TraceScenario( + name="async-anthropic-chat-bridge", + fixture=_anthropic_bridge_fixture, + mappings=BRIDGE_MAPPINGS, + asynchronous=True, + ), + TraceScenario( + name="async-anthropic-chat-bridge-stream", + fixture=_anthropic_bridge_stream_fixture, + mappings=( + *BRIDGE_MAPPINGS, + mapping(span="python_chat_stream_wrapper", python_frame=r"CustomStreamWrapper\.__init__$"), + mapping(span="python_chat_stream_next", python_frame=r"CustomStreamWrapper\.__anext__$"), + mapping( + span="python_responses_bridge_stream_iterator", + python_frame=r"LiteLLMCompletionStreamingIterator\.__init__$|LiteLLMCompletionStreamingIterator\.__anext__$", + ), + ), + asynchronous=True, + ), + ), +) diff --git a/tests/rust-python-harness/strategies/trace_parity/sdk/test_core_scenario_matrix.py b/tests/rust-python-harness/strategies/trace_parity/sdk/test_core_scenario_matrix.py new file mode 100644 index 00000000000..d0921398795 --- /dev/null +++ b/tests/rust-python-harness/strategies/trace_parity/sdk/test_core_scenario_matrix.py @@ -0,0 +1,63 @@ +from __future__ import annotations + +from importlib import import_module +from typing import Final, cast + +from ..models import TraceSuite + + +def _suite(module: str) -> TraceSuite: + loaded: Final = import_module(module) + candidate: Final = cast(object, getattr(loaded, "TRACE_SUITE")) + assert isinstance(candidate, TraceSuite) + return candidate + + +def test_core_sdk_scenario_matrix_keeps_distinct_migration_paths() -> None: + chat: Final = _suite("tests.rust-python-harness.strategies.trace_parity.sdk.chat_completions.case") + messages: Final = _suite("tests.rust-python-harness.strategies.trace_parity.sdk.messages.case") + responses: Final = _suite("tests.rust-python-harness.strategies.trace_parity.sdk.responses.case") + + assert {(scenario.name, scenario.asynchronous) for scenario in chat.scenarios} >= { + ("sync-anthropic", False), + ("async-anthropic", True), + ("sync-anthropic-stream", False), + ("async-anthropic-stream", True), + ("async-anthropic-provider-error", True), + ("async-anthropic-stream-error", True), + ("sync-bedrock", False), + ("async-bedrock", True), + ("sync-bedrock-event-stream", False), + ("async-bedrock-event-stream", True), + } + assert {(scenario.name, scenario.asynchronous) for scenario in messages.scenarios} >= { + ("async-anthropic-stream", True), + ("async-azure-ai-stream", True), + ("async-bedrock-event-stream", True), + ("async-bedrock-event-stream-error", True), + ("async-bedrock-invalid-thinking-retry", True), + ("sync-unsupported", False), + } + assert {(scenario.name, scenario.asynchronous) for scenario in responses.scenarios} >= { + ("sync-openai", False), + ("async-openai", True), + ("sync-openai-stream", False), + ("async-openai-stream", True), + ("async-openai-provider-error", True), + ("async-openai-stream-failed", True), + ("async-azure", True), + ("async-anthropic-chat-bridge", True), + ("async-anthropic-chat-bridge-stream", True), + } + + +def test_core_gateway_matrix_keeps_downstream_streams_separate() -> None: + modules: Final = ( + "tests.rust-python-harness.strategies.trace_parity.gateway.chat_completions.case", + "tests.rust-python-harness.strategies.trace_parity.gateway.messages.case", + "tests.rust-python-harness.strategies.trace_parity.gateway.responses.case", + ) + + for module in modules: + suite = _suite(module) + assert any("downstream-stream" in scenario.name for scenario in suite.scenarios) diff --git a/tests/rust-python-harness/strategies/trace_parity/sdk/transcription/case.py b/tests/rust-python-harness/strategies/trace_parity/sdk/transcription/case.py index 2fb1054d10a..3b4d2e1447d 100644 --- a/tests/rust-python-harness/strategies/trace_parity/sdk/transcription/case.py +++ b/tests/rust-python-harness/strategies/trace_parity/sdk/transcription/case.py @@ -92,11 +92,16 @@ TRACE_SUITE: Final = TraceSuite( route=SPEC, scenarios=( TraceScenario( - name="bedrock", + name="sync-bedrock", fixture=_fixture, - mappings=MAPPINGS, - sync_mappings=SYNC_MAPPINGS, - async_mappings=ASYNC_MAPPINGS, + mappings=SYNC_MAPPINGS, + asynchronous=False, + ), + TraceScenario( + name="async-bedrock", + fixture=_fixture, + mappings=ASYNC_MAPPINGS, + asynchronous=True, ), ), ) diff --git a/tests/rust-python-harness/strategies/trace_parity/test_reporting.py b/tests/rust-python-harness/strategies/trace_parity/test_reporting.py index 68264064c92..22cc87592b8 100644 --- a/tests/rust-python-harness/strategies/trace_parity/test_reporting.py +++ b/tests/rust-python-harness/strategies/trace_parity/test_reporting.py @@ -1,58 +1,46 @@ from __future__ import annotations -from collections.abc import Sequence from typing import Final, Literal import pytest from ...shared.reporting.models import CaseResult, Coverage, HarnessCase, ResultArtifact, RunStatus from ...shared.reporting.strategy import ModuleCaseSpec, NotImplementedCaseSpec -from ...shared.tracing.steps import PipelineStep, TraceContract, TraceMapping, mapping +from ...shared.tracing.steps import PipelineStep from . import reporting -from .reporting import TRACE_COMPARISON_ARTIFACT, TraceComparisonArtifact, render_trace_results - -MAPPINGS: Final = ( - mapping(rust_span="ocr", python_frame=r"ocr/main\.py:\d+ a?ocr$"), - mapping(rust_span="http_request", python_frame=r"AsyncHTTPHandler\.post$"), -) +from .reporting import TRACE_ARTIFACT, TraceArtifact, render_trace_results -def _result(comparison: TraceComparisonArtifact) -> CaseResult: +def _result(trace: TraceArtifact) -> CaseResult: case: Final = HarnessCase( strategy_id="trace_parity", strategy_label="Trace parity", - sdk_function=comparison.sdk_function, + sdk_function=trace.sdk_function, spec=ModuleCaseSpec(coverage=Coverage.PARTIAL, module="example"), - surface=comparison.surface, + surface=trace.surface, ) result: Final = CaseResult(case=case) - nodeid: Final = f"trace:sdk:{comparison.sdk_function}:{comparison.scenario}:{comparison.mode}" + nodeid: Final = f"trace:{trace.surface}:{trace.sdk_function}:{trace.scenario}" result.collected.add(nodeid) - result.record( - nodeid, - RunStatus.PASSED, - artifacts=(ResultArtifact(TRACE_COMPARISON_ARTIFACT, comparison.model_dump_json()),), - ) + result.record(nodeid, RunStatus.PASSED, artifacts=(ResultArtifact(TRACE_ARTIFACT, trace.model_dump_json()),)) return result -def _comparison( +def _trace( python: tuple[PipelineStep, ...], rust: tuple[PipelineStep, ...], *, - mappings: Sequence[TraceMapping] = MAPPINGS, rust_error: str | None = None, -) -> TraceComparisonArtifact: - return TraceComparisonArtifact.from_traces( + engine: Literal["python", "rust", "both"] = "both", + scenario: str = "sync-default", +) -> TraceArtifact: + return TraceArtifact.from_traces( + engine=engine, surface="sdk", sdk_function="ocr", - scenario="default", - mode="sync", - mappings=mappings, - contract=TraceContract(), + scenario=scenario, python=python, rust=rust, - python_unmatched=796, rust_error=rust_error, ) @@ -67,107 +55,55 @@ def _events(*items: tuple[str, int, str | None]) -> tuple[PipelineStep, ...]: return tuple(steps) -def test_renderer_shows_matching_python_and_rust_paths() -> None: - rust: Final = _events(("ocr", 0, None), ("http_request", 1, None)) +def test_renderer_prints_python_and_rust_traces_independently() -> None: python: Final = _events( ("ocr", 0, "ocr/main.py:88 aocr"), - ("http_request", 1, "http_handler.py:673 AsyncHTTPHandler.post"), + ("python_prepare", 1, "prep.py:1 python_prepare"), ) - - section: Final = render_trace_results((_result(_comparison(python, rust)),))[0] - report: Final = "\n\n".join(section.blocks) - - assert section.title == "SDK trace comparisons" - assert "Case: ocr" in report - assert "PYTHON (2 steps)\n1 aocr (ocr/main.py:88)\n2 AsyncHTTPHandler.post (http_handler.py:673)" in report - assert "RUST (2 steps)\nocr -> 1 aocr\n http_request -> 2 AsyncHTTPHandler.post" in report - assert "Mapping (identifier -> span)" not in report - assert "Trace: MATCH" in report - assert "Same steps, order, and nesting" in report - assert "Unseen mappings:" not in report - - -def test_renderer_reports_mappings_that_matched_nothing() -> None: - events: Final = _events(("ocr", 0, None)) - - section: Final = render_trace_results((_result(_comparison(events, events)),))[0] - report: Final = "\n\n".join(section.blocks) - - assert "Unseen mappings: http_request" in report - assert "Contract: FAIL" in report - - -def test_renderer_numbers_repeated_span_occurrences() -> None: - mappings: Final = (MAPPINGS[0], MAPPINGS[1]) - rust: Final = _events(("ocr", 0, None), ("http_request", 1, None), ("http_request", 1, None)) - python: Final = _events( - ("ocr", 0, "ocr/main.py:88 aocr"), - ("http_request", 1, "http_handler.py:673 AsyncHTTPHandler.post"), - ("http_request", 1, "http_handler.py:673 AsyncHTTPHandler.post"), - ) - - report: Final = "\n\n".join(render_trace_results((_result(_comparison(python, rust, mappings=mappings)),))[0].blocks) - - assert "http_request#2" in report - - -def test_renderer_accepts_declared_engine_specific_steps() -> None: - mappings: Final = ( - *MAPPINGS[:1], - mapping(span="python_prepare", python_frame=r"python_prepare$"), - mapping(rust_span="rust_prepare"), - ) - python: Final = _events(("ocr", 0, None), ("python_prepare", 1, "prep.py:1 python_prepare")) rust: Final = _events(("ocr", 0, None), ("rust_prepare", 1, None)) - section: Final = render_trace_results((_result(_comparison(python, rust, mappings=mappings)),))[0] + section: Final = render_trace_results((_result(_trace(python, rust)),))[0] report: Final = "\n\n".join(section.blocks) - assert "2 python_prepare (prep.py:1) [python only]" in report - assert "rust_prepare -> [rust only]" in report - assert "Trace: MATCH" in report - assert "Contract: PASS" in report + assert section.title == "SDK traces" + assert "PYTHON (2 steps)\n1 aocr (ocr/main.py:88)\n2 python_prepare (prep.py:1)" in report + assert "RUST (2 steps)\n1 ocr\n2 rust_prepare" in report + assert "python only" not in report + assert "rust only" not in report + assert " -> " not in report + assert "Trace: MATCH" not in report + assert "Trace: DRIFT" not in report + assert "Contract:" not in report -def test_unavailable_check_reports_mode_from_nodeid() -> None: - case: Final = HarnessCase( - strategy_id="trace_parity", - strategy_label="Trace parity", - sdk_function="ocr", - spec=ModuleCaseSpec(coverage=Coverage.PARTIAL, module="example"), - surface="sdk", - ) - result: Final = CaseResult(case=case) - result.collected.add("trace:sdk:ocr:default:sync") - result.record("trace:sdk:ocr:default:sync", RunStatus.ERROR) +@pytest.mark.parametrize( + ("engine", "present", "absent"), + (("python", "PYTHON (1 steps)", "RUST"), ("rust", "RUST (1 steps)", "PYTHON")), +) +def test_renderer_prints_only_selected_engine(engine: Literal["python", "rust"], present: str, absent: str) -> None: + events: Final = _events(("ocr", 0, None)) - section: Final = render_trace_results((result,))[0] - report: Final = "\n\n".join(section.blocks) + report: Final = "\n\n".join(render_trace_results((_result(_trace(events, events, engine=engine)),))[0].blocks) - assert "Case: ocr" in report - assert "Scenario: default / Mode: sync" in report - assert "Trace: NOT AVAILABLE\nTest outcome: error" in report - assert "unknown mode" not in report + assert present in report + assert absent not in report def test_renderer_keeps_collected_trace_when_one_engine_errors() -> None: - python: Final = _events( - ("ocr", 0, "ocr/main.py:88 aocr"), - ("http_request", 1, "http_handler.py:673 AsyncHTTPHandler.post"), + python: Final = _events(("ocr", 0, "ocr/main.py:88 aocr")) + + report: Final = "\n\n".join( + render_trace_results( + (_result(_trace(python, (), rust_error="rust: native Rust bridge must include the trace-parity feature")),) + )[0].blocks ) - section: Final = render_trace_results( - (_result(_comparison(python, (), rust_error="rust: native Rust bridge must include the trace-parity feature")),) - )[0] - report: Final = "\n\n".join(section.blocks) - - assert "PYTHON (2 steps)\n1 aocr (ocr/main.py:88) [python only]" in report + assert "PYTHON (1 steps)\n1 aocr (ocr/main.py:88)" in report assert "Rust error: rust: native Rust bridge must include the trace-parity feature" in report assert "hint: rebuild the native bridge with the trace-parity feature" in report - assert "Contract: FAIL" in report -def test_renderer_groups_all_modes_under_one_case_header() -> None: +def test_unavailable_trace_reports_scenario_from_nodeid() -> None: case: Final = HarnessCase( strategy_id="trace_parity", strategy_label="Trace parity", @@ -176,76 +112,55 @@ def test_renderer_groups_all_modes_under_one_case_header() -> None: surface="sdk", ) result: Final = CaseResult(case=case) - events: Final = _events(("ocr", 0, None)) - modes: Final[tuple[Literal["sync", "async"], ...]] = ("sync", "async") - for mode in modes: - nodeid = f"trace:sdk:ocr:default:{mode}" - result.collected.add(nodeid) - comparison = TraceComparisonArtifact.from_traces( - surface="sdk", - sdk_function="ocr", - scenario="default", - mode=mode, - mappings=MAPPINGS, - contract=TraceContract(), - python=events, - rust=events, - python_unmatched=0, - ) - result.record( - nodeid, - RunStatus.PASSED, - artifacts=(ResultArtifact(TRACE_COMPARISON_ARTIFACT, comparison.model_dump_json()),), - ) + result.collected.add("trace:sdk:ocr:async-error") + result.record("trace:sdk:ocr:async-error", RunStatus.ERROR) - section: Final = render_trace_results((result,))[0] + report: Final = "\n\n".join(render_trace_results((result,))[0].blocks) + + assert "Scenario: async-error" in report + assert "Trace: NOT AVAILABLE\nTest outcome: error" in report + + +def test_renderer_groups_scenarios_under_one_case_header() -> None: + result: Final = _result(_trace(_events(("ocr", 0, None)), (), scenario="sync-default")) + async_trace: Final = _trace((), _events(("ocr", 0, None)), scenario="async-default") + nodeid: Final = "trace:sdk:ocr:async-default" + result.collected.add(nodeid) + result.record(nodeid, RunStatus.PASSED, artifacts=(ResultArtifact(TRACE_ARTIFACT, async_trace.model_dump_json()),)) + + report: Final = render_trace_results((result,))[0].blocks[0] - assert len(section.blocks) == 1 - report: Final = section.blocks[0] assert report.count("Case: ocr") == 1 - assert "Scenario: default / Mode: sync" in report - assert "Scenario: default / Mode: async" in report + assert "Scenario: sync-default" in report + assert "Scenario: async-default" in report def test_renderer_colors_every_trace_line_in_a_terminal(monkeypatch: pytest.MonkeyPatch) -> None: - rust: Final = _events(("ocr", 0, None), ("http_request", 1, None)) - python: Final = _events( - ("ocr", 0, "ocr/main.py:88 aocr"), - ("http_request", 1, "http_handler.py:673 AsyncHTTPHandler.post"), - ) + events: Final = _events(("ocr", 0, "ocr/main.py:88 aocr")) monkeypatch.setattr(reporting.sys.stdout, "isatty", lambda: True) monkeypatch.delenv("NO_COLOR", raising=False) - section: Final = render_trace_results((_result(_comparison(python, rust)),))[0] - report: Final = "\n\n".join(section.blocks) + report: Final = "\n\n".join(render_trace_results((_result(_trace(events, events)),))[0].blocks) - assert "\033[36mPYTHON\033[0m (2 steps)" in report + assert "\033[36mPYTHON\033[0m (1 steps)" in report assert "\033[36m1 aocr (ocr/main.py:88)\033[0m" in report - assert "\033[33mRUST\033[0m (2 steps)" in report - assert "\033[33mocr\033[0m -> \033[36m1 aocr\033[0m" in report - assert "\033[33mhttp_request\033[0m -> \033[36m2 AsyncHTTPHandler.post\033[0m" in report + assert "\033[33mRUST\033[0m (1 steps)" in report + assert "\033[33m1 ocr\033[0m" in report -def test_renderer_groups_cases_and_unavailable_entries_by_surface() -> None: - events: Final = _events(("ocr", 0, None)) - gateway_results: Final = tuple( - CaseResult( - case=HarnessCase( - strategy_id="trace_parity", - strategy_label="Trace parity", - sdk_function=sdk_function, - spec=NotImplementedCaseSpec(reason=f"No {sdk_function} case is registered."), - surface="gateway", - ), - status=RunStatus.NOT_IMPLEMENTED, - ) - for sdk_function in ("ocr", "messages") +def test_renderer_groups_unavailable_entries_by_surface() -> None: + gateway_result: Final = CaseResult( + case=HarnessCase( + strategy_id="trace_parity", + strategy_label="Trace parity", + sdk_function="messages", + spec=NotImplementedCaseSpec(reason="No messages case is registered."), + surface="gateway", + ), + status=RunStatus.NOT_IMPLEMENTED, ) - sections: Final = render_trace_results((_result(_comparison(events, events)), *gateway_results)) + sections: Final = render_trace_results((_result(_trace((), ())), gateway_result)) - assert tuple(section.title for section in sections) == ("SDK trace comparisons", "GATEWAY trace comparisons") - gateway_report: Final = "\n\n".join(sections[1].blocks) - assert gateway_report.count("Not implemented") == 1 - assert "- ocr: No ocr case is registered." in gateway_report - assert "- messages: No messages case is registered." in gateway_report + assert tuple(section.title for section in sections) == ("SDK traces", "GATEWAY traces") + assert "- messages: No messages case is registered." in "\n\n".join(sections[1].blocks) diff --git a/tests/rust-python-harness/strategies/trace_parity/test_runner.py b/tests/rust-python-harness/strategies/trace_parity/test_runner.py index 0fcc5860ff2..4992854e21d 100644 --- a/tests/rust-python-harness/strategies/trace_parity/test_runner.py +++ b/tests/rust-python-harness/strategies/trace_parity/test_runner.py @@ -1,12 +1,20 @@ from __future__ import annotations -from typing import Final +import importlib +from pathlib import Path +from typing import Final, cast + +import pytest + +import litellm from ...shared.reporting.models import Coverage, HarnessCase, HarnessRun, RunStatus, SdkFunction, Surface from ...shared.reporting.strategy import ModuleCaseSpec -from ...shared.tracing.steps import Engine +from ...shared.tracing.steps import Engine, PipelineStep from .models import GatewayRouteSpec, RouteFixture, RouteSpec, TraceScenario, TraceSuite -from .runner import run_trace_mode, scenario_nodeids, validate_trace_suite +from .reporting import TraceArtifact +from .runner import run_trace_cases, run_trace_scenario, runner_selection, scenario_nodeids, validate_trace_suite +from .sdk.execution import execute_trace def _fixture(_engine: Engine, _base_url: str) -> RouteFixture: @@ -27,46 +35,98 @@ def test_scenario_filtering_and_occurrence_node_ids() -> None: suite: Final = TraceSuite( route=RouteSpec("ocr", ("ocr", "aocr"), ("ocr", "aocr"), _fixture), scenarios=( - TraceScenario("one", _fixture, (), modes=("sync", "async")), - TraceScenario("two", _fixture, (), modes=("async",)), + TraceScenario("sync-one", _fixture, (), asynchronous=False), + TraceScenario("async-one", _fixture, (), asynchronous=True), + TraceScenario("async-two", _fixture, (), asynchronous=True), ), ) case: Final = _case() - nodes: Final = scenario_nodeids(suite, case, frozenset({"two"})) + nodes: Final = scenario_nodeids(suite, case, frozenset({"async-two"})) - assert tuple(nodeid for _, _, nodeid in nodes) == ("trace:sdk:ocr:two:async",) + assert tuple(nodeid for _, nodeid in nodes) == ("trace:sdk:ocr:async-two",) + + +def test_python_engine_is_separate_from_scenario_selection() -> None: + assert runner_selection(("mistral", "--engine=python")) == (frozenset({"mistral"}), "python") + + +def test_python_engine_skips_native_bridge(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + runner: Final = importlib.import_module("tests.rust-python-harness.strategies.trace_parity.runner") + case: Final = _case() + selected: list[tuple[frozenset[str], str]] = [] + + def reject_bridge(_repo_root: Path) -> str | None: + raise AssertionError("Python-only tracing must not inspect or build the native bridge") + + def capture_case( + _run: HarnessRun, + _case: HarnessCase, + scenarios: frozenset[str], + _on_update: object, + engine: str, + ) -> None: + selected.append((scenarios, engine)) + + monkeypatch.setattr(runner, "ensure_trace_bridge", reject_bridge) + monkeypatch.setattr(runner, "_run_case", capture_case) + + exit_code, _ = run_trace_cases((case,), tmp_path, lambda _: None, ("mistral", "--engine=python")) + + assert exit_code == 0 + assert selected == [(frozenset({"mistral"}), "python")] + + +def test_expected_provider_failure_omits_feedback_banner( + capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + loaded: Final = importlib.import_module("tests.rust-python-harness.strategies.trace_parity.sdk.responses.case") + suite: Final = cast(TraceSuite, loaded.TRACE_SUITE) + scenario: Final = next(item for item in suite.scenarios if item.name == "async-openai-provider-error") + monkeypatch.setattr(litellm, "suppress_debug_info", False) + assert isinstance(suite.route, RouteSpec) + + result: Final = execute_trace(suite.route, scenario, "sdk", engine="python") + + assert result.python_error is None + assert "Give Feedback / Get Help" not in capsys.readouterr().out + assert litellm.suppress_debug_info is False def test_scenario_validation_rejects_duplicate_and_unsafe_names() -> None: route: Final = RouteSpec("ocr", ("ocr", "aocr"), ("ocr", "aocr"), _fixture) duplicate: Final = TraceSuite( route=route, - scenarios=(TraceScenario("same", _fixture, ()), TraceScenario("same", _fixture, ())), + scenarios=( + TraceScenario("sync-same", _fixture, (), asynchronous=False), + TraceScenario("sync-same", _fixture, (), asynchronous=False), + ), + ) + unsafe: Final = TraceSuite( + route=route, scenarios=(TraceScenario("sync-bad:name", _fixture, (), asynchronous=False),) ) - unsafe: Final = TraceSuite(route=route, scenarios=(TraceScenario("bad:name", _fixture, ()),)) case: Final = _case() assert validate_trace_suite(duplicate, case) is not None assert validate_trace_suite(unsafe, case) is not None -def test_scenario_validation_rejects_invalid_modes_and_route_registration() -> None: - invalid_modes: Final = TraceSuite( +def test_scenario_validation_rejects_invalid_names_and_route_registration() -> None: + invalid_name: Final = TraceSuite( route=RouteSpec("ocr", ("ocr", "aocr"), ("ocr", "aocr"), _fixture), - scenarios=(TraceScenario("invalid", _fixture, (), modes=("sync", "sync")),), + scenarios=(TraceScenario("bedrock", _fixture, (), asynchronous=True),), ) wrong_function: Final = TraceSuite( route=RouteSpec("messages", ("create", "acreate"), ("messages", "amessages"), _fixture), - scenarios=(TraceScenario("one", _fixture, ()),), + scenarios=(TraceScenario("sync-one", _fixture, (), asynchronous=False),), ) wrong_surface: Final = TraceSuite( route=GatewayRouteSpec("ocr"), - scenarios=(TraceScenario("one", _fixture, ()),), + scenarios=(TraceScenario("sync-one", _fixture, (), asynchronous=False),), ) case: Final = _case() - assert "unique sync/async modes" in (validate_trace_suite(invalid_modes, case) or "") + assert "start with sync- or async-" in (validate_trace_suite(invalid_name, case) or "") assert "does not match case function" in (validate_trace_suite(wrong_function, case) or "") assert "must use RouteSpec" in (validate_trace_suite(wrong_surface, case) or "") @@ -77,11 +137,35 @@ def test_invalid_route_dispatch_records_harness_error() -> None: result: Final = run.results[case.key] suite: Final = TraceSuite( route=GatewayRouteSpec("ocr"), - scenarios=(TraceScenario("one", _fixture, (), modes=("sync",)),), + scenarios=(TraceScenario("sync-one", _fixture, (), asynchronous=False),), ) - nodeid: Final = "trace:sdk:ocr:one:sync" + nodeid: Final = "trace:sdk:ocr:sync-one" - run_trace_mode(run, result, suite, suite.scenarios[0], "sync", "sdk", nodeid, lambda _: None) + run_trace_scenario(run, result, suite, suite.scenarios[0], "sdk", nodeid, lambda _: None) assert result.outcomes[nodeid] is RunStatus.ERROR assert run.failures == [(nodeid, "gateway route cannot run on the sdk surface")] + + +def test_different_python_and_rust_traces_pass(monkeypatch: pytest.MonkeyPatch) -> None: + runner: Final = importlib.import_module("tests.rust-python-harness.strategies.trace_parity.runner") + case: Final = _case() + run: Final = HarnessRun.from_cases((case,)) + result: Final = run.results[case.key] + suite: Final = TraceSuite( + route=RouteSpec("ocr", ("ocr", "aocr"), ("ocr", "aocr"), _fixture), + scenarios=(TraceScenario("sync-one", _fixture, (), asynchronous=False),), + ) + trace: Final = TraceArtifact.from_traces( + surface="sdk", + sdk_function="ocr", + scenario="sync-one", + python=(PipelineStep(0, None, "python_step", "python.py:1 python_step"),), + rust=(PipelineStep(0, None, "rust_step", "rust_step"),), + ) + monkeypatch.setattr(runner, "_execute_scenario", lambda *_args: trace) + + run_trace_scenario(run, result, suite, suite.scenarios[0], "sdk", "trace:sdk:ocr:sync-one", lambda _: None) + + assert result.outcomes["trace:sdk:ocr:sync-one"] is RunStatus.PASSED + assert run.failures == []