mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-07 08:26:10 +00:00
198906495f
* refactor(python-bridge): split non-streaming bridge modules * refactor(python-bridge): bring shared function tracing into route layer * feat(dev): list Python route functions and call sites * feat(dev): list Rust route functions and call sites * docs(dev): record OCR parity gaps across Python and Rust * feat(dev): list executed SDK calls with runtime tracing * feat(dev): report Python vs Rust SDK pipeline steps in one CLI * feat(dev): side-by-side pipeline step report in compare CLI * fix(dev): drop invalid Final annotations in compare cell loop * feat(dev): blue python-only and yellow rust-only steps in compare CLI * feat(dev): vertical layout with section spacing in compare CLI * fix(dev): validate SDK trace stages across sync and async routes * refactor(rust): align SDK route call structure with Python * refactor(python-bridge): share sync and async route call wrappers * refactor(dev): split compare CLI into fixtures, runtime, and report modules * fix(ci): run SDK trace tests and satisfy test lint
33 lines
1.4 KiB
Python
33 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import ExitStack
|
|
from typing import Final
|
|
from urllib.error import HTTPError
|
|
from urllib.request import Request, urlopen
|
|
|
|
import pytest
|
|
|
|
from tests.sdk_function_trace.mock_provider import MockProviderResponse, mock_provider
|
|
|
|
|
|
def test_mock_provider_preserves_error_response() -> None:
|
|
response: Final = MockProviderResponse(429, (("retry-after", "2"),), b'{"error":"rate limited"}')
|
|
with mock_provider(response) as api_base:
|
|
with pytest.raises(HTTPError) as error:
|
|
urlopen(Request(api_base, data=b"{}"), timeout=5)
|
|
with error.value as received:
|
|
assert received.code == 429
|
|
assert received.headers["retry-after"] == "2"
|
|
assert received.read() == response.body
|
|
|
|
|
|
@pytest.mark.parametrize("request_count", [0, 2])
|
|
def test_mock_provider_rejects_missing_or_duplicate_requests(request_count: int) -> None:
|
|
response: Final = MockProviderResponse(200, (), b"{}")
|
|
with ExitStack() as stack:
|
|
api_base: Final = stack.enter_context(mock_provider(response))
|
|
for _ in range(request_count):
|
|
with urlopen(Request(api_base, data=b"{}"), timeout=5) as received:
|
|
assert received.read() == response.body
|
|
with pytest.raises(AssertionError, match=f"expected one provider request, received {request_count}"):
|
|
stack.close()
|