How to use Hypercorn - 10 common examples

async def test_keep_alive_timeout(event_loop: asyncio.AbstractEventLoop) -> None:
    transport = MockTransport()
    config = Config()
    config.keep_alive_timeout = 0.01
    server = HTTPServer(event_loop, config, transport, "")  # type: ignore
    server.start_keep_alive_timeout()
    assert not transport.closed.is_set()
    await asyncio.sleep(2 * config.keep_alive_timeout)
    assert transport.closed.is_set()

async def _protocol(monkeypatch: MonkeyPatch) -> H11Protocol:
    MockHTTPStream = AsyncMock()  # noqa: N806
    MockHTTPStream.return_value = AsyncMock(spec=HTTPStream)
    monkeypatch.setattr(hypercorn.protocol.h11, "HTTPStream", MockHTTPStream)
    MockEvent = AsyncMock()  # noqa: N806
    MockEvent.return_value = AsyncMock(spec=IOEvent)
    return H11Protocol(Config(), False, None, None, CoroutineMock(), CoroutineMock(), MockEvent)

async def test_startup_failure() -> None:
    lifespan = Lifespan(lifespan_failure, Config())
    with pytest.raises(LifespanFailure) as exc_info:
        async with trio.open_nursery() as lifespan_nursery:
            await lifespan_nursery.start(lifespan.handle_lifespan)
            await lifespan.wait_for_startup()

    assert str(exc_info.value) == "Lifespan failure in startup. 'Failure'"

def test_create_sockets_unix(monkeypatch: MonkeyPatch) -> None:
    mock_socket = Mock()
    monkeypatch.setattr(socket, "socket", mock_socket)
    monkeypatch.setattr(os, "chown", Mock())
    config = Config()
    config.bind = ["unix:/tmp/hypercorn.sock"]
    sockets = config.create_sockets()
    sock = sockets.insecure_sockets[0]
    mock_socket.assert_called_with(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.setsockopt.assert_called_with(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # type: ignore
    sock.bind.assert_called_with("/tmp/hypercorn.sock")  # type: ignore
    sock.setblocking.assert_called_with(False)  # type: ignore
    sock.set_inheritable.assert_called_with(True)  # type: ignore

async def test_protocol_handle_max_incomplete(monkeypatch: MonkeyPatch) -> None:
    config = Config()
    config.h11_max_incomplete_size = 5
    MockHTTPStream = AsyncMock()  # noqa: N806
    MockHTTPStream.return_value = AsyncMock(spec=HTTPStream)
    monkeypatch.setattr(hypercorn.protocol.h11, "HTTPStream", MockHTTPStream)
    MockEvent = AsyncMock()  # noqa: N806
    MockEvent.return_value = AsyncMock(spec=IOEvent)
    protocol = H11Protocol(config, False, None, None, CoroutineMock(), CoroutineMock(), MockEvent)
    await protocol.handle(RawData(data=b"GET / HTTP/1.1\r\nHost: hypercorn\r\n"))
    protocol.send.assert_called()
    assert protocol.send.call_args_list == [
        call(
            RawData(
                data=b"HTTP/1.1 400 \r\ncontent-length: 0\r\nconnection: close\r\n"
                b"date: Thu, 01 Jan 1970 01:23:20 GMT\r\nserver: hypercorn-h11\r\n\r\n"
            )
        ),

async def test_post_response_keep_alive_timeout(event_loop: asyncio.AbstractEventLoop) -> None:
    config = Config()
    config.keep_alive_timeout = 0.01
    transport = MockTransport()
    server = H11Server(echo_framework, event_loop, config, transport)  # type: ignore
    server.pause_writing()
    server.data_received(b"GET / HTTP/1.1\r\nHost: hypercorn\r\n\r\n")
    await asyncio.sleep(2 * config.keep_alive_timeout)
    assert not transport.closed.is_set()
    server.resume_writing()
    await asyncio.sleep(2 * config.keep_alive_timeout)
    assert transport.closed.is_set()

async def test_http2_request(nursery: trio._core._run.Nursery) -> None:
    client_stream, server_stream = trio.testing.memory_stream_pair()
    server_stream.transport_stream = Mock(return_value=PropertyMock(return_value=MockSocket()))
    server_stream.do_handshake = CoroutineMock()
    server_stream.selected_alpn_protocol = Mock(return_value="h2")
    server = TCPServer(sanity_framework, Config(), server_stream)
    nursery.start_soon(server.run)
    client = h2.connection.H2Connection()
    client.initiate_connection()
    await client_stream.send_all(client.data_to_send())
    stream_id = client.get_next_available_stream_id()
    client.send_headers(
        stream_id,
        [
            (":method", "GET"),
            (":path", "/"),
            (":authority", "hypercorn"),
            (":scheme", "https"),
            ("content-length", "%d" % len(SANITY_BODY)),
        ],
    )
    client.send_data(stream_id, SANITY_BODY)

def test_create_sockets_multiple(monkeypatch: MonkeyPatch) -> None:
    mock_socket = Mock()
    monkeypatch.setattr(socket, "socket", mock_socket)
    monkeypatch.setattr(os, "chown", Mock())
    config = Config()
    config.bind = ["127.0.0.1", "unix:/tmp/hypercorn.sock"]
    sockets = config.create_sockets()
    assert len(sockets.insecure_sockets) == 2

async def test_protocol_handle_h2c_upgrade(protocol: H11Protocol) -> None:
    with pytest.raises(H2CProtocolRequired) as exc_info:
        await protocol.handle(
            RawData(
                data=(
                    b"GET / HTTP/1.1\r\nHost: hypercorn\r\n"
                    b"upgrade: h2c\r\nhttp2-settings: abcd\r\n\r\nbbb"
                )
            )
        )
    assert protocol.send.call_args_list == [
        call(
            RawData(
                b"HTTP/1.1 101 \r\nupgrade: h2c\r\ndate: Thu, 01 Jan 1970 01:23:20 GMT\r\n"
                b"server: hypercorn-h11\r\n\r\n"
            )
        )
    ]
    assert exc_info.value.data == b"bbb"
    assert exc_info.value.headers == [
        (b":method", b"GET"),
        (b":path", b"/"),
        (b":authority", b"hypercorn"),
        (b"host", b"hypercorn"),
        (b"upgrade", b"h2c"),
        (b"http2-settings", b"abcd"),
    ]
    assert exc_info.value.settings == "abcd"

async def test_protocol_handle_h2_prior(protocol: H11Protocol) -> None:
    with pytest.raises(H2ProtocolAssumed) as exc_info:
        await protocol.handle(RawData(data=b"PRI * HTTP/2.0\r\n\r\nbbb"))

    assert exc_info.value.data == b"PRI * HTTP/2.0\r\n\r\nbbb"