From 1bf95121d3b069c206e1dfc6369cb4d73364299a Mon Sep 17 00:00:00 2001 From: kerry Date: Fri, 2 Oct 2026 23:03:15 +0000 Subject: [PATCH] test(spend): derive SDK user agents and strengthen audit assertions Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../spend/test_spend_log_request_tags.py | 48 ++++++++++++++----- .../test_spend_log_request_tags_chaos.py | 11 ++++- 2 files changed, 46 insertions(+), 13 deletions(-) diff --git a/tests/integration/spend/test_spend_log_request_tags.py b/tests/integration/spend/test_spend_log_request_tags.py index e064d9f6496..0c8ab3a6817 100644 --- a/tests/integration/spend/test_spend_log_request_tags.py +++ b/tests/integration/spend/test_spend_log_request_tags.py @@ -13,7 +13,7 @@ import pytest import yaml from integration._support.client import Gateway, eventually from integration._support.database import read_rows -from integration._support.process import owned_proxy +from integration._support.process import owned_proxy, owned_proxy_process from integration._support.wire import Reply, Request, wire_server from integration.spend._request_tag_helpers import ( GEMINI_MODEL, @@ -115,7 +115,7 @@ def test_pass_through_anthropic_sdk_records_header_tags(gateway: Gateway, tmp_pa assert message.id.startswith("msg_") assert len(wire.drain()) == 1 assert eventually(lambda: tags_by_key(key), lambda tags: len(tags) == 1, seconds=70) == [ - ["User-Agent: Anthropic", "User-Agent: Anthropic/Python 0.84.0", TENANT_TAG] + ["User-Agent: Anthropic", f"User-Agent: Anthropic/Python {anthropic.__version__}", TENANT_TAG] ] @@ -137,7 +137,7 @@ def test_pass_through_anthropic_sdk_stream_records_header_tags(gateway: Gateway, assert message.id.startswith("msg_") assert len(wire.drain()) == 1 assert eventually(lambda: tags_by_key(key), lambda tags: len(tags) == 1, seconds=70) == [ - ["User-Agent: Anthropic", "User-Agent: Anthropic/Python 0.84.0", TENANT_TAG] + ["User-Agent: Anthropic", f"User-Agent: Anthropic/Python {anthropic.__version__}", TENANT_TAG] ] @@ -191,7 +191,7 @@ def test_pass_through_openai_sdk_records_header_tags(gateway: Gateway, tmp_path: assert request_id.startswith("chatcmpl_") assert len(wire.drain()) == 1 assert eventually(lambda: tags_by_key(key), lambda tags: len(tags) == 1, seconds=70) == [ - ["User-Agent: OpenAI", "User-Agent: OpenAI/Python 2.33.0", TENANT_TAG] + ["User-Agent: OpenAI", f"User-Agent: OpenAI/Python {openai.__version__}", TENANT_TAG] ] @@ -284,7 +284,7 @@ def test_pass_through_openai_async_sdk_stream_records_header_tags(gateway: Gatew assert request_id.startswith("chatcmpl_") assert len(wire.drain()) == 1 assert eventually(lambda: tags_by_key(key), lambda tags: len(tags) == 1, seconds=70) == [ - ["User-Agent: AsyncOpenAI", "User-Agent: AsyncOpenAI/Python 2.33.0", TENANT_TAG] + ["User-Agent: AsyncOpenAI", f"User-Agent: AsyncOpenAI/Python {openai.__version__}", TENANT_TAG] ] @@ -903,9 +903,9 @@ def test_repeated_requests_each_record_tags(gateway: Gateway, tmp_path: Path, ro model=f"anthropic/{MODEL}", api_base=wire.url, api_key="synthetic-anthropic-key" ) key: Final = scenario.key() - ids: Final = [] - for _ in range(3): - response: Final = candidate.request( + + def repeat(_: int): + return candidate.request( "POST", route, { @@ -916,8 +916,12 @@ def test_repeated_requests_each_record_tags(gateway: Gateway, tmp_path: Path, ro key=key, headers={**SENT_HEADERS, "anthropic-version": "2023-06-01"}, ) - assert response.status_code == 200, response.text - ids.append(response.json()["id"]) + + responses: Final = tuple(repeat(index) for index in range(3)) + assert all(response.status_code == 200 for response in responses), [ + (response.status_code, response.text) for response in responses + ] + ids: Final = tuple(response.json()["id"] for response in responses) assert len(wire.drain()) == 3 for identity in ids: assert eventually( @@ -959,9 +963,10 @@ def test_guardrail_mode_tag_decider_is_unchanged_on_pass_through(gateway: Gatewa }, ) with ( - owned_proxy(gateway, tmp_path, provider_env(wire.url), config=config, workers=2) as candidate, - candidate.scenario() as scenario, + owned_proxy_process(gateway, tmp_path, provider_env(wire.url), config=config, workers=2) as owned, + owned.gateway.scenario() as scenario, ): + candidate: Final = owned.gateway model: Final = scenario.model(model=f"openai/{OPENAI_MODEL}", api_base=f"{wire.url}/v1") key: Final = scenario.key(allowed_passthrough_routes=["/custom-anthropic"]) body: Final = {"model": MODEL, "max_tokens": 16, "messages": [{"role": "user", "content": "bananablock"}]} @@ -977,3 +982,22 @@ def test_guardrail_mode_tag_decider_is_unchanged_on_pass_through(gateway: Gatewa ) assert control.status_code != 200, control.text assert len(wire.drain()) == 0, "tag-matched guardrail should have blocked before the upstream" + call_id: Final = control.headers["x-litellm-call-id"] + + def blocked_rows() -> list[dict]: + return read_rows( + 'SELECT metadata FROM "LiteLLM_SpendLogs" WHERE litellm_call_id=%s', + (call_id,), + ) + + spend_row: Final = eventually( + blocked_rows, + lambda rows: len(rows) == 1 and "bananablock" in json.dumps(rows[0]["metadata"]), + seconds=30, + return_last_on_timeout=True, + ) + if not (len(spend_row) == 1 and "bananablock" in json.dumps(spend_row[0]["metadata"])): + log_text: Final = owned.log.read_text() + assert "Content blocked: keyword 'bananablock' detected" in log_text, ( + "guardrail block text not in spend row or proxy log" + ) diff --git a/tests/integration/spend/test_spend_log_request_tags_chaos.py b/tests/integration/spend/test_spend_log_request_tags_chaos.py index 9dd86518ddb..7b0b627c915 100644 --- a/tests/integration/spend/test_spend_log_request_tags_chaos.py +++ b/tests/integration/spend/test_spend_log_request_tags_chaos.py @@ -150,9 +150,11 @@ def test_burst_across_routes_records_tags_once_per_response(gateway: Gateway, tm def test_sink_outage_does_not_lose_spend_log_tags(gateway: Gateway, tmp_path: Path) -> None: down: Final = threading.Event() delivered: Final = [] # mutable-ok: sink thread appends between drains + rejected: Final = [] # mutable-ok: sink thread appends between drains def stoppable_sink(request: Request) -> Reply: if down.is_set(): + rejected.append(request) return Reply(status=503) delivered.append(request) return Reply() @@ -214,13 +216,20 @@ def test_sink_outage_does_not_lose_spend_log_tags(gateway: Gateway, tmp_path: Pa second: Final = [ response for group in pool.map(lambda i: burst(100 + i), range(3)) for response in group ] + second_ids: Final = call_ids(second) + outage_probe: Final = eventually( + lambda: (len(rejected), events_for(second_ids)), + lambda state: state[0] >= 1 and state[1] == set(), + seconds=30, + ) + assert outage_probe[0] >= 1, "sink saw no rejection during the outage window" + assert outage_probe[1] == set(), "burst-2 event delivered to a down sink" down.clear() third: Final = _tagged_requests(candidate, key, anthropic_model, openai_model, False, 200) responses: Final = [*first, *second, *third] assert all(response.status_code == 200 for response in responses), [ (response.status_code, response.text[:200]) for response in responses ] - second_ids: Final = call_ids(second) third_ids: Final = call_ids(list(third)) recovery_probe: Final = next(iter(third_ids)) eventually(