"""Shutdown behaviour (E2E case 12) and the vllmctl CLI glue."""

from __future__ import annotations

import asyncio
import json
import os
import subprocess

import pytest

from conftest import raw_asgi
from routing import ShutdownGuardMiddleware


async def test_shutdown_guard_returns_503_when_not_started(stack):
    """uvicorn cancels in-flight tasks after timeout_graceful_shutdown; that
    CancelledError must become a depth-aware 503, not uvicorn's bare 500."""
    stack.manager.begin_shutdown()

    async def app(scope, receive, send):
        raise asyncio.CancelledError()

    guard = ShutdownGuardMiddleware(app, stack.manager)
    status, body = await raw_asgi(guard, "POST", "/v1/chat/completions")
    assert status == 503
    payload = json.loads(body)
    assert payload["error"]["code"] == "router_shutting_down"
    assert payload["error"]["retry_after_seconds"] == 5


async def test_shutdown_guard_passes_through_when_not_shutting_down(stack):
    class Boom(RuntimeError):
        pass

    async def app(scope, receive, send):
        raise Boom("nope")

    guard = ShutdownGuardMiddleware(app, stack.manager)
    with pytest.raises(Boom):
        await raw_asgi(guard, "POST", "/v1/chat/completions")


async def test_shutdown_guard_cannot_unsend_a_started_response(stack):
    """A response that already started streaming is closed, not replaced."""
    stack.manager.begin_shutdown()

    async def app(scope, receive, send):
        await send({"type": "http.response.start", "status": 200, "headers": []})
        await send({"type": "http.response.body", "body": b"partial"})
        raise asyncio.CancelledError()

    guard = ShutdownGuardMiddleware(app, stack.manager)
    with pytest.raises(asyncio.CancelledError):
        await raw_asgi(guard, "POST", "/v1/chat/completions")


async def test_shutdown_guard_forwards_normal_requests(stack):
    async def app(scope, receive, send):
        await send({"type": "http.response.start", "status": 204, "headers": []})
        await send({"type": "http.response.body", "body": b""})

    guard = ShutdownGuardMiddleware(app, stack.manager)
    status, _ = await raw_asgi(guard, "GET", "/health")
    assert status == 204


def test_vllmctl_service_name_mapping():
    """`logs router` / `restart router` must map to the compose service
    `router`; the model services map to vllm-*."""
    script = (
        "source /data/home/renbaibing/vllm/vllmctl >/dev/null 2>&1; "
        'for n in router vllm-router Router text vllm-text TEXT ocr OvisOCR2 '
        "embed Qwen3-Embedding-8B bogus; do "
        'printf "%s:%s\\n" "$n" "$(docker_service_for "$n" || echo ERR)"; done'
    )
    out = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
                         timeout=30)
    assert out.returncode == 0, out.stderr
    mapping = dict(line.split(":", 1) for line in out.stdout.strip().splitlines())
    assert mapping["router"] == "router"
    assert mapping["vllm-router"] == "router"
    assert mapping["Router"] == "router"
    assert mapping["text"] == "vllm-text"
    assert mapping["vllm-text"] == "vllm-text"
    assert mapping["TEXT"] == "vllm-text"
    assert mapping["ocr"] == "vllm-ocr"
    assert mapping["OvisOCR2"] == "vllm-ocr"
    assert mapping["embed"] == "vllm-embed"
    assert mapping["Qwen3-Embedding-8B"] == "vllm-embed"
    assert mapping["bogus"] == "ERR"


def test_vllmctl_has_no_removed_machinery():
    source = open("/data/home/renbaibing/vllm/vllmctl", encoding="utf-8").read()
    assert ".runas.py" not in source
    assert ".user.env" not in source
    assert "idle-watch on" not in source
    assert "cmd_idle_watch" not in source
    assert "admin/wake/" in source and "admin/sleep/" in source
    assert os.access("/data/home/renbaibing/vllm/vllmctl", os.X_OK)