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

cleanup dead code

Browse source
This commit is contained in:
Yujong Lee 2026-09-15 20:28:36 -07:00
parent b9fca28c6e
commit 7167bd0ff0
45 changed files with 42 additions and 3789 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
litellm-rust/Cargo.lock generated
View file

@ -1889,31 +1889,6 @@ version = "0.2.186"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
[[package]]
name = "litellm-ai-gateway"
version = "0.1.0"
dependencies = [
"axum",
"base64 0.22.1",
"futures-channel",
"futures-util",
"litellm-auth",
"litellm-config",
"litellm-core",
"reqwest 0.12.28",
"rustls 0.23.42",
"rustls-native-certs",
"serde",
"serde_json",
"sha2 0.10.9",
"subtle",
"thiserror 2.0.19",
"tokio",
"tokio-tungstenite",
"tower",
"tracing",
]
[[package]]
name = "litellm-auth"
version = "0.1.0"
@ -1993,16 +1968,6 @@ dependencies = [
"tokio",
]
[[package]]
name = "litellm-config"
version = "0.1.0"
dependencies = [
"litellm-core",
"pyo3",
"serde_json",
"thiserror 2.0.19",
]
[[package]]
name = "litellm-core"
version = "0.1.0"
@ -2032,6 +1997,7 @@ dependencies = [
"thiserror 2.0.19",
"tokio",
"tokio-tungstenite",
"tracing",
"url",
"veil",
]

1
litellm-rust/crates/auth-gcp/src/lib.rs
View file

@ -105,7 +105,6 @@ impl VertexAuth {
}
}
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
pub async fn validate_environment(
&self,
headers: Vec<(String, String)>,

5
litellm-rust/crates/core/Cargo.toml
View file

@ -29,13 +29,10 @@ subtle.workspace = true
tokio = { workspace = true, features = ["sync"] }
tokio-tungstenite.workspace = true
thiserror.workspace = true
tracing.workspace = true
sha2.workspace = true
url.workspace = true
veil.workspace = true
[features]
default = []
observability = ["dep:tracing-subscriber"]
[dev-dependencies]
rstest.workspace = true

1
litellm-rust/crates/core/src/audio_transcription/transformation.rs
View file

@ -15,7 +15,6 @@ pub enum AudioTranscriptionAuth {
pub trait AudioTranscriptionProviderConfig: Sync {
fn supported_transcription_params(&self) -> &'static [&'static str];
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
fn map_transcription_params(&self, params: &OpaqueParams) -> OpaqueParams {
params.provider_params()
}

1
litellm-rust/crates/core/src/http_utils.rs
View file

@ -51,7 +51,6 @@ pub async fn http_request(
request.send().await
}
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
pub async fn execute_http_request(
client: &reqwest::Client,
request: reqwest::Request,

1
litellm-rust/crates/core/src/llms/azure_ai/ocr/document_intelligence/transformation.rs
View file

@ -212,7 +212,6 @@ fn normalize_features(features: Option<&Value>) -> Result<Option<String>, crate:
Ok(Some(normalized.join(",")))
}
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
fn build_request(document: OcrDocument) -> Result<DocumentIntelligenceRequest, crate::ocr::Error> {
let source = document.source();
if source.is_empty() {

1
litellm-rust/crates/core/src/llms/mistral/ocr/transformation.rs
View file

@ -64,7 +64,6 @@ impl BaseOcrConfig for MistralOCRConfig {
self.get_complete_url(request.connection.api_base.as_deref())
}
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
fn transform_ocr_request(
&self,
model: &str,

12
litellm-rust/crates/core/src/llms/reducto/ocr/transformation.rs
View file

@ -127,12 +127,6 @@ impl BaseOcrConfig for ReductoParseV3Config {
.into())
}
#[tracing::instrument(
name = "async_transform_ocr_request",
target = "litellm::function_trace",
level = "trace",
skip_all
)]
async fn async_transform_ocr_request(
&self,
_model: &str,
@ -222,12 +216,6 @@ impl BaseOcrConfig for ReductoParseLegacyConfig {
.into())
}
#[tracing::instrument(
name = "async_transform_ocr_request",
target = "litellm::function_trace",
level = "trace",
skip_all
)]
async fn async_transform_ocr_request(
&self,
_model: &str,

6
litellm-rust/crates/core/src/ocr/client.rs
View file

@ -33,12 +33,6 @@ impl OcrClient {
shared_client()
}
#[tracing::instrument(
name = "ocr",
target = "litellm::function_trace",
level = "trace",
skip_all
)]
pub async fn perform(
&self,
request: LiteLLMOcrRequest,

1
litellm-rust/crates/python-bridge/Cargo.toml
View file

@ -17,7 +17,6 @@ panic-test = []
[dependencies]
futures-util.workspace = true
tracing = { workspace = true, optional = true }
litellm-core.workspace = true
litellm-auth.workspace = true
litellm-token-counter.workspace = true

2
litellm-rust/crates/python-bridge/src/lib.rs
View file

@ -122,6 +122,8 @@ mod tests {
.filter(|name| !name.starts_with('_'))
.collect();
assert_eq!(public_names, expected);
assert!(!module.hasattr("_trace").expect("module lookup should work"));
});
}

79
litellm-rust/crates/python-bridge/src/routes/audio_transcription/value.rs
View file

@ -123,82 +123,3 @@ pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
super::super::add_function(module, wrap_pyfunction!(transcription, module)?)?;
super::super::add_function(module, wrap_pyfunction!(atranscription, module)?)
}
#[cfg(feature = "trace-parity")]
mod trace {
use super::{
AudioTranscriptionInputs, Value, core_error_to_pyerr, prepare_transcription, run_async,
run_sync,
};
use pyo3::prelude::*;
#[pyfunction]
#[pyo3(signature = (model, audio, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn transcription(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] audio: Value,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] optional_params: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Py<PyAny>> {
run_sync(
py,
crate::function_trace::capture(prepare_transcription(AudioTranscriptionInputs {
model,
audio,
api_key,
api_base,
custom_llm_provider,
extra_headers,
optional_params,
timeout_seconds,
})?),
core_error_to_pyerr,
)
}
#[pyfunction]
#[pyo3(signature = (model, audio, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn atranscription(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] audio: Value,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] optional_params: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Bound<'_, PyAny>> {
run_async(
py,
crate::function_trace::capture(prepare_transcription(AudioTranscriptionInputs {
model,
audio,
api_key,
api_base,
custom_llm_provider,
extra_headers,
optional_params,
timeout_seconds,
})?),
core_error_to_pyerr,
)
}
pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
super::super::super::add_function(module, wrap_pyfunction!(transcription, module)?)?;
super::super::super::add_function(module, wrap_pyfunction!(atranscription, module)?)
}
}
#[cfg(feature = "trace-parity")]
pub(super) fn register_trace(module: &Bound<'_, PyModule>) -> PyResult<()> {
trace::register(module)
}

79
litellm-rust/crates/python-bridge/src/routes/chat_completions/value.rs
View file

@ -143,82 +143,3 @@ pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
super::super::add_function(module, wrap_pyfunction!(chat_completions, module)?)?;
super::super::add_function(module, wrap_pyfunction!(achat_completions, module)?)
}
#[cfg(feature = "trace-parity")]
mod trace {
use super::{
ChatCompletionsInputs, Value, chat_completions_error_to_pyerr, prepare_chat_completions,
run_async, run_sync,
};
use pyo3::prelude::*;
#[pyfunction]
#[pyo3(signature = (model, messages, optional_params=None, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn chat_completions(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] messages: Value,
#[pyo3(from_py_with = litellm_python_interop::from_py)] optional_params: Option<Value>,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Py<PyAny>> {
run_sync(
py,
crate::function_trace::capture(prepare_chat_completions(ChatCompletionsInputs {
model,
messages,
optional_params,
api_key,
api_base,
custom_llm_provider,
extra_headers,
timeout_seconds,
})?),
chat_completions_error_to_pyerr,
)
}
#[pyfunction]
#[pyo3(signature = (model, messages, optional_params=None, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn achat_completions(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] messages: Value,
#[pyo3(from_py_with = litellm_python_interop::from_py)] optional_params: Option<Value>,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Bound<'_, PyAny>> {
run_async(
py,
crate::function_trace::capture(prepare_chat_completions(ChatCompletionsInputs {
model,
messages,
optional_params,
api_key,
api_base,
custom_llm_provider,
extra_headers,
timeout_seconds,
})?),
chat_completions_error_to_pyerr,
)
}
pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
super::super::super::add_function(module, wrap_pyfunction!(chat_completions, module)?)?;
super::super::super::add_function(module, wrap_pyfunction!(achat_completions, module)?)
}
}
#[cfg(feature = "trace-parity")]
pub(super) fn register_trace(module: &Bound<'_, PyModule>) -> PyResult<()> {
trace::register(module)
}

74
litellm-rust/crates/python-bridge/src/routes/messages/value.rs
View file

@ -114,77 +114,3 @@ pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
super::super::add_function(module, wrap_pyfunction!(messages, module)?)?;
super::super::add_function(module, wrap_pyfunction!(amessages, module)?)
}
#[cfg(feature = "trace-parity")]
mod trace {
use super::{
MessagesInputs, Value, core_error_to_pyerr, prepare_messages, run_async, run_sync,
};
use pyo3::prelude::*;
#[pyfunction]
#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn messages(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] body: Value,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Py<PyAny>> {
run_sync(
py,
crate::function_trace::capture(prepare_messages(MessagesInputs {
model,
body,
api_key,
api_base,
custom_llm_provider,
extra_headers,
timeout_seconds,
})?),
core_error_to_pyerr,
)
}
#[pyfunction]
#[pyo3(signature = (model, body, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn amessages(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] body: Value,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Bound<'_, PyAny>> {
run_async(
py,
crate::function_trace::capture(prepare_messages(MessagesInputs {
model,
body,
api_key,
api_base,
custom_llm_provider,
extra_headers,
timeout_seconds,
})?),
core_error_to_pyerr,
)
}
pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
super::super::super::add_function(module, wrap_pyfunction!(messages, module)?)?;
super::super::super::add_function(module, wrap_pyfunction!(amessages, module)?)
}
}
#[cfg(feature = "trace-parity")]
pub(super) fn register_trace(module: &Bound<'_, PyModule>) -> PyResult<()> {
trace::register(module)
}

7
litellm-rust/crates/python-bridge/src/routes/ocr/mod.rs
View file

@ -4,16 +4,9 @@ mod errors;
mod lifecycle;
mod project;
mod request;
#[cfg(feature = "trace-parity")]
mod trace;
use pyo3::prelude::*;
pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
lifecycle::register(module)
}
#[cfg(feature = "trace-parity")]
pub(super) fn register_trace(module: &Bound<'_, PyModule>) -> PyResult<()> {
trace::register(module)
}

138
litellm-rust/crates/python-bridge/src/routes/ocr/trace.rs
View file

@ -1,138 +0,0 @@
use litellm_core::ocr::Error;
use std::future::Future;
use litellm_core::ocr::LiteLLMOcrRequest;
use pyo3::prelude::*;
use serde_json::Value;
use super::errors::to_pyerr as ocr_error_to_pyerr;
use super::request::BridgeOcrRequest;
use crate::execution::{run_async, run_sync};
use crate::marshal::{RouteOptions, RouteOptionsInputs, object_or_empty};
struct OcrInputs {
model: String,
document: Value,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
extra_headers: Option<Value>,
optional_params: Option<Value>,
input_sources: Option<Value>,
timeout_seconds: Option<f64>,
}
fn prepare_ocr(
inputs: OcrInputs,
) -> PyResult<impl Future<Output = Result<Value, Error>> + Send + 'static> {
let document = inputs.document;
let options = RouteOptions::from_python(RouteOptionsInputs {
model: inputs.model,
api_key: inputs.api_key,
api_base: inputs.api_base,
custom_llm_provider: inputs.custom_llm_provider,
extra_headers: inputs.extra_headers,
timeout_seconds: inputs.timeout_seconds,
})?;
let optional_params = object_or_empty("optional_params", inputs.optional_params)?;
let input_sources = inputs
.input_sources
.map(serde_json::from_value)
.transpose()
.map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))?
.unwrap_or_default();
Ok(async move {
let RouteOptions {
model,
api_key,
api_base,
custom_llm_provider,
extra_headers,
timeout,
} = options;
let request = LiteLLMOcrRequest::try_from(BridgeOcrRequest {
model,
document,
api_key,
api_base,
custom_llm_provider,
extra_headers,
optional_params: optional_params.into(),
input_sources,
timeout_seconds: timeout.map(|value| value.as_secs_f64()),
})?;
litellm_core::ocr::ocr(request)
.await
.map(|response| response.into_json())
})
}
#[pyfunction]
#[pyo3(signature = (model, document, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, input_sources=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn ocr(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] document: Value,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] optional_params: Option<Value>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] input_sources: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Py<PyAny>> {
run_sync(
py,
crate::function_trace::capture(prepare_ocr(OcrInputs {
model,
document,
api_key,
api_base,
custom_llm_provider,
extra_headers,
optional_params,
input_sources,
timeout_seconds,
})?),
ocr_error_to_pyerr,
)
}
#[pyfunction]
#[pyo3(signature = (model, document, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, input_sources=None, timeout_seconds=None))]
#[allow(clippy::too_many_arguments)]
fn aocr(
py: Python<'_>,
model: String,
#[pyo3(from_py_with = litellm_python_interop::from_py)] document: Value,
api_key: Option<String>,
api_base: Option<String>,
custom_llm_provider: Option<String>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] extra_headers: Option<Value>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] optional_params: Option<Value>,
#[pyo3(from_py_with = litellm_python_interop::from_py)] input_sources: Option<Value>,
timeout_seconds: Option<f64>,
) -> PyResult<Bound<'_, PyAny>> {
run_async(
py,
crate::function_trace::capture(prepare_ocr(OcrInputs {
model,
document,
api_key,
api_base,
custom_llm_provider,
extra_headers,
optional_params,
input_sources,
timeout_seconds,
})?),
ocr_error_to_pyerr,
)
}
pub(super) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
super::super::add_function(module, wrap_pyfunction!(ocr, module)?)?;
super::super::add_function(module, wrap_pyfunction!(aocr, module)?)
}

27
litellm-rust/crates/python-bridge/src/routes/tests.rs
View file

