oh-my-pi/python/robomp/tests/test_queue_dispatch.py

"""Dispatch action -> task mapping in WorkerPool._dispatch.

Regression guard for the route<->dispatch contract: `github_events.route`
queues a `pull_request.labeled` event as a `review_pr` task in `vouched_label`
mode, so `_dispatch` MUST invoke `tasks.review_pr` for that action. It
previously only handled `opened/reopened/ready_for_review`, so every vouched
PR fell through to the no-op branch and was silently marked `done`.
"""

from __future__ import annotations

import pytest

from robomp import tasks
from robomp.config import Settings
from robomp.db import Database, EventRow
from robomp.queue import WorkerPool
from robomp.slot_pool import SlotPool


class _StubGitHub:
    """Sentinel; dispatch tests stub out the task body."""


class _StubSandbox:
    natives_cache = None


class _StubGitTransport:
    pass


def _make_pool(settings: Settings, db: Database) -> WorkerPool:
    return WorkerPool(
        settings=settings,
        db=db,
        github=_StubGitHub(),  # type: ignore[arg-type]
        sandbox=_StubSandbox(),  # type: ignore[arg-type]
        git_transport=_StubGitTransport(),  # type: ignore[arg-type]
        slot_pool=SlotPool(),
    )


def _pr_row(action: str, *, delivery: str = "pr1") -> EventRow:
    return EventRow(
        delivery_id=delivery,
        event_type="pull_request",
        repo="octo/widget",
        issue_key="octo/widget#7",
        payload={"action": action, "pull_request": {"number": 7}},
        received_at="2026-01-01T00:00:00Z",
        state="running",
        attempts=1,
        last_error=None,
    )


@pytest.mark.parametrize("action", ["opened", "reopened", "ready_for_review", "labeled"])
@pytest.mark.asyncio
async def test_dispatch_routes_pr_review_actions_to_review_pr(
    settings: Settings, db: Database, monkeypatch: pytest.MonkeyPatch, action: str
) -> None:
    """Every PR action `route` can queue for review MUST reach `tasks.review_pr`.

    `labeled` is the vouched-label trigger; the others are the `open` trigger.
    """
    seen: list[str] = []

    async def fake_review_pr(*, payload, **_kwargs) -> None:
        seen.append(str(payload.get("action")))

    monkeypatch.setattr(tasks, "review_pr", fake_review_pr)

    await _make_pool(settings, db)._dispatch(_pr_row(action))  # noqa: SLF001

    assert seen == [action]


@pytest.mark.asyncio
async def test_dispatch_pr_synchronize_is_noop(
    settings: Settings, db: Database, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Actions `route` never queues for review must NOT spawn a review task."""
    called = False

    async def fake_review_pr(**_kwargs) -> None:
        nonlocal called
        called = True

    monkeypatch.setattr(tasks, "review_pr", fake_review_pr)

    await _make_pool(settings, db)._dispatch(_pr_row("synchronize"))  # noqa: SLF001

    assert called is False


@pytest.mark.asyncio
async def test_claim_promotes_deferred_submission_after_window_frees(settings: Settings, db: Database) -> None:
    assert db.record_submission(delivery_id="accepted", login="alice", repo="octo/widget")
    assert db.defer_submission_event(
        delivery_id="deferred",
        event_type="issues",
        login="alice",
        repo="octo/widget",
        issue_key="octo/widget#8",
        payload={"action": "opened", "issue": {"number": 8}},
        cap=1,
        reason="rate limit",
    )
    settings.rate_limit_window_seconds = -1

    row = await _make_pool(settings, db)._claim_next_unique()  # noqa: SLF001

    assert row is not None
    assert row.delivery_id == "deferred"
    assert row.state == "running"


@pytest.mark.asyncio
async def test_deferred_promotion_sweep_runs_while_queue_is_busy(settings: Settings, db: Database) -> None:
    """A sustained ordinary queue must not starve deferred events forever.

    The empty-queue path never fires when `claim_next_event` keeps returning
    work, so the independent sweep is the only thing that re-admits a deferred
    submission after its rolling window frees.
    """
    # Ordinary queued work for another issue keeps `claim_next_event` busy.
    assert db.record_event(
        delivery_id="busy",
        event_type="issues",
        repo="octo/widget",
        issue_key="octo/widget#1",
        payload={"action": "opened"},
    )
    assert db.record_submission(delivery_id="accepted", login="alice", repo="octo/widget")
    assert db.defer_submission_event(
        delivery_id="deferred",
        event_type="issues",
        login="alice",
        repo="octo/widget",
        issue_key="octo/widget#8",
        payload={"action": "opened", "issue": {"number": 8}},
        cap=1,
        reason="rate limit",
    )
    settings.rate_limit_window_seconds = -1

    pool = _make_pool(settings, db)
    promoted = await pool._promote_deferred()  # noqa: SLF001

    assert promoted == 1
    assert db.get_event("deferred").state == "queued"