"""Diagnostics for pytest tests that stop making progress.

This watchdog only reports. ``pytest-timeout`` owns the later hard failure, so
CI gets useful context before a wedged process is terminated fail-closed.
"""

from __future__ import annotations

import asyncio
import faulthandler
import inspect
import io
import sys
import threading
import time
import traceback
from typing import IO, Any, cast

import pytest
import pytest_asyncio


class HangWatchdog:
    """Track the active test and emit at most one diagnostic for it."""

    def __init__(self, threshold_seconds: float, *, stream: IO[str] | None = None) -> None:
        self.threshold_seconds = threshold_seconds
        self.stream: IO[str] = stream or sys.__stderr__ or sys.stderr
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None
        self._nodeid: str | None = None
        self._started_at = 0.0
        self._reported = False
        self._loop: asyncio.AbstractEventLoop | None = None

    def start(self) -> None:
        if self.threshold_seconds <= 0:
            return
        self._thread = threading.Thread(
            target=self._run,
            name="pytest-hang-watchdog",
            daemon=True,
        )
        self._thread.start()

    def stop(self) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join(timeout=2)

    def test_started(self, nodeid: str, *, now: float | None = None) -> None:
        with self._lock:
            self._nodeid = nodeid
            self._started_at = time.monotonic() if now is None else now
            self._reported = False
            self._loop = None

    def test_finished(self, nodeid: str) -> None:
        with self._lock:
            if self._nodeid == nodeid:
                self._nodeid = None
                self._loop = None

    def register_loop(self, loop: asyncio.AbstractEventLoop) -> None:
        with self._lock:
            self._loop = loop

    def unregister_loop(self, loop: asyncio.AbstractEventLoop) -> None:
        with self._lock:
            if self._loop is loop:
                self._loop = None

    def check(self, *, now: float | None = None) -> bool:
        """Emit diagnostics if due; public to allow deterministic tests."""
        current_time = time.monotonic() if now is None else now
        with self._lock:
            if (
                self._nodeid is None
                or self._reported
                or current_time - self._started_at < self.threshold_seconds
            ):
                return False
            self._reported = True
            nodeid = self._nodeid
            elapsed = current_time - self._started_at
            loop = self._loop

        print(
            "\n===== PYTEST HANG DIAGNOSTIC =====\n"
            f"current_test={nodeid}\n"
            f"elapsed_seconds={elapsed:.1f}\n"
            "Python thread stacks follow:",
            file=self.stream,
            flush=True,
        )
        try:
            faulthandler.dump_traceback(file=cast(Any, self.stream), all_threads=True)
        except (AttributeError, io.UnsupportedOperation):
            # StringIO-like streams used by unit tests have no file descriptor.
            traceback.print_stack(file=self.stream)

        if loop is not None and loop.is_running():
            loop.call_soon_threadsafe(_dump_asyncio_tasks, loop, nodeid, self.stream)
        else:
            print("No running asyncio loop registered for this test.", file=self.stream)
            _print_diagnostic_end(self.stream)
        return True

    def _run(self) -> None:
        interval = min(max(self.threshold_seconds / 10, 0.1), 1.0)
        while not self._stop.wait(interval):
            self.check()


def _dump_asyncio_tasks(
    loop: asyncio.AbstractEventLoop,
    nodeid: str,
    stream: IO[str],
) -> None:
    """Run on the event-loop thread and print task stacks without locals."""
    tasks = sorted(asyncio.all_tasks(loop), key=lambda task: task.get_name())
    print(f"Asyncio tasks for {nodeid}: {len(tasks)}", file=stream)
    for task in tasks:
        coro = task.get_coro()
        coro_name = getattr(coro, "__qualname__", type(coro).__qualname__)
        print(
            f"--- task name={task.get_name()!r} coro={coro_name} "
            f"done={task.done()} cancelled={task.cancelled()} ---",
            file=stream,
        )
        frames = task.get_stack(limit=20)
        if not frames:
            print("  <no Python stack available>", file=stream)
        for frame in frames:
            for line in traceback.format_stack(frame, limit=1):
                print(line.rstrip(), file=stream)
    stream.flush()
    _print_diagnostic_end(stream)


def _print_diagnostic_end(stream: IO[str]) -> None:
    print("===== END PYTEST HANG DIAGNOSTIC =====", file=stream, flush=True)


def _watchdog(config: pytest.Config) -> HangWatchdog | None:
    return getattr(config, "_helix_hang_watchdog", None)


def pytest_addoption(parser: pytest.Parser) -> None:
    parser.addoption(
        "--hang-diagnostics-seconds",
        action="store",
        type=float,
        default=0.0,
        help="Dump the active test, thread stacks, and asyncio tasks after N seconds.",
    )


def pytest_configure(config: pytest.Config) -> None:
    watchdog = HangWatchdog(config.getoption("hang_diagnostics_seconds"))
    setattr(config, "_helix_hang_watchdog", watchdog)
    watchdog.start()


def pytest_unconfigure(config: pytest.Config) -> None:
    watchdog = _watchdog(config)
    if watchdog is not None:
        watchdog.stop()


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_protocol(item: pytest.Item, nextitem: pytest.Item | None):
    del nextitem
    watchdog = _watchdog(item.config)
    if watchdog is not None:
        watchdog.test_started(item.nodeid)
    try:
        yield
    finally:
        if watchdog is not None:
            watchdog.test_finished(item.nodeid)


@pytest.hookimpl(trylast=True)
def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
    for item in items:
        if not isinstance(item, pytest.Function):
            continue
        obj = getattr(item, "obj", None)
        is_async = bool(item.get_closest_marker("asyncio")) or (
            obj is not None and inspect.iscoroutinefunction(obj)
        )
        if is_async and "_hang_diagnostics_event_loop" not in item.fixturenames:
            item.fixturenames.append("_hang_diagnostics_event_loop")


@pytest_asyncio.fixture
async def _hang_diagnostics_event_loop(request: pytest.FixtureRequest):
    watchdog = _watchdog(request.config)
    loop = asyncio.get_running_loop()
    if watchdog is not None:
        watchdog.register_loop(loop)
    try:
        yield
    finally:
        if watchdog is not None:
            watchdog.unregister_loop(loop)