"""Regression tests for the #91912 salvage — credential-pool handlers off-loop
and the bounded Copilot token exchange.

The 2026-08-22 incident: ``GET /api/credentials/pool`` ran ``load_pool()`` on
the uvicorn event loop; for the copilot provider that reaches
``urllib.request.urlopen`` whose ``timeout`` does not bound ``getaddrinfo``,
so a networkless host froze the whole dashboard backend for 17 minutes.
"""

from __future__ import annotations

import threading
import time
from unittest.mock import patch

import pytest

from hermes_cli import copilot_auth
import hermes_cli.web_routers.ops as _rt_ops

# ---------------------------------------------------------------------------
# _urlopen_bounded
# ---------------------------------------------------------------------------

class TestUrlopenBounded:

    def test_reraises_worker_exception(self):
        with patch("urllib.request.urlopen", side_effect=OSError("boom")):
            with pytest.raises(OSError, match="boom"):
                copilot_auth._urlopen_bounded("req", 1.0)

    def test_hard_cap_fires_on_hung_resolver_and_closes_late_response(self, monkeypatch):
        """A urlopen that hangs past timeout + grace must raise TimeoutError
        promptly, and when the abandoned worker later *succeeds* it must close
        the response instead of leaking the socket."""
        monkeypatch.setattr(copilot_auth, "_DNS_GRACE_SECONDS", 0.05)
        release = threading.Event()
        closed = threading.Event()

        class _LateResponse:
            def close(self):
                closed.set()

        def hung_urlopen(req, timeout):
            release.wait(timeout=5)
            return _LateResponse()

        with patch("urllib.request.urlopen", side_effect=hung_urlopen):
            started = time.monotonic()
            with pytest.raises(TimeoutError, match="hard cap"):
                copilot_auth._urlopen_bounded("req", 0.05)
            elapsed = time.monotonic() - started
            assert elapsed < 2.0
            release.set()
            assert closed.wait(timeout=2), "late response was not closed"

# ---------------------------------------------------------------------------
# single-flight exchange
# ---------------------------------------------------------------------------

class TestExchangeSingleFlight:
    @pytest.fixture(autouse=True)
    def _clean_caches(self, monkeypatch, tmp_path):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
        copilot_auth._jwt_cache.clear()
        copilot_auth._exchange_failure_cache.clear()
        copilot_auth._exchange_locks.clear()
        yield
        copilot_auth._jwt_cache.clear()
        copilot_auth._exchange_failure_cache.clear()
        copilot_auth._exchange_locks.clear()

    def test_concurrent_callers_share_one_exchange(self, monkeypatch):
        """N concurrent callers for the same token must perform ONE network
        exchange; the rest wait on the lock and hit the populated cache."""
        monkeypatch.setattr(copilot_auth, "_load_jwt_from_disk", lambda fp: None)
        monkeypatch.setattr(copilot_auth, "_save_jwt_to_disk", lambda *a, **k: None)
        calls = []
        gate = threading.Event()

        class _Resp:
            def __enter__(self):
                return self

            def __exit__(self, *exc):
                return False

            def read(self):
                return b'{"token": "tid=1;exp=9", "expires_at": 4102444800}'

        def fake_bounded(req, timeout):
            calls.append(threading.get_ident())
            gate.wait(timeout=5)  # hold the first exchange open while others queue
            return _Resp()

        monkeypatch.setattr(copilot_auth, "_urlopen_bounded", fake_bounded)

        results = []
        threads = [
            threading.Thread(target=lambda: results.append(copilot_auth.exchange_copilot_token("ghu_" + "x" * 30)))
            for _ in range(8)
        ]
        for t in threads:
            t.start()
        time.sleep(0.2)  # let every caller reach the lock
        assert len(calls) == 1
        gate.set()
        for t in threads:
            t.join(timeout=5)

        assert len(calls) == 1
        assert len(results) == 8
        assert {r[0] for r in results} == {"tid=1;exp=9"}

    def test_waiters_observe_negative_cache_after_failed_exchange(self, monkeypatch):
        """When the single in-flight exchange fails, queued callers must not
        each start their own exchange — they see the negative cache."""
        monkeypatch.setattr(copilot_auth, "_load_jwt_from_disk", lambda fp: None)
        monkeypatch.setattr(copilot_auth, "_EXCHANGE_MAX_ATTEMPTS", 1)
        calls = []
        gate = threading.Event()

        def fake_bounded(req, timeout):
            calls.append(1)
            gate.wait(timeout=5)
            raise TimeoutError("hard cap")

        monkeypatch.setattr(copilot_auth, "_urlopen_bounded", fake_bounded)
        errors = []

        def run():
            try:
                copilot_auth.exchange_copilot_token("ghu_" + "y" * 30)
            except ValueError as exc:
                errors.append(str(exc))

        threads = [threading.Thread(target=run) for _ in range(5)]
        for t in threads:
            t.start()
        time.sleep(0.2)
        gate.set()
        for t in threads:
            t.join(timeout=5)

        assert len(calls) == 1
        assert len(errors) == 5
        assert any("recently failed" in e for e in errors)

    def test_negative_cache_short_circuits_before_taking_the_lock(self, monkeypatch):
        """While one exchange holds the lock, a caller whose fingerprint is
        already in the failure cache must raise immediately rather than park
        an executor thread behind the holder."""
        fp = copilot_auth._token_fingerprint("ghu_" + "z" * 30)
        copilot_auth._exchange_failure_cache[fp] = time.time() + 60
        lock = copilot_auth._exchange_lock_for(fp)
        lock.acquire()  # simulate an in-flight holder
        try:
            started = time.monotonic()
            with pytest.raises(ValueError, match="recently failed"):
                copilot_auth.exchange_copilot_token("ghu_" + "z" * 30)
            assert time.monotonic() - started < 0.5
        finally:
            lock.release()

# ---------------------------------------------------------------------------
# web_server credential-pool handlers off the loop
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_list_credential_pool_runs_off_event_loop(monkeypatch):
    import hermes_cli.auth as auth_mod

    loop_thread = threading.get_ident()
    seen = {}

    def fake_read_pool(*args, **kwargs):
        seen["thread"] = threading.get_ident()
        return {}

    monkeypatch.setattr(auth_mod, "read_credential_pool", fake_read_pool)
    result = await _rt_ops.list_credential_pool()

    assert result == {"providers": []}
    assert seen["thread"] != loop_thread