scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

test(e2e): add logging e2e coverage (s3_v2, gcs_bucket, team langfuse callback, datadog failure) (#38552)

Browse source 
* test: add logging e2e coverage (s3_v2, gcs_bucket, team langfuse callback, datadog failure)

Five new live e2e scenarios raising Logging & Guardrails registry coverage:
s3_v2 success and failure objects read back from the real S3 bucket,
gcs_bucket success record read back through the GCS JSON API (with
nextPageToken pagination and per-request bearer minting), team-scoped
Langfuse callback delivery with non-team isolation, and DataDog failure
event delivery queried by indexed model_group. datadog_reader gains
query-based variants of the marker search; the langfuse cell is a new
registry row. Bucket readers settle past a full flush interval so a
late duplicate cannot hide from the exactly-one assertions

* test: cover clock-skew day prefix in gcs read-back and retry team callback propagation

* test: key the s3 failure read-back on the provider error, not payload absence

* chore: rerun ci

* chore: rerun ci after config sync

* chore: rerun ci with pr lane env

* chore: rerun ci

* chore: rerun ci

* chore: rerun ci

* chore: rerun ci

* chore: rerun ci

* test: add guardrail e2e coverage (presidio masking, bedrock post and during call, moderation on messages) (#38553)

* test: add guardrail e2e coverage (presidio masking, bedrock post/during, moderation on messages)

* test: require the phone placeholder positively in the presidio masking predicate

* test: count only the 400 verdict body as a bedrock post_call block

* test(e2e): exempt the guardrail config echo from the post_call leak assertion

* test(e2e): pin the fail-closed contract for an unknown guardrail name (skipped, product gap)

* test(e2e): tolerate the readiness 503 from a transient db blip in the callback-config probes
This commit is contained in:
yucheng-berri 2026-08-29 09:43:44 -07:00 • committed by GitHub
parent b2fe72e70c
commit f0fadb7f99
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 1374 additions and 108 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
tests/e2e/coverage_registry/guardrail.yaml
View file

@ -32,3 +32,4 @@
- {id: guardrail.mcp_security.pre_call.blocks, module: guardrail, tier: P2, hook_point: pre_call, assertions: [blocks], exercised_on: [mcp_operations], source: "guardrail_hooks/mcp_security", rationale: "MCP protocol security"}
- {id: guardrail.llm_as_a_judge.pre_call.blocks, module: guardrail, tier: P2, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "guardrail_hooks/llm_as_a_judge", rationale: "LLM-based judgment guardrail"}
- {id: guardrail.litellm_content_filter.pre_mcp_call.blocks, module: guardrail, tier: P1, hook_point: pre_mcp_call, assertions: [blocks], exercised_on: [mcp_operations], source: "guardrail_hooks/litellm_content_filter/content_filter.py:_scan_mcp_tool_call_arguments", rationale: "A general content-filter guardrail configured mode=pre_mcp_call blocks a banned keyword in an MCP tool call's arguments before it reaches the upstream MCP server; a clean argument passes"}
- {id: guardrail.dispatch.pre_call.rejects_unknown_name, module: guardrail, tier: P1, hook_point: pre_call, assertions: [blocks], exercised_on: [chat_completions], source: "proxy guardrail dispatch (per-request `guardrails` selector)", rationale: "A request naming a guardrail this proxy does not serve must fail closed with a 4xx; today it is silently served unguarded, so a typo'd name drops the protection the caller asked for"}

1
tests/e2e/coverage_registry/logging.yaml
View file

@ -24,3 +24,4 @@
- {id: logging.focus.success.writes_object, module: logging, tier: P1, event: success, assertions: [writes_object], exercised_on: [chat_completions, messages], source: "integrations/focus/focus_logger.py", rationale: "Cost mgmt multi-destination export"}
- {id: logging.niche_integrations.success.logs_spend, module: logging, tier: P2, event: success, assertions: [logs_spend], exercised_on: [chat_completions], source: grammar, rationale: "SMOKE cohort: athina/galileo/deepeval/langtrace/weave/lunary/humanloop/traceloop/helicone/argilla/newrelic/sqs/supabase/dynamodb/agentops/lago/etc"}
- {id: logging.niche_integrations.failure.logs_spend, module: logging, tier: P2, event: failure, assertions: [logs_spend], exercised_on: [chat_completions], source: grammar, rationale: "SMOKE niche failure path"}
- {id: logging.langfuse.success.logs_spend, module: logging, tier: P1, event: success, assertions: [logs_spend], exercised_on: [chat_completions], source: "integrations/langfuse/langfuse_otel.py", rationale: "Team-scoped Langfuse delivery via /team/callback; LangChain-ecosystem evals spend"}

64
tests/e2e/guardrails/guardrails_client.py
View file

@ -68,11 +68,27 @@ class BlockCodeExecutionParamsBody(GuardrailParamsBase):
guardrail: Literal["block_code_execution"] = "block_code_execution"
class PresidioParamsBody(GuardrailParamsBase):
"""Presidio PII guardrail params. `presidio_filter_scope="input"` keeps the
registration to a single callback on the configured mode; the default
("both") also registers a second post_call output-masking callback, which a
pre_call- or logging_only-scoped test must not drag in. `output_parse_pii`
stays unset/False: True would unmask the response back to the caller."""
guardrail: Literal["presidio"] = "presidio"
presidio_analyzer_api_base: str
presidio_anonymizer_api_base: str
presidio_filter_scope: Literal["input", "output", "both"] | None = None
presidio_language: str | None = None
output_parse_pii: bool | None = None
GuardrailParamsBody = (
ContentFilterParamsBody
| BedrockGuardrailParamsBody
| OpenAIModerationParamsBody
| BlockCodeExecutionParamsBody
| PresidioParamsBody
)
@ -253,6 +269,30 @@ class GuardrailsClient:
),
)
def chat_stream_raw(
self,
key: str,
model: str,
text: str,
*,
guardrails: list[str] | None = None,
max_tokens: int = 64,
) -> StreamingResponse:
"""Drive /chat/completions with stream=true, returning the raw HTTP
outcome (status, headers, SSE events) via the shared ProxyClient stream
sender - a streamed guardrail block is judged on status and stream
shape, not a typed body."""
return self.proxy.chat_stream(
key,
ChatBody(
model=model,
messages=[ChatMessage(role="user", content=text)],
max_tokens=max_tokens,
stream=True,
guardrails=guardrails,
),
)
def messages(
self,
key: str,
@ -318,7 +358,7 @@ def build_client(proxy: ProxyClient) -> GuardrailsClient:
return GuardrailsClient(proxy=proxy)
def poll_until_blocked(call: Callable[[], Result[ChatResponse]]) -> Result[ChatResponse]:
def poll_until_blocked[R: BaseModel](call: Callable[[], Result[R]]) -> Result[R]:
"""Retry a call that a guardrail should reject until it is, returning the last result.
Registering a guardrail is a control-plane write; the data-plane worker that
@ -337,3 +377,25 @@ def poll_until_blocked(call: Callable[[], Result[ChatResponse]]) -> Result[ChatR
time.sleep(POLL_INTERVAL)
last = call()
return last
#: Statuses a stream poll keeps retrying through instead of returning as "the
#: block": network failures (-1), key propagation (401), rate limits (429) -
#: transient rig noise, not a guardrail verdict.
_TRANSIENT_STREAM_STATUSES = frozenset({-1, 401, 429})
def poll_until_blocked_stream(call: Callable[[], StreamingResponse]) -> StreamingResponse:
"""poll_until_blocked for raw/streamed sends, which return a StreamingResponse
instead of a Result: retry while the call still succeeds (the data-plane worker
has not picked the new guardrail up yet) or fails with a transient status,
returning the first guardrail-shaped non-2xx outcome or the last result at
the deadline."""
deadline = time.monotonic() + POLL_TIMEOUT
last = call()
while time.monotonic() < deadline:
if not last.ok and last.status_code not in _TRANSIENT_STREAM_STATUSES:
return last
time.sleep(POLL_INTERVAL)
last = call()
return last

92
tests/e2e/guardrails/test_bedrock_guardrail_e2e.py
View file

@ -1,9 +1,12 @@
"""Live e2e: Bedrock ApplyGuardrail pre_call blocks denied input on chat.
"""Live e2e: Bedrock ApplyGuardrail blocks on chat, pre_call and post_call.
Registers a default-on bedrock guardrail via POST /guardrails with identifier/
pre_call registers a bedrock guardrail via POST /guardrails with identifier/
version from env, then sends a prompt the guardrail's configured policy denies.
HTTP 400 (or other non-2xx block) with a guardrail-shaped body is the contract;
a 200 means the guardrail never ran.
a 200 means the guardrail never ran. post_call scans the MODEL OUTPUT only, so
its test makes the model echo the word the guardrail's word policy denies
(BEDROCK_GUARDRAIL_BLOCKED_WORD, default FORBIDDENWORD) and the block must
arrive without leaking the model's text.
No AWS keys are passed: the gateway signs ApplyGuardrail with its own
pod-identity role, since the static AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY
@ -12,17 +15,37 @@ env vars are deliberately absent from the gateway (they hijack RDS IAM auth).
from __future__ import annotations
import json
import os
from typing import Final
import pytest
from e2e_config import unique_marker
from e2e_http import UnknownApiError
from guardrails_client import GuardrailsClient, poll_until_blocked
from guardrails_client import (
BedrockGuardrailParamsBody,
GuardrailsClient,
poll_until_blocked,
)
from lifecycle import ResourceManager
from pydantic import JsonValue, TypeAdapter
pytestmark = pytest.mark.e2e
_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
def _without_assessments(value: JsonValue) -> JsonValue:
"""The assessments echo guardrail CONFIG, not content: the stage guardrail's
topic policy is itself named after the denied word, so its label lands in
every assessment listing and would trip a leak check aimed at model output."""
if isinstance(value, dict):
return {key: _without_assessments(child) for key, child in value.items() if key != "assessments"}
if isinstance(value, list):
return [_without_assessments(item) for item in value]
return value
MODEL = "gemini-2.5-flash"
# Matches the word/topic policy the guardrail this suite points at actually denies.
# Content filters are not assumed: the guardrail resource carries no contentPolicy,
@ -42,23 +65,17 @@ class TestBedrockGuardrail:
version = os.environ["BEDROCK_GUARDRAIL_VERSION"]
name = f"e2e-bedrock-guard-{unique_marker()}"
guardrail_id = client.create_bedrock_guardrail(
name, identifier=identifier, version=version
)
guardrail_id = client.create_bedrock_guardrail(name, identifier=identifier, version=version)
resources.defer(lambda: client.delete_guardrail(guardrail_id))
# Selected per request rather than registered default_on, so an upstream
# ApplyGuardrail failure surfaces here instead of 403ing every other suite
# running against this proxy.
result = poll_until_blocked(
lambda: client.chat(scoped_key, MODEL, BLOCKED_PROMPT, guardrails=[name])
)
result = poll_until_blocked(lambda: client.chat(scoped_key, MODEL, BLOCKED_PROMPT, guardrails=[name]))
match result:
case UnknownApiError(status_code=status, body=body):
assert status in {400, 403}, (
f"expected a guardrail block status, got {status}: {body[:400]}"
)
assert status in {400, 403}, f"expected a guardrail block status, got {status}: {body[:400]}"
body_lower = body.lower()
assert any(
token in body_lower
@ -72,6 +89,49 @@ class TestBedrockGuardrail:
)
), f"block body should name the guardrail reason; got: {body[:400]}"
case _:
pytest.fail(
f"bedrock default-on guardrail did not block harmful prompt; got {result}"
pytest.fail(f"bedrock default-on guardrail did not block harmful prompt; got {result}")
@pytest.mark.covers(
"guardrail.bedrock.post_call.blocks",
exercised_on=["chat_completions"],
)
def test_bedrock_post_call_blocks_denied_model_output(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
identifier = os.environ["BEDROCK_GUARDRAIL_IDENTIFIER"]
version = os.environ["BEDROCK_GUARDRAIL_VERSION"]
blocked_word = os.environ.get("BEDROCK_GUARDRAIL_BLOCKED_WORD", "FORBIDDENWORD")
name = f"e2e-bedrock-post-{unique_marker()}"
guardrail_id = client.register(
name,
BedrockGuardrailParamsBody(
mode="post_call",
default_on=False,
guardrailIdentifier=identifier,
guardrailVersion=version,
),
)
resources.defer(lambda: client.delete_guardrail(guardrail_id))
# post_call scans OUTPUT only, so the denied word has to come out of the
# model: ask it to echo the word verbatim. The word in the prompt itself
# is not scanned in this mode.
prompt = f"Reply with exactly this one word and nothing else: {blocked_word}"
result = poll_until_blocked(lambda: client.chat(scoped_key, MODEL, prompt, guardrails=[name], max_tokens=128))
match result:
case UnknownApiError(status_code=status, body=body):
# A policy block is a 400 naming the verdict; a failed
# ApplyGuardrail call surfaces as 403 "guardrail request
# failed", which must not count as a block.
assert status == 400, f"expected the guardrail block status 400, got {status}: {body[:400]}"
body_lower = body.lower()
assert any(token in body_lower for token in ("violated", "blocked", "intervened")), (
f"block body should name the guardrail verdict; got: {body[:400]}"
)
assert blocked_word not in json.dumps(_without_assessments(_JSON.validate_json(body))), (
f"the blocked model output must not leak into the error body; got: {body[:400]}"
)
case _:
pytest.fail(f"bedrock post_call guardrail did not block denied model output; got {result}")

41
tests/e2e/guardrails/test_guardrail_dispatch_e2e.py Normal file
View file

@ -0,0 +1,41 @@
"""Live e2e: the per-request `guardrails` selector must fail closed.
A request that names a guardrail is a caller asking for protection. When the
proxy does not serve that name (a typo, a deleted guardrail, or a worker that
never loaded it), answering 200 silently drops the protection the caller asked
for; the contract this test pins is a 4xx naming the unknown guardrail.
"""
from __future__ import annotations
import pytest
from e2e_config import unique_marker
from e2e_http import UnknownApiError, ValidationError
from guardrails_client import GuardrailsClient
pytestmark = pytest.mark.e2e
MODEL = "gemini-2.5-flash"
@pytest.mark.skip(
reason=(
"stage red: product gap, a request naming a guardrail the proxy does not "
"serve is silently served unguarded (200) instead of failing closed"
)
)
@pytest.mark.covers(
"guardrail.dispatch.pre_call.rejects_unknown_name",
exercised_on=["chat_completions"],
)
def test_request_naming_an_unknown_guardrail_fails_closed(client: GuardrailsClient, scoped_key: str) -> None:
result = client.chat(scoped_key, MODEL, "say hi", guardrails=[f"e2e-no-such-guardrail-{unique_marker()}"])
match result:
case UnknownApiError(status_code=status, body=body):
assert status == 400, f"expected a 400 for an unknown guardrail name, got {status}: {body[:400]}"
assert "guardrail" in body.lower(), f"the rejection should name the guardrail; got: {body[:400]}"
case ValidationError(message=message):
assert "guardrail" in message.lower(), f"the rejection should name the guardrail; got: {message[:400]}"
case _:
pytest.fail(f"a request naming an unknown guardrail must fail closed with a 4xx; got {result}")

47
tests/e2e/guardrails/test_openai_moderation_guardrail_e2e.py
View file

@ -7,7 +7,9 @@ before the upstream model runs; a prompt that trips the policy must be rejected
with HTTP 400 naming the moderation policy, and the same guardrail must let a
benign prompt through. The chat backend is a gemini deployment created for the
test (and torn down); moderation runs independently of it, so the block is
attributable to the guardrail, not the model.
attributable to the guardrail, not the model. The same pre_call contract is
also exercised through /v1/messages (Anthropic format): a flagged prompt is
rejected with a 400 naming moderation and a benign one passes.
"""
from __future__ import annotations
@ -69,3 +71,46 @@ class TestOpenAIModerationGuardrail:
"the same moderation guardrail must let a benign prompt through, but the "
f"call returned no choices: {allowed}"
)
@pytest.mark.covers(
"guardrail.openai_moderations.pre_call.blocks",
exercised_on=["messages"],
)
def test_moderation_blocks_flagged_input_on_messages(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
model = client.create_backend_model(resources, prefix="e2e-moderation-msg-backend")
name = f"e2e-openai-moderation-msg-{unique_marker()}"
guardrail_id = client.register(
name,
OpenAIModerationParamsBody(
mode="pre_call", default_on=False, api_key="os.environ/OPENAI_API_KEY"
),
)
resources.defer(lambda: client.delete_guardrail(guardrail_id))
blocked = poll_until_blocked(
lambda: client.messages(scoped_key, model, FLAGGED_PROMPT, guardrails=[name])
)
match blocked:
case UnknownApiError(status_code=400, body=body):
assert "moderation" in body.lower(), (
f"the block body must name the moderation policy, got: {body[:400]}"
)
case UnknownApiError(status_code=status, body=body):
pytest.fail(
f"expected a 400 moderation block on /v1/messages, got {status}: {body[:400]}"
)
case _:
pytest.fail(
f"openai moderation did not block a flagged /v1/messages prompt; got {blocked}"
)
allowed = unwrap(
client.messages(scoped_key, model, BENIGN_PROMPT, guardrails=[name], max_tokens=64)
)
assert allowed.content or allowed.choices, (
"the same moderation guardrail must let a benign /v1/messages prompt through, but "
f"the response carried neither content nor choices: {allowed}"
)

184
tests/e2e/guardrails/test_presidio_masking_e2e.py Normal file
View file

@ -0,0 +1,184 @@
"""Live e2e: the Presidio PII guardrail masks, per its configured hook point.
pre_call: the guardrail calls the Presidio analyzer/anonymizer on the request
messages BEFORE the model runs, so the model only ever sees placeholders like
<EMAIL_ADDRESS>. A prompt asking the model to repeat a fake email + phone back
must come back with the placeholders echoed and the raw PII absent, on
/chat/completions and on /v1/messages (Anthropic format).
The analyzer/anonymizer endpoints come from PRESIDIO_ANALYZER_API_BASE /
PRESIDIO_ANONYMIZER_API_BASE; missing env is a hard failure, never a skip.
Each guardrail registers with presidio_filter_scope="input" so only the
configured hook's callback exists (the default "both" adds a second post_call
output masker), and is deleted on teardown.
"""
from __future__ import annotations
import os
import time
from collections.abc import Callable
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import Result, Success
from guardrails_client import GuardrailsClient, PresidioParamsBody
from lifecycle import ResourceManager
from models import AnthropicMessagesResponse, ChatResponse
pytestmark = pytest.mark.e2e
MODEL = "gemini-2.5-flash"
# A guardrail created via POST /guardrails reaches the worker that served the
# create immediately, but every other worker only picks it up on its next
# periodic DB sync (~30s), so the first requests can be served unguarded.
GUARDRAIL_PROPAGATION_DEADLINE_SECONDS = 40.0
GUARDRAIL_PROPAGATION_POLL_INTERVAL_SECONDS = 5.0
# Presidio's anonymizer replaces a detected entity with its unnumbered type
# placeholder, e.g. <EMAIL_ADDRESS>. The pre_call assertions match on the bare
# token because the model is echoing the masked prompt and may not preserve the
# angle brackets; the logged payload keeps the placeholder verbatim.
MASKED_EMAIL_TOKEN = "EMAIL_ADDRESS"
MASKED_PHONE_TOKEN = "PHONE_NUMBER"
# Fictional NANP 555 number; a standard format Presidio's phone recognizer detects.
FAKE_PHONE = "+1 415-555-0134"
def _presidio_bases() -> tuple[str, str]:
analyzer = os.environ.get("PRESIDIO_ANALYZER_API_BASE", "").strip()
anonymizer = os.environ.get("PRESIDIO_ANONYMIZER_API_BASE", "").strip()
if not analyzer or not anonymizer:
pytest.fail(
"Presidio e2e requires PRESIDIO_ANALYZER_API_BASE and PRESIDIO_ANONYMIZER_API_BASE "
"(the running Presidio analyzer/anonymizer services); missing env is a hard failure, not a skip"
)
return analyzer, anonymizer
def _register_presidio(
client: GuardrailsClient,
resources: ResourceManager,
*,
name: str,
) -> None:
analyzer, anonymizer = _presidio_bases()
guardrail_id = client.register(
name,
PresidioParamsBody(
mode="pre_call",
default_on=False,
presidio_analyzer_api_base=analyzer,
presidio_anonymizer_api_base=anonymizer,
presidio_filter_scope="input",
),
)
resources.defer(lambda: client.delete_guardrail(guardrail_id))
def _fake_email() -> str:
return f"jane.doe.{unique_marker()}@example.com"
def _pii_prompt(marker: str, email: str) -> str:
return (
f"{marker} Repeat this sentence back to me exactly, word for word: "
f"My email address is {email} and my phone number is {FAKE_PHONE}."
)
def _first_content(response: ChatResponse) -> str:
if not response.choices:
return ""
message = response.choices[0].message
return (message.content if message else None) or ""
def _messages_text(response: AnthropicMessagesResponse) -> str:
"""The text of a /v1/messages answer, whichever shape the proxy produced
(Anthropic-native content blocks or OpenAI-normalized choices)."""
parts: list[str] = []
for block in response.content or []:
if block.text:
parts.append(block.text)
for choice in response.choices or []:
if choice.message and choice.message.content:
parts.append(choice.message.content)
return "\n".join(parts)
def _assert_eventually_masked[R: BaseModel](
fetch: Callable[[], Result[R]], extract: Callable[[R], str], *, email: str
) -> None:
"""Retry the call until the response comes back masked, to the propagation
deadline. An unmasked early response is in-flight guardrail propagation, not
a failure, and neither is a transient non-Success (a replica that has not
reloaded the guardrail answers 404, the live model can rate-limit) - only a
response that still carries the raw PII at the deadline is."""
deadline = time.monotonic() + GUARDRAIL_PROPAGATION_DEADLINE_SECONDS
last: str = "<no successful response yet>"
while True:
result = fetch()
match result:
case Success(data=data):
content = extract(data)
last = content
masked = MASKED_EMAIL_TOKEN in content and MASKED_PHONE_TOKEN in content and email not in content
if masked:
assert FAKE_PHONE not in content, (
f"the raw phone number must be masked before the model sees it, but the "
f"response echoed it: {content[:300]!r}"
)
return
case _:
last = f"<non-Success result: {result}>"
if time.monotonic() >= deadline:
pytest.fail(
f"presidio pre_call guardrail never masked the PII within "
f"{GUARDRAIL_PROPAGATION_DEADLINE_SECONDS}s; last observation: {last[:300]!r}"
)
time.sleep(GUARDRAIL_PROPAGATION_POLL_INTERVAL_SECONDS)
class TestPresidioPreCallMasking:
@pytest.mark.covers(
"guardrail.presidio.pre_call.masks",
exercised_on=["chat_completions"],
)
def test_pre_call_masks_pii_on_chat_completions(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
name = f"e2e-presidio-pre-chat-{unique_marker()}"
_register_presidio(client, resources, name=name)
email = _fake_email()
prompt = _pii_prompt(unique_marker(), email)
_assert_eventually_masked(
lambda: client.chat(scoped_key, MODEL, prompt, guardrails=[name], max_tokens=128),
_first_content,
email=email,
)
@pytest.mark.covers(
"guardrail.presidio.pre_call.masks",
exercised_on=["messages"],
)
def test_pre_call_masks_pii_on_messages(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
name = f"e2e-presidio-pre-msg-{unique_marker()}"
_register_presidio(client, resources, name=name)
email = _fake_email()
prompt = _pii_prompt(unique_marker(), email)
_assert_eventually_masked(
lambda: client.messages(scoped_key, MODEL, prompt, guardrails=[name], max_tokens=128),
_messages_text,
email=email,
)

87
tests/e2e/guardrails/test_streaming_guardrail_e2e.py Normal file
View file

@ -0,0 +1,87 @@
"""Live e2e: a Bedrock guardrail in during_call mode blocks a streamed chat.
during_call runs the Bedrock ApplyGuardrail INPUT scan in an asyncio.gather
alongside the LLM call (common_request_processing.py); when the scan flags the
prompt, the raised block cancels the LLM task before the stream ever starts, so
the client sees a non-2xx JSON error - not an SSE stream, not an in-stream
error frame - and zero content chunks are delivered.
The prompt deliberately contains the exact word the guardrail's word policy
denies (BEDROCK_GUARDRAIL_BLOCKED_WORD, default FORBIDDENWORD), so the INPUT
scan intervenes deterministically. Identifier/version come from
BEDROCK_GUARDRAIL_IDENTIFIER / BEDROCK_GUARDRAIL_VERSION like the rest of the
bedrock suite; no AWS keys are passed (the gateway signs with pod identity).
The guardrail registers default_on=False and is selected per request, so an
upstream ApplyGuardrail failure surfaces here instead of 403ing other suites.
"""
from __future__ import annotations
import os
import pytest
from e2e_config import unique_marker
from guardrails_client import (
BedrockGuardrailParamsBody,
GuardrailsClient,
poll_until_blocked_stream,
)
from lifecycle import ResourceManager
pytestmark = pytest.mark.e2e
MODEL = "gemini-2.5-flash"
class TestBedrockDuringCallStreaming:
@pytest.mark.covers(
"guardrail.bedrock.during.blocks",
exercised_on=["chat_completions"],
)
def test_during_call_blocks_stream_before_first_chunk(
self, client: GuardrailsClient, resources: ResourceManager, scoped_key: str
) -> None:
identifier = os.environ["BEDROCK_GUARDRAIL_IDENTIFIER"]
version = os.environ["BEDROCK_GUARDRAIL_VERSION"]
blocked_word = os.environ.get("BEDROCK_GUARDRAIL_BLOCKED_WORD", "FORBIDDENWORD")
name = f"e2e-bedrock-during-{unique_marker()}"
guardrail_id = client.register(
name,
BedrockGuardrailParamsBody(
mode="during_call",
default_on=False,
guardrailIdentifier=identifier,
guardrailVersion=version,
),
)
resources.defer(lambda: client.delete_guardrail(guardrail_id))
# The denied word sits in the INPUT: during_call scans the request
# messages while the model call runs, and the flag must win the race
# by cancelling the stream outright.
prompt = f"Please use the word {blocked_word} in a sentence."
result = poll_until_blocked_stream(
lambda: client.chat_stream_raw(scoped_key, MODEL, prompt, guardrails=[name], max_tokens=64)
)
assert not result.ok, (
f"the during_call guardrail never blocked the streamed request; got a "
f"{result.status_code} with {result.chunks} chunks"
)
assert result.status_code == 400, (
f"a during_call block surfaces as HTTP 400 before the stream starts, got "
f"{result.status_code}: {result.body[:400]}"
)
assert result.chunks == 0 and not result.stream_events, (
f"no content chunk may be delivered on a during_call block, but "
f"{result.chunks} chunks arrived: {result.stream_events[:3]}"
)
assert "text/event-stream" not in (result.content_type or ""), (
f"the block must be a JSON error response, not an SSE stream; got content-type {result.content_type!r}"
)
body_lower = result.body.lower()
assert any(token in body_lower for token in ("guardrail", "violated", "blocked", "bedrock", "intervened")), (
f"block body should name the guardrail reason; got: {result.body[:400]}"
)

4
tests/e2e/logging/conftest.py
View file

@ -47,6 +47,4 @@ def dd_logs() -> DdLogsReader:
def datadog_creds() -> None:
"""Require Datadog shipping credentials. Hard-fail when absent; never skip."""
if not (os.getenv("DD_API_KEY") and os.getenv("DD_SITE")):
pytest.fail(
"Datadog e2e requires DD_API_KEY and DD_SITE; missing credentials is a hard failure, not a skip"
)
pytest.fail("Datadog e2e requires DD_API_KEY and DD_SITE; missing credentials is a hard failure, not a skip")

33
tests/e2e/logging/datadog_reader.py
View file

@ -97,15 +97,22 @@ class DdLogsReader:
indexed ``message`` empty, so a plain full-text query matches nothing;
``*:`` extends the scan to every attribute (the marker sits in the
prompt, e.g. ``messages.content``, wherever the route's payload puts
it). More than one hit for one call IS the duplicate-delivery bug, so
this never collapses to a single event. A 429 backs off and retries -
the search budget is org-wide, so another consumer can empty it under
us - while any other failure stays a hard fail."""
it)."""
return self.events_for_query(f"*:*{marker}*")
def events_for_query(self, query: str) -> list[DdLogEvent]:
"""Every ingested event the search query matches (failure payloads
carry no prompt to mark, so failure scenarios query indexed attributes
like ``@model_group:...`` instead of a body marker). More than one hit
for one call IS the duplicate-delivery bug, so this never collapses to
a single event. A 429 backs off and retries - the search budget is
org-wide, so another consumer can empty it under us - while any other
failure stays a hard fail."""
for _ in range(_RATE_LIMIT_RETRIES):
result = post(
URL(f"https://api.{self.site}/api/v2/logs/events/search"),
headers=_DdAuthHeaders(api_key=self.api_key, app_key=self.app_key),
json=_SearchRequest(filter=_SearchFilter(query=f"*:*{marker}*")),
json=_SearchRequest(filter=_SearchFilter(query=query)),
response_type=_SearchResponse,
timeout=30.0,
)
@ -123,6 +130,10 @@ class DdLogsReader:
)
def poll_events_for_marker(self, marker: str) -> list[DdLogEvent]:
"""``poll_events_for_query`` over the every-attribute marker scan."""
return self.poll_events_for_query(f"*:*{marker}*")
def poll_events_for_query(self, query: str) -> list[DdLogEvent]:
"""Poll until at least one matching event is searchable (the callback
flushes in periodic batches and DataDog ingestion adds seconds of lag),
then keep re-reading for DD_SETTLE_SECONDS so a late duplicate cannot
@ -132,15 +143,13 @@ class DdLogsReader:
request budget. At the deadline the last result is returned as-is."""
deadline = time.monotonic() + POLL_TIMEOUT
while time.monotonic() < deadline:
events = self.events_for_marker(marker)
events = self.events_for_query(query)
if events:
return self._settled_events_for_marker(marker, events)
return self._settled_events_for_query(query, events)
time.sleep(DD_SEARCH_INTERVAL)
return self.events_for_marker(marker)
return self.events_for_query(query)
def _settled_events_for_marker(
self, marker: str, events: list[DdLogEvent]
) -> list[DdLogEvent]:
def _settled_events_for_query(self, query: str, events: list[DdLogEvent]) -> list[DdLogEvent]:
"""Re-read at every search interval until the settle window closes; a
duplicate ends the watch early because more waiting cannot clear it.
@ -151,7 +160,7 @@ class DdLogsReader:
last_nonempty = events
while time.monotonic() < settle_deadline:
time.sleep(DD_SEARCH_INTERVAL)
latest = self.events_for_marker(marker)
latest = self.events_for_query(query)
if not latest:
continue
if len(latest) > 1:

220
tests/e2e/logging/gcs_reader.py Normal file
View file

@ -0,0 +1,220 @@
"""Read-back for the gcs_bucket logging test against the real GCS bucket.
The proxy ships StandardLoggingPayload objects with its own service account
(litellm_settings.callbacks: ["gcs_bucket"] + GCS_BUCKET_NAME), and the test
reads them back through the GCS JSON API. Auth is a self-signed service-account
JWT (RS256 via PyJWT + cryptography, both litellm proxy dependencies the
runner installs) minted per request and sent directly as the Bearer token -
Google accepts that for storage.googleapis.com with no token exchange, which
keeps every HTTP read inside ``e2e_http``.
The default gcs_bucket mode batches payloads into ``{date}/batch-{id}.ndjson``
objects; unbatched mode writes ``{date}/{response_id}`` per call. The reader
handles both: it polls the day's listing, downloads the direct object when
present, and otherwise scans batch objects fresh enough to hold the call.
Missing configuration is a hard failure, never a skip.
"""
from __future__ import annotations
import os
import time
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import quote
import jwt
import pytest
from pydantic import BaseModel, ConfigDict, Field
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT
from e2e_http import URL, Headers, probe
_GCS_API = "https://storage.googleapis.com"
#: Tolerance for clock skew between this host and GCS object timestamps.
_SKEW = timedelta(seconds=120)
#: How long to keep re-reading after the first match before trusting the
#: exactly-one assertion: past one full gcs_bucket flush interval (~20s), so
#: a duplicate shipped by a later flush is seen, plus listing-latency margin.
GCS_SETTLE_SECONDS = 45.0
class _ServiceAccount(BaseModel):
model_config = ConfigDict(extra="ignore")
client_email: str
private_key: str
class _GcsAuthHeaders(Headers):
authorization: str = Field(serialization_alias="Authorization")
class _GcsObject(BaseModel):
model_config = ConfigDict(extra="ignore")
name: str
updated: datetime | None = None
class _GcsListResponse(BaseModel):
model_config = ConfigDict(extra="ignore")
items: list[_GcsObject] = []
next_page_token: str | None = Field(default=None, validation_alias="nextPageToken")
class _GcsListParams(BaseModel):
prefix: str
max_results: int = Field(default=1000, serialization_alias="maxResults")
page_token: str | None = Field(default=None, serialization_alias="pageToken")
class _GcsMediaParams(BaseModel):
alt: str = "media"
class GcsLogRecord(BaseModel):
"""The StandardLoggingPayload fields the gcs scenario pins."""
model_config = ConfigDict(extra="ignore")
id: str
status: str
model_group: str | None = None
response_cost: float | None = None
total_tokens: int | None = None
error_str: str | None = None
def _mint_bearer(account: _ServiceAccount) -> str:
"""Self-signed service-account JWT: for Google APIs a token whose ``aud``
is the service endpoint authorizes directly, no oauth2 token exchange.
Minted per request so a long session never outlives one token's expiry."""
now = int(time.time())
claims: dict[str, str | int] = {
"iss": account.client_email,
"sub": account.client_email,
"aud": f"{_GCS_API}/",
"iat": now,
"exp": now + 3600,
}
return jwt.encode(claims, account.private_key, algorithm="RS256")
@dataclass(frozen=True, slots=True)
class GcsLogReader:
bucket: str
account: _ServiceAccount
def _headers(self) -> _GcsAuthHeaders:
return _GcsAuthHeaders(authorization=f"Bearer {_mint_bearer(self.account)}")
def _list(self, prefix: str) -> list[_GcsObject]:
"""Every object under ``prefix``, following ``nextPageToken`` - the
shared day prefix accumulates all of the proxy's traffic, and a fresh
record past the 1000-object page cap must still be seen."""
items: list[_GcsObject] = []
page_token: str | None = None
while True:
result = probe(
URL(f"{_GCS_API}/storage/v1/b/{self.bucket}/o"),
headers=self._headers(),
params=_GcsListParams(prefix=prefix, page_token=page_token),
)
if result.status_code != 200:
pytest.fail(
f"GCS object listing for gs://{self.bucket}/{prefix} failed "
f"({result.status_code}): {result.body[:300]}"
)
page = _GcsListResponse.model_validate_json(result.body)
items.extend(page.items)
page_token = page.next_page_token
if not page_token:
return items
def _download(self, name: str) -> str:
result = probe(
URL(f"{_GCS_API}/storage/v1/b/{self.bucket}/o/{quote(name, safe='')}"),
headers=self._headers(),
params=_GcsMediaParams(),
)
if result.status_code != 200:
pytest.fail(
f"GCS object download gs://{self.bucket}/{name} failed ({result.status_code}): {result.body[:300]}"
)
return result.body
def records_for_response_id(self, response_id: str, *, since: datetime) -> list[GcsLogRecord]:
"""Every payload written for ``response_id``: the direct
``{date}/{response_id}`` object plus any hit inside batch NDJSON
objects updated after ``since``. More than one hit is the
duplicate-delivery bug, so this never collapses to a single record."""
records: list[GcsLogRecord] = []
window_start = since - _SKEW
for day_offset in (-1, 0, 1):
day = (since + timedelta(days=day_offset)).strftime("%Y-%m-%d")
for obj in self._list(f"{day}/"):
if obj.name == f"{day}/{response_id}":
records.append(GcsLogRecord.model_validate_json(self._download(obj.name)))
continue
is_fresh_batch = f"{day}/batch-" in obj.name and obj.updated is not None and obj.updated >= window_start
if is_fresh_batch:
records.extend(
GcsLogRecord.model_validate_json(line)
for line in self._download(obj.name).splitlines()
if response_id in line
)
return records
def poll_records_for_response_id(self, response_id: str, *, since: datetime) -> list[GcsLogRecord]:
"""Poll until the payload is readable (the gcs_bucket callback flushes
on a ~20s timer), then keep re-reading for GCS_SETTLE_SECONDS - past a
full flush interval - so a duplicate shipped by a later flush cannot
hide from the exactly-one assertion. A duplicate ends the settle early
because more waiting cannot clear it."""
deadline = time.monotonic() + POLL_TIMEOUT
while time.monotonic() < deadline:
records = self.records_for_response_id(response_id, since=since)
if records:
return self._settled_records(response_id, since=since, first=records)
time.sleep(POLL_INTERVAL)
return []
def _settled_records(self, response_id: str, *, since: datetime, first: list[GcsLogRecord]) -> list[GcsLogRecord]:
"""Re-read at every poll interval until the settle window closes; a
transiently empty re-read never downgrades what was already seen."""
settle_deadline = time.monotonic() + GCS_SETTLE_SECONDS
latest = first
while time.monotonic() < settle_deadline and len(latest) <= 1:
time.sleep(POLL_INTERVAL)
latest = self.records_for_response_id(response_id, since=since) or latest
return latest
def utc_now() -> datetime:
return datetime.now(timezone.utc)
def build_gcs_reader() -> GcsLogReader:
bucket = os.environ.get("GCS_BUCKET_NAME", "")
if not bucket:
pytest.fail(
"GCS_BUCKET_NAME must be set: the gcs test reads the proxy's gcs_bucket "
"delivery back from the real bucket (the cluster secret manager injects "
"it; locally set it in tests/e2e/.env)"
)
raw = ""
credentials_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "")
if credentials_path and Path(credentials_path).is_file():
raw = Path(credentials_path).read_text()
else:
raw = os.environ.get("VERTEXAI_CREDENTIALS", "")
if not raw:
pytest.fail(
"GCS read-back needs a service-account key: set "
"GOOGLE_APPLICATION_CREDENTIALS (path) or VERTEXAI_CREDENTIALS (JSON), "
"as the cluster secret manager does"
)
return GcsLogReader(bucket=bucket, account=_ServiceAccount.model_validate_json(raw))

40
tests/e2e/logging/logging_client.py
View file

@ -480,12 +480,8 @@ class LoggingClient:
stream=True if stream else None,
)
if stream:
return self.proxy.transport.stream(
"/v1/messages", headers=self.proxy.transport.bearer(key), json=body
)
return self.proxy.transport.send(
"/v1/messages", headers=self.proxy.transport.bearer(key), json=body
)
return self.proxy.transport.stream("/v1/messages", headers=self.proxy.transport.bearer(key), json=body)
return self.proxy.transport.send("/v1/messages", headers=self.proxy.transport.bearer(key), json=body)
def responses_raw(
self, key: str, model: str, text: str, *, max_output_tokens: int = 64, stream: bool = False
@ -499,12 +495,8 @@ class LoggingClient:
model=model, input=text, max_output_tokens=max_output_tokens, stream=True if stream else None
)
if stream:
return self.proxy.transport.stream(
"/v1/responses", headers=self.proxy.transport.bearer(key), json=body
)
return self.proxy.transport.send(
"/v1/responses", headers=self.proxy.transport.bearer(key), json=body
)
return self.proxy.transport.stream("/v1/responses", headers=self.proxy.transport.bearer(key), json=body)
return self.proxy.transport.send("/v1/responses", headers=self.proxy.transport.bearer(key), json=body)
def scrape_metrics(self) -> str:
return self.proxy.probe("/metrics", params=NoBody()).body
@ -530,9 +522,7 @@ class LoggingClient:
return False
return True
rows = self.proxy.poll_logs_for_key(
key, min_rows=1, predicate=lambda rs: any(_matches(r) for r in rs)
)
rows = self.proxy.poll_logs_for_key(key, min_rows=1, predicate=lambda rs: any(_matches(r) for r in rs))
for row in rows:
if _matches(row):
return row
@ -593,9 +583,7 @@ class LoggingClient:
deadline = time.monotonic() + POLL_TIMEOUT
last: LangfuseObservation | None = None
while time.monotonic() < deadline:
last = self.find_langfuse_observation(
creds, key_alias=key_alias, prompt_marker=prompt_marker
)
last = self.find_langfuse_observation(creds, key_alias=key_alias, prompt_marker=prompt_marker)
if last is not None:
cost = observation_spend(last)
if not require_positive_cost or (cost is not None and cost > 0):
@ -611,9 +599,7 @@ class LoggingClient:
prompt_marker: str,
) -> list[LangfuseObservation]:
"""Generation plus any sibling/child observations (guardrail spans, etc.)."""
gen = self.poll_langfuse_observation(
creds, key_alias=key_alias, prompt_marker=prompt_marker
)
gen = self.poll_langfuse_observation(creds, key_alias=key_alias, prompt_marker=prompt_marker)
if gen is None or not gen.trace_id:
return [] if gen is None else [gen]
return self.list_langfuse_observations(creds, trace_id=gen.trace_id) or [gen]
@ -636,3 +622,15 @@ def first_ok(client: LoggingClient, send: Callable[[], StreamingResponse]) -> St
def build_logging_client(proxy: ProxyClient) -> LoggingClient:
return LoggingClient(proxy=proxy)
def readiness_details_body(client: LoggingClient) -> str:
"""/health/readiness/details, tolerating the 503 it serves while the
ephemeral stack's DB leg blips: the recorded state the logging suites check
here is the callback list, which the body carries either way."""
result = client.proxy.probe("/health/readiness/details", params=NoBody())
db_blip = result.status_code == 503 and '"db":"disconnected"' in result.body
assert result.status_code == 200 or db_blip, (
f"/health/readiness/details must answer 200, got {result.status_code}: {result.body[:300]}"
)
return result.body

