import asyncio import logging import threading from types import SimpleNamespace import pytest from robomp import tasks from robomp.github_client import IssueInfo, RepoInfo async def test_triage_issue_keeps_event_loop_live_while_workspace_setup_blocks(db, settings, monkeypatch, tmp_path): async def _resolve_repo_and_issue(_github, _payload): repo = RepoInfo( full_name="octo/widget", default_branch="main", clone_url="https://x/octo/widget.git", private=False, ) issue = IssueInfo( repo="octo/widget", number=1, title="bug", body="b", state="open", author="alice", labels=(), is_pull_request=False, ) return repo, issue monkeypatch.setattr(tasks, "_resolve_repo_and_issue", _resolve_repo_and_issue) async def _no_closing(*a, **k): return () github = SimpleNamespace(list_closing_pull_requests=_no_closing) entered = threading.Event() release = threading.Event() captured: dict[str, object] = {} def _blocking_ensure(**_kwargs): entered.set() # True ONLY if a concurrent coroutine set `release` while we blocked here. # Blocks a WORKER THREAD (via to_thread) in the fixed code; blocks the # LOOP itself in the broken code. captured["release_seen_in_time"] = release.wait(1.0) return SimpleNamespace(branch="farm/x/y", session_dir=str(tmp_path / "sess")) sandbox = SimpleNamespace(natives_cache=None, ensure_workspace=_blocking_ensure) async def _noop_run_task(**_kwargs): return None monkeypatch.setattr(tasks, "run_task", _noop_run_task) async def _releaser(): # Waits (off-loop) until ensure_workspace has actually started, then # releases it. This coroutine can ONLY make progress if the event loop # is live while ensure_workspace is blocking. await asyncio.to_thread(entered.wait, 1.0) assert entered.is_set(), "ensure_workspace never started" release.set() triage_task = asyncio.create_task( tasks.triage_issue( settings=settings, db=db, github=github, sandbox=sandbox, git_transport=SimpleNamespace(), payload={}, delivery_id="d1", ) ) releaser_task = asyncio.create_task(_releaser()) await asyncio.wait_for(triage_task, timeout=3.0) await asyncio.wait_for(releaser_task, timeout=1.0) assert captured.get("release_seen_in_time") is True, ( "event loop was frozen during ensure_workspace: the concurrent releaser " "could not run, so release.wait timed out (this is the pre-fix hang)" ) async def test_run_workspace_op_drains_thread_before_propagating_cancel(): started = threading.Event() proceed = threading.Event() finished = threading.Event() def slow_op(**_kwargs): started.set() # Block on the worker thread until the test releases us. assert proceed.wait(2.0), "proceed was never set — test bug" finished.set() return "done" task = asyncio.create_task(tasks._run_workspace_op(slow_op)) # Wait (off-loop) until the worker thread is actually running. await asyncio.to_thread(started.wait, 1.0) assert started.is_set() async def pump(turns: int = 20) -> None: # Deterministically advance the loop without a wall-clock sleep: each # sleep(0) drains the ready queue, so a DETACHING (pre-fix) helper would # resolve `task` within these turns. A draining helper keeps it pending # while the worker thread is still blocked on `proceed`. for _ in range(turns): await asyncio.sleep(0) # Cancel the AWAITING coroutine while the thread is mid-flight, then a SECOND # time while it is still blocked. The repeated cancel must land on the drain # loop's re-`await` and be swallowed by its `continue` branch, NOT abandon # the thread. The whole sequence runs under try/finally so any failed assert # still releases the worker and cannot leak a blocked thread into later tests. try: task.cancel() await pump() assert not task.done(), "helper propagated the first cancel before the thread completed (thread abandoned)" task.cancel() await pump() # The thread is still blocked on `proceed`, so it has not finished and # the task has not resolved despite two cancels. assert not finished.is_set(), "thread finished before we released it — impossible unless abandoned" assert not task.done(), "helper abandoned the thread after a repeated cancel" finally: proceed.set() # The helper must now let the thread finish, THEN raise CancelledError. with pytest.raises(asyncio.CancelledError): await task # Deterministic in the fixed helper: the thread completed before the cancel propagated. assert finished.is_set(), "thread did not complete before cancellation propagated" async def test_run_workspace_op_logs_worker_exception_on_concurrent_cancel(caplog): started = threading.Event() proceed = threading.Event() boom = RuntimeError("git exploded") def failing_op(**_kwargs): started.set() assert proceed.wait(2.0), "proceed was never set — test bug" raise boom task = asyncio.create_task(tasks._run_workspace_op(failing_op)) await asyncio.to_thread(started.wait, 1.0) assert started.is_set() # Cancel the caller while the worker is still blocked (mid-flight), so the # helper enters its cancel-drain loop and is awaiting the shielded inner. task.cancel() await asyncio.sleep(0.05) with caplog.at_level(logging.WARNING, logger="robomp.tasks"): # Release the worker so inner completes WITH an exception while the # helper is draining -> the drain's `await shield(inner)` re-raises boom, # breaks the loop, and the guarded log.warning must fire. proceed.set() with pytest.raises(asyncio.CancelledError): await task warnings = [r for r in caplog.records if r.levelno == logging.WARNING] assert warnings, "worker exception during cancel was not logged" assert any(r.exc_info and r.exc_info[1] is boom for r in warnings), ( "the worker's exception was not attached to the warning" )