import asyncio
from unittest.mock import AsyncMock

import httpx
import pytest

from src import gitea_proxy
from src import main


@pytest.mark.anyio
async def test_log_issue_time_uses_canonical_shared_issue_endpoint_and_seconds():
    requests = []

    async def handler(request):
        requests.append(request)
        return httpx.Response(201, json={"id": 1})

    gitea_proxy.start_client(transport=httpx.MockTransport(handler))
    try:
        await gitea_proxy.log_issue_time("pull:stackchain/stackchain-dashboard:586:", 2520)
    finally:
        await gitea_proxy.stop_client()

    assert len(requests) == 1
    assert requests[0].method == "POST"
    assert requests[0].url.path == "/api/v1/repos/stackchain/stackchain-dashboard/issues/586/times"
    assert requests[0].read() == b'{"time":2520}'


@pytest.mark.anyio
async def test_gitea_transport_is_reused_across_requests_and_closed():
    client_ids = []

    async def handler(request):
        client_ids.append(id(gitea_proxy._client))
        return httpx.Response(200, json={"path": request.url.path})

    client = gitea_proxy.start_client(transport=httpx.MockTransport(handler))
    try:
        first = await gitea_proxy.fetch("user")
        second = await gitea_proxy.fetch("user/repos")
    finally:
        await gitea_proxy.stop_client()

    assert first == {"path": "/api/v1/user"}
    assert second == {"path": "/api/v1/user/repos"}
    assert client_ids == [id(client), id(client)]
    assert client.is_closed


@pytest.mark.anyio
async def test_gitea_transport_coalesces_identical_concurrent_gets():
    calls = 0
    release = asyncio.Event()

    async def handler(request):
        nonlocal calls
        calls += 1
        await release.wait()
        return httpx.Response(200, json={"path": request.url.path})

    gitea_proxy.start_client(
        transport=httpx.MockTransport(handler),
        max_concurrency=4,
        admission_timeout=0.1,
    )
    try:
        requests = [asyncio.create_task(gitea_proxy.fetch("user")) for _ in range(20)]
        await asyncio.sleep(0)
        release.set()
        results = await asyncio.gather(*requests)
    finally:
        await gitea_proxy.stop_client()

    assert calls == 1
    assert results == [{"path": "/api/v1/user"}] * 20


@pytest.mark.anyio
async def test_cancelling_one_coalesced_waiter_keeps_the_shared_get_running():
    started = asyncio.Event()
    release = asyncio.Event()

    async def handler(request):
        started.set()
        await release.wait()
        return httpx.Response(200, json={"path": request.url.path})

    gitea_proxy.start_client(transport=httpx.MockTransport(handler))
    owner = asyncio.create_task(gitea_proxy.fetch("user"))
    try:
        await started.wait()
        waiter = asyncio.create_task(gitea_proxy.fetch("user"))
        await asyncio.sleep(0)
        waiter.cancel()
        with pytest.raises(asyncio.CancelledError):
            await waiter
        release.set()
        assert await owner == {"path": "/api/v1/user"}
    finally:
        release.set()
        await asyncio.gather(owner, return_exceptions=True)
        await gitea_proxy.stop_client()


@pytest.mark.anyio
async def test_gitea_transport_rejects_reads_beyond_bounded_capacity():
    active = 0
    peak = 0
    saturated = asyncio.Event()
    release = asyncio.Event()

    async def handler(request):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        if active == 2:
            saturated.set()
        try:
            await release.wait()
            return httpx.Response(200, json={"path": request.url.path})
        finally:
            active -= 1

    gitea_proxy.start_client(
        transport=httpx.MockTransport(handler),
        max_concurrency=3,
        admission_timeout=0.02,
    )
    first = asyncio.create_task(gitea_proxy.fetch("first"))
    second = asyncio.create_task(gitea_proxy.fetch("second"))
    try:
        await asyncio.wait_for(saturated.wait(), timeout=0.2)
        with pytest.raises(gitea_proxy.GiteaOverloadedError):
            await gitea_proxy.fetch("third")
    finally:
        release.set()
        await asyncio.gather(first, second)
        await gitea_proxy.stop_client()

    assert peak == 2


