"""Contract tests for the FastAPI ask/stream service — httpx.AsyncClient over ASGITransport,
no real socket, for seven of them; the streaming-is-incremental test speaks ASGI to the app
directly, because ASGITransport buffers the whole body before the client sees any of it. The
ninth builds and runs the actual Docker image; it is marked slow and skips itself when no
docker daemon answers — a passing suite on a machine without docker is 8 passed, 1 skipped.
(verify-local-exercise.mjs counts a skip as a failure, so on such a machine it is run with
VERIFY_K="not dockerfile", which it reports as a filtered, narrower certification.)
"""

from __future__ import annotations

import asyncio
import re
import shutil
import subprocess
import time
import urllib.error
import urllib.request
from pathlib import Path

import pytest

UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.I)


async def test_ask_returns_200_with_answer_and_request_id(client):
    r = await client.post("/ask", json={"prompt": "say hi"})
    assert r.status_code == 200, f"expected 200, got {r.status_code}: {r.text}"
    body = r.json()
    assert isinstance(body.get("answer"), str) and body["answer"], (
        f"expected a non-empty 'answer' string, got {body!r}"
    )
    assert body.get("request_id"), f"response is missing a non-empty 'request_id': {body!r}"


async def test_ask_missing_prompt_returns_422(client):
    r = await client.post("/ask", json={})
    assert r.status_code == 422, f"missing 'prompt' should 422, got {r.status_code}: {r.text}"


async def test_stream_sets_event_stream_content_type_and_five_tokens_then_done(client):
    events: list[str] = []
    async with client.stream("GET", "/stream", params={"prompt": "say hi"}) as r:
        ctype = r.headers.get("content-type", "")
        assert ctype.startswith("text/event-stream"), f"expected text/event-stream, got {ctype!r}"
        buf = ""
        async for chunk in r.aiter_text():
            buf += chunk
    for block in buf.strip().split("\n\n"):
        lines = [ln for ln in block.split("\n") if ln]
        if lines:
            events.append(lines[0])
    tokens = [e for e in events if e == "event: token"]
    assert len(tokens) == 5, f"expected 5 'event: token' frames, got {len(tokens)}: {events!r}"
    assert events[-1] == "event: done", (
        f"expected the stream to end with 'event: done', got {events!r}"
    )


async def test_stream_delivers_chunks_incrementally_not_buffered(app, model):
    # httpx's ASGITransport runs the app to completion and only then hands back the collected
    # body, so a client-side read cannot tell a stream from a buffer. Speak ASGI to the app
    # directly instead and read the model's counter at the moment the first body chunk leaves.
    scope = {
        "type": "http", "asgi": {"version": "3.0"}, "http_version": "1.1", "method": "GET",
        "scheme": "http", "path": "/stream", "raw_path": b"/stream", "root_path": "",
        "query_string": b"prompt=say+hi", "headers": [(b"host", b"test")],
        "client": ("127.0.0.1", 50000), "server": ("test", 80),
    }
    request_sent = False

    async def receive():
        nonlocal request_sent
        if not request_sent:
            request_sent = True
            return {"type": "http.request", "body": b"", "more_body": False}
        await asyncio.Event().wait()  # the client never hangs up; the server cancels this wait

    first: tuple[bytes, int] | None = None

    async def send(message):
        nonlocal first
        if message["type"] == "http.response.body" and message.get("body") and first is None:
            first = (message["body"], model.emitted)

    await app(scope, receive, send)
    assert first is not None, "never observed a chunk from the stream"
    body, emitted = first
    assert body.startswith(b"event: token"), (
        f"expected the first chunk to be a token frame, got {body!r}"
    )
    assert emitted < 5, (
        f"the model had already emitted all {emitted} tokens by the time the first chunk "
        "left the app — the response is buffering the whole body instead of streaming it"
    )


async def test_healthz_ok_and_readyz_503_until_loaded(client, model):
    h = await client.get("/healthz")
    assert h.status_code == 200, f"/healthz should always be 200, got {h.status_code}"

    r1 = await client.get("/readyz")
    assert r1.status_code == 503, f"/readyz before load() should be 503, got {r1.status_code}"
    assert r1.json().get("ready") is False, (
        f"/readyz body should say ready: false, got {r1.json()!r}"
    )

    model.load()
    r2 = await client.get("/readyz")
    assert r2.status_code == 200, f"/readyz after load() should be 200, got {r2.status_code}"
    assert r2.json().get("ready") is True, f"/readyz body should say ready: true, got {r2.json()!r}"


async def test_incoming_request_id_is_echoed_in_response(client):
    rid = "11111111-1111-4111-8111-111111111111"
    r = await client.post("/ask", json={"prompt": "hi"}, headers={"X-Request-Id": rid})
    assert r.headers.get("x-request-id") == rid, (
        "expected the incoming X-Request-Id echoed back, "
        f"got header {r.headers.get('x-request-id')!r}"
    )
    assert r.json().get("request_id") == rid, (
        f"expected the response body's request_id to match the incoming header, got {r.json()!r}"
    )


async def test_generated_request_id_is_valid_uuid4(client):
    r = await client.post("/ask", json={"prompt": "hi"})
    rid = r.json().get("request_id", "")
    assert UUID_RE.match(rid), f"expected request_id to look like a uuid4, got {rid!r}"
    assert r.headers.get("x-request-id") == rid, (
        "expected the same id on the X-Request-Id response header, "
        f"got {r.headers.get('x-request-id')!r}"
    )


async def test_log_line_carries_same_request_id_as_response(client, caplog):
    caplog.set_level("INFO", logger="askapi")
    r = await client.post("/ask", json={"prompt": "hi"})
    rid = r.json()["request_id"]
    ids_seen = [getattr(rec, "request_id", None) for rec in caplog.records]
    assert rid in ids_seen, f"no log record carried request_id={rid!r} — records had {ids_seen!r}"


def _docker_daemon_up() -> bool:
    if not shutil.which("docker"):
        return False
    try:
        r = subprocess.run(["docker", "info"], capture_output=True, timeout=10)
        return r.returncode == 0
    except Exception:
        return False


@pytest.mark.slow
@pytest.mark.skipif(not _docker_daemon_up(), reason="docker daemon not available")
def test_dockerfile_builds_and_serves_healthz():
    root = Path(__file__).resolve().parents[1]
    tag = "askapi-verify:latest"
    build = subprocess.run(
        ["docker", "build", "-t", tag, str(root)], capture_output=True, text=True, timeout=600
    )
    assert build.returncode == 0, (
        f"docker build failed:\n{build.stdout[-2000:]}\n{build.stderr[-2000:]}"
    )
    run = subprocess.run(
        ["docker", "run", "--rm", "-d", "-p", "18099:8000", tag],
        capture_output=True, text=True, timeout=30,
    )
    assert run.returncode == 0, f"docker run failed: {run.stderr}"
    cid = run.stdout.strip()
    try:
        ok, last_err = False, None
        for _ in range(40):
            try:
                with urllib.request.urlopen("http://127.0.0.1:18099/healthz", timeout=1) as resp:
                    body = resp.read()
                    assert resp.status == 200, f"/healthz returned {resp.status}"
                    assert b'"status"' in body, f"/healthz body is missing 'status': {body!r}"
                    ok = True
                    break
            except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
                last_err = e
                time.sleep(0.5)
        assert ok, f"the container never answered /healthz on :18099: {last_err}"
    finally:
        subprocess.run(["docker", "rm", "-f", cid], capture_output=True)