"""Contract tests for f1api's connection-pool layer — py-10 already covers the request/response
shapes and the middleware, so these are new material: pool concurrency, draining on shutdown,
readiness tied to the pool instead of a model, and request ids surviving a real await point
(`pool.acquire()`) instead of the synchronous handoff py-10's stub model made trivial.

Every test builds its own `FakePool` (different tests need different `max_size` / delay to make
their assertion deterministic) and its own app around it — there is no shared `client` fixture.
Concurrency is real: httpx.AsyncClient over ASGITransport, no sockets, in one event loop.
"""

from __future__ import annotations

import asyncio
import re
from contextlib import asynccontextmanager
from pathlib import Path

from httpx import ASGITransport, AsyncClient

from f1api.app import create_app
from f1api.pool import FakePool


@asynccontextmanager
async def running(pool: FakePool):
    app = create_app(pool)
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        yield app, client


async def test_no_sync_db_calls_or_blocking_sleep_in_request_path():
    src = Path(__file__).resolve().parents[1] / "src"
    offenders = []
    for path in sorted(src.rglob("*.py")):
        text = path.read_text()
        if "psycopg2" in text:
            offenders.append(f"{path.name}: imports/uses psycopg2, a blocking driver")
        if re.search(r"\btime\.sleep\(", text):
            offenders.append(f"{path.name}: calls the blocking time.sleep")
    assert not offenders, "blocking call(s) found in the request path:\n" + "\n".join(offenders)

    # A static scan alone can't tell "no blocking calls" from "no calls at all" — an empty
    # starter passes it vacuously. Prove /ask actually goes through the pool for real too.
    pool = FakePool(min_size=1, max_size=2, acquire_delay=0.0)
    async with running(pool) as (_app, client):
        r = await client.post("/ask", json={"prompt": "ping"})
    assert r.status_code == 200, f"expected /ask to work at all, got {r.status_code}"
    assert pool.peak_in_use >= 1, "/ask never checked out a connection from the pool"


async def test_pool_size_is_respected_under_concurrent_requests():
    pool = FakePool(min_size=1, max_size=3, acquire_delay=0.03)
    async with running(pool) as (_app, client):
        await asyncio.gather(*(client.post("/ask", json={"prompt": f"q{i}"}) for i in range(9)))
    assert pool.peak_in_use == pool.max_size, (
        f"expected exactly {pool.max_size} connections checked out at once under contention "
        f"(9 requests, a pool of {pool.max_size}), peaked at {pool.peak_in_use} — either the "
        "pool cap leaked or /ask never went through the pool at all"
    )


async def test_100_concurrent_requests_complete_under_the_fake_pool():
    pool = FakePool(min_size=2, max_size=8, acquire_delay=0.005)
    async with running(pool) as (_app, client):
        results = await asyncio.gather(
            *(client.post("/ask", json={"prompt": f"q{i}"}) for i in range(100))
        )
    codes = [r.status_code for r in results]
    assert codes.count(200) == 100, f"expected all 100 requests to succeed, got {codes}"


async def test_graceful_shutdown_drains_inflight_requests():
    pool = FakePool(min_size=1, max_size=4, acquire_delay=0.05)
    async with running(pool) as (app, client):
        tasks = [
            asyncio.create_task(client.post("/ask", json={"prompt": f"q{i}"})) for i in range(4)
        ]
        await asyncio.sleep(0.01)  # let each request reach pool.acquire() and start its delay
        assert pool.in_use > 0, "test setup bug: requests should already hold a connection"
        results = await asyncio.gather(app.state.shutdown(), *tasks)
    responses = results[1:]
    codes = [r.status_code for r in responses]
    assert codes.count(200) == 4, (
        f"in-flight requests should all still succeed across a graceful shutdown, got {codes}"
    )
    assert pool.closed is True, "shutdown should close the pool once every request has drained"


async def test_request_ids_preserved_under_concurrent_requests():
    pool = FakePool(min_size=1, max_size=4, acquire_delay=0.02)
    ids = [f"1111111{i}-1111-4111-8111-11111111111{i}" for i in range(4)]
    async with running(pool) as (_app, client):

        async def call(rid: str):
            return rid, await client.post(
                "/ask", json={"prompt": "hi"}, headers={"X-Request-Id": rid}
            )

        results = await asyncio.gather(*(call(rid) for rid in ids))
    for rid, r in results:
        assert r.status_code == 200, f"{rid}: expected 200, got {r.status_code}"
        assert r.headers.get("x-request-id") == rid, (
            f"expected X-Request-Id {rid!r} echoed back, got {r.headers.get('x-request-id')!r}"
        )
        assert r.json().get("request_id") == rid, (
            f"expected the response body's request_id to match {rid!r}, got {r.json()!r} — "
            "a concurrent request's id leaked into this one"
        )


async def test_readyz_flips_to_503_when_pool_is_lost():
    pool = FakePool(min_size=1, max_size=2, acquire_delay=0.0)
    async with running(pool) as (_app, client):
        r1 = await client.get("/readyz")
        assert r1.status_code == 200, (
            f"expected 200 while the pool is healthy, got {r1.status_code}"
        )
        assert r1.json().get("ready") is True, f"expected ready: true, got {r1.json()!r}"

        pool.simulate_loss()
        r2 = await client.get("/readyz")
        assert r2.status_code == 503, f"expected 503 once the pool is lost, got {r2.status_code}"
        assert r2.json().get("ready") is False, f"expected ready: false, got {r2.json()!r}"