"""unit test: a session's close cannot park the caller indefinitely

run from the repo root:
    uv run pytest tests/test_session_close_budget.py

offline — no network. curl_cffi's close awaits the multi handle with no timeout
of its own, so a wedged transfer would park whatever closes the session,
including an `async with` exit. These build a real session and wedge the
backend's close, asserting the bound lives on the session itself so every call
site gets it without arranging anything.
"""

import asyncio
import logging

from aioproxies import AioProxies

from monitor_utils import sessions
from monitor_utils.sessions import BoundedCloseSession, get_default_session


def _proxies(tmp_path) -> AioProxies:
    """A one-entry pool in the host:port:user:pass format monitors ship."""
    proxy_file = tmp_path / "proxies.txt"
    proxy_file.write_text("proxy.example.io:8080:user:secret\n")
    return AioProxies.from_file(path=str(proxy_file))


async def _wedge_close(session) -> asyncio.Event:
    """Replace the backend's close with one that never returns.

    Builds the backend session first so TLSSession.close actually reaches the
    backend rather than short-circuiting on an unbuilt session.
    """

    session.session  # force lazy construction
    started = asyncio.Event()

    async def wedged(inner_session):
        started.set()
        await asyncio.sleep(3600)

    session.backend.close = wedged
    return started


async def _really_close(session, original_close):
    """Close for real, so the test does not leak the curl handle it allocated."""
    session.backend.close = original_close
    await session.close()


def test_close_is_abandoned_when_it_exceeds_the_budget(monkeypatch, caplog, tmp_path):
    """A close that never returns is abandoned and logged, not awaited forever."""

    monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05)
    proxies = _proxies(tmp_path)

    async def run():
        session = get_default_session(proxies, timeout=8)
        original_close = session.backend.close
        started = await _wedge_close(session)

        # wait_for is the assertion: an unbounded close never returns here
        await asyncio.wait_for(session.close(), timeout=5)

        assert started.is_set()
        await _really_close(session, original_close)

    with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
        asyncio.run(run())

    assert any(
        record.message == "session close exceeded budget, abandoning session"
        for record in caplog.records
    )


def test_async_with_exit_is_bounded(monkeypatch, caplog, tmp_path):
    """The bound applies through `async with`, which is how monitors close."""

    monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05)
    proxies = _proxies(tmp_path)

    async def run():
        session = get_default_session(proxies, timeout=8)
        original_close = session.backend.close
        started = await _wedge_close(session)

        async with session:
            pass

        assert started.is_set()
        await _really_close(session, original_close)

    with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
        asyncio.run(asyncio.wait_for(run(), timeout=5))

    assert any(
        record.message == "session close exceeded budget, abandoning session"
        for record in caplog.records
    )


def test_healthy_close_completes_and_is_not_logged(caplog, tmp_path):
    """The normal path closes fully and reports nothing."""

    proxies = _proxies(tmp_path)

    async def run():
        session = get_default_session(proxies, timeout=8)
        session.session  # force lazy construction so close has work to do
        await session.close()
        return session

    with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
        session = asyncio.run(run())

    assert session._is_closed()
    assert not caplog.records


def test_get_default_session_returns_a_bounded_session(tmp_path):
    """The factory hands out the bounded type, not a bare TLSSession."""

    proxies = _proxies(tmp_path)

    async def run():
        session = get_default_session(proxies, timeout=8)
        try:
            return isinstance(session, BoundedCloseSession)
        finally:
            await session.close()

    assert asyncio.run(run())