diff --git a/tests/e2e/proxy_client.py b/tests/e2e/proxy_client.py index 5050b6fce68..961db4590e2 100644 --- a/tests/e2e/proxy_client.py +++ b/tests/e2e/proxy_client.py @@ -443,14 +443,16 @@ class ProxyClient: # ---- spend read-back ------------------------------------------------ - def spend_logs(self, params: SpendLogsParams) -> list[SpendLogRow]: - result = self.transport.get( + def _read_spend_logs(self, params: SpendLogsParams) -> Result[SpendLogs]: + return self.transport.get( "/spend/logs", headers=self.transport.master, params=params, response_type=SpendLogs, ) - match result: + + def spend_logs(self, params: SpendLogsParams) -> list[SpendLogRow]: + match self._read_spend_logs(params): case Success(data=logs): return logs.root case _: @@ -481,7 +483,7 @@ class ProxyClient: def poll_logs_for_key( self, key: str, *, min_rows: int = 1, predicate: RowsPredicate | None = None ) -> list[SpendLogRow]: - return self._poll(lambda: self.spend_logs(SpendLogsParams(api_key=key)), min_rows, predicate) + return self._poll(lambda: self._read_spend_logs(SpendLogsParams(api_key=key)), min_rows, predicate) def poll_logs_for_request_id( self, @@ -491,24 +493,47 @@ class ProxyClient: predicate: RowsPredicate | None = None, ) -> list[SpendLogRow]: return self._poll( - lambda: self.spend_logs(SpendLogsParams(request_id=request_id)), + lambda: self._read_spend_logs(SpendLogsParams(request_id=request_id)), min_rows, predicate, ) def _poll( self, - fetch: Callable[[], list[SpendLogRow]], + fetch: Callable[[], Result[SpendLogs]], min_rows: int, predicate: RowsPredicate | None, ) -> list[SpendLogRow]: + """Poll /spend/logs to the deadline, returning the last rows the proxy served. + + A read that never succeeded is not evidence the rows are absent. /spend/logs + 500s and times out under load (see test_spend_logs_endpoint_returns_spend), and + counting those as "no rows yet" makes an unreadable endpoint indistinguishable + from a spend row that was never written - the caller then asserts an empty list + and blames the write path for a failure on the read path. A single successful + read anywhere in the window is enough to trust the emptiness, so a transient + blip still polls through; a window where every read failed raises instead. + """ deadline = time.monotonic() + self.poll_timeout rows: list[SpendLogRow] = [] + last_failure: Result[SpendLogs] | None = None + served = False while time.monotonic() < deadline: - rows = fetch() - if len(rows) >= min_rows and (predicate is None or predicate(rows)): - return rows + match fetch(): + case Success(data=logs): + served = True + rows = logs.root + if len(rows) >= min_rows and (predicate is None or predicate(rows)): + return rows + case failure: + last_failure = failure time.sleep(self.poll_interval) + if not served and last_failure is not None: + raise AssertionError( + f"/spend/logs never returned a successful read in {self.poll_timeout}s of " + f"polling, so an empty result says nothing about the spend rows. " + f"Last read: {last_failure}" + ) return rows # ---- route probe ---------------------------------------------------- diff --git a/tests/e2e/quota_management/spend_tracking/test_spend_tracking_e2e.py b/tests/e2e/quota_management/spend_tracking/test_spend_tracking_e2e.py index 6a0032981fd..d5e014f4737 100644 --- a/tests/e2e/quota_management/spend_tracking/test_spend_tracking_e2e.py +++ b/tests/e2e/quota_management/spend_tracking/test_spend_tracking_e2e.py @@ -106,9 +106,26 @@ def test_streaming_chat_completion_tracks_spend( f"count to three {unique_marker()}", max_tokens=64, ) + # A stream commits its HTTP 200 before the upstream completes, so result.ok alone + # says nothing about what the stream did: a break mid-flight comes back as a 200 + # carrying an SSE error frame and no [DONE] (async_data_generator's except arm). + # Judge the stream on its own terms here, or a broken stream is only noticed + # E2E_POLL_TIMEOUT later as a missing spend row, blaming the wrong subsystem. assert ( result.ok ), f"stream failed (status {result.status_code}): {result.body[:300]}" + assert result.is_streaming, ( + f"expected an SSE stream from /chat/completions, got content-type " + f"{result.content_type!r}" + ) + assert result.chunks > 0, "no SSE events were consumed from the stream" + assert ( + result.stream_error is None + ), f"the stream carried an error event: {result.stream_error}" + assert result.stream_done, ( + f"the stream never terminated with [DONE], so it broke mid-flight: " + f"chunks={result.chunks} last_events={result.stream_events[-2:]}" + ) rows = client.poll_logs_for_key( scoped_key, predicate=lambda rs: any((r.spend or 0) > 0 for r in rs) diff --git a/tests/e2e/test_proxy_client.py b/tests/e2e/test_proxy_client.py new file mode 100644 index 00000000000..ae7135d2094 --- /dev/null +++ b/tests/e2e/test_proxy_client.py @@ -0,0 +1,195 @@ +"""Harness coverage for ProxyClient's spend-log poll contract. + +No proxy needed and no ``e2e`` marker: this pins how the poller reports an +unreadable /spend/logs. The poller is what every spend, budget and logging +suite reads its assertions off, so conflating "the endpoint answered and there +are no rows" with "the endpoint never answered" mislabels a read-path outage as +missing spend, and the resulting failure message accuses the wrong subsystem +120s after the fact. The transport is injected, so nothing here talks HTTP or +monkeypatches anything. +""" + +from __future__ import annotations + +from collections.abc import Iterator, Sequence +from dataclasses import dataclass + +import pytest +from pydantic import BaseModel + +from e2e_http import ( + AuthHeaders, + BinaryStream, + ProbeResult, + Result, + StreamingResponse, + Success, + UnknownApiError, +) +from models import SpendLogRow, SpendLogs +from proxy_client import ProxyClient + + +class _UnusedTransport: + """Every Transport member the poll path does not touch. + + ProxyClient depends on the whole Transport Protocol, so a fake must satisfy + all of it to type-check. Raising keeps the fake honest: a test that reaches + one of these is exercising something it did not set up. + """ + + def post[R: BaseModel](self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R]) -> Result[R]: + raise NotImplementedError(path) + + def stream(self, path: str, *, headers: BaseModel, json: BaseModel) -> StreamingResponse: + raise NotImplementedError(path) + + def stream_binary( + self, path: str, *, headers: BaseModel, json: BaseModel, chunk_size: int = 8192 + ) -> BinaryStream: + raise NotImplementedError(path) + + def send( + self, + path: str, + *, + headers: BaseModel, + json: BaseModel, + params: BaseModel | None = None, + stream: bool = False, + ) -> StreamingResponse: + raise NotImplementedError(path) + + def delete[R: BaseModel]( + self, + path: str, + *, + headers: BaseModel, + json: BaseModel, + response_type: type[R], + params: BaseModel | None = None, + ) -> Result[R]: + raise NotImplementedError(path) + + def patch[R: BaseModel]( + self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R] + ) -> Result[R]: + raise NotImplementedError(path) + + def put[R: BaseModel]( + self, path: str, *, headers: BaseModel, json: BaseModel, response_type: type[R] + ) -> Result[R]: + raise NotImplementedError(path) + + def probe(self, path: str, *, params: BaseModel) -> ProbeResult: + raise NotImplementedError(path) + + def upload[R: BaseModel]( + self, + path: str, + *, + headers: BaseModel, + form: BaseModel, + filename: str, + content: bytes, + file_content_type: str = "application/jsonl", + file_field: str = "file", + params: BaseModel | None = None, + response_type: type[R], + ) -> Result[R]: + raise NotImplementedError(path) + + def download(self, path: str, *, headers: BaseModel) -> StreamingResponse: + raise NotImplementedError(path) + + def bearer(self, key: str) -> AuthHeaders: + raise NotImplementedError(key) + + @property + def master(self) -> AuthHeaders: + return AuthHeaders(authorization="Bearer sk-fake-master") + + +@dataclass +class ScriptedReads(_UnusedTransport): + """Serves one scripted /spend/logs outcome per GET, then repeats the last.""" + + reads: Sequence[Result[SpendLogs]] + calls: int = 0 + + def get[R: BaseModel]( + self, + path: str, + *, + headers: BaseModel, + params: BaseModel, + response_type: type[R], + timeout: float | None = None, + ) -> Result[R]: + assert path == "/spend/logs", f"fake only serves /spend/logs, got {path}" + outcome = self.reads[min(self.calls, len(self.reads) - 1)] + self.calls += 1 + return outcome # pyright: ignore[reportReturnType] # scripted per-test as Result[SpendLogs] + + +def _row(request_id: str, spend: float) -> SpendLogRow: + return SpendLogRow(request_id=request_id, spend=spend) + + +def _served(rows: Iterator[SpendLogRow] | list[SpendLogRow]) -> Success[SpendLogs]: + return Success(status_code=200, data=SpendLogs(list(rows))) + + +def _client(reads: Sequence[Result[SpendLogs]]) -> tuple[ProxyClient, ScriptedReads]: + transport = ScriptedReads(reads=reads) + # Short budget with no sleep: the contract under test is which outcome the + # window produces, not how long it waits. + return ProxyClient(transport=transport, poll_timeout=0.05, poll_interval=0.0), transport + + +class TestSpendLogPollDistinguishesUnreadableFromEmpty: + def test_rows_are_returned_once_the_predicate_holds(self) -> None: + rows = [_row("r1", 0.01)] + client, transport = _client([_served([]), _served(rows)]) + + polled = client.poll_logs_for_key("sk-test", predicate=lambda rs: any((r.spend or 0) > 0 for r in rs)) + + assert [r.request_id for r in polled] == ["r1"] + assert transport.calls == 2, "poller must keep reading until the predicate holds" + + def test_a_window_of_only_failed_reads_raises_instead_of_reporting_no_rows(self) -> None: + """The regression: 24 consecutive 500s used to come back as an empty list, so + the caller asserted "no spend row was written" for a request whose row it had + never been able to read.""" + client, _ = _client([UnknownApiError(status_code=500, body='{"error":"deadlock detected"}')]) + + with pytest.raises(AssertionError) as failure: + client.poll_logs_for_key("sk-test") + + message = str(failure.value) + assert "never returned a successful read" in message + assert "500" in message and "deadlock detected" in message, ( + f"the failure must carry the read that failed, not just its own verdict: {message}" + ) + + def test_a_readable_endpoint_with_no_rows_still_reports_no_rows(self) -> None: + """The other side of the same contract: emptiness the proxy actually served is + a real answer about the spend pipeline and must reach the caller unchanged.""" + client, _ = _client([_served([])]) + + assert client.poll_logs_for_key("sk-test") == [] + + def test_a_transient_read_failure_is_polled_through(self) -> None: + rows = [_row("r1", 0.02)] + client, transport = _client([UnknownApiError(status_code=500, body="blip"), _served(rows)]) + + polled = client.poll_logs_for_key("sk-test") + + assert [r.request_id for r in polled] == ["r1"] + assert transport.calls == 2, "one failed read must not end the poll" + + def test_request_id_polling_shares_the_contract(self) -> None: + client, _ = _client([UnknownApiError(status_code=503, body="upstream unavailable")]) + + with pytest.raises(AssertionError, match="never returned a successful read"): + client.poll_logs_for_request_id("chatcmpl-abc")