From 8405e2f9b711ffc9bdabf0f0289051a6813bf5db Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Sun, 19 Jul 2026 07:20:59 +0200 Subject: [PATCH] =?UTF-8?q?sync-v2:=20queue-isolation=20routing=20test=20(?= =?UTF-8?q?acceptance=20=C2=A710.5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Assert bulk backfills route to the bulk queue and interactive user actions to the interactive queue, and that enqueue_job stamps _queue_name accordingly — the mechanism that keeps a backfill burst from starving sends/flag changes. Co-Authored-By: Claude Opus 4.8 --- backend/tests/test_queue_routing.py | 53 +++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 backend/tests/test_queue_routing.py 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), + ]