How to use h11 - 10 common examples

async def test_post_request(serving_app: Quart, event_loop: asyncio.AbstractEventLoop) -> None:
    connection = MockConnection(serving_app, event_loop)
    await connection.send(
        h11.Request(
            method='POST', target='/echo',
            headers=BASIC_HEADERS + [('content-length', str(len(BASIC_DATA.encode())))],
        ),
    )
    await connection.send(h11.Data(data=BASIC_DATA.encode()))
    await connection.send(h11.EndOfMessage())
    await connection.transport.closed.wait()
    response, *data, end = connection.get_events()
    assert isinstance(response, h11.Response)
    assert response.status_code == 200
    assert all(isinstance(datum, h11.Data) for datum in data)
    assert b''.join(datum.data for datum in data).decode() == BASIC_DATA
    assert isinstance(end, h11.EndOfMessage)

async def test_bad_framework_http(path: str, event_loop: asyncio.AbstractEventLoop) -> None:
    connection = MockHTTPConnection(path, event_loop, framework=bad_framework)
    await asyncio.sleep(0)  # Yield to allow the server to process
    await connection.transport.closed.wait()
    response, *_ = connection.get_events()
    assert isinstance(response, h11.Response)
    assert response.status_code == 500

def test_response_headers_2(example):
    resp, _ = example.send(h11.Request(method='GET', target='/',
                                       headers=[('Host', 'example')]),
                           h11.EndOfMessage())
    assert (b'set-cookie', b'sessionid=123456') in resp.headers
    assert (b'set-cookie', b'__adtrack=abcdef') in resp.headers

def test_forwarding_requests_2(example):
    resp, _, _ = example.send(h11.Request(method='GET', target='/',
                                          headers=[('Host', 'example')]),
                              h11.EndOfMessage())
    # ``develop1.example`` is unreachable
    assert 500 <= resp.status_code <= 599

def test_request_bytes_iterable(self):
        # Assert that we send the first set of body bytes with the request packet.
        body_bytes = [b"Hello, ", b"world!"]
        body_size = 13
        request = Request(
            method=b"POST",
            target="post",
            body=body_bytes,
            headers={"Content-Length": body_size},
        )
        request.add_host("httpbin.org", port=80, scheme="http")
        state_machine = h11.Connection(our_role=h11.CLIENT)
        iterable = _request_bytes_iterable(request, state_machine)
        first_packet, second_packet = next(iterable), next(iterable)
        assert request.method in first_packet
        assert body_bytes[0] in first_packet
        assert body_bytes[1] in second_packet
        with pytest.raises(StopIteration):
            next(iterable)

def __init__(self, *, framework: ASGIFramework = echo_framework) -> None:
        self.client_stream, server_stream = trio.testing.memory_stream_pair()
        server_stream.socket = MockSocket()
        self.client = h11.Connection(h11.CLIENT)
        self.server = H11Server(framework, Config(), server_stream)

def _make_handshake(
    request_headers: Headers,
    accept_headers: Optional[Headers] = None,
    subprotocol: Optional[str] = None,
    extensions: Optional[List[Extension]] = None,
) -> Tuple[h11.InformationalResponse, bytes]:
    client = h11.Connection(h11.CLIENT)
    server = WSConnection(SERVER)
    nonce = generate_nonce()
    server.receive_data(
        client.send(
            h11.Request(
                method="GET",
                target="/",
                headers=[
                    (b"Host", b"localhost"),
                    (b"Connection", b"Keep-Alive, Upgrade"),
                    (b"Upgrade", b"WebSocket"),
                    (b"Sec-WebSocket-Version", b"13"),
                    (b"Sec-WebSocket-Key", nonce),
                ]
                + request_headers,
            )

def test_handshake_rejection_with_body() -> None:
    events = _make_handshake_rejection(400, body=b"Hello")
    assert events == [
        h11.Response(headers=[(b"content-length", b"5")], status_code=400),
        h11.Data(data=b"Hello"),
        h11.EndOfMessage(),
    ]

async def test_protocol_handle_request(protocol: H11Protocol) -> None:
    client = h11.Connection(h11.CLIENT)
    await protocol.handle(
        RawData(data=client.send(h11.Request(method="GET", target="/?a=b", headers=BASIC_HEADERS)))
    )
    protocol.stream.handle.assert_called()
    assert protocol.stream.handle.call_args_list == [
        call(
            Request(
                stream_id=1,
                headers=[(b"host", b"hypercorn"), (b"connection", b"close")],
                http_version="1.1",
                method="GET",
                raw_path=b"/?a=b",
            )
        ),
        call(EndBody(stream_id=1)),
    ]

def _make_connection_request(request_headers: Headers, method: str = "GET") -> Request:
    client = h11.Connection(h11.CLIENT)
    server = WSConnection(SERVER)
    server.receive_data(
        client.send(h11.Request(method=method, target="/", headers=request_headers))
    )
    event = next(server.events())
    assert isinstance(event, Request)
    return event