airflow_scheduler/.venv/Lib/site-packages/h11/tests/test_events.py

from http import HTTPStatus

import pytest

from .. import _events
from .._events import (
    ConnectionClosed,
    Data,
    EndOfMessage,
    Event,
    InformationalResponse,
    Request,
    Response,
)
from .._util import LocalProtocolError


def test_events() -> None:
    with pytest.raises(LocalProtocolError):
        # Missing Host:
        req = Request(
            method="GET", target="/", headers=[("a", "b")], http_version="1.1"
        )
    # But this is okay (HTTP/1.0)
    req = Request(method="GET", target="/", headers=[("a", "b")], http_version="1.0")
    # fields are normalized
    assert req.method == b"GET"
    assert req.target == b"/"
    assert req.headers == [(b"a", b"b")]
    assert req.http_version == b"1.0"

    # This is also okay -- has a Host (with weird capitalization, which is ok)
    req = Request(
        method="GET",
        target="/",
        headers=[("a", "b"), ("hOSt", "example.com")],
        http_version="1.1",
    )
    # we normalize header capitalization
    assert req.headers == [(b"a", b"b"), (b"host", b"example.com")]

    # Multiple host is bad too
    with pytest.raises(LocalProtocolError):
        req = Request(
            method="GET",
            target="/",
            headers=[("Host", "a"), ("Host", "a")],
            http_version="1.1",
        )
    # Even for HTTP/1.0
    with pytest.raises(LocalProtocolError):
        req = Request(
            method="GET",
            target="/",
            headers=[("Host", "a"), ("Host", "a")],
            http_version="1.0",
        )

    # Header values are validated
    for bad_char in "\x00\r\n\f\v":
        with pytest.raises(LocalProtocolError):
            req = Request(
                method="GET",
                target="/",
                headers=[("Host", "a"), ("Foo", "asd" + bad_char)],
                http_version="1.0",
            )

    # But for compatibility we allow non-whitespace control characters, even
    # though they're forbidden by the spec.
    Request(
        method="GET",
        target="/",
        headers=[("Host", "a"), ("Foo", "asd\x01\x02\x7f")],
        http_version="1.0",
    )

    # Request target is validated
    for bad_byte in b"\x00\x20\x7f\xee":
        target = bytearray(b"/")
        target.append(bad_byte)
        with pytest.raises(LocalProtocolError):
            Request(
                method="GET", target=target, headers=[("Host", "a")], http_version="1.1"
            )

    # Request method is validated
    with pytest.raises(LocalProtocolError):
        Request(
            method="GET / HTTP/1.1",
            target=target,
            headers=[("Host", "a")],
            http_version="1.1",
        )

    ir = InformationalResponse(status_code=100, headers=[("Host", "a")])
    assert ir.status_code == 100
    assert ir.headers == [(b"host", b"a")]
    assert ir.http_version == b"1.1"

    with pytest.raises(LocalProtocolError):
        InformationalResponse(status_code=200, headers=[("Host", "a")])

    resp = Response(status_code=204, headers=[], http_version="1.0")  # type: ignore[arg-type]
    assert resp.status_code == 204
    assert resp.headers == []
    assert resp.http_version == b"1.0"

    with pytest.raises(LocalProtocolError):
        resp = Response(status_code=100, headers=[], http_version="1.0")  # type: ignore[arg-type]

    with pytest.raises(LocalProtocolError):
        Response(status_code="100", headers=[], http_version="1.0")  # type: ignore[arg-type]

    with pytest.raises(LocalProtocolError):
        InformationalResponse(status_code=b"100", headers=[], http_version="1.0")  # type: ignore[arg-type]

    d = Data(data=b"asdf")
    assert d.data == b"asdf"

    eom = EndOfMessage()
    assert eom.headers == []

    cc = ConnectionClosed()
    assert repr(cc) == "ConnectionClosed()"


def test_intenum_status_code() -> None:
    # https://github.com/python-hyper/h11/issues/72

    r = Response(status_code=HTTPStatus.OK, headers=[], http_version="1.0")  # type: ignore[arg-type]
    assert r.status_code == HTTPStatus.OK
    assert type(r.status_code) is not type(HTTPStatus.OK)
    assert type(r.status_code) is int


def test_header_casing() -> None:
    r = Request(
        method="GET",
        target="/",
        headers=[("Host", "example.org"), ("Connection", "keep-alive")],
        http_version="1.1",
    )
    assert len(r.headers) == 2
    assert r.headers[0] == (b"host", b"example.org")
    assert r.headers == [(b"host", b"example.org"), (b"connection", b"keep-alive")]
    assert r.headers.raw_items() == [
        (b"Host", b"example.org"),
        (b"Connection", b"keep-alive"),
    ]