@ -158,9 +158,7 @@ fn route_input_validation_preserves_left_to_right_order() {
PyModule::new(py, "invalid_payload").expect("invalid payload should be created");
let error = module
.getattr("transcription")
.and_then(|function| {
function.call(("model", &invalid_payload), Some(&headers_kwargs))
})
.and_then(|function| function.call(("model", &invalid_payload), Some(&headers_kwargs)))
.expect_err("payload should be validated before headers");
assert!(!error.to_string().contains("extra_headers"));
});
@ -241,26 +239,3 @@ fn chat_completions_decline_keeps_existing_reasons() {
);
});
}
#[cfg(feature = "trace-parity")]
#[test]
fn trace_routes_preserve_the_direct_route_signatures() {
Python::initialize();
Python::attach(|py| {
let parent = PyModule::new(py, "routes").expect("module should be created");
let module = PyModule::new(py, "_trace").expect("trace module should be created");
ocr::register_trace(&module).expect("trace OCR route should register");
parent
.add_submodule(&module)
.expect("trace module should be attached");
let signature: String = module
.getattr("ocr")
.and_then(|function| function.getattr("__text_signature__"))
.and_then(|signature| signature.extract())
.expect("trace signature should be available");
assert_eq!(
signature,
"(model, document, api_key=None, api_base=None, custom_llm_provider=None, extra_headers=None, optional_params=None, input_sources=None, timeout_seconds=None)"
);
});
}

23
tests/rust-python-harness/AGENTS.md
View file

