Function _test_echo

test/mitmproxy/proxy/layers/quic/test__stream_layers.py:436–452

(
    playbook: tutils.Playbook, tssl: SSLTest, conn: connection.Connection
)

def _test_echo(
    playbook: tutils.Playbook, tssl: SSLTest, conn: connection.Connection
) -> None:
    tssl.quic.send_datagram_frame(b"Hello World")
    data = tutils.Placeholder(bytes)
    assert (
        playbook
        >> events.DataReceived(conn, tssl.read())
        << commands.SendData(conn, data)
    )
    tssl.write(data())
    while event := tssl.quic.next_event():
        if isinstance(event, quic_events.DatagramFrameReceived):
            assert event.data == b"hello world"
            break
    else:
        raise AssertionError()


def finish_handshake(