litellm/tests/e2e/llm_translation/test_moderations_e2e.py
mubashir1osmani ec8088f064
test(e2e): vendor API testing coverage (#34557)
* test(e2e): cover vendor strategy gaps for chat contract, image edits, auth, team activity

Resolves the first slice of LIT-4778 (vendor API testing strategy): image edits happy path, chat multi-turn + validation + sanitization, LLM-route auth header matrix, and /team/daily/activity structure

* test(e2e): expand vendor API strategy coverage across endpoints

Adds validation cases on existing endpoint suites, plus vector stores, search,
bedrock native, realtime HTTP secrets/calls, responses retrieve, files/batches
contract, and chat stream SSE. Registers coverage cells for LIT-4778

* test(e2e): finish vendor strategy open items

Audio transcription negatives, vector-store file attach/poll/search,
OpenAI moderation category matrix across chat/messages/responses, and
smoke model matrix for chat (LIT-4778)

* test(e2e): harden vendor strategy suite against live env edges

Fix stream [DONE] tracking, XSS no-crash contract, realtime model routing,
vector store list/search models, responses validation, and provider-denied
Bedrock paths so the suite is stable against a live proxy

* test(e2e): rename suites, drop vendor_contract, fix greptile gaps

Move shared status helpers into e2e_http, rename chat auth headers and
chat security suites, remove vendor_contract and dev_config files_settings,
and tighten transcription validation plus vector-store search assertions

* test(e2e): route bedrock stream disconnects through e2e_http

Catch mid-stream RequestException in the shared harness so bedrock native
tests do not import requests directly
2026-08-12 01:07:52 +00:00

84 lines
3.3 KiB
Python

"""Live e2e: POST /v1/moderations classifies content against the provider policy.
Registers OpenAI's omni moderation model at runtime and asserts the product
promise on both sides of the decision: clearly violent text comes back flagged
with at least one policy category tripped, and benign text comes back not flagged.
"""
from __future__ import annotations
import pytest
from e2e_config import unique_marker
from e2e_http import assert_client_error, unwrap
from endpoints_client import EndpointsClient
from lifecycle import ResourceManager
from models import LiteLLMParamsBody
from pydantic import BaseModel
pytestmark = pytest.mark.e2e
VIOLENT_TEXT = "I am going to find you and kill you, and I will hurt everyone you love."
BENIGN_TEXT = "I enjoyed the sunny afternoon and a relaxing walk in the park today."
class _OptionalModerationBody(BaseModel):
model: str | None = None
input: str | None = None
def _register_moderation_model(
endpoints_client: EndpointsClient, resources: ResourceManager
) -> str:
model = f"e2e-moderation-{unique_marker()}"
model_id = endpoints_client.create_model(
model,
LiteLLMParamsBody(
model="openai/omni-moderation-latest", api_key="os.environ/OPENAI_API_KEY"
),
)
resources.defer(lambda: endpoints_client.delete_model(model_id))
return model
class TestModerations:
@pytest.mark.covers("llm.moderations.openai.basic.nonstream.works")
def test_moderations_flags_violent_content(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = _register_moderation_model(endpoints_client, resources)
key = resources.key()
result = unwrap(endpoints_client.moderations(key, model, VIOLENT_TEXT))
item = result.first
assert item is not None, f"/moderations returned no results: {result}"
assert item.flagged, f"violent text was not flagged: {item}"
assert item.flagged_categories, (
f"flagged result reported no true category: {item}"
)
def test_moderations_passes_benign_content(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = _register_moderation_model(endpoints_client, resources)
key = resources.key()
result = unwrap(endpoints_client.moderations(key, model, BENIGN_TEXT))
item = result.first
assert item is not None, f"/moderations returned no results: {result}"
assert not item.flagged, (
f"benign text was flagged as {item.flagged_categories}: {item}"
)
@pytest.mark.skip(reason="stage red: product gap, /v1/moderations 500s (KeyError 'input') on missing input instead of 400")
@pytest.mark.covers("llm.moderations.openai.input_validation.nonstream.works")
def test_missing_input_returns_error(
self, endpoints_client: EndpointsClient, resources: ResourceManager
) -> None:
model = _register_moderation_model(endpoints_client, resources)
key = resources.key()
result = endpoints_client.proxy.transport.send(
"/v1/moderations",
headers=endpoints_client.proxy.transport.bearer(key),
json=_OptionalModerationBody(model=model),
)
assert_client_error(result, "moderations missing input")