115
tests/e2e/logging/s3_reader.py Normal file
View file

@ -0,0 +1,115 @@
"""Read-back for the s3 logging tests against the real S3 bucket the proxy
ships StandardLoggingPayload objects to (litellm_settings.callbacks: ["s3_v2"]).
Delivery is judged on what actually landed in the bucket: the proxy writes
with its own credentials exactly as in production, and the tests list and
download the objects back with boto3 (already a litellm proxy dependency, so
the e2e runner image carries it; it is an AWS SDK, not a raw HTTP client, so
the e2e_http-only transport rule is untouched). The bucket comes from
S3_LOGS_BUCKET_NAME - on the cluster the secret manager injects it, locally
tests/e2e/.env provides it. Missing configuration is a hard failure, never a
skip.
"""
from __future__ import annotations
import os
import time
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING
import boto3
import pytest
from pydantic import BaseModel, ConfigDict
from e2e_config import POLL_INTERVAL, POLL_TIMEOUT
if TYPE_CHECKING:
from types_boto3_s3.client import S3Client
#: How long to keep re-reading after the first match before trusting the
#: exactly-one assertion: past one full s3_v2 flush interval (~10s), so a
#: duplicate shipped by a LATER flush is seen, plus listing-latency margin.
#: The DataDog reader settles the same way (DD_SETTLE_SECONDS).
S3_SETTLE_SECONDS = 25.0
class S3LogRecord(BaseModel):
"""The StandardLoggingPayload fields the s3 scenarios pin."""
model_config = ConfigDict(extra="ignore")
id: str
status: str
model_group: str | None = None
response_cost: float | None = None
total_tokens: int | None = None
error_str: str | None = None
@dataclass(frozen=True, slots=True)
class S3LogReader:
bucket: str
client: S3Client
def list_keys(self, prefix: str) -> list[str]:
response = self.client.list_objects_v2(Bucket=self.bucket, Prefix=prefix)
return [obj["Key"] for obj in response.get("Contents", []) if "Key" in obj]
def read_record(self, key: str) -> S3LogRecord:
body = self.client.get_object(Bucket=self.bucket, Key=key)["Body"].read()
return S3LogRecord.model_validate_json(body)
def records_matching(self, *, prefix: str, predicate: Callable[[S3LogRecord], bool]) -> list[S3LogRecord]:
return [record for record in map(self.read_record, self.list_keys(prefix)) if predicate(record)]
def poll_records(self, *, prefix: str, predicate: Callable[[S3LogRecord], bool]) -> list[S3LogRecord]:
"""Poll until at least one matching object is listed (the s3_v2
callback flushes on a ~10s timer), then keep re-reading for
S3_SETTLE_SECONDS - past a full flush interval - so a duplicate
shipped by a later flush cannot hide from the exactly-one assertion.
One blind spot is inherent: a duplicate write that reuses the exact
same object key overwrites the first object and no listing can see
it; distinct-key duplicates are what this catches. At the deadline an
empty list is returned and the caller's assertion carries the failure
message."""
deadline = time.monotonic() + POLL_TIMEOUT
while time.monotonic() < deadline:
records = self.records_matching(prefix=prefix, predicate=predicate)
if records:
return self._settled_records(prefix=prefix, predicate=predicate, first=records)
time.sleep(POLL_INTERVAL)
return []
def _settled_records(
self, *, prefix: str, predicate: Callable[[S3LogRecord], bool], first: list[S3LogRecord]
) -> list[S3LogRecord]:
"""Re-read at every poll interval until the settle window closes; a
duplicate ends the watch early because more waiting cannot clear it.
A transiently empty re-read never downgrades what was already seen."""
settle_deadline = time.monotonic() + S3_SETTLE_SECONDS
latest = first
while time.monotonic() < settle_deadline and len(latest) <= 1:
time.sleep(POLL_INTERVAL)
latest = self.records_matching(prefix=prefix, predicate=predicate) or latest
return latest
def build_s3_reader() -> S3LogReader:
bucket = os.environ.get("S3_LOGS_BUCKET_NAME", "")
if not bucket:
pytest.fail(
"S3_LOGS_BUCKET_NAME must be set: the s3 tests read the proxy's s3_v2 "
"delivery back from the real bucket (the cluster secret manager injects "
"it; locally set it in tests/e2e/.env to the same bucket "
"s3_callback_params.s3_bucket_name names)"
)
region = os.environ.get("AWS_REGION_NAME") or os.environ.get("AWS_REGION") or "us-east-1"
return S3LogReader(
bucket=bucket,
# boto3.client's overload set covers every AWS service; the ones without
# installed stubs type as Unknown, so the member is "partially unknown"
# even though the s3 overload itself resolves to S3Client.
client=boto3.client("s3", region_name=region), # pyright: ignore[reportUnknownMemberType]
)

