test(proxy_behavior): scaffold session-scoped async ASGI client + liveness smoke

Slice 2 of the management-endpoints behavior-pinning effort. New top-level dir
tests/proxy_behavior/management/ outside every existing pytest glob.

conftest.py initialises the proxy app once per session against the DATABASE_URL
the harness boots Postgres at, wraps it in httpx.AsyncClient via in-process
ASGITransport. The one smoke test asserts /health/liveliness returns 200, which
exercises the full FastAPI middleware stack against a real app — no mocks.

Plan: https://www.notion.so/36643b8acdab8128a581ced0f6a4744d
This commit is contained in:
Yuneng Jiang 2026-05-19 21:21:00 -07:00
parent e59e34bed3
commit 8af57a0873
No known key found for this signature in database
4 changed files with 61 additions and 0 deletions

View file

View file

@ -0,0 +1,55 @@
"""Session-scoped async ASGI client for behavior-pinning tests.
The proxy app is initialised once per pytest session against the real Postgres
pointed at by ``DATABASE_URL``. No mocks: auth runs, prisma runs, integrations
run. Tests assert at the HTTP boundary.
"""
import os
import tempfile
from typing import AsyncIterator
import httpx
import pytest_asyncio
import yaml
MASTER_KEY = "sk-1234"
def _write_minimal_proxy_config() -> str:
config = {
"general_settings": {"master_key": MASTER_KEY},
"litellm_settings": {},
}
database_url = os.environ.get("DATABASE_URL")
if database_url:
config["general_settings"]["database_url"] = database_url
f = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False)
yaml.dump(config, f)
f.close()
return f.name
@pytest_asyncio.fixture(scope="session")
async def proxy_app():
from litellm.proxy.proxy_server import (
app,
cleanup_router_config_variables,
initialize,
)
cleanup_router_config_variables()
config_path = _write_minimal_proxy_config()
await initialize(config=config_path)
yield app
@pytest_asyncio.fixture(scope="session")
async def proxy_client(proxy_app) -> AsyncIterator[httpx.AsyncClient]:
transport = httpx.ASGITransport(app=proxy_app)
async with httpx.AsyncClient(
transport=transport, base_url="http://testserver"
) as client:
yield client

View file

@ -0,0 +1,6 @@
"""Smoke tests proving the harness boots and talks to the proxy app."""
async def test_liveliness(proxy_client):
resp = await proxy_client.get("/health/liveliness")
assert resp.status_code == 200