@pytest.mark.anyio
async def test_gitea_transport_reserves_one_slot_for_mutations_and_bounds_them():
    active = 0
    peak = 0
    reads_started = asyncio.Event()
    mutation_started = asyncio.Event()
    release = asyncio.Event()

    async def handler(request):
        nonlocal active, peak
        active += 1
        peak = max(peak, active)
        if active == 2:
            reads_started.set()
        if request.method == "POST":
            mutation_started.set()
        try:
            await release.wait()
            return httpx.Response(200, json={"method": request.method})
        finally:
            active -= 1

    client = gitea_proxy.start_client(
        transport=httpx.MockTransport(handler),
        max_concurrency=3,
        admission_timeout=0.02,
    )
    reads = [
        asyncio.create_task(gitea_proxy.fetch("first")),
        asyncio.create_task(gitea_proxy.fetch("second")),
    ]
    mutation = None
    try:
        await asyncio.wait_for(reads_started.wait(), timeout=0.2)
        mutation = asyncio.create_task(client.post("/api/v1/mutate", json={"value": 1}))
        await asyncio.wait_for(mutation_started.wait(), timeout=0.2)
        with pytest.raises(gitea_proxy.GiteaOverloadedError):
            await asyncio.wait_for(
                client.post("/api/v1/mutate", json={"value": 1}), timeout=0.1
            )
    finally:
        release.set()
        await asyncio.gather(*reads, *(task for task in [mutation] if task is not None))
        await gitea_proxy.stop_client()

    assert peak == 3


@pytest.mark.anyio
async def test_gitea_transport_reads_bulkhead_configuration_from_environment(monkeypatch):
    monkeypatch.setenv("GITEA_MAX_CONCURRENCY", "12")
    monkeypatch.setenv("GITEA_ADMISSION_TIMEOUT_SECONDS", "0.75")

    client = gitea_proxy.start_client(
        transport=httpx.MockTransport(lambda request: httpx.Response(200, json={}))
    )
    try:
        assert client.max_concurrency == 12
        assert client.admission_timeout == 0.75
    finally:
        await gitea_proxy.stop_client()


@pytest.mark.anyio
async def test_streaming_reads_share_the_same_bounded_read_capacity():
    active = 0
    saturated = asyncio.Event()
    release = asyncio.Event()

    async def handler(request):
        nonlocal active
        active += 1
        if active == 2:
            saturated.set()
        try:
            await release.wait()
            return httpx.Response(200, content=b"diff")
        finally:
            active -= 1

    gitea_proxy.start_client(
        transport=httpx.MockTransport(handler),
        max_concurrency=3,
        admission_timeout=0.02,
    )
    streams = [
        asyncio.create_task(gitea_proxy.fetch_text("first", 100)),
        asyncio.create_task(gitea_proxy.fetch_text("second", 100)),
    ]
    try:
        await asyncio.wait_for(saturated.wait(), timeout=0.2)
        with pytest.raises(gitea_proxy.GiteaOverloadedError):
            await asyncio.wait_for(gitea_proxy.fetch("third"), timeout=0.1)
    finally:
        release.set()
        await asyncio.gather(*streams)
        await gitea_proxy.stop_client()


@pytest.mark.anyio
async def test_application_lifespan_opens_and_closes_gitea_transport(monkeypatch):
    calls = []

    monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
    monkeypatch.setattr(main, "_check_readiness", AsyncMock())

    async def stop_client():
        calls.append("stop")

    monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)

    async with main.app.router.lifespan_context(main.app):
        assert calls == ["start"]

    assert calls == ["start", "stop"]


