scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-04 02:31:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 69

test: keep provider request snapshots immutable

Browse source
This commit is contained in:
Yuneng Jiang 2026-09-14 17:27:18 -07:00
parent 51db940514
commit 37bde0bdbe
No known key found for this signature in database
1 changed files with 11 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
tests/e2e/test_provider_edge.py
View file

@ -31,6 +31,7 @@ from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from types import MappingProxyType
from typing import Final
import pytest
@ -85,7 +86,10 @@ class _FakeProvider(ThreadingHTTPServer):
super().__init__(bind, _FakeProviderHandler)
self.hits: list[str] = []
self.echo_request = echo_request
self.requests: list[tuple[dict[str, str], bytes]] = []
self.requests: tuple[tuple[Mapping[str, str], bytes], ...] = ()
def capture_request(self, headers: Mapping[str, str], body: bytes) -> None:
self.requests = (*self.requests, (MappingProxyType(dict(headers)), body))
class _FakeProviderHandler(BaseHTTPRequestHandler):
@ -103,8 +107,12 @@ class _FakeProviderHandler(BaseHTTPRequestHandler):
length = int(self.headers.get("content-length") or "0")
body = self.rfile.read(length) if length else b""
provider.hits.append(f"{self.command} {self.path}")
provider.requests.append((dict(self.headers.items()), body))
payload = json.dumps({"echo": body.decode("utf-8"), "path": self.path, "hit": len(provider.hits)} if provider.echo_request else {"ok": True}).encode()
provider.capture_request(dict(self.headers.items()), body)
payload: Final = json.dumps(
{"echo": body.decode("utf-8"), "path": self.path, "hit": len(provider.hits)}
if provider.echo_request
else {"ok": True}
).encode()
self.send_response(200)
self.send_header("content-type", "application/json")
self.send_header("content-length", str(len(payload)))