vllm-frontdoor/router/tests/test_paths.py

"""Allowlist: only /v1/*, /health (and /metrics when enabled) are public."""

from __future__ import annotations

import httpx

from app import build_apps
from conftest import raw_asgi

DEV_ENDPOINTS = [
    "/sleep",
    "/wake_up",
    "/collective_rpc",
    "/reset_prefix_cache",
    "/is_sleeping",
]
ADMIN_PATHS = ["/admin/status", "/admin/wake/text", "/admin/sleep/text?level=1"]


async def test_dev_endpoints_404_on_public(pub, backend):
    for path in DEV_ENDPOINTS:
        r = await pub.post(path)
        assert r.status_code == 404, path
        assert r.json()["error"]["code"] == "not_found"
    for path in DEV_ENDPOINTS:
        r = await pub.get(path)
        assert r.status_code == 404, path
    assert backend.calls == []


async def test_admin_not_on_public(pub, backend):
    for path in ADMIN_PATHS:
        r = await pub.request("POST", path)
        assert r.status_code == 404, path
        r = await pub.get(path)
        assert r.status_code == 404, path
    assert backend.calls == []


async def test_docs_and_openapi_404(pub):
    for path in ("/docs", "/redoc", "/openapi.json", "/"):
        r = await pub.get(path)
        assert r.status_code == 404, path


async def test_dotted_paths_404(pub):
    for path in ("/v1/../sleep", "/v1/..%2fsleep", "/health/../sleep", "/./sleep"):
        r = await pub.post(path)
        assert r.status_code == 404, path


async def test_raw_traversal_404(stack, backend):
    """Percent-encoded traversal that a normal client would normalise."""
    cases = [
        ("POST", "/v1%2f..%2fsleep"),
        ("POST", "/v1%2F..%2Fsleep"),
        ("GET", "/v1/%2e%2e/wake_up"),
        ("POST", "/v1/%2e%2e%2fcollective_rpc"),
        ("POST", "/v1//../sleep"),
        ("GET", "/v1/models/../../is_sleeping"),
    ]
    for method, raw in cases:
        status, body = await raw_asgi(stack.public, method, raw,
                                      headers=[("content-type", "application/json")],
                                      body=b"{}")
        assert status == 404, (method, raw, body)
    assert backend.calls == []


async def test_normalised_traversal_still_reaches_v1(stack, backend):
    """A traversal that lands inside /v1 must still work (not over-block)."""
    status, body = await raw_asgi(
        stack.public, "POST", "/v1/../v1/chat/completions",
        headers=[("content-type", "application/json")],
        body=b'{"model": "text"}',
    )
    assert status == 200, body
    assert backend.count("POST vllm-text/v1/chat/completions") == 1


async def test_public_health(pub, backend):
    r = await pub.get("/health")
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "ok"
    assert set(body["services"]) == {"text", "ocr", "embed"}
    # liveness must not depend on backend state
    backend.services["vllm-text"]["reachable"] = False
    r = await pub.get("/health")
    assert r.status_code == 200


async def test_public_models_lists_all_three(pub, backend):
    r = await pub.get("/v1/models")
    assert r.status_code == 200
    body = r.json()
    assert body["object"] == "list"
    assert [m["id"] for m in body["data"]] == [
        "Qwen3.6-35B-A3B-FP8", "OvisOCR2", "Qwen3-Embedding-8B",
    ]
    # answered by the router itself, no backend contact
    assert backend.calls == []


async def test_metrics_disabled_by_default(pub):
    assert (await pub.get("/metrics")).status_code == 404


async def test_metrics_enabled_when_configured(stack, pub):
    stack.cfg.metrics_enabled = True
    r = await pub.get("/metrics")
    assert r.status_code == 200
    assert "vllm_router_service_depth" in r.text


async def test_unknown_http_method_on_v1(stack):
    status, _ = await raw_asgi(stack.public, "PROPFIND", "/v1/chat/completions")
    assert status in (404, 405)


async def test_admin_listener_is_a_separate_app(stack):
    """The admin app answers /admin/*; the public app must not."""
    status, body = await raw_asgi(stack.admin, "GET", "/admin/status")
    assert status == 200
    status, _ = await raw_asgi(stack.public, "GET", "/admin/status")
    assert status == 404


async def test_request_body_cap(cfg, backend):
    """ROUTER_MAX_BODY_BYTES must be wired as middleware (413, not OOM)."""
    cfg.max_body_bytes = 64
    public_app, _admin, manager = build_apps(cfg, transport=backend)
    ok_body = b'{"model": "text", "messages": []}'
    assert len(ok_body) < 64
    try:
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=public_app),
                                     base_url="http://router.test") as client:
            r = await client.post("/v1/chat/completions", content=ok_body,
                                  headers={"content-type": "application/json"})
            assert r.status_code == 200
            r = await client.post("/v1/chat/completions", content=ok_body + b" " * 512,
                                  headers={"content-type": "application/json"})
            assert r.status_code == 413
    finally:
        await manager.proxy_client.aclose()
        await manager.ctrl_client.aclose()