Class StreamEventCaptureTracker

tests/core/test_application.py:166–207

class StreamEventCaptureTracker(PreStartStreamHook, PostStreamItemHook, PostEndStreamHook):
    def post_end_stream(
        self,
        *,
        action: str,
        sequence_id: int,
        app_id: str,
        partition_key: Optional[str],
        **future_kwargs: Any,
    ):
        self.post_end_stream_calls.append((action, locals()))

    def __init__(self):
        self.pre_start_stream_calls = []
        self.post_stream_item_calls = []
        self.post_end_stream_calls = []

    def pre_start_stream(
        self,
        *,
        action: str,
        sequence_id: int,
        app_id: str,
        partition_key: Optional[str],
        **future_kwargs: Any,
    ):
        self.pre_start_stream_calls.append((action, locals()))

    def post_stream_item(
        self,
        *,
        item: Any,
        item_index: int,
        stream_initialize_time: datetime.datetime,
        first_stream_item_start_time: datetime.datetime,
        action: str,
        sequence_id: int,
        app_id: str,
        partition_key: Optional[str],
        **future_kwargs: Any,
    ):
        self.post_stream_item_calls.append((action, locals()))


class StreamEventCaptureTrackerAsync(