128
tests/e2e/logging/test_datadog_log_e2e.py
View file

@ -19,15 +19,16 @@ received).
from __future__ import annotations
import math
import time
import pytest
from pydantic import BaseModel, ConfigDict
from datadog_reader import DdLogEvent, DdLogsReader
from e2e_config import CHEAP_ANTHROPIC_MODEL, CHEAP_OPENAI_MODEL, unique_marker
from e2e_http import NoBody
from lifecycle import ResourceManager
from logging_client import LoggingClient, first_ok
from logging_client import INVALID_UPSTREAM_API_KEY, LoggingClient, first_ok, readiness_details_body
from models import LiteLLMParamsBody
pytestmark = pytest.mark.e2e
@ -46,19 +47,17 @@ class _DdMessagePayload(BaseModel):
status: str
call_type: str
stream: bool | None = None
error_str: str | None = None
def _assert_datadog_configured(client: LoggingClient) -> None:
"""Recorded state: the proxy reports the DataDog callback among its active
callbacks, so a missing destination config fails here, before any
delivery-based assertion can time out confusingly."""
result = client.proxy.probe("/health/readiness/details", params=NoBody())
assert result.status_code == 200, (
f"/health/readiness/details must answer 200, got {result.status_code}: {result.body[:300]}"
)
assert DD_LOGGER_NAME in result.body, (
body = readiness_details_body(client)
assert DD_LOGGER_NAME in body, (
f"the proxy must report the {DD_LOGGER_NAME} callback active "
f"(callbacks + DD_* env in the compose config); got: {result.body[:400]}"
f"(callbacks + DD_* env in the compose config); got: {body[:400]}"
)
@ -89,18 +88,14 @@ def _assert_exactly_one_event(
# indexed event status from the parsed payload's status attribute
# ("success") and normalizes it to its OK severity - so "ok" is what a
# successfully ingested success event looks like on the search API.
assert event.status == "ok", (
f"success events must index at DataDog's ok severity, got {event.status!r}"
)
assert event.status == "ok", f"success events must index at DataDog's ok severity, got {event.status!r}"
payload = _DdMessagePayload.model_validate(event.attributes)
assert payload.status == "success", f"payload status must be success, got {payload.status!r}"
assert payload.model_group == model_group, (
f"payload model_group must be {model_group!r}, got {payload.model_group!r}"
)
assert payload.call_type == call_type, (
f"payload call_type must be {call_type!r}, got {payload.call_type!r}"
)
assert payload.call_type == call_type, f"payload call_type must be {call_type!r}, got {payload.call_type!r}"
assert payload.total_tokens > 0, f"payload must count real tokens, got {payload.total_tokens}"
# Relative tolerance, not bit-equality: the cost round-trips through
# DataDog's attribute indexing, whose float serialization may drift in the
@ -109,9 +104,7 @@ def _assert_exactly_one_event(
f"payload response_cost {payload.response_cost} must equal the anchor cost {cost_anchor}"
)
if expect_stream:
assert payload.stream is True, (
f"a streamed call's payload must record stream=true, got {payload.stream!r}"
)
assert payload.stream is True, f"a streamed call's payload must record stream=true, got {payload.stream!r}"
return payload
@ -211,7 +204,9 @@ class TestDataDogLogDelivery:
marker = unique_marker()
outcome = first_ok(
client,
lambda: client.chat_raw(key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", stream=True, max_tokens=16),
lambda: client.chat_raw(
key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", stream=True, max_tokens=16
),
)
assert outcome.is_streaming, f"response must be an event stream, got content-type {outcome.content_type!r}"
assert outcome.chunks > 0, "the stream must deliver at least one event"
@ -231,9 +226,7 @@ class TestDataDogLogDelivery:
cost_anchor=spend_row.spend,
expect_stream=True,
)
assert spend_row.total_tokens is not None, (
"the spend row must record total_tokens for the token cross-check"
)
assert spend_row.total_tokens is not None, "the spend row must record total_tokens for the token cross-check"
assert spend_row.total_tokens == payload.total_tokens, (
f"the spend row and the DataDog event must agree on tokens: "
f"{spend_row.total_tokens} vs {payload.total_tokens}"
@ -255,7 +248,9 @@ class TestDataDogLogDelivery:
marker = unique_marker()
outcome = first_ok(
client,
lambda: client.messages_raw(key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", max_tokens=16, stream=True),
lambda: client.messages_raw(
key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", max_tokens=16, stream=True
),
)
assert outcome.is_streaming, f"response must be an event stream, got content-type {outcome.content_type!r}"
assert outcome.chunks > 0, "the stream must deliver at least one event"
@ -275,9 +270,7 @@ class TestDataDogLogDelivery:
cost_anchor=spend_row.spend,
expect_stream=True,
)
assert spend_row.total_tokens is not None, (
"the spend row must record total_tokens for the token cross-check"
)
assert spend_row.total_tokens is not None, "the spend row must record total_tokens for the token cross-check"
assert spend_row.total_tokens == payload.total_tokens, (
f"the spend row and the DataDog event must agree on tokens: "
f"{spend_row.total_tokens} vs {payload.total_tokens}"
@ -319,10 +312,89 @@ class TestDataDogLogDelivery:
cost_anchor=spend_row.spend,
expect_stream=True,
)
assert spend_row.total_tokens is not None, (
"the spend row must record total_tokens for the token cross-check"
)
assert spend_row.total_tokens is not None, "the spend row must record total_tokens for the token cross-check"
assert spend_row.total_tokens == payload.total_tokens, (
f"the spend row and the DataDog event must agree on tokens: "
f"{spend_row.total_tokens} vs {payload.total_tokens}"
)
def _assert_exactly_one_failure_event(events: list[DdLogEvent], *, model_group: str) -> _DdMessagePayload:
"""The enforced behavior for a failed call: the intake holds exactly one
event for the deployment, sourced from litellm, indexed at an error-grade
severity (DataDog derives it from the payload's status="failure"; observed
as its "emergency" bucket), whose payload carries the provider error and
no cost."""
assert events, "no DataDog log event for the failed call reached the intake within the deadline"
assert len(events) == 1, (
f"expected exactly ONE DataDog log event for the failed call, got {len(events)} - "
"more than one event for one call is the duplicate-delivery bug"
)
event = events[0]
assert "source:litellm" in event.tags, (
f"the ingested event must carry the litellm source (shipped as ddsource), got tags {event.tags!r}"
)
assert event.status in ("error", "emergency"), (
f"failure events must index at an error-grade severity, got {event.status!r}"
)
payload = _DdMessagePayload.model_validate(event.attributes)
assert payload.status == "failure", f"payload status must be failure, got {payload.status!r}"
assert payload.model_group == model_group, (
f"payload model_group must be {model_group!r}, got {payload.model_group!r}"
)
assert not payload.response_cost, f"a failed call must not be billed, got response_cost={payload.response_cost!r}"
return payload
class TestDataDogFailureDelivery:
@pytest.mark.covers("logging.datadog.failure.exports_metric", exercised_on=["chat_completions"])
def test_failed_chat_completions_emits_one_error_event(
self, client: LoggingClient, dd_logs: DdLogsReader, resources: ResourceManager
) -> None:
"""A /chat/completions call that fails at the provider must reach the
DataDog logs intake as exactly one error-grade event carrying the
provider error - failure metrics drive alerting and SLOs, so a dropped
failure event is an invisible outage.
A deployment with an invalid upstream key lets the request pass proxy
auth and fail at the provider (the same lever as the OTEL error test).
Failure payloads carry no prompt to mark, so the read-back queries the
indexed @model_group attribute of the per-run unique deployment name;
proxy-side 401s during key propagation never reach the provider and
ship no payload, so exactly one provider failure exists for it."""
_assert_datadog_configured(client)
model_name = f"dd-err-{unique_marker()}"
model_id = client.create_model(
model_name,
LiteLLMParamsBody(model="anthropic/claude-haiku-4-5", api_key=INVALID_UPSTREAM_API_KEY),
)
resources.defer(lambda: client.delete_model(model_id))
key = client.key_with_alias(f"dd-err-key-{unique_marker()}", models=[model_name])
resources.defer(lambda: client.delete_key(key))
deadline = time.monotonic() + client.proxy.poll_timeout
while True:
outcome = client.chat_raw(key, model_name, "trigger an upstream auth failure", max_tokens=16)
assert not outcome.ok, "the call must fail; the deployment's upstream key is invalid"
assert outcome.status_code != -1, (
"network failure between the test and the proxy while provoking the provider "
"failure; retrying now could double-log the failure payload and falsely trip "
f"the exactly-one assertion - fix the rig connectivity first: {outcome.body[:200]}"
)
if "AnthropicException" in outcome.body or time.monotonic() >= deadline:
break
time.sleep(client.proxy.poll_interval)
assert "AnthropicException" in outcome.body, (
"never saw the upstream provider failure before the deadline; the key may still be "
f"propagating - last outcome {outcome.status_code}: {outcome.body[:200]}"
)
assert outcome.status_code == 401, (
f"an upstream auth failure must map to 401, got {outcome.status_code}: {outcome.body[:200]}"
)
events = dd_logs.poll_events_for_query(f"@model_group:{model_name}")
payload = _assert_exactly_one_failure_event(events, model_group=model_name)
assert payload.error_str is not None and "AnthropicException" in payload.error_str, (
f"the event must carry the provider error, got error_str={payload.error_str!r}"
)

97
tests/e2e/logging/test_gcs_log_e2e.py Normal file
View file

@ -0,0 +1,97 @@
"""Live e2e: gcs_bucket log delivery for successful calls.
Covers logging.gcs_bucket.success.writes_object: one successful
/chat/completions call must land in the real GCS bucket as exactly one
StandardLoggingPayload record (GCS is the audit-trail parallel to S3 for GCP
deployments). Delivery is judged on what is actually readable in the bucket:
the proxy writes with its production service account, and the test reads the
record back through the GCS JSON API - covering both the batched NDJSON layout
(the default) and the per-request object layout.
Both halves of the contract are asserted: the recorded state (the proxy
reports the GCSBucketLogger callback active via /health/readiness/details -
note gcs_bucket is enterprise-gated, so this also requires a license) and the
enforced behavior (the record in the bucket, cost cross-checked against the
x-litellm-response-cost header of the very response the caller received).
"""
from __future__ import annotations
import math
import pytest
from e2e_config import CHEAP_ANTHROPIC_MODEL, unique_marker
from gcs_reader import GcsLogReader, build_gcs_reader, utc_now
from lifecycle import ResourceManager
from logging_client import LoggingClient, completion_response_id, first_ok, readiness_details_body
pytestmark = pytest.mark.e2e
#: The active gcs_bucket callback's name in /health/readiness/details success_callbacks.
GCS_LOGGER_NAME = "GCSBucketLogger"
@pytest.fixture(scope="session")
def gcs_logs() -> GcsLogReader:
return build_gcs_reader()
def _assert_gcs_configured(client: LoggingClient) -> None:
"""Recorded state: the proxy reports the gcs_bucket callback among its
active callbacks, so a missing destination config (or a missing enterprise
license - gcs_bucket refuses to initialize without one) fails here, before
any delivery-based assertion can time out confusingly."""
body = readiness_details_body(client)
assert GCS_LOGGER_NAME in body, (
f"the proxy must report the {GCS_LOGGER_NAME} callback active "
f"(litellm_settings.callbacks: ['gcs_bucket'] + GCS_BUCKET_NAME env + enterprise license); "
f"got: {body[:400]}"
)
class TestGcsLogDelivery:
@pytest.mark.covers("logging.gcs_bucket.success.writes_object", exercised_on=["chat_completions"])
def test_chat_completions_writes_one_success_record(
self, client: LoggingClient, gcs_logs: GcsLogReader, resources: ResourceManager
) -> None:
"""One successful non-streaming /chat/completions call must be
readable back from the bucket as exactly one payload record carrying
the model group, the token counts, and the same cost the caller's
response header reported."""
_assert_gcs_configured(client)
alias = f"gcs-chat-{unique_marker()}"
key = client.key_with_alias(alias, models=[CHEAP_ANTHROPIC_MODEL])
resources.defer(lambda: client.delete_key(key))
since = utc_now()
marker = unique_marker()
outcome = first_ok(
client,
lambda: client.chat_raw(key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", max_tokens=16),
)
assert outcome.response_cost is not None and outcome.response_cost > 0, (
f"the response must report x-litellm-response-cost, got {outcome.response_cost!r}"
)
body_id = completion_response_id(outcome.body)
assert body_id is not None, "the completion body must carry an id (it names the gcs record)"
records = gcs_logs.poll_records_for_response_id(body_id, since=since)
assert records, f"no gcs record for response {body_id} was readable from the bucket within the deadline"
assert len(records) == 1, (
f"expected exactly ONE gcs record for the call, got {len(records)} - "
"more than one record for one call is the duplicate-delivery bug"
)
record = records[0]
assert record.id == body_id, f"record id must be the response id, got {record.id!r}"
assert record.status == "success", f"payload status must be success, got {record.status!r}"
assert record.model_group == CHEAP_ANTHROPIC_MODEL, (
f"payload model_group must be {CHEAP_ANTHROPIC_MODEL!r}, got {record.model_group!r}"
)
assert record.total_tokens is not None and record.total_tokens > 0, (
f"payload must count real tokens, got {record.total_tokens!r}"
)
assert record.response_cost is not None and math.isclose(
record.response_cost, outcome.response_cost, rel_tol=1e-9
), f"payload response_cost {record.response_cost!r} must equal the header cost {outcome.response_cost}"

35
tests/e2e/logging/test_otel_trace_e2e.py
View file

@ -23,9 +23,8 @@ import pytest
from pydantic import BaseModel, ConfigDict, ValidationError
from e2e_config import CHEAP_ANTHROPIC_MODEL, CHEAP_OPENAI_MODEL, unique_marker
from e2e_http import NoBody
from lifecycle import ResourceManager
from logging_client import INVALID_UPSTREAM_API_KEY, LoggingClient, first_ok
from logging_client import INVALID_UPSTREAM_API_KEY, LoggingClient, first_ok, readiness_details_body
from models import LiteLLMParamsBody
from otel_client import JaegerSpan, JaegerTrace, OtelReader
@ -48,11 +47,7 @@ def _assert_otel_destination_configured(client: LoggingClient) -> None:
"""Recorded state: the proxy reports the OTEL v2 logger among its active
callbacks, so a missing/failed destination config fails here, before any
traffic-based assertion can time out confusingly."""
result = client.proxy.probe("/health/readiness/details", params=NoBody())
assert result.status_code == 200, (
f"/health/readiness/details must answer 200, got {result.status_code}: {result.body[:300]}"
)
details = _ReadinessDetails.model_validate_json(result.body)
details = _ReadinessDetails.model_validate_json(readiness_details_body(client))
assert OTEL_V2_LOGGER_NAME in details.success_callbacks, (
f"the proxy must report the {OTEL_V2_LOGGER_NAME} callback active "
f"(LITELLM_OTEL_V2 + arize_phoenix preset in the compose config); got: {details.success_callbacks}"
@ -164,17 +159,14 @@ def served_genai_spans(trace: JaegerTrace, genai_span: str) -> list[JaegerSpan]:
these tests fail whenever the upstream 429s, 529s, or hands back a stale
credential on the first try."""
return [
span
for span in trace.spans
if span.operation_name == genai_span and _tag(span, ERROR_STATUS_TAG) != "ERROR"
span for span in trace.spans if span.operation_name == genai_span and _tag(span, ERROR_STATUS_TAG) != "ERROR"
]
def one_served_genai_span(trace: JaegerTrace, genai_span: str) -> JaegerSpan:
served = served_genai_spans(trace, genai_span)
assert len(served) == 1, (
f"a streamed call must produce exactly ONE served gen-AI span, got {len(served)}; "
f"spans: {trace.span_names()}"
f"a streamed call must produce exactly ONE served gen-AI span, got {len(served)}; spans: {trace.span_names()}"
)
return served[0]
@ -190,8 +182,7 @@ def _assert_real_ttft(hits: list[JaegerTrace], *, genai_span: str) -> None:
"(nothing tagged with its call id was found)"
)
assert len(hits) == 1, (
f"expected exactly ONE trace for the call, got {len(hits)}: "
f"{[(t.trace_id, t.span_names()) for t in hits]}"
f"expected exactly ONE trace for the call, got {len(hits)}: {[(t.trace_id, t.span_names()) for t in hits]}"
)
trace = hits[0]
span = one_served_genai_span(trace, genai_span)
@ -280,9 +271,7 @@ def _assert_error_span_contract(span: JaegerSpan) -> None:
"the span status description must carry the same untruncated message as error.message"
)
stack = _tag(span, "litellm.provider.error.stack_trace")
assert isinstance(stack, str) and stack, (
"the error span must carry a non-empty litellm.provider.error.stack_trace"
)
assert isinstance(stack, str) and stack, "the error span must carry a non-empty litellm.provider.error.stack_trace"
class TestOtelTraceCompleteness:
@ -313,9 +302,7 @@ class TestOtelTraceCompleteness:
resources.defer(lambda: client.delete_key(key))
marker = unique_marker()
outcome = first_ok(
client, lambda: client.chat_raw(key, MODEL, f"reply with one word {marker}", max_tokens=16)
)
outcome = first_ok(client, lambda: client.chat_raw(key, MODEL, f"reply with one word {marker}", max_tokens=16))
assert outcome.call_id is not None, "success response must carry x-litellm-call-id"
hits = otel_reader.poll_traces_for_call(
@ -520,9 +507,7 @@ class TestOtelTraceCompleteness:
route = "/v1/responses"
_assert_otel_destination_configured(client)
key = client.key_with_alias(
f"otel-stream-responses-{unique_marker()}", models=[CHEAP_OPENAI_MODEL]
)
key = client.key_with_alias(f"otel-stream-responses-{unique_marker()}", models=[CHEAP_OPENAI_MODEL])
resources.defer(lambda: client.delete_key(key))
marker = unique_marker()
@ -660,9 +645,7 @@ class TestOtelTraceCompleteness:
route = "/v1/responses"
_assert_otel_destination_configured(client)
key = client.key_with_alias(
f"otel-ttft-responses-{unique_marker()}", models=[CHEAP_OPENAI_MODEL]
)
key = client.key_with_alias(f"otel-ttft-responses-{unique_marker()}", models=[CHEAP_OPENAI_MODEL])
resources.defer(lambda: client.delete_key(key))
marker = unique_marker()

170
tests/e2e/logging/test_s3_log_e2e.py Normal file
View file

@ -0,0 +1,170 @@
"""Live e2e: s3_v2 log delivery for successful and failed calls.
Covers logging.s3.success.writes_object and logging.s3.failure.writes_object:
one /chat/completions call must land in the real S3 bucket as exactly one
StandardLoggingPayload object (the primary audit trail; the batch flush must
neither drop nor duplicate it), and a failed call must be persisted the same
way for compliance. Delivery is judged on what is actually in the bucket: the
proxy writes with its production credentials and the test lists and reads the
objects back.
Both halves of the contract are asserted: the recorded state (the proxy
reports the S3Logger callback active via /health/readiness/details) and the
enforced behavior (the object in the bucket, with the cost cross-checked
against the x-litellm-response-cost header of the very response the caller
received).
The suite requires ``s3_callback_params.s3_use_key_prefix: true`` on the proxy,
which keys objects as ``{key_alias}/{date}/time-..._{id}.json`` - a unique key
alias per test turns the poll into a cheap prefix listing.
"""
from __future__ import annotations
import math
import time
import pytest
from e2e_config import CHEAP_ANTHROPIC_MODEL, unique_marker
from lifecycle import ResourceManager
from logging_client import (
INVALID_UPSTREAM_API_KEY,
LoggingClient,
completion_response_id,
first_ok,
readiness_details_body,
)
from models import LiteLLMParamsBody
from s3_reader import S3LogReader, build_s3_reader
pytestmark = pytest.mark.e2e
#: The active s3_v2 callback's name in /health/readiness/details success_callbacks.
S3_LOGGER_NAME = "S3Logger"
@pytest.fixture(scope="session")
def s3_logs() -> S3LogReader:
return build_s3_reader()
def _assert_s3_configured(client: LoggingClient) -> None:
"""Recorded state: the proxy reports the s3_v2 callback among its active
callbacks, so a missing destination config fails here, before any
delivery-based assertion can time out confusingly."""
body = readiness_details_body(client)
assert S3_LOGGER_NAME in body, (
f"the proxy must report the {S3_LOGGER_NAME} callback active "
f"(litellm_settings.callbacks: ['s3_v2'] + s3_callback_params in the proxy config); "
f"got: {body[:400]}"
)
class TestS3LogDelivery:
@pytest.mark.covers("logging.s3.success.writes_object", exercised_on=["chat_completions"])
def test_chat_completions_writes_one_success_object(
self, client: LoggingClient, s3_logs: S3LogReader, resources: ResourceManager
) -> None:
"""One successful non-streaming /chat/completions call must land in
the bucket as exactly one payload object carrying the model group, the
token counts, and the same cost the caller's response header reported."""
_assert_s3_configured(client)
alias = f"s3-chat-{unique_marker()}"
key = client.key_with_alias(alias, models=[CHEAP_ANTHROPIC_MODEL])
resources.defer(lambda: client.delete_key(key))
marker = unique_marker()
outcome = first_ok(
client,
lambda: client.chat_raw(key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", max_tokens=16),
)
assert outcome.response_cost is not None and outcome.response_cost > 0, (
f"the response must report x-litellm-response-cost, got {outcome.response_cost!r}"
)
body_id = completion_response_id(outcome.body)
assert body_id is not None, "the completion body must carry an id (it names the s3 object)"
records = s3_logs.poll_records(prefix=f"{alias}/", predicate=lambda r: r.id == body_id)
assert records, (
f"no s3 object for response {body_id} under prefix {alias}/ reached the bucket within the deadline"
)
assert len(records) == 1, (
f"expected exactly ONE s3 object for the call, got {len(records)} - "
"more than one object for one call is the duplicate-delivery bug"
)
record = records[0]
assert record.status == "success", f"payload status must be success, got {record.status!r}"
assert record.model_group == CHEAP_ANTHROPIC_MODEL, (
f"payload model_group must be {CHEAP_ANTHROPIC_MODEL!r}, got {record.model_group!r}"
)
assert record.total_tokens is not None and record.total_tokens > 0, (
f"payload must count real tokens, got {record.total_tokens!r}"
)
assert record.response_cost is not None and math.isclose(
record.response_cost, outcome.response_cost, rel_tol=1e-9
), f"payload response_cost {record.response_cost!r} must equal the header cost {outcome.response_cost}"
@pytest.mark.covers("logging.s3.failure.writes_object", exercised_on=["chat_completions"])
def test_chat_completions_failure_writes_one_object(
self, client: LoggingClient, s3_logs: S3LogReader, resources: ResourceManager
) -> None:
"""A call that fails at the provider must be persisted to the bucket as
exactly one failure payload carrying the provider error - failed calls
are part of the audit trail, not an exemption from it.
A deployment with an invalid upstream key lets the request pass proxy
auth and fail at the provider (the same lever as the OTEL error test).
Proxy-side rejections during key/model propagation can also ship
failure payloads under this alias, but without a model_group and
without the provider error, so the read-back keys on both: only
provider-reaching calls carry them, and with this key every one of
those is the AnthropicException that ends the send loop."""
_assert_s3_configured(client)
model_name = f"s3-err-{unique_marker()}"
model_id = client.create_model(
model_name,
LiteLLMParamsBody(model="anthropic/claude-haiku-4-5", api_key=INVALID_UPSTREAM_API_KEY),
)
resources.defer(lambda: client.delete_model(model_id))
alias = f"s3-err-key-{unique_marker()}"
key = client.key_with_alias(alias, models=[model_name])
resources.defer(lambda: client.delete_key(key))
deadline = time.monotonic() + client.proxy.poll_timeout
while True:
outcome = client.chat_raw(key, model_name, "trigger an upstream auth failure", max_tokens=16)
assert not outcome.ok, "the call must fail; the deployment's upstream key is invalid"
assert outcome.status_code != -1, (
"network failure between the test and the proxy while provoking the provider "
"failure; retrying now could double-log the failure payload and falsely trip "
f"the exactly-one assertion - fix the rig connectivity first: {outcome.body[:200]}"
)
if "AnthropicException" in outcome.body or time.monotonic() >= deadline:
break
time.sleep(client.proxy.poll_interval)
assert "AnthropicException" in outcome.body, (
"never saw the upstream provider failure before the deadline; the key may still be "
f"propagating - last outcome {outcome.status_code}: {outcome.body[:200]}"
)
assert outcome.status_code == 401, (
f"an upstream auth failure must map to 401, got {outcome.status_code}: {outcome.body[:200]}"
)
records = s3_logs.poll_records(
prefix=f"{alias}/",
predicate=lambda r: (
r.status == "failure" and r.model_group == model_name and "AnthropicException" in (r.error_str or "")
),
)
assert records, (
f"no failure object for {model_name} under prefix {alias}/ reached the bucket within the deadline"
)
assert len(records) == 1, f"expected exactly ONE failure object for the call, got {len(records)}"
record = records[0]
assert record.error_str is not None and "AnthropicException" in record.error_str, (
f"the persisted failure must carry the provider error, got error_str={record.error_str!r}"
)
assert not record.response_cost, f"a failed call must not be billed, got response_cost={record.response_cost!r}"

123
tests/e2e/logging/test_team_langfuse_callback_e2e.py Normal file
View file

@ -0,0 +1,123 @@
"""Live e2e: team-scoped Langfuse callback delivery and isolation.
Covers logging.langfuse.success.logs_spend: a team configured with a Langfuse
callback via POST /team/{id}/callback must deliver its members' calls to the
real Langfuse project (generation readable back through Langfuse's own API,
with the cost agreeing with the x-litellm-response-cost header), while traffic
from keys outside the team must NOT reach that project - the isolation is the
point of team-scoped callbacks.
Both halves of the contract are asserted: the recorded state (the /team/callback
registration itself answers success) and the enforced behavior (the generation
at the destination for the team key, and its absence for the non-team key).
"""
from __future__ import annotations
import time
import pytest
from e2e_config import CHEAP_ANTHROPIC_MODEL, unique_marker
from lifecycle import ResourceManager
from logging_client import (
LangfuseCreds,
LoggingClient,
costs_agree,
first_ok,
load_langfuse_creds,
observation_spend,
)
pytestmark = pytest.mark.e2e
#: How long to keep re-checking that the non-team call never surfaces in
#: Langfuse after the team call's generation has already been ingested; the
#: positive observation bounds the pipeline's latency, so a wrong delivery
#: would be visible within the same order of magnitude.
ISOLATION_SETTLE_SECONDS = 30.0
ISOLATION_CHECK_INTERVAL_SECONDS = 5.0
@pytest.fixture(scope="session")
def langfuse_creds() -> LangfuseCreds:
return load_langfuse_creds()
class TestTeamLangfuseCallback:
@pytest.mark.covers("logging.langfuse.success.logs_spend", exercised_on=["chat_completions"])
def test_team_callback_delivers_and_isolates(
self, client: LoggingClient, langfuse_creds: LangfuseCreds, resources: ResourceManager
) -> None:
team_id = client.create_team(f"lf-team-{unique_marker()}", models=[CHEAP_ANTHROPIC_MODEL])
resources.defer(lambda: client.delete_team(team_id))
# Recorded state: the registration endpoint itself must answer success
# (add_team_langfuse_callback asserts it).
client.add_team_langfuse_callback(team_id, langfuse_creds)
team_alias = f"lf-team-key-{unique_marker()}"
team_key = client.key_with_alias(team_alias, models=[CHEAP_ANTHROPIC_MODEL], team_id=team_id)
resources.defer(lambda: client.delete_key(team_key))
solo_alias = f"lf-solo-key-{unique_marker()}"
solo_key = client.key_with_alias(solo_alias, models=[CHEAP_ANTHROPIC_MODEL])
resources.defer(lambda: client.delete_key(solo_key))
# Enforced behavior, positive half, with one propagation retry: a
# worker still holding the pre-callback team object can serve the
# first call without shipping it, and by the time the first Langfuse
# poll has timed out the team cache TTL has lapsed, so a second call
# must deliver.
team_marker = ""
team_outcome = None
observation = None
for _attempt in range(2):
team_marker = unique_marker()
team_outcome = first_ok(
client,
lambda marker=team_marker: client.chat_raw(
team_key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {marker}", max_tokens=16
),
)
assert team_outcome.response_cost is not None and team_outcome.response_cost > 0, (
f"the response must report x-litellm-response-cost, got {team_outcome.response_cost!r}"
)
observation = client.poll_langfuse_observation(
langfuse_creds,
key_alias=team_alias,
prompt_marker=team_marker,
require_positive_cost=True,
)
if observation is not None:
break
solo_marker = unique_marker()
_ = first_ok(
client,
lambda: client.chat_raw(
solo_key, CHEAP_ANTHROPIC_MODEL, f"reply with one word {solo_marker}", max_tokens=16
),
)
assert observation is not None, (
f"the team key's call (marker {team_marker}) never reached Langfuse within the deadline, "
"even after a fresh call past the team-object cache TTL"
)
assert team_outcome is not None and team_outcome.response_cost is not None
cost = observation_spend(observation)
assert cost is not None and costs_agree(team_outcome.response_cost, cost), (
f"Langfuse calculatedTotalCost {cost!r} must agree with the header cost {team_outcome.response_cost}"
)
# Enforced behavior, negative half: the non-team call must never show
# up in this project. The positive generation above has already been
# ingested, which bounds the pipeline latency, so keep re-checking for
# a settle window rather than trusting a single instant.
settle_deadline = time.monotonic() + ISOLATION_SETTLE_SECONDS
while True:
leaked = client.find_langfuse_observation(langfuse_creds, key_alias=solo_alias, prompt_marker=solo_marker)
assert leaked is None, (
f"a non-team key's call (marker {solo_marker}) reached the team's Langfuse "
f"project: {leaked.id} - team callbacks must not apply outside the team"
)
if time.monotonic() >= settle_deadline:
break
time.sleep(ISOLATION_CHECK_INTERVAL_SECONDS)