litellm/tests/local_testing/test_fake_openai_endpoint.py
yuneng-jiang 0c2e404be3
test(ci): serve /moderations from the canned OpenAI mock (#37739)
* test(ci): serve /moderations from the canned OpenAI mock

The otel proxy E2E job points its `openai/*` wildcard deployment at the
canned mock, and #37492 made `get_model_list` agree with
`get_available_deployment` on bare model names. /moderations now resolves
`omni-moderation-latest` to that wildcard deployment the way
/chat/completions already did, so the request lands on the mock, which
never implemented the route and answers a bare 404.

Add /moderations and /v1/moderations to the mock, returning an
OpenAI-shaped response with one result per input item.

* style(ci): annotate the new moderations locals as Final
2026-08-20 16:58:38 -07:00

118 lines
4.1 KiB
Python

"""Regression tests for the canned OpenAI mock that decouples the suite from a
hosted endpoint (see ``tests/fake_openai_endpoint.py`` and
``tests/_fake_openai_endpoint_server.py``).
Before this, router/completion/triton tests hardcoded a shared Railway mock as
their ``api_base``; when that single deployment went down, unrelated CI jobs
failed with ``404 Application not found``. These tests pin the two behaviors
those tests now rely on from the local stand-in (the Triton embeddings shape and
the ``slow-endpoint`` delay) and guard against the dead host creeping back in.
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Final
import httpx
import pytest
from openai.types import ModerationCreateResponse
from tests.fake_openai_endpoint import (
_LOCAL_DEFAULT,
_resolve_base,
ensure_fake_openai_endpoint,
)
_REPO_ROOT = Path(__file__).resolve().parents[2]
_MIGRATED_FILES = (
"tests/llm_translation/test_triton.py",
"tests/local_testing/test_router.py",
"tests/local_testing/test_router_custom_routing.py",
"tests/local_testing/test_router_fallback_handlers.py",
"tests/local_testing/test_router_fallbacks.py",
"tests/local_testing/test_secret_detect_hook.py",
"tests/local_testing/test_lowest_latency_routing.py",
"tests/local_testing/test_completion.py",
)
# A live ``railway.app/`` or ``railway.app"`` api_base reappearing in a migrated
# file would re-couple CI to an external service's uptime. The deliberately
# broken ``...railway.appzzzzz`` fallback URL is not a live host, so it does not
# match.
_LIVE_HOSTED_MOCK = re.compile(r"railway\.app(?:/|\")")
def test_chat_completion_shape():
base = ensure_fake_openai_endpoint()
response = httpx.post(
f"{base}/v1/chat/completions",
json={"model": "gpt-4", "messages": [{"role": "user", "content": "hi"}]},
timeout=10,
)
assert response.status_code == 200
body = response.json()
assert body["choices"][0]["message"]["content"]
assert body["usage"]["total_tokens"] == 40
def test_moderations_route_parses_as_an_openai_response():
base: Final = ensure_fake_openai_endpoint()
response: Final = httpx.post(
f"{base}/v1/moderations",
json={"input": ["I want to harm someone", "hello"], "model": "omni-moderation-latest"},
timeout=10,
)
assert response.status_code == 200
parsed: Final = ModerationCreateResponse.model_validate(response.json())
assert parsed.model == "omni-moderation-latest"
assert len(parsed.results) == 2
assert parsed.results[0].categories.violence is False
def test_triton_embeddings_route():
base = ensure_fake_openai_endpoint()
response = httpx.post(f"{base}/triton/embeddings", json={"inputs": []}, timeout=10)
assert response.status_code == 200
output = response.json()["outputs"][0]
assert output["shape"] == [1, 2]
assert output["data"] == [0.1, 0.2]
def test_slow_model_blocks_past_client_timeout():
base = ensure_fake_openai_endpoint()
with pytest.raises(httpx.TimeoutException):
httpx.post(
f"{base}/v1/chat/completions",
json={
"model": "slow-endpoint",
"messages": [{"role": "user", "content": "hi"}],
},
timeout=0.5,
)
def test_remote_env_base_resolves_to_local(monkeypatch):
monkeypatch.setenv(
"FAKE_OPENAI_API_BASE",
"https://exampleopenaiendpoint-production.up.railway.app",
)
assert _resolve_base() == _LOCAL_DEFAULT
@pytest.mark.parametrize("host", ["127.0.0.1", "localhost", "[::1]"])
def test_loopback_env_base_is_honored(monkeypatch, host):
base = f"http://{host}:9191"
monkeypatch.setenv("FAKE_OPENAI_API_BASE", base)
assert _resolve_base() == base
@pytest.mark.parametrize("relative_path", _MIGRATED_FILES)
def test_migrated_files_have_no_live_hosted_mock(relative_path):
source = (_REPO_ROOT / relative_path).read_text()
assert not _LIVE_HOSTED_MOCK.search(source), (
f"{relative_path} references the hosted Railway mock; point api_base at "
"FAKE_OPENAI_API_BASE so CI does not depend on an external service"
)