54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
 
|
"""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),
|
||
|
|
]
|