Class FakeAsyncEvents

tests/lib/tools/test_session_runner.py:173–231

class FakeAsyncEvents:
    def __init__(
        self,
        *,
        streams: list[_FakeStream | BaseException] | None = None,
        stream_events: list[_StubEvent] | None = None,
        list_events: list[_StubEvent] | None = None,
        list_events_per_call: list[list[_StubEvent]] | None = None,
        list_raises: BaseException | None = None,
        send_failures: Sequence[BaseException | None] | None = None,
    ) -> None:
        if streams is not None:
            self._streams: list[_FakeStream | BaseException] = list(streams)
        elif stream_events is not None:
            self._streams = [_FakeStream(stream_events)]
        else:
            self._streams = [_FakeStream([])]
        self._list_events = list(list_events or [])
        # When set, each ``list()`` call consumes the next entry (falling back
        # to ``list_events`` once exhausted) so reconnect tests can script a
        # different history per reconcile pass.
        self._list_events_per_call = [list(evs) for evs in (list_events_per_call or [])]
        self._list_raises = list_raises
        self._send_failures: list[BaseException | None] = list(send_failures or [])
        self.send_calls: list[dict[str, Any]] = []
        self.stream_calls: int = 0
        self.stream_headers: list[Any] = []
        self.list_headers: list[Any] = []

    async def stream(self, _session_id: str, *, extra_headers: Any = None) -> _FakeStream:
        self.stream_calls += 1
        self.stream_headers.append(extra_headers)
        if not self._streams:
            return _FakeStream([])  # block forever; mirrors a fresh connection
        nxt = self._streams.pop(0)
        if isinstance(nxt, BaseException):
            raise nxt
        return nxt

    def list(self, _session_id: str, *, limit: int = 1000, extra_headers: Any = None) -> Any:  # noqa: ARG002
        list_raises = self._list_raises
        self.list_headers.append(extra_headers)
        list_events = self._list_events_per_call.pop(0) if self._list_events_per_call else self._list_events

        async def _gen() -> Any:
            for ev in list_events:
                yield ev
            if list_raises is not None:
                raise list_raises

        return _gen()

    async def send(self, session_id: str, *, events: list[Any], extra_headers: Any = None) -> None:
        idx = len(self.send_calls)
        self.send_calls.append({"session_id": session_id, "events": events, "extra_headers": extra_headers})
        if idx < len(self._send_failures):
            err = self._send_failures[idx]
            if err is not None: