Method read

test/mitmproxy/proxy/layers/quic/test__stream_layers.py:416–426

(self)

        self.quic.receive_datagram(buf, self.address, self.now)

    def read(self) -> bytes:
        self.now = self.now + 0.1
        buf = b""
        has_data = False
        for datagram, addr in self.quic.datagrams_to_send(self.now):
            assert addr == self.address
            buf += datagram
            has_data = True
        if not has_data:
            raise AssertionError("no datagrams to send")
        return buf

    def handshake_completed(self) -> bool:
        while event := self.quic.next_event():