Testing Queue Events¶

Testing event emission and realtime stream routes requires deterministic, synchronous delivery without background buffer timers or flakiness.

Runnable Example¶

"""Unit testing task events deterministically with InMemoryQueueEventSink and immediate execution."""

from litestar_queues import QueueConfig, QueueService, TaskExecutionContext, WorkerConfig, task
from litestar_queues.events import EventDeliveryConfig, InMemoryQueueEventSink, QueueEventsConfig


@task("exports.generate_csv", queue="exports")
async def generate_csv(report_name: str, _task_context: TaskExecutionContext) -> str:
    """Emit progress and custom events during CSV generation."""
    await _task_context.progress(current=1, total=2, message="Querying rows")
    await _task_context.event("export.rows_fetched", payload={"count": 250})
    await _task_context.progress(current=2, total=2, message="Writing CSV")
    return f"{report_name}.csv"


async def main() -> None:
    """Capture and assert task events synchronously with buffer=None and immediate execution."""
    sink = InMemoryQueueEventSink()
    config = QueueConfig(
        queue_backend="memory",
        execution_backend="immediate",
        worker=WorkerConfig(placement="external"),
        events=QueueEventsConfig(delivery=EventDeliveryConfig(sinks=(sink,), buffer=None, strict=True)),
    )
    async with QueueService(config=config) as service:
        handle = await service.enqueue(generate_csv, "q4-revenue")
        assert handle.status == "completed"

    event_types = [event.type for event in sink.events]
    assert event_types == ["task.started", "task.progress", "export.rows_fetched", "task.progress", "task.completed"]

Deterministic Unit Testing Recipe¶

Combine four settings in unit tests to capture every task and lifecycle event synchronously in memory:

1. InMemoryQueueEventSink(): Appends every delivered QueueEvent to sink.events (list[QueueEvent]).
2. EventDeliveryConfig(sinks=(sink,), buffer=None, strict=True): Disables background ring-buffer batching (buffer=None) so every await ctx.progress(...) or lifecycle transition delivers to sink.events before the await returns, and sets strict=True so any sink assertion error propagates immediately.
3. execution_backend="immediate": Runs the task body inline during await service.enqueue(...).
4. WorkerConfig(placement="external"): Prevents background polling loops from starting during the test.

Testing Durable Event History¶

When testing QueueEventLog.query_events() or summarize_stages() with EventHistoryConfig(...), call await event_log.flush_events() before querying non-terminal progress or log events, or await terminal task completion (which flushes the history buffer automatically via the commit-before-live barrier):

event_log = service.get_event_log()
assert event_log is not None
await event_log.flush_events()
page = await event_log.query_events(QueueEventQuery(task_id=str(handle.id)))

Testing SSE & WebSocket Stream Endpoints¶

Use Litestar's AsyncTestClient to test EventStreamConfig routes and channel_authorizer callbacks end-to-end:

• WebSocket Streams: Connect with with client.websocket_connect("/queues/events/tasks/<task_id>") as ws: and assert on ws.receive_json().
• SSE Streams: Pass max_stream_duration=0.2 on EventStreamConfig in tests so await client.get("/queues/events/sse/tasks/<task_id>") closes cleanly after emitting initial frames without hanging the test coroutine.