Class FrameFactory

test/mitmproxy/proxy/layers/http/test_http3.py:79–352

Helper class for generating QUIC stream events and commands.

class FrameFactory:
    """Helper class for generating QUIC stream events and commands."""

    def __init__(self, conn: connection.Connection, is_client: bool) -> None:
        self.conn = conn
        self.is_client = is_client
        self.decoder = pylsqpack.Decoder(
            max_table_capacity=4096,
            blocked_streams=16,
        )
        self.decoder_placeholders: list[tutils.Placeholder[bytes]] = []
        self.encoder = pylsqpack.Encoder()
        self.encoder_placeholder: tutils.Placeholder[bytes] | None = None
        self.peer_stream_id: dict[StreamType, int] = {}
        self.local_stream_id: dict[StreamType, int] = {}

    def get_default_stream_id(self, stream_type: StreamType, for_local: bool) -> int:
        if stream_type == StreamType.CONTROL:
            stream_id = 2
        elif stream_type == StreamType.QPACK_ENCODER:
            stream_id = 6
        elif stream_type == StreamType.QPACK_DECODER:
            stream_id = 10
        else:
            raise AssertionError(stream_type)
        if self.is_client is not for_local:
            stream_id = stream_id + 1
        return stream_id

    def send_stream_type(
        self,
        stream_type: StreamType,
        stream_id: int | None = None,
    ) -> quic.SendQuicStreamData:
        assert stream_type not in self.peer_stream_id
        if stream_id is None:
            stream_id = self.get_default_stream_id(stream_type, for_local=False)
        self.peer_stream_id[stream_type] = stream_id
        return quic.SendQuicStreamData(
            connection=self.conn,
            stream_id=stream_id,
            data=encode_uint_var(stream_type),
            end_stream=False,
        )

    def receive_stream_type(
        self,
        stream_type: StreamType,
        stream_id: int | None = None,
    ) -> quic.QuicStreamDataReceived:
        assert stream_type not in self.local_stream_id
        if stream_id is None:
            stream_id = self.get_default_stream_id(stream_type, for_local=True)
        self.local_stream_id[stream_type] = stream_id
        return quic.QuicStreamDataReceived(
            connection=self.conn,
            stream_id=stream_id,
            data=encode_uint_var(stream_type),