diff --git a/backend/tests/test_queue_routing.py b/backend/tests/test_queue_routing.py new file mode 100644 index 0000000..7ea7b4a --- /dev/null +++ b/backend/tests/test_queue_routing.py @@ -0,0 +1,53 @@ +"""Interactive/bulk queue isolation (refactor §1, acceptance §10.5). + +Bulk backfills and interactive user actions land on separate arq queues (run by +separate consumers with their own concurrency), so a burst of backfills can never +starve sends/flag changes. Here we assert the routing that guarantees it. +""" + +from __future__ import annotations + +from app import queue as app_queue +from app.queue import QUEUE_BULK, QUEUE_INTERACTIVE, queue_for + +_INTERACTIVE_JOBS = [ + "push_flags_job", + "move_job", + "move_thread_to_folder_job", + "delete_thread_job", + "send_job", + "push_draft_to_imap_job", + "remove_imap_draft_job", + "refresh_account_job", +] + + +def test_bulk_backfill_routes_to_bulk_queue() -> None: + assert queue_for("sync_bootstrap_account") == QUEUE_BULK + + +def test_interactive_jobs_route_to_interactive_queue() -> None: + for name in _INTERACTIVE_JOBS: + assert queue_for(name) == QUEUE_INTERACTIVE + + +async def test_enqueue_job_sets_queue_name(monkeypatch) -> None: # type: ignore[no-untyped-def] + captured: list[tuple[str, str | None]] = [] + + class FakePool: + async def enqueue_job(self, name: str, *args: object, **kwargs: object) -> None: + captured.append((name, kwargs.get("_queue_name"))) # type: ignore[arg-type] + return None + + async def fake_pool() -> FakePool: + return FakePool() + + monkeypatch.setattr(app_queue, "get_arq_pool", fake_pool) + + await app_queue.enqueue_job("sync_bootstrap_account", "a") + await app_queue.enqueue_job("send_job", "b") + + assert captured == [ + ("sync_bootstrap_account", QUEUE_BULK), + ("send_job", QUEUE_INTERACTIVE), + ]