diff options
Diffstat (limited to 'packages/meshbay-common/tests')
| -rw-r--r-- | packages/meshbay-common/tests/test_background_tasks.py | 153 |
1 files changed, 153 insertions, 0 deletions
diff --git a/packages/meshbay-common/tests/test_background_tasks.py b/packages/meshbay-common/tests/test_background_tasks.py new file mode 100644 index 0000000..4f0ac59 --- /dev/null +++ b/packages/meshbay-common/tests/test_background_tasks.py @@ -0,0 +1,153 @@ +""" +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__ |