mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-24 00:52:24 +00:00
c37fe46534
* test(vcr): guard leaked cassette patches and make injected-transport embedding tests immune * test(vcr): derive the leak guard's patch points from vcrpy's own reset list * test(vcr): share CapturingTransport and switch the encoding_format embedding test to it --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
25 lines
926 B
Python
25 lines
926 B
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from typing import Final
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, TypeAdapter
|
|
|
|
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
|
|
|
|
|
|
class CapturingTransport(httpx.AsyncBaseTransport, httpx.BaseTransport):
|
|
def __init__(self, response: BaseModel) -> None:
|
|
self._response: Final = response
|
|
self.request_bodies: tuple[Mapping[str, object], ...] = ()
|
|
|
|
def handle_request(self, request: httpx.Request) -> httpx.Response:
|
|
return self._respond(request.read())
|
|
|
|
async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
|
|
return self._respond(await request.aread())
|
|
|
|
def _respond(self, body: bytes) -> httpx.Response:
|
|
self.request_bodies = (*self.request_bodies, _JSON_OBJECT.validate_json(body))
|
|
return httpx.Response(200, json=self._response.model_dump(mode="json"))
|