Function event_processing_harness

tests/benchmarks/test_event_processing.py:28–99

Set up the full event processing pipeline for benchmarking. Creates a ``BaseStateEventProcessor`` wired to a real ``StateManagerMemory`` with mock emit callbacks. Events are enqueued directly and deltas are collected via the emit callback. Yields: An async callable that en

()

@pytest_asyncio.fixture
async def event_processing_harness():
    """Set up the full event processing pipeline for benchmarking.

    Creates a ``BaseStateEventProcessor`` wired to a real
    ``StateManagerMemory`` with mock emit callbacks.  Events are
    enqueued directly and deltas are collected via the emit callback.

    Yields:
        An async callable that enqueues the given number of events
        and waits for all expected deltas.
    """
    emitted_deltas: list[tuple[str, Mapping[str, Mapping[str, Any]]]] = []

    async def emit_delta_impl(  # noqa: RUF029
        token: str, delta: Mapping[str, Mapping[str, Any]]
    ) -> None:
        emitted_deltas.append((token, delta))

    async def emit_event_impl(token: str, *events: Event) -> None:
        pass

    def handle_backend_exception(ex: Exception) -> None:
        formatted_exc = "\n".join(traceback.format_exception(ex))
        pytest.fail(f"Event processor raised an unexpected exception:\n{formatted_exc}")

    processor = BaseStateEventProcessor(
        backend_exception_handler=handle_backend_exception,
        graceful_shutdown_timeout=5,
    )
    # Mock _rehydrate so the processor doesn't try to push full state
    # to a non-existent frontend on the first event.
    with mock.patch.object(processor, "_rehydrate", new=mock.AsyncMock()):
        state_manager = StateManagerMemory()
        root_context = EventContext(
            token="",
            state_manager=state_manager,
            enqueue_impl=processor.enqueue_many,
            emit_delta_impl=emit_delta_impl,
            emit_event_impl=emit_event_impl,
        )
        processor._root_context = root_context

        token = "benchmark-token"
        handler_name = format_event_handler(BenchmarkState.event_handlers["increment"])
        event = Event(
            name=handler_name,
            router_data={
                "query": {},
                "path": "/",
            },
            payload={},
        )

        async def run_events(num_events: int, num_expected_deltas: int) -> None:
            """Enqueue events and wait for all deltas to be emitted.

            Args:
                num_events: Number of increment events to enqueue.