"""
Tasks nobody holds, across every package.

asyncio keeps only a **weak** reference to a task, so a coroutine started with
`asyncio.ensure_future(...)` whose result is thrown away can be collected while
it is still running. Nothing raises: the loop logs "Task was destroyed but it is
pending!" and whatever that task was doing simply stops.

The node already had a guard for this — written after an abandoned stream task
never reached `async with sem.__aexit__` and lost a transcode slot for good —
and it read one file, `webrtc_server.py`, because that is where the defect was
found. Meanwhile the hub answered `chat_notify` with a discarded
`ensure_future` (a notification for every member of a group, dropped whenever
the collector got there first), the node's indexer scheduled every real-time
index update the same way (a file copied in that never appears until a
reconciliation sweep an unbounded time later), and seventeen more sites did it
in `daemon.py`, `ops.py` and the loopback API — a reload, an enrichment pass, a
swarm registration.

So this one walks **every package's source**. A guard that stops at the edge of
the file where the bug was found is a guard against that bug, not against its
class.
"""

import ast
import asyncio
import logging
from pathlib import Path

import pytest
from meshbay_common import background

REPO = Path(__file__).resolve().parents[3]
PACKAGES = ("meshbay-common", "meshbay-hub", "meshbay-node")

SPAWNERS = {"ensure_future", "create_task"}

# `spawn()` is the implementation of the rule and is the one place allowed to
# call the primitive with nothing holding the result — that is what it is for.
ALLOWED = {Path("meshbay-common/src/meshbay_common/background.py")}


def _sources() -> list[Path]:
    out: list[Path] = []
    for pkg in PACKAGES:
        src = REPO / "packages" / pkg / "src"
        if src.is_dir():
            out.extend(sorted(src.rglob("*.py")))
    return out


def test_there_are_sources_to_check():
    """A rule applied to nothing passes forever."""
    files = _sources()
    assert len(files) > 20, f"only found {len(files)} source files — wrong root?"


def _discarded(tree: ast.AST) -> list[int]:
    """Lines where a task is started and its handle dropped on the floor.

    An `Expr` statement is a call whose value goes nowhere; a `Lambda` whose
    whole body is such a call is the same thing wearing a callback's clothes
    (`signal.add_signal_handler(SIGHUP, lambda: ensure_future(reload()))`).
    Anything assigned, returned, awaited or collected into a list is held by
    its caller and is not this defect.
    """
    found: list[int] = []

    def is_spawn(node: ast.AST) -> bool:
        return (isinstance(node, ast.Call)
                and isinstance(node.func, ast.Attribute)
                and node.func.attr in SPAWNERS
                and isinstance(node.func.value, ast.Name)
                and node.func.value.id == "asyncio")

    for node in ast.walk(tree):
        if isinstance(node, ast.Expr) and is_spawn(node.value):
            found.append(node.lineno)
        elif isinstance(node, ast.Lambda) and is_spawn(node.body):
            found.append(node.lineno)
    return found


def test_no_task_is_started_and_forgotten():
    offenders: list[str] = []
    for path in _sources():
        rel = path.relative_to(REPO / "packages")
        if rel in ALLOWED:
            continue
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for line in _discarded(tree):
            offenders.append(f"{rel}:{line}")

    assert not offenders, (
        "a task whose result is discarded can be collected mid-flight and the "
        "work silently does not happen. Use meshbay_common.background.spawn(), "
        "or keep the handle where the owner can also cancel it:\n  "
        + "\n  ".join(offenders))


# ── The helper itself ────────────────────────────────────────────────────────

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

    async def work():
        started.set()
        await release.wait()

    task = background.spawn(work())
    await started.wait()
    assert background.pending() == 1, "nothing is holding the task"
    # The only strong reference the test itself holds, gone: if `spawn` is not
    # keeping one, the task is now collectable.
    del task
    release.set()
    await asyncio.sleep(0)
    await asyncio.sleep(0)
    assert background.pending() == 0, "the set grows for the life of the process"


@pytest.mark.asyncio
async def test_a_failure_is_logged_where_it_happened(caplog):
    """Nobody awaits this task, so the log is the only place it can surface."""
    async def work():
        raise RuntimeError("the background work failed")

    with caplog.at_level(logging.ERROR, logger=background.log.name):
        # Pinned rather than assumed: an earlier test in the run may have
        # raised this logger's level or disabled it outright.
        background.log.disabled = False
        background.log.setLevel(logging.NOTSET)
        task = background.spawn(work(), what="the work")
        with pytest.raises(RuntimeError):
            await task

    assert any("the work" in r.getMessage() for r in caplog.records), caplog.text


@pytest.mark.asyncio
async def test_the_work_names_itself_without_a_label():
    async def a_named_coroutine():
        raise RuntimeError("boom")

    task = background.spawn(a_named_coroutine())
    with pytest.raises(RuntimeError):
        await task
    # The default label is the coroutine's own name, so no call site has to
    # carry one for a log line to be readable.
    assert "a_named_coroutine" in a_named_coroutine.__qualname__