@pytest.mark.anyio
async def test_application_shutdown_finishes_snapshot_before_closing_transport(monkeypatch):
    calls = []
    started = asyncio.Event()

    async def active_snapshot():
        started.set()
        try:
            await asyncio.Event().wait()
        finally:
            calls.append("snapshot cancelled")

    monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
    monkeypatch.setattr(main, "_check_readiness", AsyncMock())

    async def stop_client():
        assert main._live_snapshot_task is not None
        state = "done" if main._live_snapshot_task.done() else "active"
        calls.append(f"stop ({state})")

    monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)

    try:
        async with main.app.router.lifespan_context(main.app):
            main._live_snapshot_task = asyncio.create_task(active_snapshot())
            await started.wait()
    finally:
        task = main._live_snapshot_task
        if task is not None and not task.done():
            task.cancel()
            with pytest.raises(asyncio.CancelledError):
                await task
        main._live_snapshot_task = None

    assert calls == ["start", "snapshot cancelled", "stop (done)"]


@pytest.mark.anyio
async def test_application_shutdown_cancels_available_work_scan_before_transport(monkeypatch):
    calls = []
    started = asyncio.Event()

    async def active_scan():
        started.set()
        try:
            await asyncio.Event().wait()
        finally:
            calls.append("available scan cancelled")

    monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
    monkeypatch.setattr(main, "_check_readiness", AsyncMock())

    async def stop_client():
        task = main._available_issue_snapshot_task
        calls.append(f"stop ({'done' if task is not None and task.done() else 'active'})")

    monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)

    try:
        async with main.app.router.lifespan_context(main.app):
            main._available_issue_snapshot_task = asyncio.create_task(active_scan())
            await started.wait()
    finally:
        task = main._available_issue_snapshot_task
        if task is not None and not task.done():
            task.cancel()
            with pytest.raises(asyncio.CancelledError):
                await task
        main._available_issue_snapshot_task = None

    assert calls == ["start", "available scan cancelled", "stop (done)"]


@pytest.mark.anyio
async def test_application_shutdown_drains_authored_mutation_before_transport(monkeypatch):
    calls = []
    started = asyncio.Event()
    release = asyncio.Event()

    async def authored_mutation():
        started.set()
        await release.wait()
        calls.append("mutation persisted")
        return {"id": 461}

    monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
    monkeypatch.setattr(main, "_check_readiness", AsyncMock())

    async def stop_client():
        calls.append("stop")

    monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)
    context = main.app.router.lifespan_context(main.app)
    await context.__aenter__()
    task = asyncio.create_task(authored_mutation())
    main._authored_action_operations["shutdown-drain-461"] = (
        ("comment", "stackchain/dashboard", 461),
        task,
        0.0,
    )
    await started.wait()

    shutdown = asyncio.create_task(context.__aexit__(None, None, None))
    await asyncio.sleep(0)

    assert not shutdown.done()
    assert calls == ["start"]

    release.set()
    await shutdown

    assert calls == ["start", "mutation persisted", "stop"]


@pytest.mark.anyio
async def test_application_shutdown_bounds_and_settles_stalled_authored_mutation(monkeypatch):
    calls = []
    started = asyncio.Event()

    async def stalled_mutation():
        started.set()
        try:
            await asyncio.Event().wait()
        finally:
            calls.append("mutation cancelled")

    monkeypatch.setattr(main, "AUTHORED_ACTION_SHUTDOWN_GRACE_SECONDS", 0.01)
    monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
    monkeypatch.setattr(main, "_check_readiness", AsyncMock())

    async def stop_client():
        calls.append("stop")

    monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)
    context = main.app.router.lifespan_context(main.app)
    await context.__aenter__()
    task = asyncio.create_task(stalled_mutation())
    main._authored_action_operations["stalled-shutdown-461"] = (
        ("review", "stackchain/dashboard", 461),
        task,
        0.0,
    )
    await started.wait()

    await asyncio.wait_for(context.__aexit__(None, None, None), timeout=0.2)

    assert task.cancelled()
    assert main._authored_action_operations == {}
    assert calls == ["start", "mutation cancelled", "stop"]