@ -15,16 +15,6 @@ tests/rust-python-harness/
│ │ ├── sdk/
│ │ │ └── ocr/
│ │
│ ├── trace_parity/
│ │ ├── __init__.py
│ │ ├── models.py
│ │ ├── reporting.py
│ │ └── sdk/
│ │ ├── chat_completions/
│ │ ├── messages/
│ │ ├── ocr/
│ │ └── transcription/
│ │
│ ├── unit_tests_parity/
│ │ ├── __init__.py
│ │ ├── reporting.py
@ -37,7 +27,6 @@ tests/rust-python-harness/
│
└── shared/
├── parity/
├── tracing/
├── reporting/
│ └── strategy.py
└── unit_runners/
@ -53,14 +42,12 @@ 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/` profiles the Python call stack and prints every collected Python call under `litellm/`; it never collects Rust spans and never rebuilds the native extension
- E2E and trace strategies load their registered module cases and run surface-specific execution from their folders
- `shared/unit_runners/contracts.py` owns the typed per-function unit contracts consumed by `unit_tests_parity` and `unit_tests_rust`
- `unit_tests_parity/runner.py` runs each contract's `unit_parity_scope` with `LITELLM_RUST=0` and `LITELLM_RUST=1` in separate processes and requires matching outcomes, including failures; exclusions require a reason in the contract
- `unit_tests_rust/runner.py` runs each contract's focused Cargo test suite; native Rust unit tests stay beside their implementation
- E2E strategies load their registered module cases and run surface-specific execution from their folders
- `unit_tests_parity/runner.py` runs each suite with `LITELLM_RUST=0` and `LITELLM_RUST=1` in separate processes and requires matching outcomes, including failures; exclusions require a reason
- `unit_tests_rust/runner.py` runs each focused Cargo test suite; native Rust unit tests stay beside their implementation
- `shared/unit_runners/suite_runner.py` runs typed suites registered in code with nodeids of the form `suite:<strategy_id>:<function>:<suite>`
- Every strategy declares its report sections and presentation in its own `reporting.py`; shared reporting code only provides reusable models and cell-formatting primitives
- `shared/` contains reusable parity, tracing, reporting primitives, and unit-runner machinery
- `shared/` contains reusable parity, reporting, and unit-runner machinery
- Keep fixtures with their owning API and existing Python tests in their current locations
- Each strategy folder carries an `AGENTS.md` one-liner stating what it should be doing
- Run the harness's own checks with `uv run pytest -o consider_namespace_packages=true tests/rust-python-harness/shared tests/rust-python-harness/cli tests/rust-python-harness/strategies/trace_parity tests/rust-python-harness/strategies/unit_tests_parity tests/rust-python-harness/strategies/unit_tests_rust tests/test_rust_python_harness.py -q`
- Run the harness's own checks with `uv run pytest -o consider_namespace_packages=true tests/rust-python-harness/shared tests/rust-python-harness/cli tests/rust-python-harness/strategies/unit_tests_parity tests/rust-python-harness/strategies/unit_tests_rust tests/test_rust_python_harness.py -q`

61
tests/rust-python-harness/cli/test_cli.py
View file

@ -59,7 +59,7 @@ def _strategy_source(
"from pathlib import Path\n"
"strategy = importlib.import_module('tests.rust-python-harness.shared.reporting.strategy')\n"
"models = importlib.import_module('tests.rust-python-harness.shared.reporting.models')\n"
"runner = importlib.import_module('tests.rust-python-harness.strategies.trace_parity.runner')\n"
"runner = importlib.import_module('tests.rust-python-harness.strategies.e2e_parity.runner')\n"
"rendering = importlib.import_module('tests.rust-python-harness.shared.reporting.rendering')\n"
"def render(results):\n"
" return (rendering.ReportSection('Example outcomes', "
@ -68,7 +68,7 @@ def _strategy_source(
"STRATEGY = strategy.StrategyDefinition("
f"id={strategy_id!r}, order=1, label='Example strategy', description='Example description', "
"directory=Path(__file__).parent, runnable_spec=strategy.SuiteCaseSpec, cases=CASES, "
f"run=runner.run_trace_cases, render=render, surfaces={surfaces!r})\n"
f"run=runner.run_e2e_cases, render=render, surfaces={surfaces!r})\n"
)
@ -89,12 +89,11 @@ def test_should_load_surface_aware_and_function_only_strategies() -> None:
assert [strategy.id for strategy in strategies] == [
"e2e_parity",
"trace_parity",
"unit_tests_parity",
"unit_tests_rust",
]
for strategy in strategies:
expected: Final = tuple(
expected = tuple(
(surface, function) for surface in (strategy.definition.surfaces or (None,)) for function in SDK_FUNCTIONS
)
assert tuple((case.surface, case.sdk_function) for case in strategy.cases) == expected
@ -102,19 +101,21 @@ def test_should_load_surface_aware_and_function_only_strategies() -> None:
def test_unit_strategies_use_function_only_cases() -> None:
strategies: Final = {
strategy.id: strategy for strategy in load_catalog() if strategy.id in {"unit_tests_parity", "unit_tests_rust"}
strategy.id: strategy
for strategy in load_catalog()
if strategy.id in {"unit_tests_parity", "unit_tests_rust"}
}
for sdk_function in SDK_FUNCTIONS:
cases: Final = tuple(
cases = tuple(
case for strategy in strategies.values() for case in strategy.cases if case.sdk_function == sdk_function
)
assert len(cases) == 2
assert all(case.surface is None for case in cases)
expected_parity: Final = (
expected_parity = (
CaseDisposition.RUNNABLE if sdk_function in UNIT_PARITY_SUITES else CaseDisposition.NOT_IMPLEMENTED
)
expected_rust: Final = (
expected_rust = (
CaseDisposition.RUNNABLE if sdk_function in RUST_SUITES else CaseDisposition.NOT_IMPLEMENTED
)
assert cases[0].spec.disposition is expected_parity
@ -133,7 +134,7 @@ def test_every_strategy_folder_complies() -> None:
assert folders == {strategy.id for strategy in strategies}
for strategy in strategies:
definition: Final = strategy.definition
definition = strategy.definition
assert isinstance(definition, StrategyDefinition)
assert definition.directory == strategy.directory
assert not (strategy.directory / "strategy.json").exists()
@ -234,7 +235,6 @@ 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": "traces",
"unit_tests_parity": "Python backend parity outcomes",
"unit_tests_rust": "Native Rust unit-test outcomes",
}
@ -253,7 +253,6 @@ def test_every_unavailable_case_finishes_and_explains_itself() -> None:
("strategy_id", "present", "absent"),
(
("e2e_parity", "--surface", "--pytest-arg"),
("trace_parity", "--surface", "--pytest-arg"),
("unit_tests_parity", "--pytest-arg", "--surface"),
("unit_tests_rust", "--function", "--surface"),
),
@ -280,7 +279,6 @@ def test_run_help_lists_all_and_every_strategy(capsys: pytest.CaptureFixture[str
for command in (
"all",
"e2e_parity",
"trace_parity",
"unit_tests_parity",
"unit_tests_rust",
):
@ -348,25 +346,6 @@ def test_strategy_command_forwards_repeated_filters_and_runner_arguments(
]
def test_trace_command_forwards_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"]) == 0
assert captured == [("async-mistral",)]
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] = []
@ -402,23 +381,9 @@ def test_run_all_selects_every_declared_case_once(monkeypatch: pytest.MonkeyPatc
monkeypatch.setattr(cli, "run_command", capture_run)
assert main(["run", "all", "--function", "ocr"]) == 0
assert len(selected) == 6
assert len(selected) == 4
assert sum(case.surface is None for case in selected) == 2
assert sum(case.surface is not None for case in selected) == 4
def test_run_reports_not_implemented_surface_as_not_run(
capsys: pytest.CaptureFixture[str],
) -> None:
exit_code: Final = main(["run", "trace_parity", "--surface", "gateway", "--function", "ocr"])
captured: Final = capsys.readouterr()
assert exit_code == 0
assert "Result: NOT RUN" in captured.out
assert "Harness support: 0/1 cases implemented" in captured.out
assert "Cases: 1 selected, 1 not implemented, 0 skipped" in captured.out
assert "Not implemented" in captured.out
assert "No gateway OCR trace-parity case is registered." in captured.out
assert sum(case.surface is not None for case in selected) == 2
def test_keyboard_interrupt_exits_cleanly(
@ -458,7 +423,7 @@ def test_runner_interrupt_skips_the_completion_report(
monkeypatch.setattr(commands, "run_strategies", interrupt_run)
exit_code: Final = main(["run", "trace_parity", "--surface", "gateway"])
exit_code: Final = main(["run", "e2e_parity", "--surface", "gateway"])
captured: Final = capsys.readouterr()
assert exit_code == 130

0
tests/rust-python-harness/shared/tracing/__init__.py
View file

197
tests/rust-python-harness/shared/tracing/profiler.py
View file

@ -1,197 +0,0 @@
from __future__ import annotations
import sys
import threading
import warnings
from collections.abc import Callable, Generator, Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from types import CodeType, FrameType, FunctionType, MappingProxyType
from typing import Final
@dataclass(frozen=True, slots=True)
class FunctionTraceEvent:
id: int
parent_id: int | None
function: str
module_path: str | None = None
file: str | None = None
line: int | None = None
@property
def raw(self) -> str:
location: Final = f"{self.file}:{self.line}" if self.file is not None and self.line is not None else ""
qualified: Final = f"{self.module_path}::{self.function}" if self.module_path is not None else self.function
return f"{location} {qualified}" if location else qualified
class PythonProfiler:
def __init__(self, source_root: Path) -> None:
self._source_root: Final = str(source_root.resolve()) + "/"
self._seen_frames: Final[set[FrameType]] = set()
self._event_ids: Final[dict[FrameType, int]] = {}
self._lock: Final = threading.Lock()
self.events: Final[list[FunctionTraceEvent]] = []
def __call__(self, frame: FrameType, event: str, _arg: object) -> None:
if event != "call" or frame in self._seen_frames:
return
function_name: Final = self.function_name(frame)
if function_name is None:
return
with self._lock:
event_id: Final = len(self.events)
parent_id: Final = next(
(self._event_ids[ancestor] for ancestor in _frame_ancestors(frame) if ancestor in self._event_ids),
None,
)
self._seen_frames.add(frame)
self._event_ids[frame] = event_id
self.events.append(FunctionTraceEvent(id=event_id, parent_id=parent_id, function=function_name))
def function_name(self, frame: FrameType) -> str | None:
code: Final = frame.f_code
if not code.co_filename.startswith(self._source_root):
return None
relative: Final = code.co_filename.removeprefix(self._source_root)
return f"{relative}:{code.co_firstlineno} {_qualified_name(frame)}"
class PythonFunctionUsageProfiler:
def __init__(self, source_root: Path, functions: frozenset[str]) -> None:
self._source_root: Final = str(source_root.resolve()) + "/"
self._functions: Final = functions
self.called: Final[set[str]] = set()
def __call__(self, frame: FrameType, event: str, _arg: object) -> None:
if event != "call":
return
code: Final = frame.f_code
if not code.co_filename.startswith(self._source_root):
return
relative: Final = code.co_filename.removeprefix(self._source_root)
function: Final = f"{relative}:{code.co_firstlineno} {_qualified_name(frame)}"
if function in self._functions:
self.called.add(function)
def _qualified_name(frame: FrameType) -> str:
code: Final = frame.f_code
native: Final = getattr(code, "co_qualname", None)
if isinstance(native, str):
return native
enclosing: Final = next(
(
name
for ancestor in _frame_ancestors(frame)
for declared_code, name in _declared_functions(ancestor.f_locals, frozenset())
if declared_code is code
),
None,
)
if enclosing is not None:
return enclosing
module_name: Final = frame.f_globals.get("__name__")
if not isinstance(module_name, str):
return code.co_name
return _module_qualnames(module_name).get(code, code.co_name)
@lru_cache(maxsize=None)
def _module_qualnames(module_name: str) -> Mapping[CodeType, str]:
module: Final = sys.modules.get(module_name)
if module is None:
return MappingProxyType({})
return MappingProxyType(dict(_declared_functions(vars(module), frozenset())))
def _declared_functions(namespace: Mapping[str, object], visited: frozenset[int]) -> Iterator[tuple[CodeType, str]]:
for attribute in tuple(namespace.values()):
for value in _accessors(attribute):
if isinstance(value, FunctionType):
yield from ((wrapped.__code__, wrapped.__qualname__) for wrapped in _unwrapped(value))
elif isinstance(value, type) and id(value) not in visited:
yield from _declared_functions(dict(vars(value)), visited | {id(value)})
def _unwrapped(function: FunctionType) -> Iterator[FunctionType]:
yield function
inner: Final = getattr(function, "__wrapped__", None)
if isinstance(inner, FunctionType):
yield from _unwrapped(inner)
def _accessors(value: object) -> tuple[object, ...]:
if isinstance(value, (staticmethod, classmethod)):
return (value.__func__,)
if isinstance(value, property):
return tuple(accessor for accessor in (value.fget, value.fset, value.fdel) if accessor is not None)
return (value,)
def _frame_ancestors(frame: FrameType) -> Generator[FrameType]:
ancestor: Final = frame.f_back
if ancestor is not None:
yield ancestor
yield from _frame_ancestors(ancestor)
@contextmanager
def _installed_profiler(profiler: Callable[[FrameType, str, object], None], *, threads: bool) -> Generator[None]:
if threads and sys.version_info >= (3, 12):
tool_id: Final = next((slot for slot in (2, 3, 4, 0, 1, 5) if sys.monitoring.get_tool(slot) is None), None)
if tool_id is None:
raise RuntimeError("no sys.monitoring tool ID is available for Python trace collection")
def started(_code: CodeType, _offset: int) -> None:
profiler(sys._getframe(1), "call", None)
sys.monitoring.use_tool_id(tool_id, "litellm-python-trace")
try:
sys.monitoring.register_callback(tool_id, sys.monitoring.events.PY_START, started)
sys.monitoring.set_events(tool_id, sys.monitoring.events.PY_START)
yield
finally:
sys.monitoring.set_events(tool_id, 0)
sys.monitoring.register_callback(tool_id, sys.monitoring.events.PY_START, None)
sys.monitoring.free_tool_id(tool_id)
return
if threads:
warnings.warn(
"Python <3.12 cannot trace existing worker threads; use Python 3.12+ for complete threaded traces",
RuntimeWarning,
stacklevel=3,
)
previous_thread: Final = threading.getprofile()
if threads:
threading.setprofile(profiler)
previous: Final = sys.getprofile()
sys.setprofile(profiler)
try:
yield
finally:
sys.setprofile(previous)
if threads:
threading.setprofile(previous_thread)
@contextmanager
def profile_python(source_root: Path, *, threads: bool = False) -> Generator[PythonProfiler]:
profiler: Final = PythonProfiler(source_root)
with _installed_profiler(profiler, threads=threads):
yield profiler
@contextmanager
def profile_python_function_usage(
source_root: Path,
functions: frozenset[str],
*,
threads: bool = False,
) -> Generator[PythonFunctionUsageProfiler]:
profiler: Final = PythonFunctionUsageProfiler(source_root, functions)
with _installed_profiler(profiler, threads=threads):
yield profiler

339
tests/rust-python-harness/shared/tracing/pytest_usage.py
View file

@ -1,339 +0,0 @@
from __future__ import annotations
import argparse
import ast
import importlib
import inspect
import os
import subprocess
import sys
import tempfile
import warnings
from collections.abc import Generator, Sequence
from pathlib import Path
from types import CodeType
from typing import TYPE_CHECKING, Final
from pluggy import HookimplMarker
from pydantic import BaseModel, ConfigDict
from .profiler import profile_python_function_usage
if TYPE_CHECKING:
import pytest
hookimpl: Final = HookimplMarker("pytest")
class PythonFunctionIdentity(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
file: str
line: int
qualname: str
@property
def raw(self) -> str:
return f"{self.file}:{self.line} {self.qualname}"
@property
def key(self) -> str:
return f"{self.file}::{self.qualname}"
@classmethod
def from_trace(cls, raw: str) -> PythonFunctionIdentity:
location, separator, qualname = raw.partition(" ")
file, line_separator, line = location.rpartition(":")
if not separator or not line_separator or not file or not line.isdigit() or not qualname:
raise ValueError(f"Unrecognized Python trace function: {raw}")
return cls(file=file, line=int(line), qualname=qualname)
class PythonFunctionReference(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
module: str
qualname: str
@property
def owner(self) -> str:
return self.qualname.partition(".")[0]
def resolve(self, source_root: Path) -> PythonFunctionIdentity:
value: object = importlib.import_module(self.module)
for component in self.qualname.split("."):
value = getattr(value, component)
if not callable(value):
raise ValueError(f"Python function is not callable: {self.module}:{self.qualname}")
function: Final = inspect.unwrap(value)
code: Final = getattr(function, "__code__", None)
qualname: Final = getattr(function, "__qualname__", None)
if not isinstance(code, CodeType) or not isinstance(qualname, str):
raise ValueError(f"Python function has no code object: {self.module}:{self.qualname}")
source: Final = Path(code.co_filename).resolve()
try:
relative: Final = source.relative_to(source_root.resolve())
except ValueError as error:
raise ValueError(f"Python function is outside {source_root}: {source}") from error
return PythonFunctionIdentity(
file=relative.as_posix(),
line=code.co_firstlineno,
qualname=qualname,
)
class RustFunctionIdentity(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
file: str
line: int
module_path: str
function: str
@property
def test_module(self) -> str:
_, separator, module = self.module_path.partition("::")
if not separator:
raise ValueError(f"Rust function has no crate-qualified module: {self.module_path}")
return f"{module}::tests"
@classmethod
def from_trace(cls, raw: str) -> RustFunctionIdentity:
location, separator, qualified = raw.partition(" ")
file, line_separator, line = location.rpartition(":")
module_path, function_separator, function = qualified.rpartition("::")
if (
not separator
or not line_separator
or not function_separator
or not file
or not line.isdigit()
or not module_path
or not function
):
raise ValueError(f"Unrecognized Rust trace function: {raw}")
return cls(file=file, line=int(line), module_path=module_path, function=function)
class PythonFunctionUsage(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
function: PythonFunctionIdentity
tests: tuple[str, ...]
class PythonUsageReport(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
usages: tuple[PythonFunctionUsage, ...]
collected_tests: tuple[str, ...]
exit_code: int
problems: tuple[str, ...] = ()
def candidate_test_files(
functions: Sequence[PythonFunctionReference | PythonFunctionIdentity],
search_roots: Sequence[str],
repo_root: Path,
*,
exclude_roots: Sequence[str] = (),
) -> tuple[str, ...]:
owners: Final = frozenset(
function.owner if isinstance(function, PythonFunctionReference) else function.qualname.partition(".")[0]
for function in functions
if "." in function.qualname
and (
isinstance(function, PythonFunctionReference)
or function.file.startswith("ocr/")
or "/ocr/" in function.file
)
)
top_level_functions: Final = frozenset(
function.qualname for function in functions if "." not in function.qualname and function.qualname.isidentifier()
)
candidates: Final = tuple(
path.relative_to(repo_root).as_posix()
for root in search_roots
for path in sorted((repo_root / root).rglob("test*.py"))
if not any(
path == repo_root / excluded or path.is_relative_to(repo_root / excluded) for excluded in exclude_roots
)
if _references_function(path, owners, top_level_functions)
)
return tuple(dict.fromkeys(candidates))
def _references_function(path: Path, owners: frozenset[str], top_level_functions: frozenset[str]) -> bool:
contents: Final = path.read_text(errors="ignore")
if any(owner in contents for owner in owners):
return True
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore", SyntaxWarning)
tree: Final = ast.parse(contents)
except SyntaxError:
return False
aliases: Final = frozenset(
alias.asname or alias.name
for node in ast.walk(tree)
if isinstance(node, ast.ImportFrom)
for alias in node.names
if alias.name in top_level_functions
)
names: Final = top_level_functions | aliases
return any(
isinstance(node, ast.Call)
and (
(isinstance(node.func, ast.Name) and node.func.id in names)
or (isinstance(node.func, ast.Attribute) and node.func.attr in top_level_functions)
)
for node in ast.walk(tree)
)
class _WorkerConfig(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
functions: tuple[PythonFunctionIdentity, ...]
source_root: Path
output: Path
pytest_args: tuple[str, ...]
class _FunctionUsagePlugin:
def __init__(self, functions: tuple[PythonFunctionIdentity, ...], source_root: Path) -> None:
self._functions: Final = functions
self._function_names: Final = frozenset(function.raw for function in functions)
self._source_root: Final = source_root
self._tests_by_function: Final[dict[str, set[str]]] = {function.raw: set() for function in functions}
self.collected_tests: tuple[str, ...] = ()
self.problems: tuple[str, ...] = ()
def pytest_collection_finish(self, session: pytest.Session) -> None:
self.collected_tests = tuple(item.nodeid for item in session.items)
def pytest_collectreport(self, report: pytest.CollectReport) -> None:
if report.failed:
self.problems = (*self.problems, str(report.longrepr))
@hookimpl(hookwrapper=True)
def pytest_runtest_protocol(self, item: pytest.Item, nextitem: pytest.Item | None) -> Generator[None, object, None]:
del nextitem
with profile_python_function_usage(self._source_root, self._function_names, threads=True) as profiler:
yield
for function in self._functions:
if function.raw in profiler.called:
self._tests_by_function[function.raw].add(item.nodeid)
def usages(self) -> tuple[PythonFunctionUsage, ...]:
return tuple(
PythonFunctionUsage(
function=function,
tests=tuple(sorted(self._tests_by_function[function.raw])),
)
for function in self._functions
)
def collect_python_function_tests(
functions: Sequence[PythonFunctionIdentity],
selectors: Sequence[str],
repo_root: Path,
*,
source_root: Path | None = None,
exclusions: Sequence[str] = (),
) -> PythonUsageReport:
selected_functions: Final = tuple(dict.fromkeys(functions))
if not selected_functions:
raise ValueError("Python function discovery needs at least one function")
if not selectors:
raise ValueError("Python function discovery needs at least one test selector")
with tempfile.TemporaryDirectory(prefix="litellm-function-tests-") as directory:
temporary: Final = Path(directory)
config_path: Final = temporary / "config.json"
output_path: Final = temporary / "report.json"
config: Final = _WorkerConfig(
functions=selected_functions,
source_root=source_root or repo_root / "litellm",
output=output_path,
pytest_args=tuple(
(
"-o",
"consider_namespace_packages=true",
"-p",
"no:cacheprovider",
*selectors,
*(f"--deselect={nodeid}" for nodeid in exclusions),
)
),
)
config_path.write_text(config.model_dump_json())
import_roots: Final = tuple(
dict.fromkeys(
(
str(repo_root),
str(source_root or repo_root / "litellm"),
*(
str(path.parent if path.suffix == ".py" else path)
for selector in selectors
if (path := repo_root / selector.partition("::")[0]).exists()
),
os.environ.get("PYTHONPATH", ""),
)
)
)
env: Final = {
**os.environ,
"PYTHONPATH": os.pathsep.join(import_roots),
}
try:
result: Final = subprocess.run(
(sys.executable, "-m", __name__, str(config_path)),
cwd=repo_root,
env=env,
capture_output=True,
text=True,
timeout=600,
check=False,
)
except (OSError, subprocess.TimeoutExpired) as error:
return PythonUsageReport(usages=(), collected_tests=(), exit_code=1, problems=(str(error),))
if not output_path.exists():
return PythonUsageReport(
usages=(),
collected_tests=(),
exit_code=result.returncode or 1,
problems=((result.stdout + result.stderr).strip(),),
)
report: Final = PythonUsageReport.model_validate_json(output_path.read_text())
process_output: Final = (result.stdout + result.stderr).strip()
if result.returncode and not report.problems and process_output:
return report.model_copy(update={"problems": (process_output,)})
return report
def _run_worker(config: _WorkerConfig) -> int:
import pytest
plugin: Final = _FunctionUsagePlugin(config.functions, config.source_root)
exit_code: Final = int(pytest.main(list(config.pytest_args), plugins=[plugin]))
report: Final = PythonUsageReport(
usages=plugin.usages(),
collected_tests=plugin.collected_tests,
exit_code=exit_code,
problems=plugin.problems,
)
config.output.write_text(report.model_dump_json())
return exit_code
def main(argv: Sequence[str] | None = None) -> int:
parser: Final = argparse.ArgumentParser()
parser.add_argument("config", type=Path)
namespace: Final = parser.parse_args(argv)
config: Final = _WorkerConfig.model_validate_json(namespace.config.read_text())
return _run_worker(config)
if __name__ == "__main__":
raise SystemExit(main())

39
tests/rust-python-harness/shared/tracing/steps.py
View file

@ -1,39 +0,0 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from .profiler import FunctionTraceEvent
@dataclass(frozen=True, slots=True)
class PipelineStep:
id: int
parent_id: int | None
span: str
raw: str
def pipeline_projection(events: Sequence[FunctionTraceEvent]) -> tuple[PipelineStep, ...]:
raw_parents: dict[int, int | None] = {}
projected_ids: set[int] = set()
shown: list[PipelineStep] = []
for event in events:
if event.id in raw_parents:
raise ValueError(f"duplicate trace event id {event.id}")
if event.parent_id is not None and event.parent_id not in raw_parents:
raise ValueError(f"trace event {event.id} references unknown or later parent {event.parent_id}")
raw_parents[event.id] = event.parent_id
parent_id: int | None = event.parent_id
while parent_id is not None and parent_id not in projected_ids:
parent_id = raw_parents[parent_id]
shown.append(PipelineStep(event.id, parent_id, event.function, event.raw))
projected_ids.add(event.id)
return tuple(shown)
def trace_depths(steps: Sequence[PipelineStep]) -> dict[int, int]:
depths: dict[int, int] = {}
for step in steps:
depths[step.id] = 0 if step.parent_id is None else depths[step.parent_id] + 1
return depths

236
tests/rust-python-harness/shared/tracing/test_profiler.py
View file

@ -1,236 +0,0 @@
from __future__ import annotations
import asyncio
import sys
import threading
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from functools import wraps
from pathlib import Path
from types import FrameType, FunctionType
from typing import Final, ParamSpec, TypeVar, cast
import pytest
from .profiler import (
FunctionTraceEvent,
PythonProfiler,
_module_qualnames,
profile_python,
profile_python_function_usage,
)
_P = ParamSpec("_P")
_T = TypeVar("_T")
def _passthrough(function: Callable[_P, _T]) -> Callable[_P, _T]:
@wraps(function)
def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
return function(*args, **kwargs)
return wrapper
class Decorated:
@_passthrough
def call(self) -> None:
return None
def _events_named(profiler: PythonProfiler, name: str) -> tuple[FunctionTraceEvent, ...]:
return tuple(event for event in profiler.events if event.function.endswith(name))
@pytest.mark.parametrize("threads", (False, True))
def test_profiler_keeps_repeated_calls(threads: bool) -> None:
def called() -> None:
return None
with profile_python(Path(__file__).parent, threads=threads) as profiler:
called()
called()
assert len(_events_named(profiler, "called")) == 2
def test_profiler_qualifies_decorated_methods_by_class() -> None:
with profile_python(Path(__file__).parent) as profiler:
Decorated().call()
assert any(event.function.endswith(" Decorated.call") for event in profiler.events)
assert _module_qualnames(__name__)[cast(FunctionType, Decorated.call.__wrapped__).__code__] == "Decorated.call"
@pytest.mark.parametrize("threads", (False, True))
def test_profiler_records_real_frame_ancestry(threads: bool) -> None:
def called() -> None:
return None
def outer() -> None:
called()
with profile_python(Path(__file__).parent, threads=threads) as profiler:
outer()
outer_event, called_event = (event for event in profiler.events if event.function.endswith(("outer", "called")))
assert called_event.parent_id == outer_event.id
def test_profiler_restores_previous_profiler_after_failure() -> None:
previous: Final = sys.getprofile()
with pytest.raises(RuntimeError, match="stop"):
with profile_python(Path(__file__).parent):
raise RuntimeError("stop")
assert sys.getprofile() is previous
@pytest.mark.parametrize("threads", (False, True))
def test_profiler_does_not_count_coroutine_resumption_as_another_call(threads: bool) -> None:
async def suspended() -> None:
await asyncio.sleep(0)
await asyncio.sleep(0)
with profile_python(Path(__file__).parent, threads=threads) as profiler:
asyncio.run(suspended())
assert len(_events_named(profiler, "suspended")) == 1
@pytest.mark.parametrize("threads", (False, True))
def test_profiler_preserves_parent_across_coroutine_suspension(threads: bool) -> None:
def called() -> None:
return None
async def suspended() -> None:
await asyncio.sleep(0)
called()
with profile_python(Path(__file__).parent, threads=threads) as profiler:
asyncio.run(suspended())
suspended_event: Final = _events_named(profiler, "suspended")[0]
called_event: Final = _events_named(profiler, "called")[0]
assert called_event.parent_id == suspended_event.id
def test_profiler_captures_worker_threads_when_enabled() -> None:
def called() -> None:
return None
with profile_python(Path(__file__).parent, threads=True) as profiler:
thread: Final = threading.Thread(target=called)
thread.start()
thread.join()
called_event: Final = _events_named(profiler, "called")[0]
assert called_event.parent_id is None
@pytest.mark.skipif(sys.version_info < (3, 12), reason="existing worker capture requires sys.monitoring")
@pytest.mark.parametrize("prewarm", (False, True))
def test_profiler_captures_reused_workers_without_leaking_between_sessions(prewarm: bool) -> None:
def called() -> None:
return None
with ThreadPoolExecutor(max_workers=1) as executor:
if prewarm:
executor.submit(called).result(timeout=5)
with profile_python(Path(__file__).parent, threads=True) as first:
executor.submit(called).result(timeout=5)
executor.submit(called).result(timeout=5)
with profile_python(Path(__file__).parent, threads=True) as second:
executor.submit(called).result(timeout=5)
executor.submit(called).result(timeout=5)
assert len(_events_named(first, "called")) == 1
assert len(_events_named(second, "called")) == 1
def test_profiler_restores_main_and_worker_hooks_after_failure() -> None:
previous: Final = sys.getprofile()
previous_thread: Final = threading.getprofile()
with ThreadPoolExecutor(max_workers=1) as executor:
worker_previous: Final = executor.submit(sys.getprofile).result(timeout=5)
with pytest.raises(RuntimeError, match="stop"):
with profile_python(Path(__file__).parent, threads=True):
raise RuntimeError("stop")
assert executor.submit(sys.getprofile).result(timeout=5) is worker_previous
assert sys.getprofile() is previous
assert threading.getprofile() is previous_thread
@pytest.mark.skipif(sys.version_info < (3, 12), reason="existing worker capture requires sys.monitoring")
def test_function_usage_profiler_captures_reused_workers() -> None:
def selected() -> None:
return None
function: Final = f"{Path(__file__).name}:{selected.__code__.co_firstlineno} {selected.__qualname__}"
with ThreadPoolExecutor(max_workers=1) as executor:
executor.submit(selected).result(timeout=5)
with profile_python_function_usage(Path(__file__).parent, frozenset((function,)), threads=True) as profiler:
executor.submit(selected).result(timeout=5)
assert profiler.called == {function}
@pytest.mark.skipif(sys.version_info < (3, 12), reason="independent thread hooks require sys.monitoring")
def test_threaded_profiler_preserves_custom_worker_hook_and_releases_monitoring_slot() -> None:
def worker_hook(_frame: FrameType, _event: str, _arg: object) -> None:
return None
def fail_with_profile(executor: ThreadPoolExecutor) -> None:
with profile_python(Path(__file__).parent, threads=True):
assert executor.submit(sys.getprofile).result(timeout=5) is worker_hook
raise RuntimeError("stop")
tools_before: Final = tuple(sys.monitoring.get_tool(slot) for slot in range(6))
with ThreadPoolExecutor(max_workers=1, initializer=lambda: sys.setprofile(worker_hook)) as executor:
assert executor.submit(sys.getprofile).result(timeout=5) is worker_hook
with pytest.raises(RuntimeError, match="stop"):
fail_with_profile(executor)
assert executor.submit(sys.getprofile).result(timeout=5) is worker_hook
assert tuple(sys.monitoring.get_tool(slot) for slot in range(6)) == tools_before
@pytest.mark.skipif(sys.version_info < (3, 12), reason="existing worker capture requires sys.monitoring")
def test_threaded_profiler_keeps_concurrent_event_ids_and_parent_links() -> None:
def child() -> None:
return None
def parent() -> None:
child()
with ThreadPoolExecutor(max_workers=4) as executor:
with profile_python(Path(__file__).parent, threads=True) as profiler:
futures: Final = tuple(executor.submit(parent) for _ in range(200))
for future in futures:
future.result(timeout=5)
parent_ids: Final = frozenset(event.id for event in _events_named(profiler, "parent"))
children: Final = _events_named(profiler, "child")
assert len(parent_ids) == len(children) == 200
assert frozenset(event.parent_id for event in children) == parent_ids
assert tuple(event.id for event in profiler.events) == tuple(range(len(profiler.events)))
def test_function_usage_profiler_records_only_selected_functions() -> None:
def selected() -> None:
return None
def ignored() -> None:
return None
source_root: Final = Path(__file__).parent
function: Final = f"{Path(__file__).name}:{selected.__code__.co_firstlineno} {selected.__qualname__}"
with profile_python_function_usage(source_root, frozenset((function,))) as profiler:
selected()
ignored()
assert profiler.called == {function}

168
tests/rust-python-harness/shared/tracing/test_pytest_usage.py
View file

@ -1,168 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import Final
import pytest
from .pytest_usage import (
PythonFunctionIdentity,
PythonFunctionReference,
RustFunctionIdentity,
candidate_test_files,
collect_python_function_tests,
)
def test_collects_parameterized_tests_that_execute_function(tmp_path: Path) -> None:
(tmp_path / "pytest.ini").write_text("[pytest]\n")
(tmp_path / "source.py").write_text("def target():\n return 1\n\ndef other():\n return 2\n")
(tmp_path / "test_source.py").write_text(
"import pytest\n"
"from source import other, target\n"
"@pytest.mark.parametrize('value', [1, 2])\n"
"def test_target(value): assert target() + value > 0\n"
"def test_other(): assert other() == 2\n"
)
target: Final = PythonFunctionIdentity(file="source.py", line=1, qualname="target")
report: Final = collect_python_function_tests(
(target,),
("test_source.py",),
tmp_path,
source_root=tmp_path,
)
assert report.exit_code == 0, report.problems
assert report.usages[0].tests == (
"test_source.py::test_target[1]",
"test_source.py::test_target[2]",
)
def test_collects_async_and_threaded_function_calls(tmp_path: Path) -> None:
(tmp_path / "pytest.ini").write_text("[pytest]\n")
(tmp_path / "source.py").write_text(
"async def async_target():\n return 1\n\ndef threaded_target():\n return 2\n"
)
(tmp_path / "test_source.py").write_text(
"import asyncio\n"
"from threading import Thread\n"
"from source import async_target, threaded_target\n"
"def test_async(): assert asyncio.run(async_target()) == 1\n"
"def test_thread():\n"
" thread = Thread(target=threaded_target)\n"
" thread.start()\n"
" thread.join()\n"
)
functions: Final = (
PythonFunctionIdentity(file="source.py", line=1, qualname="async_target"),
PythonFunctionIdentity(file="source.py", line=4, qualname="threaded_target"),
)
report: Final = collect_python_function_tests(
functions,
("test_source.py",),
tmp_path,
source_root=tmp_path,
)
assert report.exit_code == 0, report.problems
assert report.usages[0].tests == ("test_source.py::test_async",)
assert report.usages[1].tests == ("test_source.py::test_thread",)
def test_adds_candidate_directory_to_worker_import_path(tmp_path: Path) -> None:
(tmp_path / "pytest.ini").write_text("[pytest]\n")
source: Final = tmp_path / "source"
tests: Final = tmp_path / "tests"
source.mkdir()
tests.mkdir()
(source / "implementation.py").write_text("def target():\n return 1\n")
(tests / "helper.py").write_text("VALUE = 1\n")
(tests / "test_source.py").write_text(
"from helper import VALUE\nfrom implementation import target\ndef test_target(): assert target() == VALUE\n"
)
target: Final = PythonFunctionIdentity(file="implementation.py", line=1, qualname="target")
report: Final = collect_python_function_tests(
(target,),
("tests/test_source.py",),
tmp_path,
source_root=source,
)
assert report.exit_code == 0, report.problems
assert report.usages[0].tests == ("tests/test_source.py::test_target",)
def test_parses_function_identity_from_trace() -> None:
function: Final = PythonFunctionIdentity.from_trace("llms/mistral/ocr/transformation.py:72 Config.map")
assert function.file == "llms/mistral/ocr/transformation.py"
assert function.line == 72
assert function.qualname == "Config.map"
def test_resolves_function_and_finds_candidate_test_files(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
package: Final = tmp_path / "package"
tests: Final = tmp_path / "tests"
package.mkdir()
tests.mkdir()
(package / "__init__.py").write_text("")
(package / "implementation.py").write_text("class Config:\n def transform(self):\n return 1\n")
(tests / "test_implementation.py").write_text("from package.implementation import Config\n")
(tests / "test_unrelated.py").write_text("def test_other(): pass\n")
monkeypatch.syspath_prepend(tmp_path)
reference: Final = PythonFunctionReference(module="package.implementation", qualname="Config.transform")
function: Final = reference.resolve(tmp_path)
candidates: Final = candidate_test_files((reference,), ("tests",), tmp_path)
assert function.file == "package/implementation.py"
assert function.qualname == "Config.transform"
assert candidates == ("tests/test_implementation.py",)
def test_candidate_test_files_excludes_harness_roots(tmp_path: Path) -> None:
tests: Final = tmp_path / "tests"
harness: Final = tests / "harness"
harness.mkdir(parents=True)
(tests / "test_implementation.py").write_text("from package.implementation import Config\n")
(harness / "test_fixture.py").write_text("from package.implementation import Config\n")
function: Final = PythonFunctionReference(module="package.implementation", qualname="Config.transform")
candidates: Final = candidate_test_files(
(function,),
("tests",),
tmp_path,
exclude_roots=("tests/harness",),
)
assert candidates == ("tests/test_implementation.py",)
def test_candidate_test_files_finds_top_level_calls_and_import_aliases(tmp_path: Path) -> None:
tests: Final = tmp_path / "tests"
tests.mkdir()
(tests / "test_attribute.py").write_text("import package\ndef test_call(): package.ocr()\n")
(tests / "test_alias.py").write_text("from package import ocr as run_ocr\ndef test_call(): run_ocr()\n")
(tests / "test_unrelated.py").write_text("def test_call(): return 'ocr'\n")
function: Final = PythonFunctionIdentity(file="ocr/main.py", line=1, qualname="ocr")
candidates: Final = candidate_test_files((function,), ("tests",), tmp_path)
assert candidates == (
"tests/test_alias.py",
"tests/test_attribute.py",
)
def test_parses_rust_function_identity_and_derives_test_module() -> None:
function: Final = RustFunctionIdentity.from_trace(
"crates/core/src/providers/mistral/ocr/transformation.rs:73 "
"litellm_core::providers::mistral::ocr::transformation::supported_ocr_params"
)
assert function.file == "crates/core/src/providers/mistral/ocr/transformation.rs"
assert function.test_module == "providers::mistral::ocr::transformation::tests"

44
tests/rust-python-harness/shared/tracing/test_steps.py
View file

@ -1,44 +0,0 @@
from __future__ import annotations
from typing import Final
import pytest
from .profiler import FunctionTraceEvent
from .steps import pipeline_projection, trace_depths
def event(event_id: int, function: str, parent_id: int | None = None) -> FunctionTraceEvent:
return FunctionTraceEvent(event_id, parent_id, function)
def test_projection_keeps_every_call_and_parent() -> None:
events: Final = (
event(0, "module.py:1 entry"),
event(1, "module.py:2 internal_helper", 0),
event(2, "module.py:3 nested", 1),
event(3, "module.py:2 internal_helper", 0),
)
steps: Final = pipeline_projection(events)
assert tuple((step.id, step.parent_id, step.span, step.raw) for step in steps) == tuple(
(item.id, item.parent_id, item.function, item.raw) for item in events
)
def test_projection_preserves_repeated_occurrences() -> None:
steps: Final = pipeline_projection((event(0, "route"), event(1, "http", 0), event(2, "http", 0)))
assert [step.span for step in steps] == ["route", "http", "http"]
def test_projection_preserves_multiple_roots() -> None:
steps: Final = pipeline_projection((event(0, "route"), event(1, "request")))
assert trace_depths(steps) == {0: 0, 1: 0}
def test_projection_rejects_duplicate_and_unknown_parent_ids() -> None:
with pytest.raises(ValueError, match="duplicate trace event id"):
pipeline_projection((event(0, "route"), event(0, "request")))
with pytest.raises(ValueError, match="unknown or later parent"):
pipeline_projection((event(1, "request", 0),))

1
tests/rust-python-harness/strategies/trace_parity/AGENTS.md
View file

@ -1 +0,0 @@
Prints every collected Python call under litellm/ from live traces against replayed HTTP responses. API-key and Vertex credentials scenarios exercise separate authentication paths; credentials scenarios replay the token exchange locally.

115
tests/rust-python-harness/strategies/trace_parity/__init__.py
View file

@ -1,115 +0,0 @@
from pathlib import Path
from typing import Final
from ...shared.reporting.models import SURFACES, Coverage
from ...shared.reporting.strategy import (
CaseDefinition,
ModuleCaseSpec,
NotImplementedCaseSpec,
RunnerArgumentDefinition,
StrategyDefinition,
)
from .reporting import render_trace_results
from .runner import run_trace_cases
CASES: Final[tuple[CaseDefinition, ...]] = (
CaseDefinition(
"ocr",
ModuleCaseSpec(
coverage=Coverage.PARTIAL,
module="tests.rust-python-harness.strategies.trace_parity.sdk.ocr.case",
),
surface="sdk",
),
CaseDefinition(
"messages",
ModuleCaseSpec(
coverage=Coverage.PARTIAL,
module="tests.rust-python-harness.strategies.trace_parity.sdk.messages.case",
note="Success paths are async; sync tracing captures the currently unsupported behavior.",
),
surface="sdk",
),
CaseDefinition(
"responses",
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(
"count_tokens",
NotImplementedCaseSpec(reason="No token-count trace-parity case is registered."),
surface="sdk",
),
CaseDefinition(
"chat_completions",
ModuleCaseSpec(
coverage=Coverage.PARTIAL,
module="tests.rust-python-harness.strategies.trace_parity.sdk.chat_completions.case",
),
surface="sdk",
),
CaseDefinition(
"transcription",
ModuleCaseSpec(
coverage=Coverage.PARTIAL,
module="tests.rust-python-harness.strategies.trace_parity.sdk.transcription.case",
note=(
"The Python SDK delegates this provider to the Rust pipeline, so only dispatch is visible "
"to the Python profiler."
),
),
surface="sdk",
),
CaseDefinition(
"ocr",
NotImplementedCaseSpec(reason="No gateway OCR trace-parity case is registered."),
surface="gateway",
),
CaseDefinition(
"messages",
NotImplementedCaseSpec(reason="No gateway Messages trace-parity case is registered."),
surface="gateway",
),
CaseDefinition(
"responses",
NotImplementedCaseSpec(reason="No gateway Responses trace-parity case is registered."),
surface="gateway",
),
CaseDefinition(
"count_tokens",
NotImplementedCaseSpec(reason="No gateway token-count trace-parity case is registered."),
surface="gateway",
),
CaseDefinition(
"chat_completions",
NotImplementedCaseSpec(reason="No gateway chat trace-parity case is registered."),
surface="gateway",
),
CaseDefinition(
"transcription",
NotImplementedCaseSpec(reason="No gateway transcription trace-parity case is registered."),
surface="gateway",
),
)
STRATEGY: Final = StrategyDefinition(
id="trace_parity",
order=20,
label="Traces",
description="Print Python profiler frames for representative pipeline scenarios.",
directory=Path(__file__).parent,
runnable_spec=ModuleCaseSpec,
cases=CASES,
run=run_trace_cases,
render=render_trace_results,
surfaces=SURFACES,
runner_argument=RunnerArgumentDefinition(
option="--scenario",
metavar="NAME",
help="run only this named trace scenario; repeat to select more than one",
),
)

177
tests/rust-python-harness/strategies/trace_parity/fixtures.py
View file

@ -1,177 +0,0 @@
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 = tuple(aws_event_stream_frame(event) for event in events)
body: Final = (
b"".join((*frames[:-1], frames[-1][:-1] + bytes((frames[-1][-1] ^ 0xFF,))))
if corrupt_last_frame
else b"".join(frames)
)
return RecordedHttpStreamResponse(
kind="http_stream",
status_code=200,
headers=AWS_EVENT_STREAM_HEADERS,
chunks=(RecordedStreamChunk.from_bytes(body),),
)
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}),
)

68
tests/rust-python-harness/strategies/trace_parity/models.py
View file

@ -1,68 +0,0 @@
from __future__ import annotations
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from typing import Final, Literal, cast
from ...shared.parity.recorded_http import RecordedResponse
from ...shared.reporting.models import SdkFunction
TraceFailureSource = Literal["python", "harness"]
@dataclass(frozen=True, slots=True)
class RouteFixture:
kwargs: dict[str, object]
provider_responses: tuple[RecordedResponse, ...]
expected_failure: bool = False
consume_stream: bool = False
environment: tuple[tuple[str, str], ...] = ()
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,
environment=self.environment,
)
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)
class RouteSpec:
route: SdkFunction
python_entrypoints: tuple[str, str]
fixture: Callable[[str], RouteFixture]
@dataclass(frozen=True, slots=True)
class TraceScenario:
name: str
fixture: Callable[[str], RouteFixture]
asynchronous: bool
@dataclass(frozen=True, slots=True)
class TraceSuite:
route: RouteSpec
scenarios: tuple[TraceScenario, ...]
@dataclass(frozen=True, slots=True)
class TraceExecutionFailure:
engine: TraceFailureSource
message: str

182
tests/rust-python-harness/strategies/trace_parity/reporting.py
View file

@ -1,182 +0,0 @@
from __future__ import annotations
import os
import sys
from collections.abc import Sequence
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, trace_depths
TRACE_ARTIFACT: Final = "trace"
_COLORS: Final[dict[str, str]] = {"red": "31", "cyan": "36"}
_RESET: Final = "\033[0m"
def _paint(text: str, color: str) -> str:
if not sys.stdout.isatty() or os.environ.get("NO_COLOR"):
return text
return f"\033[{_COLORS[color]}m{text}{_RESET}"
class TraceEventArtifact(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
id: int
parent_id: int | None
span: str
raw: str
def step(self) -> PipelineStep:
return PipelineStep(self.id, self.parent_id, self.span, self.raw)
class TraceArtifact(BaseModel):
model_config = ConfigDict(frozen=True, extra="forbid")
surface: Surface
sdk_function: SdkFunction
scenario: str
python: tuple[TraceEventArtifact, ...]
python_error: str | None = None
@classmethod
def from_traces(
cls,
*,
surface: Surface,
sdk_function: SdkFunction,
scenario: str,
python: Sequence[PipelineStep],
python_error: str | None = None,
) -> TraceArtifact:
return cls(
surface=surface,
sdk_function=sdk_function,
scenario=scenario,
python=tuple(
TraceEventArtifact(id=step.id, parent_id=step.parent_id, span=step.span, raw=step.raw)
for step in python
),
python_error=python_error,
)
def python_steps(self) -> tuple[PipelineStep, ...]:
return tuple(event.step() for event in self.python)
def has_errors(self) -> bool:
return self.python_error is not None
def _split_raw(raw: str) -> tuple[str, str]:
location, separator, name = raw.partition(" ")
if separator:
return name, location
return raw, ""
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 ""
return _paint(f"{index} {' ' * depth}{name}{suffix}", "cyan")
def _python_lines(steps: tuple[PipelineStep, ...]) -> str:
depths: Final = trace_depths(steps)
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 _error_lines(artifact: TraceArtifact) -> tuple[str, ...]:
if artifact.python_error is None:
return ()
return (_paint(f"Python error: {artifact.python_error}", "red"),)
def _render_trace(artifact: TraceArtifact) -> str:
return "\n\n".join((_python_lines(artifact.python_steps()), *_error_lines(artifact)))
def _scenario(nodeid: str) -> str:
parts: Final = nodeid.split(":")
return parts[-1] if len(parts) >= 4 else "default"
def _unavailable(status: RunStatus) -> str:
return f"Trace: NOT AVAILABLE\nTest outcome: {status.value}"
def _render_artifact(body: str) -> str:
try:
artifact: Final = TraceArtifact.model_validate_json(body)
except ValidationError as error:
return f"Trace artifact is invalid: {error}"
return _render_trace(artifact)
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_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)}"
return f"{label}\n{'-' * len(label)}\n\n{body}"
def _case_block(result: CaseResult) -> str:
header: Final = f"Case: {result.case.sdk_function}"
outcomes: Final = tuple(result.outcomes.items()) or (
(nodeid, RunStatus.NOT_RUN) for nodeid in sorted(result.collected)
)
sections: Final = tuple(_scenario_section(result, nodeid, status) for nodeid, status in outcomes)
return "\n\n".join((f"{header}\n{'=' * len(header)}", *sections))
def _unavailable_block(title: str, lines: tuple[str, ...]) -> str | None:
if not lines:
return None
return f"{title}\n{'-' * len(title)}\n" + "\n".join(lines)
def _surface_section(surface: Surface, results: Sequence[CaseResult]) -> ReportSection | None:
selected: Final = tuple(result for result in results if result.case.surface == surface)
if not selected:
return None
outcome_blocks: Final = tuple(_case_block(result) for result in selected if result.outcomes)
not_implemented: Final = _unavailable_block(
"Not implemented",
tuple(
f"- {result.case.sdk_function}: {spec.reason}"
for result in selected
if isinstance((spec := result.case.spec), NotImplementedCaseSpec)
),
)
skipped: Final = _unavailable_block(
"Skipped",
tuple(
f"- {result.case.sdk_function}: {spec.reason}"
for result in selected
if isinstance((spec := result.case.spec), SkippedCaseSpec)
),
)
blocks: Final = (
*outcome_blocks,
*((not_implemented,) if not_implemented else ()),
*((skipped,) if skipped else ()),
)
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("Traces", ("No traces selected",)),)

158
tests/rust-python-harness/strategies/trace_parity/runner.py
View file

@ -1,158 +0,0 @@
from __future__ import annotations
import importlib
from collections.abc import Sequence
from pathlib import Path
from time import monotonic
from typing import Final
from ...shared.reporting.models import CaseResult, HarnessCase, HarnessRun, ResultArtifact, RunStatus, Surface
from ...shared.reporting.strategy import ModuleCaseSpec, UpdateCallback
from .models import (
TraceExecutionFailure,
TraceScenario,
TraceSuite,
)
from .reporting import TRACE_ARTIFACT, TraceArtifact
from .sdk.execution import execute_trace
def _load_case(reference: str, harness_case: HarnessCase) -> TraceSuite | TraceExecutionFailure:
try:
module: Final = importlib.import_module(reference)
except Exception as error:
return TraceExecutionFailure("harness", f"cannot import {reference}: {type(error).__name__}: {error}")
suite: Final = getattr(module, "TRACE_SUITE", None)
if not isinstance(suite, TraceSuite):
return TraceExecutionFailure("harness", f"{reference} must export TRACE_SUITE: TraceSuite")
validation_error: Final = validate_trace_suite(suite, harness_case)
if validation_error is not None:
return TraceExecutionFailure("harness", f"{reference} {validation_error}")
return suite
def validate_trace_suite(suite: TraceSuite, harness_case: HarnessCase) -> str | None:
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_names: Final = tuple(
scenario.name
for scenario in suite.scenarios
if not scenario.name.startswith("async-" if scenario.asynchronous else "sync-")
)
if invalid_names:
return f"scenario names must start with sync- or async-: {', '.join(invalid_names)}"
surface: Final = harness_case.surface
if surface != "sdk":
return "requires the sdk surface"
if suite.route.route != harness_case.sdk_function:
return f"route {suite.route.route} does not match case function {harness_case.sdk_function}"
return None
def scenario_nodeids(
trace_suite: TraceSuite,
harness_case: HarnessCase,
selected_scenarios: frozenset[str] = frozenset(),
) -> tuple[tuple[TraceScenario, str], ...]:
surface: Final = harness_case.surface
if surface is None:
return ()
return tuple(
(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
)
def _record_setup_failure(run: HarnessRun, case: HarnessCase, message: str, stage: str) -> None:
result: Final = run.results[case.key]
nodeid: Final = f"trace:{case.surface}:{case.sdk_function}:{stage}"
result.collected.add(nodeid)
result.record(nodeid, RunStatus.ERROR)
run.failures.append((nodeid, message))
def run_trace_scenario(
run: HarnessRun,
result: CaseResult,
trace_suite: TraceSuite,
scenario: TraceScenario,
surface: Surface,
nodeid: str,
on_update: UpdateCallback,
) -> None:
started_at: Final = monotonic()
trace: Final = _execute_scenario(trace_suite, scenario, surface)
duration: Final = monotonic() - started_at
if isinstance(trace, TraceExecutionFailure):
result.record(nodeid, RunStatus.ERROR, duration)
run.failures.append((nodeid, trace.message))
on_update(run)
return
artifact: Final = ResultArtifact(TRACE_ARTIFACT, trace.model_dump_json())
if trace.has_errors():
result.record(nodeid, RunStatus.ERROR, duration, (artifact,))
run.failures.append((nodeid, trace.python_error or ""))
else:
result.record(nodeid, RunStatus.PASSED, duration, (artifact,))
on_update(run)
def _execute_scenario(
trace_suite: TraceSuite,
scenario: TraceScenario,
surface: Surface,
) -> TraceArtifact | TraceExecutionFailure:
if surface != "sdk":
return TraceExecutionFailure("harness", "trace scenarios only run on the sdk surface")
return execute_trace(trace_suite.route, scenario, surface)
def _run_case(
run: HarnessRun,
harness_case: HarnessCase,
selected_scenarios: frozenset[str],
on_update: UpdateCallback,
) -> None:
result: Final = run.results[harness_case.key]
spec: Final = harness_case.spec
if not isinstance(spec, ModuleCaseSpec):
return
surface: Final = harness_case.surface
if surface is None:
return
trace_suite: Final = _load_case(spec.module, harness_case)
if isinstance(trace_suite, TraceExecutionFailure):
_record_setup_failure(run, harness_case, trace_suite.message, "load")
on_update(run)
return
nodeids: Final = scenario_nodeids(trace_suite, harness_case, selected_scenarios)
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, nodeid in nodeids:
run_trace_scenario(run, result, trace_suite, scenario, surface, nodeid, on_update)
def run_trace_cases(
cases: Sequence[HarnessCase],
repo_root: Path,
on_update: UpdateCallback,
runner_args: Sequence[str] = (),
) -> tuple[int, HarnessRun]:
del repo_root
selected_scenarios: Final = frozenset(runner_args)
run: Final = HarnessRun.from_cases(cases)
for harness_case in cases:
_run_case(run, harness_case, selected_scenarios, on_update)
run.finished_at = monotonic()
on_update(run)
failed: Final = any(
result.status in {RunStatus.ERROR, RunStatus.FAILED, RunStatus.MISSING} for result in run.results.values()
)
return int(failed), run

161
tests/rust-python-harness/strategies/trace_parity/sdk/chat_completions/case.py
View file

@ -1,161 +0,0 @@
from __future__ import annotations
from typing import Final
from ...fixtures import (
anthropic_response_body,
anthropic_stream_events,
aws_event_stream_response,
json_response,
sse_response,
)
from ...models import RouteFixture, RouteSpec, TraceScenario, TraceSuite
def _anthropic_fixture(_base_url: str) -> RouteFixture:
return RouteFixture(
kwargs={
"model": "anthropic/claude-sonnet-5",
"messages": [{"role": "user", "content": "hello"}],
"max_tokens": 16,
},
provider_responses=(json_response(anthropic_response_body()),),
)
def _bedrock_fixture(_base_url: str) -> RouteFixture:
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",
"aws_region_name": "us-east-1",
}
return RouteFixture(
kwargs={
"model": "bedrock/us-east-1/anthropic.claude-v2",
"messages": [{"role": "user", "content": "hello"}],
**credentials,
"max_tokens": 16,
},
provider_responses=(json_response(response),),
)
def _anthropic_stream_fixture(_base_url: str) -> RouteFixture:
fixture: Final = _anthropic_fixture(_base_url)
return fixture.derive(
kwargs={"stream": True},
provider_responses=(sse_response(anthropic_stream_events()),),
consume_stream=True,
)
def _bedrock_stream_fixture(_base_url: str) -> RouteFixture:
fixture: Final = _bedrock_fixture(_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(_base_url: str) -> RouteFixture:
fixture: Final = _anthropic_fixture(_base_url)
return fixture.derive(
provider_responses=(
json_response(
{"type": "error", "error": {"type": "invalid_request_error", "message": "bad request"}},
status=400,
),
),
expected_failure=True,
)
def _stream_error_fixture(base_url: str) -> RouteFixture:
fixture: Final = _anthropic_fixture(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,
)
SPEC: Final = RouteSpec(
"chat_completions",
("completion", "acompletion"),
_anthropic_fixture,
)
TRACE_SUITE: Final = TraceSuite(
route=SPEC,
scenarios=(
TraceScenario(
name="sync-anthropic",
fixture=_anthropic_fixture,
asynchronous=False,
),
TraceScenario(
name="async-anthropic",
fixture=_anthropic_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-anthropic-stream",
fixture=_anthropic_stream_fixture,
asynchronous=False,
),
TraceScenario(
name="async-anthropic-stream",
fixture=_anthropic_stream_fixture,
asynchronous=True,
),
TraceScenario(
name="async-anthropic-provider-error",
fixture=_provider_error_fixture,
asynchronous=True,
),
TraceScenario(
name="async-anthropic-stream-error",
fixture=_stream_error_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-bedrock",
fixture=_bedrock_fixture,
asynchronous=False,
),
TraceScenario(
name="async-bedrock",
fixture=_bedrock_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-bedrock-event-stream",
fixture=_bedrock_stream_fixture,
asynchronous=False,
),
TraceScenario(
name="async-bedrock-event-stream",
fixture=_bedrock_stream_fixture,
asynchronous=True,
),
),
)

148
tests/rust-python-harness/strategies/trace_parity/sdk/execution.py
View file

@ -1,148 +0,0 @@
from __future__ import annotations
import asyncio
import os
from collections.abc import AsyncIterable, Awaitable, Iterable
from dataclasses import dataclass
from pathlib import Path
from typing import Final, Protocol, cast
from unittest.mock import patch
from ....shared.parity.replay import replay_server
from ....shared.reporting.models import Surface
from ....shared.tracing.profiler import FunctionTraceEvent, profile_python
from ....shared.tracing.steps import pipeline_projection
from ..models import RouteFixture, RouteSpec, TraceExecutionFailure, TraceScenario
from ..reporting import TraceArtifact
class SdkCall(Protocol):
def __call__(self, **kwargs: object) -> object: ...
@dataclass(frozen=True, slots=True)
class _CollectedTrace:
events: tuple[FunctionTraceEvent, ...]
error: str | None = None
def _invoke(
function: SdkCall,
kwargs: dict[str, object],
*,
asynchronous: bool,
consume_stream: bool = False,
) -> object:
async def invoke_async() -> object:
try:
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
await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10)
await GLOBAL_LOGGING_WORKER.stop()
if asynchronous:
return asyncio.run(invoke_async())
response: Final = function(**kwargs)
if consume_stream and isinstance(response, Iterable):
return tuple(cast(Iterable[object], response))
return response
def _entrypoint(spec: RouteSpec, *, asynchronous: bool) -> SdkCall:
import litellm
from litellm.anthropic_interface import messages as sdk_messages
owner: Final = sdk_messages if spec.route == "messages" else litellm
return cast(SdkCall, getattr(owner, spec.python_entrypoints[int(asynchronous)]))
def _collect(
function: SdkCall,
fixture: RouteFixture,
*,
asynchronous: bool,
) -> _CollectedTrace:
import litellm
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, fixture.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)
def collect_trace(spec: RouteSpec, *, asynchronous: bool) -> tuple[FunctionTraceEvent, ...] | TraceExecutionFailure:
function: Final = _entrypoint(spec, asynchronous=asynchronous)
try:
with replay_server() as provider:
base_fixture: Final = spec.fixture(provider.url)
for response in base_fixture.provider_responses:
provider.enqueue_response(response)
fixture: Final = RouteFixture(
kwargs={
"api_key": "test-key",
**base_fixture.kwargs,
"api_base": provider.url,
"timeout": 5,
},
provider_responses=base_fixture.provider_responses,
expected_failure=base_fixture.expected_failure,
consume_stream=base_fixture.consume_stream,
environment=base_fixture.environment,
)
with patch.dict(os.environ, fixture.environment):
collected: Final = _collect(function, fixture, asynchronous=asynchronous)
provider.take_requests(len(fixture.provider_responses))
except Exception as error:
return TraceExecutionFailure("python", f"{type(error).__name__}: {error}")
if fixture.expected_failure and collected.error is None:
return TraceExecutionFailure("python", "call succeeded but the scenario expects failure")
if not fixture.expected_failure and collected.error is not None:
return TraceExecutionFailure("python", collected.error)
if not collected.events:
return TraceExecutionFailure("python", "trace is empty")
return collected.events
def execute_trace(route: RouteSpec, scenario: TraceScenario, surface: Surface) -> TraceArtifact:
scenario_route: Final = RouteSpec(
route=route.route,
python_entrypoints=route.python_entrypoints,
fixture=scenario.fixture,
)
python_trace: Final = collect_trace(scenario_route, asynchronous=scenario.asynchronous)
python_error: Final = None if isinstance(python_trace, tuple) else f"{python_trace.engine}: {python_trace.message}"
python_events: Final = python_trace if isinstance(python_trace, tuple) else ()
try:
python: Final = pipeline_projection(python_events)
except ValueError as error:
return TraceArtifact.from_traces(
surface=surface,
sdk_function=route.route,
scenario=scenario.name,
python=(),
python_error=f"harness: {error}",
)
return TraceArtifact.from_traces(
surface=surface,
sdk_function=route.route,
scenario=scenario.name,
python=python,
python_error=python_error,
)

178
tests/rust-python-harness/strategies/trace_parity/sdk/messages/case.py
View file

@ -1,178 +0,0 @@
from __future__ import annotations
from typing import Final
from ...fixtures import (
anthropic_response_body,
anthropic_stream_events,
aws_event_stream_response,
json_response,
sse_response,
)
from ...models import RouteFixture, RouteSpec, TraceScenario, TraceSuite
def _fixture(provider: str) -> RouteFixture:
conversation: Final = {"messages": [{"role": "user", "content": "hello"}], "max_tokens": 16}
return RouteFixture(
kwargs={
"model": f"{provider}/claude-sonnet-5",
**conversation,
},
provider_responses=(json_response(anthropic_response_body()),),
)
def _anthropic_fixture(_base_url: str) -> RouteFixture:
return _fixture("anthropic")
def _azure_fixture(_base_url: str) -> RouteFixture:
return _fixture("azure_ai")
def _bedrock_kwargs() -> dict[str, object]:
conversation: Final = {"messages": [{"role": "user", "content": "hello"}], "max_tokens": 16}
return {
"model": "bedrock/anthropic.claude-3-sonnet-20240229-v1:0",
**conversation,
"aws_access_key_id": "AKIAIOSFODNN7EXAMPLE",
"aws_secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
"aws_region_name": "us-east-1",
}
def _bedrock_fixture(_base_url: str) -> RouteFixture:
response_fixture: Final = _fixture("anthropic")
return RouteFixture(kwargs=_bedrock_kwargs(), provider_responses=response_fixture.provider_responses)
def _bedrock_retry_fixture(_base_url: str) -> RouteFixture:
success_fixture: Final = _bedrock_fixture(_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(), "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(_base_url: str) -> RouteFixture:
fixture: Final = _fixture("anthropic")
return fixture.derive(kwargs={"mock_response": "hello from mock"}, provider_responses=())
def _provider_error_fixture(_base_url: str) -> RouteFixture:
fixture: Final = _fixture("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(_base_url: str) -> RouteFixture:
fixture: Final = _fixture("anthropic")
return fixture.derive(provider_responses=(), expected_failure=True)
def _stream_fixture_for(provider: str) -> RouteFixture:
fixture: Final = _fixture(provider)
return fixture.derive(
kwargs={"stream": True},
provider_responses=(sse_response(anthropic_stream_events()),),
consume_stream=True,
)
def _stream_fixture(_base_url: str) -> RouteFixture:
return _stream_fixture_for("anthropic")
def _azure_stream_fixture(_base_url: str) -> RouteFixture:
return _stream_fixture_for("azure_ai")
def _bedrock_stream_fixture(base_url: str) -> RouteFixture:
fixture: Final = _bedrock_fixture(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(base_url: str) -> RouteFixture:
fixture: Final = _bedrock_fixture(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"), _anthropic_fixture)
TRACE_SUITE: Final = TraceSuite(
route=SPEC,
scenarios=(
TraceScenario(name="async-anthropic", fixture=_anthropic_fixture, asynchronous=True),
TraceScenario(name="async-azure-ai", fixture=_azure_fixture, asynchronous=True),
TraceScenario(name="async-bedrock", fixture=_bedrock_fixture, asynchronous=True),
TraceScenario(
name="async-bedrock-invalid-thinking-retry",
fixture=_bedrock_retry_fixture,
asynchronous=True,
),
TraceScenario(name="async-mock-response", fixture=_mock_fixture, asynchronous=True),
TraceScenario(
name="async-anthropic-provider-error",
fixture=_provider_error_fixture,
asynchronous=True,
),
TraceScenario(
name="async-anthropic-stream",
fixture=_stream_fixture,
asynchronous=True,
),
TraceScenario(
name="async-azure-ai-stream",
fixture=_azure_stream_fixture,
asynchronous=True,
),
TraceScenario(
name="async-bedrock-event-stream",
fixture=_bedrock_stream_fixture,
asynchronous=True,
),
TraceScenario(
name="async-bedrock-event-stream-error",
fixture=_bedrock_stream_error_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-unsupported",
fixture=_sync_unsupported_fixture,
asynchronous=False,
),
),
)

374
tests/rust-python-harness/strategies/trace_parity/sdk/ocr/case.py
View file

@ -1,374 +0,0 @@
from __future__ import annotations
import json
from typing import Final
from .....shared.parity.recorded_http import HttpHeader, RecordedHttpResponse
from ...models import RouteFixture, RouteSpec, TraceScenario, TraceSuite
SUCCESS_CALLBACK_SYNC_MAPPING: Final = mapping(
rust_span="success_callback",
python_frame=r"BoundedLoggingThreadPoolExecutor\.submit$",
)
SUCCESS_CALLBACK_ASYNC_MAPPING: Final = mapping(
rust_span="success_callback",
python_frame=r"Logging\.async_success_handler$",
)
FAILURE_CALLBACK_MAPPING: Final = mapping(
rust_span="failure_callback",
python_frame=r"Logging\.(?:async_)?failure_handler$",
)
IGNORED_SUCCESS_CALLBACK_MAPPING: Final = mapping(rust_span="success_callback")
SYNC_MAPPINGS: Final = (
*COMMON_MAPPINGS,
mapping(rust_span="execute_ocr_provider_call", python_frame=r"BaseLLMHTTPHandler\.ocr$"),
mapping(span="python_transform_ocr_response_wrapper", python_frame=r"BaseLLMHTTPHandler\._transform_ocr_response$"),
mapping(
rust_span="transform_ocr_response",
python_frame=r"MistralOCRConfig\.transform_ocr_response$",
),
)
ASYNC_MAPPINGS: Final = (
*COMMON_MAPPINGS,
mapping(span="python_ocr_wrapper", python_frame=r"BaseLLMHTTPHandler\.ocr$"),
mapping(rust_span="execute_ocr_provider_call", python_frame=r"BaseLLMHTTPHandler\.async_ocr$"),
mapping(
rust_span="transform_ocr_response",
python_frame=r"MistralOCRConfig\.transform_ocr_response$",
),
)
PUBLIC_RUST_DISPATCH_MAPPINGS: Final = (
mapping(span="public_sdk_entrypoint", python_frame=r"ocr/main\.py:\d+ a?ocr$"),
mapping(span="public_request", python_frame=r"rust_bridge/ocr\.py:\d+ bind_request$"),
mapping(span="bind_request", python_frame=r"rust_bridge/ocr\.py:\d+ _bind_request$"),
mapping(span="rust_ocr_enabled", python_frame=r"rust_bridge/configuration\.py:\d+ rust_ocr_enabled$"),
mapping(span="load_native_bridge", python_frame=r"rust_bridge/bindings\.py:\d+ NativeBinding\.load$"),
mapping(span="native_call_setup", python_frame=r"rust_bridge/lifecycle\.py:\d+ setup$"),
mapping(span="native_response", python_frame=r"rust_bridge/ocr\.py:\d+ build_response$"),
mapping(span="native_call_finalize", python_frame=r"rust_bridge/lifecycle\.py:\d+ finalize$"),
mapping(
span="native_success_bookkeeping",
python_frame=r"rust_bridge/lifecycle\.py:\d+ success_bookkeeping$",
),
*(mapping(rust_span=item.rust) for item in SYNC_MAPPINGS if item.rust is not None),
)
CALLBACK_SUCCESS_SYNC_MAPPINGS: Final = (*SYNC_MAPPINGS, SUCCESS_CALLBACK_SYNC_MAPPING)
CALLBACK_SUCCESS_ASYNC_MAPPINGS: Final = (*ASYNC_MAPPINGS, SUCCESS_CALLBACK_ASYNC_MAPPING)
CALLBACK_FAILURE_SYNC_MAPPINGS: Final = (
*COMMON_MAPPINGS,
mapping(rust_span="execute_ocr_provider_call", python_frame=r"BaseLLMHTTPHandler\.ocr$"),
FAILURE_CALLBACK_MAPPING,
)
CALLBACK_FAILURE_ASYNC_MAPPINGS: Final = (
*COMMON_MAPPINGS,
mapping(span="python_ocr_wrapper", python_frame=r"BaseLLMHTTPHandler\.ocr$"),
mapping(rust_span="execute_ocr_provider_call", python_frame=r"BaseLLMHTTPHandler\.async_ocr$"),
FAILURE_CALLBACK_MAPPING,
)
AZURE_COMMON_MAPPINGS: Final = (
*COMMON_MAPPINGS[:7],
mapping(
rust_span="transform_ocr_request",
python_frame=(
r"AzureAIOCRConfig\.(?:async_)?transform_ocr_request$"
r"|MistralOCRConfig\.transform_ocr_request$"
),
),
COMMON_MAPPINGS[-1],
)
AZURE_SYNC_MAPPINGS: Final = (
*AZURE_COMMON_MAPPINGS,
mapping(rust_span="execute_ocr_provider_call", python_frame=r"BaseLLMHTTPHandler\.ocr$"),
mapping(
span="python_transform_ocr_response_wrapper",
python_frame=r"BaseLLMHTTPHandler\._transform_ocr_response$",
),
mapping(
rust_span="transform_ocr_response",
python_frame=r"MistralOCRConfig\.transform_ocr_response$",
),
)
AZURE_ASYNC_MAPPINGS: Final = (
*AZURE_COMMON_MAPPINGS,
mapping(span="python_ocr_wrapper", python_frame=r"BaseLLMHTTPHandler\.ocr$"),
mapping(rust_span="execute_ocr_provider_call", python_frame=r"BaseLLMHTTPHandler\.async_ocr$"),
mapping(
rust_span="transform_ocr_response",
python_frame=r"MistralOCRConfig\.transform_ocr_response$",
),
)
def _fixture(engine: Engine, model: str, document: dict[str, str] | None = None) -> RouteFixture:
response: Final = json.dumps(
{
"pages": [{"index": 0, "markdown": "hello"}],
"model": "mistral-ocr-latest",
"usage_info": {"pages_processed": 1},
}
).encode()
return RouteFixture(
kwargs={
"model": model,
"document": document or {"type": "document_url", "document_url": "https://example.com/document.pdf"},
"pages": [0],
},
provider_responses=(
RecordedHttpResponse.from_bytes(
200, (HttpHeader(name="content-type", value="application/json"),), response
),
),
)
def _mistral_fixture(_base_url: str) -> RouteFixture:
return _fixture("mistral/mistral-ocr-latest")
def _callback_fixture(*, failure: bool) -> RouteFixture:
fixture: Final = _fixture("mistral/mistral-ocr-latest")
provider_responses: Final = (
(
RecordedHttpResponse.from_bytes(
400,
(HttpHeader(name="content-type", value="application/json"),),
b'{"message":"trace callback provider failure"}',
),
)
if failure
else fixture.provider_responses
)
return RouteFixture(
kwargs=fixture.kwargs,
provider_responses=provider_responses,
expected_failure=failure,
)
def _mistral_callback_success_fixture(_base_url: str) -> RouteFixture:
return _callback_fixture(failure=False)
def _mistral_callback_failure_fixture(_base_url: str) -> RouteFixture:
return _callback_fixture(failure=True)
def _azure_fixture(_base_url: str) -> RouteFixture:
return _fixture(
"azure_ai/pixtral-12b-2409",
{"type": "image_url", "image_url": "data:image/png;base64,aGVsbG8="},
)
def _vertex_deepseek_fixture(_base_url: str) -> RouteFixture:
vertex: Final = {"vertex_project": "trace-project", "vertex_location": "us-central1"}
return RouteFixture(
kwargs={
"model": "vertex_ai/deepseek-ocr-maas",
"document": {"type": "image_url", "image_url": "data:image/png;base64,aGVsbG8="},
**vertex,
},
provider_responses=(
RecordedHttpResponse.from_bytes(
200,
(HttpHeader(name="content-type", value="application/json"),),
json.dumps(
{
"choices": [{"message": {"role": "assistant", "content": "hello"}}],
"usage": {"prompt_tokens": 1, "completion_tokens": 1},
}
).encode(),
),
),
)
def _vertex_deepseek_credentials_fixture(base_url: str) -> RouteFixture:
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
fixture: Final = _vertex_deepseek_fixture(base_url)
private_key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
credentials: Final = json.dumps(
{
"type": "service_account",
"project_id": "trace-project",
"private_key_id": "trace-key",
"private_key": private_key.private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
).decode(),
"client_email": "trace@trace-project.iam.gserviceaccount.com",
"token_uri": f"{base_url}/token",
}
)
return RouteFixture(
kwargs={**fixture.kwargs, "api_key": None},
environment=(("VERTEXAI_CREDENTIALS", credentials), ("VERTEX_AI_API_KEY", "")),
provider_responses=(
RecordedHttpResponse.from_bytes(
200,
(HttpHeader(name="content-type", value="application/json"),),
b'{"access_token":"trace-token","token_type":"Bearer","expires_in":3600}',
),
*fixture.provider_responses,
),
)
def _cohere_fixture(_base_url: str) -> RouteFixture:
return RouteFixture(
kwargs={
"model": "cohere/parse-v5.0",
"document": {"type": "image_url", "image_url": "data:image/png;base64,aGVsbG8="},
"output_format": "blocks",
},
provider_responses=(
RecordedHttpResponse.from_bytes(
200,
(HttpHeader(name="content-type", value="application/json"),),
json.dumps(
{
"pages": [{"index": 0, "blocks": [{"type": "text", "text": {"content": "hello"}}]}],
"meta": {"billed_units": {"pages": 1}},
}
).encode(),
),
),
)
def _azure_document_intelligence_fixture(base_url: str) -> RouteFixture:
completed: Final = json.dumps(
{
"status": "succeeded",
"analyzeResult": {
"content": "hello",
"pages": [
{
"pageNumber": 1,
"width": 8.5,
"height": 11,
"unit": "inch",
"lines": [{"content": "hello"}],
}
],
},
}
).encode()
return RouteFixture(
kwargs={
"model": "azure_ai/doc-intelligence/prebuilt-read",
"document": {
"type": "document_url",
"document_url": "data:application/pdf;base64,aGVsbG8=",
},
"pages": [0],
},
provider_responses=(
RecordedHttpResponse.from_bytes(
202,
(
HttpHeader(name="content-type", value="application/json"),
HttpHeader(name="operation-location", value=f"{base_url}/operations/trace"),
),
b"{}",
),
RecordedHttpResponse.from_bytes(
200,
(HttpHeader(name="content-type", value="application/json"),),
completed,
),
),
)
SPEC: Final = RouteSpec("ocr", ("ocr", "aocr"), _mistral_fixture)
TRACE_SUITE: Final = TraceSuite(
route=SPEC,
scenarios=(
TraceScenario(
name="sync-mistral",
fixture=_mistral_fixture,
asynchronous=False,
),
TraceScenario(
name="async-mistral",
fixture=_mistral_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-mistral-callback-success",
fixture=_mistral_callback_success_fixture,
asynchronous=False,
),
TraceScenario(
name="async-mistral-callback-success",
fixture=_mistral_callback_success_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-mistral-callback-failure",
fixture=_mistral_callback_failure_fixture,
asynchronous=False,
),
TraceScenario(
name="async-mistral-callback-failure",
fixture=_mistral_callback_failure_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-azure-ai",
fixture=_azure_fixture,
asynchronous=False,
),
TraceScenario(
name="async-azure-ai",
fixture=_azure_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-azure-document-intelligence",
fixture=_azure_document_intelligence_fixture,
asynchronous=False,
),
TraceScenario(
name="async-azure-document-intelligence",
fixture=_azure_document_intelligence_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-vertex-deepseek",
fixture=_vertex_deepseek_fixture,
asynchronous=False,
),
TraceScenario(
name="async-vertex-deepseek",
fixture=_vertex_deepseek_fixture,
asynchronous=True,
),
TraceScenario(
name="sync-vertex-deepseek-credentials",
fixture=_vertex_deepseek_credentials_fixture,
asynchronous=False,
),
TraceScenario(
name="async-vertex-deepseek-credentials",
fixture=_vertex_deepseek_credentials_fixture,
asynchronous=True,
),
TraceScenario(
name="async-cohere",
fixture=_cohere_fixture,
asynchronous=True,
),
),
)

136
tests/rust-python-harness/strategies/trace_parity/sdk/responses/case.py
View file

@ -1,136 +0,0 @@
from __future__ import annotations
from typing import Final
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
def _native_fixture(provider: str) -> RouteFixture:
model: Final = "gpt-5"
return RouteFixture(
kwargs={
"model": f"{provider}/{model}",
"input": "hello",
},
provider_responses=(json_response(responses_body(model=model)),),
)
def _openai_fixture(_base_url: str) -> RouteFixture:
return _native_fixture("openai")
def _azure_fixture(_base_url: str) -> RouteFixture:
fixture: Final = _native_fixture("azure")
return fixture.derive(kwargs={"api_version": "2025-04-01-preview"})
def _openai_stream_fixture(_base_url: str) -> RouteFixture:
fixture: Final = _openai_fixture(_base_url)
return fixture.derive(
kwargs={"stream": True},
provider_responses=(sse_response(responses_stream_events()),),
consume_stream=True,
)
def _provider_error_fixture(_base_url: str) -> RouteFixture:
fixture: Final = _openai_fixture(_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(base_url: str) -> RouteFixture:
fixture: Final = _openai_fixture(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(_base_url: str) -> RouteFixture:
return RouteFixture(
kwargs={
"model": "anthropic/claude-sonnet-5",
"input": "hello",
"max_output_tokens": 16,
},
provider_responses=(json_response(anthropic_response_body()),),
)
def _anthropic_bridge_stream_fixture(_base_url: str) -> RouteFixture:
fixture: Final = _anthropic_bridge_fixture(_base_url)
return fixture.derive(
kwargs={"stream": True},
provider_responses=(sse_response(anthropic_stream_events()),),
consume_stream=True,
)
SPEC: Final = RouteSpec("responses", ("responses", "aresponses"), _openai_fixture)
TRACE_SUITE: Final = TraceSuite(
route=SPEC,
scenarios=(
TraceScenario(name="sync-openai", fixture=_openai_fixture, asynchronous=False),
TraceScenario(name="async-openai", fixture=_openai_fixture, asynchronous=True),
TraceScenario(
name="sync-openai-stream",
fixture=_openai_stream_fixture,
asynchronous=False,
),
TraceScenario(
name="async-openai-stream",
fixture=_openai_stream_fixture,
asynchronous=True,
),
TraceScenario(
name="async-openai-provider-error",
fixture=_provider_error_fixture,
asynchronous=True,
),
TraceScenario(
name="async-openai-stream-failed",
fixture=_stream_failed_fixture,
asynchronous=True,
),
TraceScenario(name="async-azure", fixture=_azure_fixture, asynchronous=True),
TraceScenario(
name="async-anthropic-chat-bridge",
fixture=_anthropic_bridge_fixture,
asynchronous=True,
),
TraceScenario(
name="async-anthropic-chat-bridge-stream",
fixture=_anthropic_bridge_stream_fixture,
asynchronous=True,
),
),
)

57
tests/rust-python-harness/strategies/trace_parity/sdk/test_core_scenario_matrix.py
View file

@ -1,57 +0,0 @@
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")
ocr: Final = _suite("tests.rust-python-harness.strategies.trace_parity.sdk.ocr.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 ocr.scenarios} >= {
("async-cohere", True),
("sync-vertex-deepseek", False),
("async-vertex-deepseek", True),
}
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),
}

67
tests/rust-python-harness/strategies/trace_parity/sdk/transcription/case.py
View file

@ -1,67 +0,0 @@
from __future__ import annotations
import base64
import io
import json
import wave
from typing import Final
from .....shared.parity.recorded_http import HttpHeader, RecordedHttpResponse
from ...models import RouteFixture, RouteSpec, TraceScenario, TraceSuite
def _audio_bytes() -> bytes:
with io.BytesIO() as buffer:
with wave.open(buffer, "wb") as audio:
audio.setnchannels(1)
audio.setsampwidth(2)
audio.setframerate(16000)
audio.writeframes(b"\x00\x00" * 1600)
return buffer.getvalue()
def _fixture(_base_url: str) -> RouteFixture:
credentials: Final = {
"aws_access_key_id": "test-access",
"aws_secret_access_key": "test-secret",
"aws_region_name": "us-east-1",
}
audio: Final = _audio_bytes()
payload: Final = {"file": ("sample.wav", audio, "audio/wav"), **credentials}
response: Final = json.dumps(
{
"output": {"message": {"role": "assistant", "content": [{"text": "hello"}]}},
"stopReason": "end_turn",
"usage": {"inputTokens": 2, "outputTokens": 3, "totalTokens": 5},
}
).encode()
return RouteFixture(
kwargs={"model": "bedrock/mistral.voxtral-mini-3b-2507", **payload},
provider_responses=(
RecordedHttpResponse.from_bytes(
200, (HttpHeader(name="content-type", value="application/json"),), response
),
),
)
SPEC: Final = RouteSpec(
"transcription",
("transcription", "atranscription"),
_fixture,
)
TRACE_SUITE: Final = TraceSuite(
route=SPEC,
scenarios=(
TraceScenario(
name="sync-bedrock",
fixture=_fixture,
asynchronous=False,
),
TraceScenario(
name="async-bedrock",
fixture=_fixture,
asynchronous=True,
),
),
)

137
tests/rust-python-harness/strategies/trace_parity/test_reporting.py
View file

@ -1,137 +0,0 @@
from __future__ import annotations
from typing import Final
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
from . import reporting
from .reporting import TRACE_ARTIFACT, TraceArtifact, render_trace_results
def _result(trace: TraceArtifact) -> CaseResult:
case: Final = HarnessCase(
strategy_id="trace_parity",
strategy_label="Trace parity",
sdk_function=trace.sdk_function,
spec=ModuleCaseSpec(coverage=Coverage.PARTIAL, module="example"),
surface=trace.surface,
)
result: Final = CaseResult(case=case)
nodeid: Final = f"trace:{trace.surface}:{trace.sdk_function}:{trace.scenario}"
result.collected.add(nodeid)
result.record(nodeid, RunStatus.PASSED, artifacts=(ResultArtifact(TRACE_ARTIFACT, trace.model_dump_json()),))
return result
def _trace(
python: tuple[PipelineStep, ...],
*,
python_error: str | None = None,
scenario: str = "sync-default",
) -> TraceArtifact:
return TraceArtifact.from_traces(
surface="sdk",
sdk_function="ocr",
scenario=scenario,
python=python,
python_error=python_error,
)
def _events(*items: tuple[str, int, str | None]) -> tuple[PipelineStep, ...]:
parents: dict[int, int] = {}
steps: list[PipelineStep] = []
for event_id, (span, depth, raw) in enumerate(items):
parent_id = parents.get(depth - 1) if depth else None
steps.append(PipelineStep(event_id, parent_id, span, raw if raw is not None else span))
parents[depth] = event_id
return tuple(steps)
def test_renderer_prints_the_python_trace() -> None:
python: Final = _events(
("ocr", 0, "ocr/main.py:88 aocr"),
("python_prepare", 1, "prep.py:1 python_prepare"),
)
section: Final = render_trace_results((_result(_trace(python)),))[0]
report: Final = "\n\n".join(section.blocks)
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" not in report
def test_renderer_keeps_collected_trace_when_python_errors() -> None:
python: Final = _events(("ocr", 0, "ocr/main.py:88 aocr"))
report: Final = "\n\n".join(
render_trace_results((_result(_trace(python, python_error="python: replay server closed")),))[0].blocks
)
assert "PYTHON (1 steps)\n1 aocr (ocr/main.py:88)" in report
assert "Python error: python: replay server closed" in report
def test_unavailable_trace_reports_scenario_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:async-error")
result.record("trace:sdk:ocr:async-error", RunStatus.ERROR)
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 report.count("Case: ocr") == 1
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:
events: Final = _events(("ocr", 0, "ocr/main.py:88 aocr"))
monkeypatch.setattr(reporting.sys.stdout, "isatty", lambda: True)
monkeypatch.delenv("NO_COLOR", raising=False)
report: Final = "\n\n".join(render_trace_results((_result(_trace(events)),))[0].blocks)
assert "\033[36mPYTHON\033[0m (1 steps)" in report
assert "\033[36m1 aocr (ocr/main.py:88)\033[0m" in report
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(_trace(())), gateway_result))
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)

242
tests/rust-python-harness/strategies/trace_parity/test_runner.py
View file

@ -1,242 +0,0 @@
from __future__ import annotations
import importlib
import os
from pathlib import Path
from types import SimpleNamespace
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.profiler import FunctionTraceEvent
from ...shared.tracing.steps import PipelineStep
from .models import RouteFixture, RouteSpec, TraceScenario, TraceSuite
from .reporting import TraceArtifact
from .runner import run_trace_cases, run_trace_scenario, scenario_nodeids, validate_trace_suite
from .sdk.execution import SdkCall, collect_trace, execute_trace
def _fixture(_base_url: str) -> RouteFixture:
return RouteFixture(kwargs={}, provider_responses=())
def _case(*, surface: Surface = "sdk", function: SdkFunction = "ocr") -> HarnessCase:
return HarnessCase(
strategy_id="trace_parity",
strategy_label="Trace parity",
sdk_function=function,
spec=ModuleCaseSpec(coverage=Coverage.PARTIAL, module="example"),
surface=surface,
)
def test_scenario_filtering_and_occurrence_node_ids() -> None:
suite: Final = TraceSuite(
route=RouteSpec("ocr", ("ocr", "aocr"), _fixture),
scenarios=(
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({"async-two"}))
assert tuple(nodeid for _, nodeid in nodes) == ("trace:sdk:ocr:async-two",)
def test_runner_arguments_select_scenarios(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[frozenset[str]] = []
def capture_case(
_run: HarnessRun,
_case: HarnessCase,
scenarios: frozenset[str],
_on_update: object,
) -> None:
selected.append(scenarios)
monkeypatch.setattr(runner, "_run_case", capture_case)
exit_code, _ = run_trace_cases((case,), tmp_path, lambda _: None, ("mistral",))
assert exit_code == 0
assert selected == [frozenset({"mistral"})]
def test_python_trace_preserves_native_ocr_dispatch_setting(monkeypatch: pytest.MonkeyPatch) -> None:
execution: Final = importlib.import_module("tests.rust-python-harness.strategies.trace_parity.sdk.execution")
route: Final = RouteSpec("ocr", ("ocr", "aocr"), _fixture)
observed: list[str | None] = []
def collect(
_function: SdkCall,
_fixture: RouteFixture,
*,
asynchronous: bool,
) -> SimpleNamespace:
observed.append(os.environ.get("LITELLM_RUST"))
return SimpleNamespace(
events=(FunctionTraceEvent(0, None, "aocr" if asynchronous else "ocr"),),
error=None,
)
monkeypatch.setattr(execution, "_collect", collect)
monkeypatch.setenv("LITELLM_RUST", "0")
collect_trace(route, asynchronous=False)
monkeypatch.setenv("LITELLM_RUST", "1")
collect_trace(route, asynchronous=True)
assert observed == ["0", "1"]
assert os.environ["LITELLM_RUST"] == "1"
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)
result: Final = execute_trace(suite.route, scenario, "sdk")
assert result.python_error is None
assert "Give Feedback / Get Help" not in capsys.readouterr().out
assert litellm.suppress_debug_info is False
@pytest.mark.parametrize("asynchronous", (False, True))
def test_vertex_trace_keeps_unmapped_helpers_and_parents(asynchronous: bool) -> None:
loaded: Final = importlib.import_module("tests.rust-python-harness.strategies.trace_parity.sdk.ocr.case")
suite: Final = cast(TraceSuite, loaded.TRACE_SUITE)
name: Final = f"{'async' if asynchronous else 'sync'}-vertex-deepseek"
scenario: Final = next(item for item in suite.scenarios if item.name == name)
trace: Final = execute_trace(suite.route, scenario, "sdk")
assert trace.python_error is None
url: Final = next(
event for event in trace.python if event.raw.endswith(" VertexAIDeepSeekOCRConfig.get_complete_url")
)
project: Final = next(
event for event in trace.python if event.raw.endswith(" VertexBase.safe_get_vertex_ai_project")
)
location: Final = next(
event for event in trace.python if event.raw.endswith(" VertexBase.safe_get_vertex_ai_location")
)
assert project.parent_id == location.parent_id == url.id
assert not any(event.raw.endswith(" VertexBase.get_access_token") for event in trace.python)
@pytest.mark.parametrize("asynchronous", (False, True))
def test_vertex_credentials_trace_runs_real_auth_helpers(asynchronous: bool, monkeypatch: pytest.MonkeyPatch) -> None:
loaded: Final = importlib.import_module("tests.rust-python-harness.strategies.trace_parity.sdk.ocr.case")
suite: Final = cast(TraceSuite, loaded.TRACE_SUITE)
name: Final = f"{'async' if asynchronous else 'sync'}-vertex-deepseek-credentials"
scenario: Final = next(item for item in suite.scenarios if item.name == name)
monkeypatch.setenv("VERTEXAI_CREDENTIALS", "original-credentials")
monkeypatch.setenv("VERTEX_AI_API_KEY", "original-api-key")
trace: Final = execute_trace(suite.route, scenario, "sdk")
assert trace.python_error is None
validate: Final = next(
event for event in trace.python if event.raw.endswith(" VertexAIDeepSeekOCRConfig.validate_environment")
)
helpers: Final = (
"VertexBase.safe_get_vertex_ai_project",
"VertexBase.safe_get_vertex_ai_credentials",
"VertexBase.get_access_token",
)
assert tuple(event.raw.split(" ", 1)[1] for event in trace.python if event.parent_id == validate.id) == helpers
token: Final = next(event for event in trace.python if event.raw.endswith(" VertexBase.get_access_token"))
load: Final = next(event for event in trace.python if event.raw.endswith(" VertexBase.load_auth"))
refresh: Final = next(event for event in trace.python if event.raw.endswith(" VertexBase.refresh_auth"))
assert load.parent_id == token.id
assert refresh.parent_id == load.id
assert os.environ["VERTEXAI_CREDENTIALS"] == "original-credentials"
assert os.environ["VERTEX_AI_API_KEY"] == "original-api-key"
def test_scenario_validation_rejects_duplicate_and_unsafe_names() -> None:
route: Final = RouteSpec("ocr", ("ocr", "aocr"), _fixture)
duplicate: Final = TraceSuite(
route=route,
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),))
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_names_and_route_registration() -> None:
invalid_name: Final = TraceSuite(
route=RouteSpec("ocr", ("ocr", "aocr"), _fixture),
scenarios=(TraceScenario("bedrock", _fixture, asynchronous=True),),
)
wrong_function: Final = TraceSuite(
route=RouteSpec("messages", ("create", "acreate"), _fixture),
scenarios=(TraceScenario("sync-one", _fixture, asynchronous=False),),
)
wrong_surface: Final = TraceSuite(
route=RouteSpec("ocr", ("ocr", "aocr"), _fixture),
scenarios=(TraceScenario("sync-one", _fixture, asynchronous=False),),
)
case: Final = _case()
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 "requires the sdk surface" in (validate_trace_suite(wrong_surface, _case(surface="gateway")) or "")
def test_invalid_route_dispatch_records_harness_error() -> None:
case: Final = _case()
run: Final = HarnessRun.from_cases((case,))
result: Final = run.results[case.key]
suite: Final = TraceSuite(
route=RouteSpec("ocr", ("ocr", "aocr"), _fixture),
scenarios=(TraceScenario("sync-one", _fixture, asynchronous=False),),
)
nodeid: Final = "trace:gateway:ocr:sync-one"
run_trace_scenario(run, result, suite, suite.scenarios[0], "gateway", nodeid, lambda _: None)
assert result.outcomes[nodeid] is RunStatus.ERROR
assert run.failures == [(nodeid, "trace scenarios only run on the sdk surface")]
def test_python_trace_without_errors_passes(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"), _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"),),
)
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 == []

24
tests/rust-python-harness/strategies/unit_tests_parity/__init__.py
View file

@ -19,20 +19,22 @@ from ...shared.unit_runners.suite_runner import run_suites
from .reporting import render_unit_parity_results
from .runner import UnitParityExclusion, UnitParitySuite, run_suite
UNIT_PARITY_SUITES: Final[Mapping[SdkFunction, UnitParitySuite]] = MappingProxyType(
{
sdk_function: UnitParitySuite(
python_selectors=contract.unit_parity.python_selectors,
exclusions=tuple(
UnitParityExclusion(
nodeid=exclusion.nodeid,
reason=exclusion.reason,
)
for exclusion in contract.unit_parity.exclusions
"ocr": UnitParitySuite(
python_selectors=(
"tests/test_litellm/llms/azure_ai/test_azure_document_intelligence_ocr_transformation.py",
"tests/test_litellm/llms/mistral/ocr",
"tests/test_litellm/llms/ocr",
"tests/test_litellm/ocr",
),
)
for sdk_function, contract in UNIT_TEST_CONTRACTS.items()
exclusions=(
UnitParityExclusion(
nodeid="tests/test_litellm/ocr/test_rust_bridge.py::test_rust_toggles_flag",
reason="This test asserts the process-level backend flag selected by the parity runner.",
),
),
),
}
)

2
tests/rust-python-harness/strategies/unit_tests_rust/AGENTS.md
View file

@ -1 +1 @@
Runs the focused native Cargo test suite for each mapped API.
Runs the focused native Cargo test suite for each registered API.

10
tests/rust-python-harness/strategies/unit_tests_rust/__init__.py
View file

@ -18,16 +18,8 @@ from ...shared.unit_runners.suite_runner import run_suites
from .reporting import render_rust_unit_results
from .runner import RustSuite, run_suite
RUST_SUITES: Final[Mapping[SdkFunction, RustSuite]] = MappingProxyType(
{
sdk_function: RustSuite(
cargo_manifest=contract.rust.cargo_manifest,
cargo_filter=contract.rust.cargo_filter,
cargo_package=contract.rust.cargo_package,
)
for sdk_function, contract in UNIT_TEST_CONTRACTS.items()
}
{"ocr": RustSuite(cargo_manifest="litellm-rust/Cargo.toml", cargo_filter="ocr")}
)

16
tests/test_rust_python_harness.py
View file

@ -1,6 +1,7 @@
from __future__ import annotations
import importlib
from pathlib import Path
from typing import Final
import pytest
@ -8,10 +9,9 @@ import pytest
models = importlib.import_module("tests.rust-python-harness.shared.reporting.models")
strategy_module = importlib.import_module("tests.rust-python-harness.shared.reporting.strategy")
ui = importlib.import_module("tests.rust-python-harness.shared.reporting.ui")
contracts = importlib.import_module("tests.rust-python-harness.shared.unit_runners.contracts")
cli = importlib.import_module("tests.rust-python-harness.cli")
UNIT_TEST_CONTRACTS = contracts.UNIT_TEST_CONTRACTS
REPO_ROOT = Path(__file__).resolve().parents[1]
CaseResult = models.CaseResult
Coverage = models.Coverage
HarnessCase = models.HarnessCase
@ -37,10 +37,6 @@ def _case(module: str = "tests.example") -> HarnessCase:
"module",
[
"tests.rust-python-harness.strategies.e2e_parity.sdk.ocr.test_sdk_parity",
"tests.rust-python-harness.strategies.trace_parity.sdk.ocr.case",
"tests.rust-python-harness.strategies.trace_parity.sdk.messages.case",
"tests.rust-python-harness.strategies.trace_parity.sdk.chat_completions.case",
"tests.rust-python-harness.strategies.trace_parity.sdk.transcription.case",
],
)
def test_implemented_namespace_case_modules_remain_importable(module: str) -> None:
@ -108,14 +104,10 @@ def test_should_format_developer_facing_run_context() -> None:
assert _format_duration(1.25) == "1.2s"
def test_should_leave_functions_without_unit_test_contracts_unimplemented() -> None:
assert "messages" not in UNIT_TEST_CONTRACTS
def test_strategy_subcommand_accepts_function_filter(capsys: pytest.CaptureFixture[str]) -> None:
exit_code: Final = cli.main(["run", "unit_tests_rust", "--function", "messages"])
exit_code: Final = cli.main(["run", "unit_tests_parity", "--function", "messages"])
captured: Final = capsys.readouterr()
assert exit_code == 0
assert "- messages: not_implemented" in captured.out
assert "unit_tests_rust:messages: not_implemented" not in captured.out
assert "unit_tests_parity:messages: not_implemented" not in captured.out