test(messages): add rust vs python messages parity characterization suite (#49610)

Runs every case through the Python /v1/messages path and the Rust path against one recording upstream and compares dispatch, upstream request, outcome, hidden params, standard logging fields and callback kwargs. Known Rust differences are strict xfails tagged with their inventory ID.

tests/rust-python-harness/strategies/e2e_parity/__init__.py

@@ -20,7 +20,7 @@ CASES: Final[tuple[CaseDefinition, ...]] = (
     CaseDefinition(
         "messages",
         NotImplementedCaseSpec(
-            reason="Bridge unit tests exist, but no standalone end-to-end parity case is registered."
+            reason="Parity runs as the pytest suite in sdk/messages/test_native_parity.py, not as a runner case."
         ),
         surface="sdk",
     ),

tests/rust-python-harness/strategies/e2e_parity/sdk/messages/conftest.py

@@ -0,0 +1,34 @@
+from collections.abc import AsyncIterator, Generator
+from importlib import import_module
+from pathlib import Path
+from typing import Final
+
+import pytest
+import pytest_asyncio
+
+from tests.test_litellm_rust.support.isolation import isolated_test_state
+from tests.test_litellm_rust.support.recording_server import RecordingServer, recording_service
+
+PARITY_ROOT: Final = Path(__file__).parent
+
+
+@pytest_asyncio.fixture(autouse=True, loop_scope="function")
+async def isolate_test_state() -> AsyncIterator[None]:
+    async with isolated_test_state():
+        yield
+
+
+@pytest.fixture
+def recording_server() -> Generator[RecordingServer]:
+    with recording_service() as server:
+        yield server
+
+
+def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
+    try:
+        import_module("litellm.rust_bridge._native")
+    except ImportError:
+        skip: Final = pytest.mark.skip(reason="requires a compiled litellm.rust_bridge._native extension")
+        for item in items:
+            if PARITY_ROOT in item.path.parents:
+                item.add_marker(skip)

tests/test_litellm_rust/support/recording_server.py

@@ -25,6 +25,7 @@ class ResponseSpec:
     status: int = 200
     headers: dict[str, str] = field(default_factory=dict)
     delay: float = 0
+    chunk_delay: float = 0
     events: tuple[tuple[str, object], ...] = ()
 
     def payloads(self) -> tuple[bytes, ...]:
@@ -89,7 +90,9 @@ def recording_service() -> Iterator[RecordingServer]:
                 self.send_header(name, value)
             self.end_headers()
             try:
-                for payload in payloads:
+                for index, payload in enumerate(payloads):
+                    if index and response.chunk_delay:
+                        time.sleep(response.chunk_delay)
                     self.wfile.write(payload)
                     self.wfile.flush()
             except (BrokenPipeError, ConnectionResetError):