# Tests for aiohttp/protocol.py

import asyncio
import gzip
import platform
import re
import sys
import zlib
from collections.abc import Iterable, Iterator
from contextlib import suppress
from typing import Any
from unittest import mock
from urllib.parse import quote

import pytest
from multidict import CIMultiDict
from yarl import URL

import aiohttp
from aiohttp import http_exceptions, streams
from aiohttp.base_protocol import BaseProtocol
from aiohttp.client_proto import ResponseHandler
from aiohttp.helpers import DEFAULT_CHUNK_SIZE, NO_EXTENSIONS, HeadersDictProxy
from aiohttp.http_parser import (
    DeflateBuffer,
    HeadersParser,
    HttpParser,
    HttpPayloadParser,
    HttpRequestParser,
    HttpRequestParserPy,
    HttpResponseParser,
    HttpResponseParserPy,
    PayloadState,
)
from aiohttp.http_writer import HttpVersion
from aiohttp.web_protocol import RequestHandler
from aiohttp.web_request import Request
from aiohttp.web_server import Server

try:
    try:
        import brotlicffi as brotli
    except ImportError:
        import brotli
except ImportError:
    brotli = None

try:
    if sys.version_info >= (3, 14):
        import compression.zstd as zstandard  # noqa: I900
    else:
        import backports.zstd as zstandard
except ImportError:
    zstandard = None  # type: ignore[assignment]

REQUEST_PARSERS = [HttpRequestParserPy]
RESPONSE_PARSERS = [HttpResponseParserPy]

with suppress(ImportError):
    from aiohttp.http_parser import HttpRequestParserC, HttpResponseParserC

    REQUEST_PARSERS.append(HttpRequestParserC)
    RESPONSE_PARSERS.append(HttpResponseParserC)


@pytest.fixture
def server() -> Any:
    return mock.create_autospec(
        Server,
        request_factory=mock.Mock(),
        request_handler=mock.AsyncMock(),
        instance=True,
    )


@pytest.fixture
def protocol() -> Any:
    return mock.create_autospec(
        BaseProtocol,
        spec_set=True,
        instance=True,
    )


def _gen_ids(parsers: Iterable[type[HttpParser[Any]]]) -> list[str]:
    return [
        "py-parser" if parser.__module__ == "aiohttp.http_parser" else "c-parser"
        for parser in parsers
    ]


@pytest.fixture(params=REQUEST_PARSERS, ids=_gen_ids(REQUEST_PARSERS))
def parser(
    event_loop: asyncio.AbstractEventLoop,
    server: Server[Request],
    request: pytest.FixtureRequest,
) -> Iterator[HttpRequestParser]:
    protocol = RequestHandler(server, loop=event_loop)

    # Parser implementations
    parser = request.param(
        protocol,
        event_loop,
        DEFAULT_CHUNK_SIZE,
        max_line_size=8190,
        max_headers=128,
        max_field_size=8190,
    )
    protocol._force_close = False
    protocol._parser = parser
    with mock.patch.object(protocol, "transport", True):
        yield parser


@pytest.fixture(params=REQUEST_PARSERS, ids=_gen_ids(REQUEST_PARSERS))
def request_cls(request: pytest.FixtureRequest) -> type[HttpRequestParser]:
    # Request Parser class
    return request.param  # type: ignore[no-any-return]


@pytest.fixture(params=RESPONSE_PARSERS, ids=_gen_ids(RESPONSE_PARSERS))
def response(
    event_loop: asyncio.AbstractEventLoop,
    request: pytest.FixtureRequest,
) -> HttpResponseParser:
    protocol = ResponseHandler(event_loop)

    # Parser implementations
    parser = request.param(
        protocol,
        event_loop,
        DEFAULT_CHUNK_SIZE,
        max_line_size=8190,
        max_headers=128,
        max_field_size=8190,
        read_until_eof=True,
    )
    protocol._parser = parser
    return parser  # type: ignore[no-any-return]


@pytest.fixture(params=RESPONSE_PARSERS, ids=_gen_ids(RESPONSE_PARSERS))
def response_cls(request: pytest.FixtureRequest) -> type[HttpResponseParser]:
    # Parser implementations
    return request.param  # type: ignore[no-any-return]


@pytest.mark.skipif(NO_EXTENSIONS, reason="Extensions available but not imported")
def test_c_parser_loaded() -> None:
    assert "HttpRequestParserC" in dir(aiohttp.http_parser)
    assert "HttpResponseParserC" in dir(aiohttp.http_parser)
    assert "RawRequestMessageC" in dir(aiohttp.http_parser)
    assert "RawResponseMessageC" in dir(aiohttp.http_parser)


_PIPELINED_GET = b"GET / HTTP/1.1\r\nHost: a\r\n\r\n"


def _build_request_parser(
    request_cls: type[HttpRequestParser],
    protocol: BaseProtocol,
    loop: asyncio.AbstractEventLoop,
    max_msg_queue_size: int,
) -> HttpRequestParser:
    return request_cls(
        protocol,
        loop,
        DEFAULT_CHUNK_SIZE,
        max_line_size=8190,
        max_headers=128,
        max_field_size=8190,
        max_msg_queue_size=max_msg_queue_size,
    )


def test_max_msg_queue_size_caps_emitted_messages(
    request_cls: type[HttpRequestParser],
    protocol: BaseProtocol,
    event_loop: asyncio.AbstractEventLoop,
) -> None:
    parser = _build_request_parser(request_cls, protocol, event_loop, 4)
    messages, upgraded, _tail = parser.feed_data(_PIPELINED_GET * 10)
    assert len(messages) == 4
    assert not upgraded


def test_max_msg_queue_size_resumes_after_consume(
    request_cls: type[HttpRequestParser],
    protocol: BaseProtocol,
    event_loop: asyncio.AbstractEventLoop,
) -> None:
    limit = 4
    total = 10
    parser = _build_request_parser(request_cls, protocol, event_loop, limit)
    messages, _upgraded, _tail = parser.feed_data(_PIPELINED_GET * total)
    seen = 0
    while messages:
        assert len(messages) <= limit
        seen += len(messages)
        for _msg, _payload in messages:
            parser.message_consumed()
        messages, _upgraded, _tail = parser.feed_data(b"")
    assert seen == total


def test_max_msg_queue_size_zero_is_unbounded(
    request_cls: type[HttpRequestParser],
    protocol: BaseProtocol,
    event_loop: asyncio.AbstractEventLoop,
) -> None:
    parser = _build_request_parser(request_cls, protocol, event_loop, 0)
    messages, _upgraded, _tail = parser.feed_data(_PIPELINED_GET * 50)
    assert len(messages) == 50


def test_message_consumed_underflow_is_ignored(
    request_cls: type[HttpRequestParser],
    protocol: BaseProtocol,
    event_loop: asyncio.AbstractEventLoop,
) -> None:
    parser = _build_request_parser(request_cls, protocol, event_loop, 4)
    # No message is in flight; consuming must not underflow the counter.
    parser.message_consumed()
    messages, _upgraded, _tail = parser.feed_data(_PIPELINED_GET * 4)
    assert len(messages) == 4


def test_parse_headers(parser: HttpRequestParser) -> None:
    text = b"""GET /test HTTP/1.1\r
Host: a\r
test: a line\r
test2: data\r
\r
"""
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    msg = messages[0][0]

    assert list(msg.headers.items()) == [
        ("Host", "a"),
        ("test", "a line"),
        ("test2", "data"),
    ]
    assert msg.raw_headers == (
        (b"Host", b"a"),
        (b"test", b"a line"),
        (b"test2", b"data"),
    )
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade


def test_reject_obsolete_line_folding(parser: HttpRequestParser) -> None:
    text = b"""GET /test HTTP/1.1\r
Host: a\r
test: line\r
 Content-Length: 48\r
test2: data\r
\r
"""
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


@pytest.mark.skipif(NO_EXTENSIONS, reason="Only tests C parser.")
def test_invalid_character(
    event_loop: asyncio.AbstractEventLoop,
    server: Server[Request],
    request: pytest.FixtureRequest,
) -> None:
    protocol = RequestHandler(server, loop=event_loop)

    parser = HttpRequestParserC(
        protocol,
        event_loop,
        2**16,
        max_line_size=8190,
        max_field_size=8190,
    )
    protocol._parser = parser
    text = b"POST / HTTP/1.1\r\nHost: localhost:8080\r\nSet-Cookie: abc\x01def\r\n\r\n"
    error_detail = re.escape(r""":

    b'Set-Cookie: abc\x01def'
                     ^""")
    with pytest.raises(http_exceptions.BadHttpMessage, match=error_detail):
        parser.feed_data(text)


@pytest.mark.skipif(NO_EXTENSIONS, reason="Only tests C parser.")
def test_invalid_linebreak(
    event_loop: asyncio.AbstractEventLoop,
    server: Server[Request],
    request: pytest.FixtureRequest,
) -> None:
    protocol = RequestHandler(server, loop=event_loop)

    parser = HttpRequestParserC(
        protocol,
        event_loop,
        2**16,
        max_line_size=8190,
        max_field_size=8190,
    )
    protocol._parser = parser
    text = b"GET /world HTTP/1.1\r\nHost: 127.0.0.1\n\r\n"
    error_detail = re.escape(r""":

    b'Host: 127.0.0.1\n'
                     ^""")
    with pytest.raises(http_exceptions.BadHttpMessage, match=error_detail):
        parser.feed_data(text)


def test_partial_request_split_still_parses(parser: HttpRequestParser) -> None:
    """A legitimate request split mid-line must still stream."""
    messages, _, _ = parser.feed_data(b"GET /split HTTP/1.1\r\nHo")
    assert not messages
    messages, _, _ = parser.feed_data(b"st: a\r\n\r\n")
    assert len(messages) == 1
    assert messages[0][0].path == "/split"


@pytest.mark.parametrize(
    "attacker",
    (
        pytest.param(
            b"GET /attacker HTTP/1.1\nHost: a\nX-Foo: bar\n\n", id="request_line"
        ),
        pytest.param(
            b"POST /attacker HTTP/1.1\r\nHost: a\r\n"
            b"Transfer-Encoding: chunked\r\n\r\n5\n",
            id="chunk_size",
        ),
        pytest.param(
            b"POST /attacker HTTP/1.1\r\nHost: a\r\n"
            b"Transfer-Encoding: chunked\r\n\r\n0\r\nX-Trailer: bad\n",
            id="trailer",
        ),
    ),
)
def test_reject_bare_lf_at_point_seen(
    parser: HttpRequestParser, attacker: bytes
) -> None:
    """A bare LF where CRLF is required is rejected at the point it's seen."""
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(attacker)


def test_cve_2023_37276(parser: HttpRequestParser) -> None:
    text = b"""POST / HTTP/1.1\r\nHost: localhost:8080\r\nX-Abc: \rxTransfer-Encoding: chunked\r\n\r\n"""
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


@pytest.mark.parametrize(
    "rfc9110_5_6_2_token_delim",
    r'"(),/:;<=>?@[\]{}',
)
def test_bad_header_name(
    parser: HttpRequestParser, rfc9110_5_6_2_token_delim: str
) -> None:
    text = f"POST / HTTP/1.1\r\nHost: a\r\nhead{rfc9110_5_6_2_token_delim}er: val\r\n\r\n".encode()
    if rfc9110_5_6_2_token_delim == ":":
        # Inserting colon into header just splits name/value earlier.
        parser.feed_data(text)
        return

    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


@pytest.mark.parametrize(
    ("hdr_vals", "expected"),
    (
        (
            ('"http://example.com/a.html,foo", apples',),
            ("http://example.com/a.html,foo", "apples"),
        ),
        (("bananas, apples",), ("bananas", "apples")),
        (("bananas", "apples"), ("bananas", "apples")),
        (
            ('"http://example.com/a.html,foo", "apples"',),
            ("http://example.com/a.html,foo", "apples"),
        ),
        (
            ('"Sat, 04 May 1996", "Wed, 14 Sep 2005"',),
            ("Sat, 04 May 1996", "Wed, 14 Sep 2005"),
        ),
        (("foo,bar,baz",), ("foo", "bar", "baz")),
        (('"applebanna, this',), ('"applebanna', "this")),
        (('fooo", "bar"',), ('fooo"', "bar")),
        ((" spam , eggs ",), ("spam", "eggs")),
        ((" spam ", " eggs "), ("spam", "eggs")),
        ((r'spam"foo\"bar"',), (r'spam"foo\"bar"',)),
        # https://www.rfc-editor.org/rfc/rfc9110.html#name-recipient-requirements
        (("foo, ,bar,",), ("foo", "bar")),
        ((",   , ",), ()),
        (("",), ()),
        # Escaped characters in quoted strings
        # https://www.rfc-editor.org/rfc/rfc9110.html#section-5.6.4-3
        ((r'"foo\"bar"',), ('foo"bar',)),
        ((r'"foo\\\"bar"',), (r"foo\"bar",)),
        ((r'"foo\\", bar',), ("foo\\", "bar")),
        # Comments: https://www.rfc-editor.org/rfc/rfc9110.html#section-5.6.5
        ((r"foo (bar\"spam\)eggs,\\baz)",), (r'foo (bar"spam)eggs,\baz)',)),
        # Not a comment (requires whitespace)
        (("foo(bar,spam)",), ("foo(bar", "spam)")),
        # Parameters: https://www.rfc-editor.org/rfc/rfc9110.html#section-5.6.6
        (("text/html;charset=utf-8",), ("text/html;charset=utf-8",)),
        (('Text/HTML; Charset="utf-8"',), ('Text/HTML; Charset="utf-8"',)),
        (
            ("text/plain; q=0.5, text/html, text/x-dvi; q=0.8; format=flowed, */*",),
            (
                "text/plain; q=0.5",
                "text/html",
                "text/x-dvi; q=0.8; format=flowed",
                "*/*",
            ),
        ),
        ((r'foo; bar="spam,\"eggs", baz',), ('foo; bar="spam,"eggs"', "baz")),
        ((r'foo;bar="spam\\\",eggs",baz',), ('foo;bar="spam\\",eggs"', "baz")),
        # Not valid parameters
        (('foo; bar ="spam,eggs"',), ('foo; bar ="spam', 'eggs"')),
        (('foo;bar= "spam,eggs"',), ('foo;bar= "spam', 'eggs"')),
    ),
)
def test_list_headers(
    parser: HttpRequestParser, hdr_vals: tuple[str], expected: tuple[str, ...]
) -> None:
    headers = "\r\n".join(f"Foo: {v}" for v in hdr_vals)
    text = f"POST / HTTP/1.1\r\nHost: a\r\n{headers}\r\n\r\n".encode()
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.headers.getall("Foo") == expected


@pytest.mark.parametrize(
    "hdr",
    (
        "Content-Length: -5",  # https://www.rfc-editor.org/rfc/rfc9110.html#name-content-length
        "Content-Length: +256",
        "Content-Length: \N{SUPERSCRIPT ONE}",
        "Content-Length: \N{MATHEMATICAL DOUBLE-STRUCK DIGIT ONE}",
        "Foo: abc\rdef",  # https://www.rfc-editor.org/rfc/rfc9110.html#section-5.5-5
        "Bar: abc\ndef",
        "Baz: abc\x00def",
        "Foo : bar",  # https://www.rfc-editor.org/rfc/rfc9112.html#section-5.1-2
        "Foo\t: bar",
        "\xffoo: bar",
        "Foo: abc\x01def",  # CTL bytes forbidden per RFC 9110 §5.5
        "Foo: abc\x7fdef",  # DEL is also a CTL byte
        "Foo: abc\x1fdef",
    ),
)
def test_bad_headers(parser: HttpRequestParser, hdr: str) -> None:
    text = f"POST / HTTP/1.1\r\nHost: a\r\n{hdr}\r\n\r\n".encode()
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_ctl_host_header_bad_characters(parser: HttpRequestParser) -> None:
    """CTL byte in Host header must be rejected."""
    text = b"GET /test HTTP/1.1\r\nHost: trusted.example\x01@bad.test\r\n\r\n"
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_unpaired_surrogate_in_header_py(
    event_loop: asyncio.AbstractEventLoop, server: Server[Request]
) -> None:
    protocol = RequestHandler(server, loop=event_loop)

    parser = HttpRequestParserPy(
        protocol,
        event_loop,
        2**16,
        max_line_size=8190,
        max_field_size=8190,
    )
    protocol._parser = parser
    text = b"POST / HTTP/1.1\r\nHost: a\r\n\xff\r\n\r\n"
    message = None
    try:
        parser.feed_data(text)
    except http_exceptions.InvalidHeader as e:
        message = e.message.encode("utf-8")
    assert message is not None


def test_content_length_transfer_encoding(parser: HttpRequestParser) -> None:
    text = (
        b"GET / HTTP/1.1\r\nHost: a\r\nContent-Length: 5\r\nTransfer-Encoding: a\r\n\r\n"
        + b"apple\r\n"
    )
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


@pytest.mark.parametrize(
    "hdr",
    (
        "Content-Length",
        "Host",
        "Transfer-Encoding",
    ),
)
def test_duplicate_singleton_header_rejected(
    parser: HttpRequestParser, hdr: str
) -> None:
    val1, val2 = ("1", "2") if hdr == "Content-Length" else ("value1", "value2")
    text = (
        f"GET /test HTTP/1.1\r\n"
        f"Host: example.com\r\n"
        f"{hdr}: {val1}\r\n"
        f"{hdr}: {val2}\r\n"
        "\r\n"
    ).encode()
    with pytest.raises(http_exceptions.BadHttpMessage, match="Duplicate"):
        parser.feed_data(text)


@pytest.mark.parametrize(
    "hdr",
    (
        "Content-Location",
        "Content-Range",
        "Content-Type",
        "ETag",
        "Max-Forwards",
        "Server",
        "User-Agent",
    ),
)
def test_duplicate_non_security_singleton_header_rejected_strict(
    parser: HttpRequestParser, hdr: str
) -> None:
    """Non-security singletons are rejected in strict mode (requests)."""
    text = (
        f"GET /test HTTP/1.1\r\n"
        f"Host: example.com\r\n"
        f"{hdr}: value1\r\n"
        f"{hdr}: value2\r\n"
        "\r\n"
    ).encode()
    with pytest.raises(http_exceptions.BadHttpMessage, match="Duplicate"):
        parser.feed_data(text)


@pytest.mark.parametrize(
    "hdr",
    (
        # Content-Length is excluded because llhttp rejects duplicates
        # at the C level before our singleton check runs.
        "Content-Location",
        "Content-Range",
        "Content-Type",
        "ETag",
        "Max-Forwards",
        "Server",
        "Transfer-Encoding",
        "User-Agent",
    ),
)
def test_duplicate_singleton_header_accepted_in_lax_mode(
    response: HttpResponseParser, hdr: str
) -> None:
    """All singleton duplicates are accepted in lax mode (response parser default)."""
    text = (f"HTTP/1.1 200 OK\r\n{hdr}: value1\r\n{hdr}: value2\r\n\r\n").encode()
    messages, upgrade, tail = response.feed_data(text)
    assert len(messages) == 1


def test_duplicate_host_header_rejected(parser: HttpRequestParser) -> None:
    text = (
        b"GET /admin HTTP/1.1\r\n"
        b"Host: admin.example\r\n"
        b"Host: public.example\r\n"
        b"\r\n"
    )
    with pytest.raises(http_exceptions.BadHttpMessage, match="Duplicate.*Host"):
        parser.feed_data(text)


def test_missing_host_header_rejected(parser: HttpRequestParser) -> None:
    text = b"GET /admin HTTP/1.1\r\n\r\n"
    with pytest.raises(http_exceptions.BadHttpMessage, match="Missing 'Host' header"):
        parser.feed_data(text)


@pytest.mark.parametrize(
    ("hdr1", "hdr2"),
    (
        ("content-length", "Content-Length"),
        ("Content-Length", "content-length"),
        ("transfer-encoding", "Transfer-Encoding"),
        ("Transfer-Encoding", "transfer-encoding"),
    ),
)
def test_duplicate_singleton_header_different_casing_rejected(
    parser: HttpRequestParser, hdr1: str, hdr2: str
) -> None:
    """Singleton check must be case-insensitive per RFC 9110."""
    val1, val2 = ("1", "2") if "content-length" in hdr1.lower() else ("v1", "v2")
    text = (
        f"GET /test HTTP/1.1\r\n"
        f"Host: example.com\r\n"
        f"{hdr1}: {val1}\r\n"
        f"{hdr2}: {val2}\r\n"
        "\r\n"
    ).encode()
    with pytest.raises(http_exceptions.BadHttpMessage, match="Duplicate"):
        parser.feed_data(text)


def test_duplicate_host_header_different_casing_rejected(
    parser: HttpRequestParser,
) -> None:
    """Duplicate Host with different casing must also be rejected."""
    text = (
        b"GET /test HTTP/1.1\r\n"
        b"host: evil.example\r\n"
        b"Host: good.example\r\n"
        b"\r\n"
    )
    with pytest.raises(http_exceptions.BadHttpMessage, match="Duplicate"):
        parser.feed_data(text)


def test_bad_chunked(parser: HttpRequestParser) -> None:
    """Test that invalid chunked encoding doesn't allow content-length to be used."""
    text = (
        b"GET / HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked\r\n\r\n0_2e\r\n\r\n"
        + b"GET / HTTP/1.1\r\nHost: a\r\nContent-Length: 5\r\n\r\n0\r\n\r\n"
    )
    with pytest.raises(http_exceptions.BadHttpMessage, match="0_2e"):
        parser.feed_data(text)


def test_whitespace_before_header(parser: HttpRequestParser) -> None:
    text = b"GET / HTTP/1.1\r\nHost: a\r\n\tContent-Length: 1\r\n\r\nX"
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_whitespace_around_header_value(parser: HttpRequestParser) -> None:
    text = b"GET / HTTP/1.1\r\nHost: a\r\ntest: \t value \t \r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.headers["test"] == "value"
    assert msg.raw_headers == ((b"Host", b"a"), (b"test", b"value"))


@pytest.fixture
def xfail_c_parser_status(request: pytest.FixtureRequest) -> None:
    if isinstance(request.getfixturevalue("parser"), HttpRequestParserPy):
        return
    request.node.add_marker(
        pytest.mark.xfail(
            reason="Regression test for Py parser. May match C behaviour later.",
            raises=http_exceptions.BadStatusLine,
        )
    )


@pytest.mark.usefixtures("xfail_c_parser_status")
def test_parse_unusual_request_line(parser: HttpRequestParser) -> None:
    text = b"#smol //a HTTP/1.3\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    msg, _ = messages[0]
    assert msg.compression is None
    assert not msg.upgrade
    assert msg.method == "#SMOL"
    assert msg.path == "//a"
    assert msg.version == (1, 3)


def test_py_parser_normalises_method_to_uppercase(
    event_loop: asyncio.AbstractEventLoop, server: Server[Request]
) -> None:
    """Test Python parser canonicalises method tokens.

    llhttp rejects lowercase upstream, so this only applies to the Python parser.
    """
    protocol = RequestHandler(server, loop=event_loop)
    parser = HttpRequestParserPy(
        protocol,
        event_loop,
        2**16,
        max_line_size=8190,
        max_field_size=8190,
    )
    protocol._parser = parser
    text = b"get /test HTTP/1.1\r\nHost: a\r\n\r\n"
    messages, _upgrade, _tail = parser.feed_data(text)
    assert len(messages) == 1
    msg, _ = messages[0]
    assert msg.method == "GET"


def test_parse(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    msg, _ = messages[0]
    assert msg.compression is None
    assert not msg.upgrade
    assert msg.method == "GET"
    assert msg.path == "/test"
    assert msg.version == (1, 1)
    assert msg.headers["Host"] == "a"


def test_parse_query_method(parser: HttpRequestParser) -> None:
    """QUERY has the highest llhttp method id; both parsers must decode its name.

    Regression test for a stale hand-maintained method count in the C parser
    that mapped the last method(s) to "<unknown>".
    """
    text = b"QUERY /test HTTP/1.1\r\nHost: a\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    msg = messages[0][0]
    assert msg.method == "QUERY"
    assert msg.path == "/test"
    assert msg.version == (1, 1)


async def test_parse_body(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nContent-Length: 4\r\n\r\nbody"
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    _, payload = messages[0]
    body = await payload.read(4)
    assert body == b"body"


async def test_parse_body_with_CRLF(parser: HttpRequestParser) -> None:
    text = b"\r\nGET /test HTTP/1.1\r\nHost: a\r\nContent-Length: 4\r\n\r\nbody"
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    _, payload = messages[0]
    body = await payload.read(4)
    assert body == b"body"


def test_parse_delayed(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 0
    assert not upgrade

    messages, upgrade, tail = parser.feed_data(b"\r\n")
    assert len(messages) == 1
    msg = messages[0][0]
    assert msg.method == "GET"


def test_headers_multi_feed(parser: HttpRequestParser) -> None:
    text1 = b"GET /test HTTP/1.1\r\n"
    text2 = b"Host: a\r\n"
    text3 = b"test: line"
    text4 = b" continue\r\n\r\n"

    messages, upgrade, tail = parser.feed_data(text1)
    assert len(messages) == 0

    messages, upgrade, tail = parser.feed_data(text2)
    assert len(messages) == 0

    messages, upgrade, tail = parser.feed_data(text3)
    assert len(messages) == 0

    messages, upgrade, tail = parser.feed_data(text4)
    assert len(messages) == 1

    msg = messages[0][0]
    assert list(msg.headers.items()) == [("Host", "a"), ("test", "line continue")]
    assert msg.raw_headers == ((b"Host", b"a"), (b"test", b"line continue"))
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade


def test_headers_split_field(parser: HttpRequestParser) -> None:
    text1 = b"GET /test HTTP/1.1\r\nHost: a\r\n"
    text2 = b"t"
    text3 = b"es"
    text4 = b"t: value\r\n\r\n"

    messages, upgrade, tail = parser.feed_data(text1)
    messages, upgrade, tail = parser.feed_data(text2)
    messages, upgrade, tail = parser.feed_data(text3)
    assert len(messages) == 0
    messages, upgrade, tail = parser.feed_data(text4)
    assert len(messages) == 1

    msg = messages[0][0]
    assert list(msg.headers.items()) == [("Host", "a"), ("test", "value")]
    assert msg.raw_headers == ((b"Host", b"a"), (b"test", b"value"))
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade


def test_parse_headers_multi(parser: HttpRequestParser) -> None:
    text = (
        b"GET /test HTTP/1.1\r\nHost: a\r\n"
        b"Set-Cookie: c1=cookie1\r\n"
        b"Set-Cookie: c2=cookie2\r\n\r\n"
    )

    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    msg = messages[0][0]

    assert tuple(msg.headers.items()) == (
        ("Host", "a"),
        ("Set-Cookie", "c1=cookie1, c2=cookie2"),
    )
    assert msg.raw_headers == (
        (b"Host", b"a"),
        (b"Set-Cookie", b"c1=cookie1"),
        (b"Set-Cookie", b"c2=cookie2"),
    )
    assert not msg.should_close
    assert msg.compression is None


def test_conn_default_1_0(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.0\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.should_close


def test_conn_default_1_1(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close


def test_conn_close(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nconnection: close\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.should_close


def test_conn_close_1_0(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.0\r\nconnection: close\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.should_close


def test_conn_keep_alive_1_0(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.0\r\nconnection: keep-alive\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close


def test_conn_keep_alive_1_1(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nconnection: keep-alive\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close


def test_conn_close_comma_list(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nconnection: close, keep-alive\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.should_close


def test_conn_close_multiple_headers(parser: HttpRequestParser) -> None:
    text = (
        b"GET /test HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"connection: keep-alive\r\n"
        b"connection: close\r\n\r\n"
    )
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.should_close


def test_conn_other_1_0(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.0\r\nconnection: test\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.should_close


def test_conn_other_1_1(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nconnection: test\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close


def test_request_chunked(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntransfer-encoding: chunked\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg, payload = messages[0]
    assert msg.chunked
    assert not upgrade
    assert isinstance(payload, streams.StreamReader)


def test_te_header_non_ascii(parser: HttpRequestParser) -> None:
    # K = Kelvin sign, not valid ascii.
    text = "GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunKed\r\n\r\n"
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text.encode())


def test_upgrade_header_non_ascii(parser: HttpRequestParser) -> None:
    # K = Kelvin sign, not valid ascii.
    text = "GET /test HTTP/1.1\r\nHost: a\r\nUpgrade: websocKet\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text.encode())
    assert not upgrade


@pytest.mark.parametrize(
    ("connection", "expected"),
    [
        ("upgrade", True),
        ("upgrade, keep-alive", True),  # other tokens alongside upgrade
        ("keep-alive, upgrade", True),  # upgrade not first
        ("Upgrade, Keep-Alive", True),  # case-insensitive
        ("keep-alive", False),  # no upgrade token
        ("keep-alive, notupgrade", False),  # substring is not a token
    ],
)
def test_response_upgrade_token_in_connection_list(
    response: HttpResponseParser, connection: str, expected: bool
) -> None:
    # RFC 9110 §7.6.1: Connection is a comma-separated token list, so the parser
    # must set msg.upgrade for a 101 response whenever "upgrade" appears as a
    # token, regardless of position, case, or neighbouring tokens.
    text = (
        b"HTTP/1.1 101 Switching Protocols\r\n"
        b"Upgrade: websocket\r\n"
        b"Connection: " + connection.encode() + b"\r\n\r\n"
    )
    messages, upgrade, tail = response.feed_data(text)
    msg = messages[0][0]
    assert msg.upgrade == expected
    assert upgrade == expected


def test_request_te_chunked_with_content_length(parser: HttpRequestParser) -> None:
    text = (
        b"GET /test HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"content-length: 1234\r\n"
        b"transfer-encoding: chunked\r\n\r\n"
    )
    with pytest.raises(
        http_exceptions.BadHttpMessage,
        match="Transfer-Encoding can't be present with Content-Length",
    ):
        parser.feed_data(text)


def test_request_te_chunked123(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntransfer-encoding: chunked123\r\n\r\n"
    with pytest.raises(
        http_exceptions.BadHttpMessage,
        match="Request has invalid `Transfer-Encoding`",
    ):
        parser.feed_data(text)


def test_request_te_empty_list_invalid(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: ,  \t ,\r\n\r\n"
    with pytest.raises(
        http_exceptions.BadHttpMessage,
        match="Request has invalid `Transfer-Encoding`",
    ):
        parser.feed_data(text)


async def test_request_te_last_chunked(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: not, chunked\r\n\r\n1\r\nT\r\n3\r\nest\r\n0\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.3
    assert await messages[0][1].read() == b"Test"


def test_request_te_first_chunked(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked, not\r\n\r\n1\r\nT\r\n3\r\nest\r\n0\r\n\r\n"
    # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.3
    with pytest.raises(
        http_exceptions.BadHttpMessage,
        match="nvalid `Transfer-Encoding`",
    ):
        parser.feed_data(text)


def test_request_te_duplicate_chunked(parser: HttpRequestParser) -> None:
    """Reject duplicate chunked Transfer-Encoding per RFC 9112 section 7.1."""
    text = b"POST / HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked, chunked\r\n\r\n0\r\n\r\n"
    # https://www.rfc-editor.org/rfc/rfc9112#section-7.1-3
    with pytest.raises(
        http_exceptions.BadHttpMessage,
        match="duplicate `chunked` Transfer-Encoding|nvalid `Transfer-Encoding`",
    ):
        parser.feed_data(text)


def test_conn_upgrade(parser: HttpRequestParser) -> None:
    text = (
        b"GET /test HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"connection: upgrade\r\n"
        b"upgrade: websocket\r\n\r\n"
    )
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close
    assert msg.upgrade
    assert upgrade


def test_conn_upgrade_comma_list(parser: HttpRequestParser) -> None:
    text = (
        b"GET /test HTTP/1.1\r\n"
        b"host: a\r\n"
        b"connection: keep-alive, upgrade\r\n"
        b"upgrade: websocket\r\n\r\n"
    )
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close
    assert msg.upgrade
    assert upgrade


def test_conn_upgrade_multiple_headers(parser: HttpRequestParser) -> None:
    text = (
        b"GET /test HTTP/1.1\r\n"
        b"host: a\r\n"
        b"connection: keep-alive\r\n"
        b"connection: upgrade\r\n"
        b"upgrade: websocket\r\n\r\n"
    )
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close
    assert msg.upgrade
    assert upgrade


def test_bad_upgrade(parser: HttpRequestParser) -> None:
    """Test not upgraded if missing Upgrade header."""
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nconnection: upgrade\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.upgrade
    assert not upgrade


def test_compression_empty(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-encoding: \r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.compression is None


def test_compression_deflate(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-encoding: deflate\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.compression == "deflate"


def test_compression_gzip(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-encoding: gzip\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.compression == "gzip"


@pytest.mark.skipif(brotli is None, reason="brotli is not installed")
def test_compression_brotli(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-encoding: br\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.compression == "br"


@pytest.mark.skipif(zstandard is None, reason="zstandard is not installed")
def test_compression_zstd(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-encoding: zstd\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.compression == "zstd"


@pytest.mark.parametrize(
    "enc",
    (
        "zﬆd".encode(),  # "ﬆ".upper() == "ST"
        "deﬂate".encode(),  # "ﬂ".upper() == "FL"
    ),
)
def test_compression_non_ascii(parser: HttpRequestParser, enc: bytes) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-encoding: " + enc + b"\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    # Non-ascii input should not evaluate to a valid encoding scheme.
    assert msg.compression is None


def test_compression_unknown(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-encoding: compress\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.compression is None


def test_url_connect(parser: HttpRequestParser) -> None:
    text = b"CONNECT www.google.com HTTP/1.1\r\nHost: a\r\ncontent-length: 0\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg, payload = messages[0]
    assert upgrade
    assert msg.url == URL.build(authority="www.google.com")


def test_headers_connect(parser: HttpRequestParser) -> None:
    text = b"CONNECT www.google.com HTTP/1.1\r\nHost: a\r\ncontent-length: 0\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg, payload = messages[0]
    assert upgrade
    assert isinstance(payload, streams.StreamReader)


def test_url_absolute(parser: HttpRequestParser) -> None:
    text = (
        b"GET https://www.google.com/path/to.html HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"content-length: 0\r\n\r\n"
    )
    messages, upgrade, tail = parser.feed_data(text)
    msg, payload = messages[0]
    assert not upgrade
    assert msg.method == "GET"
    assert msg.url == URL("https://www.google.com/path/to.html")


def test_url_authority_form_only_connect(parser: HttpRequestParser) -> None:
    # https://www.rfc-editor.org/info/rfc9112/#section-3.2.3-1
    with pytest.raises(http_exceptions.InvalidURLError):
        parser.feed_data(b"GET www.google.com:443 HTTP/1.1\r\nHost: a\r\n\r\n")


def test_headers_old_websocket_key1(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nSEC-WEBSOCKET-KEY1: line\r\n\r\n"

    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_headers_content_length_err_1(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-length: line\r\n\r\n"

    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_headers_content_length_err_2(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ncontent-length: -1\r\n\r\n"

    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


_pad: dict[bytes, str] = {
    b"": "empty",
    # not a typo. Python likes triple zero
    b"\000": "NUL",
    b" ": "SP",
    b"  ": "SPSP",
    # not a typo: both 0xa0 and 0x0a in case of 8-bit fun
    b"\n": "LF",
    b"\xa0": "NBSP",
    b"\t ": "TABSP",
}


@pytest.mark.parametrize("hdr", [b"", b"foo"], ids=["name-empty", "with-name"])
@pytest.mark.parametrize("pad2", _pad.keys(), ids=["post-" + n for n in _pad.values()])
@pytest.mark.parametrize("pad1", _pad.keys(), ids=["pre-" + n for n in _pad.values()])
def test_invalid_header_spacing(
    parser: HttpRequestParser, pad1: bytes, pad2: bytes, hdr: bytes
) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\n%s%s%s: value\r\n\r\n" % (pad1, hdr, pad2)
    if pad1 == pad2 == b"" and hdr != b"":
        # one entry in param matrix is correct: non-empty name, not padded
        parser.feed_data(text)
        return

    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_empty_header_name(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\n:test\r\n\r\n"
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_invalid_header(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntest line\r\n\r\n"
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


def test_invalid_name(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntest[]: line\r\n\r\n"

    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(text)


@pytest.mark.parametrize("size", [40960, 8191])
def test_max_header_field_size(parser: HttpRequestParser, size: int) -> None:
    name = b"t" * size
    text = b"GET /test HTTP/1.1\r\n" + name + b":data\r\n\r\n"

    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        for i in range(0, len(text), 5000):  # pragma: no branch
            parser.feed_data(text[i : i + 5000])


def test_max_header_size_under_limit(parser: HttpRequestParser) -> None:
    name = b"t" * 8185
    text = b"GET /test HTTP/1.1\r\nHost: a\r\n" + name + b":data\r\n\r\n"

    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.method == "GET"
    assert msg.path == "/test"
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict([("Host", "a"), (name.decode(), "data")])
    assert msg.raw_headers == ((b"Host", b"a"), (name, b"data"))
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert msg.url == URL("/test")


@pytest.mark.parametrize("size", [40960, 8191])
def test_max_header_value_size(parser: HttpRequestParser, size: int) -> None:
    name = b"t" * size
    text = b"GET /test HTTP/1.1\r\ndata:" + name + b"\r\n\r\n"

    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        for i in range(0, len(text), 4000):  # pragma: no branch
            parser.feed_data(text[i : i + 4000])


def test_max_header_combined_size(parser: HttpRequestParser) -> None:
    k = b"t" * 4100
    text = b"GET /test HTTP/1.1\r\n" + k + b":" + k + b"\r\n\r\n"

    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        parser.feed_data(text)


@pytest.mark.parametrize("size", [40960, 8191])
async def test_max_trailer_size(parser: HttpRequestParser, size: int) -> None:
    value = b"t" * size
    text = (
        b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked\r\n\r\n"
        + hex(4000)[2:].encode()
        + b"\r\n"
        + b"b" * 4000
        + b"\r\n0\r\ntest: "
        + value
        + b"\r\n\r\n"
    )

    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        payload = None
        for i in range(0, len(text), 3000):  # pragma: no branch
            messages, upgrade, tail = parser.feed_data(text[i : i + 3000])
            if messages:
                payload = messages[0][-1]
        # Trailers are not seen until payload is read.
        assert payload is not None
        await payload.read()


@pytest.mark.parametrize("headers,trailers", ((129, 0), (0, 129), (64, 65)))
async def test_max_headers(
    parser: HttpRequestParser, headers: int, trailers: int
) -> None:
    text = (
        b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked"
        + b"".join(b"\r\nHeader-%d: Value" % i for i in range(headers))
        + b"\r\n\r\n4\r\ntest\r\n0"
        + b"".join(b"\r\nTrailer-%d: Value" % i for i in range(trailers))
        + b"\r\n\r\n"
    )

    match = "Too many (headers|trailers) received"
    with pytest.raises(http_exceptions.BadHttpMessage, match=match):
        messages, upgrade, tail = parser.feed_data(text)
        # Trailers are not seen until payload is read.
        await messages[0][-1].read()


def test_max_header_value_size_under_limit(parser: HttpRequestParser) -> None:
    value = b"A" * 8185
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ndata:" + value + b"\r\n\r\n"

    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.method == "GET"
    assert msg.path == "/test"
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict([("Host", "a"), ("data", value.decode())])
    assert msg.raw_headers == ((b"Host", b"a"), (b"data", value))
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert msg.url == URL("/test")


async def test_chunk_splits_after_pause(parser: HttpRequestParser) -> None:
    num_chunks = 20000  # comfortably above the 16385 pause threshold
    text = (
        b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked\r\n\r\n"
        + b"1\r\nb\r\n" * num_chunks
        + b"0\r\n\r\n"
    )

    messages, upgrade, tail = parser.feed_data(text)
    payload = messages[0][-1]
    # Payload should have paused reading and stopped receiving new chunks after 16k.
    assert payload._http_chunk_splits is not None
    assert len(payload._http_chunk_splits) == 16385
    # We should still get the full result after read(), as it will continue processing.
    result = await payload.read()
    # Compare len first, as it's easier to debug in diff.
    assert len(result) == num_chunks
    assert result == b"b" * num_chunks


async def test_compressed_with_tail(response: HttpResponseParser) -> None:
    """Test compressed content-length body followed by a second response.

    With 2 responses arriving in one call and the first compressed, this should
    trigger decompression pausing with the second response being saved as the tail.
    Verify that the second response is resumed from the tail.
    """
    # Must be large enough to exceed high water mark.
    original = b"x" * 1024 * 1024
    compressed = zlib.compress(original)
    resp1 = (
        b"HTTP/1.1 200 OK\r\n"
        b"Content-Length: " + str(len(compressed)).encode() + b"\r\n"
        b"Content-Encoding: deflate\r\n"
        b"\r\n"
    ) + compressed
    resp2 = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"

    msgs, upgrade, tail = response.feed_data(resp1 + resp2)
    payload = msgs[0][-1]
    result = await payload.read()
    assert len(result) == len(original)
    assert result == original

    payload = response.protocol._buffer[0][-1]
    result = await payload.read()
    assert result == b"ok"


async def test_decompress_error_while_draining_pending_data(
    response: HttpResponseParser,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Decompressor failure on the pending-data drain must not crash.

    Regression test for https://github.com/aio-libs/aiohttp/issues/13203
    where the C parser segfaulted on Python 3.12+ when the decompressor
    raised while draining pending decompressed data, as brotlicffi 1.2
    does on its empty-input drain call.
    """

    class _BrokenDecompressor:
        def __init__(self, *args: object, **kwargs: object) -> None:
            self._calls = 0

        def decompress_sync(self, data: bytes, max_length: int = 0) -> bytes:
            self._calls += 1
            if self._calls > 1:
                raise ValueError("decoder is broken")
            # Must be large enough to exceed the high water mark so the
            # parser pauses with pending data still available.
            return b"x" * (1024 * 1024)

        @property
        def data_available(self) -> bool:
            return True

    monkeypatch.setattr("aiohttp.http_parser.ZLibDecompressor", _BrokenDecompressor)

    headers = (
        b"HTTP/1.1 200 OK\r\n"
        b"Content-Length: 100\r\n"
        b"Content-Encoding: deflate\r\n"
        b"\r\n"
    )
    # First byte 0x78 keeps the zlib header sniff from swapping decompressors.
    msgs, upgrade, tail = response.feed_data(headers + b"\x78\x9c" + b"a" * 8)
    payload = msgs[0][-1]

    # The next feed drains pending data first, which hits the broken decoder.
    # The parser must not raise; the error is delivered via the payload.
    response.feed_data(b"b" * 10)
    assert isinstance(payload.exception(), http_exceptions.ContentEncodingError)


async def test_two_content_length_responses_in_one_call(
    response: HttpResponseParser,
) -> None:
    """Two complete responses in a single feed_data call.

    The first payload completes with tail data for the second, hitting the
    PAYLOAD_COMPLETE branch that resets the parser for the next message.
    """
    resp1 = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello"
    resp2 = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nworld"

    msgs, upgrade, tail = response.feed_data(resp1 + resp2)
    assert len(msgs) == 2
    assert await msgs[0][-1].read() == b"hello"
    assert await msgs[1][-1].read() == b"world"


@pytest.mark.usefixtures("parametrize_zlib_backend")
async def test_compressed_zlib_64kb(response_cls: type[HttpResponseParser]) -> None:
    loop = asyncio.get_running_loop()
    protocol = ResponseHandler(loop)
    response = response_cls(
        protocol,
        loop,
        # 64KiB limit triggered a bug with isal implementation not returning all data.
        2**16,
        max_line_size=8190,
        max_headers=128,
        max_field_size=8190,
    )
    protocol._parser = response

    original = b"".join(
        bytes((*range(0, i), *range(i, 0, -1))) for _ in range(255) for i in range(255)
    )
    compressed = zlib.compress(original)
    headers = (
        b"HTTP/1.1 200 OK\r\n"
        b"Content-Length: " + str(len(compressed)).encode() + b"\r\n"
        b"Content-Encoding: deflate\r\n"
        b"\r\n"
    )

    msgs, upgrade, tail = response.feed_data(headers + compressed)
    payload = msgs[0][-1]
    result = await payload.read()
    assert len(result) == len(original)
    assert result == original


async def test_compressed_chunked_with_pending(response: HttpResponseParser) -> None:
    """Test chunked + compressed where the decompressor needs to resume from pause.

    We need to verify that chunked messages continue parsing correctly after
    a pause and resume in the decompression.
    """
    # Must be large enough to exceed high water mark.
    original = b"A" * 1024 * 1024
    compressed = zlib.compress(original)
    chunk_data = hex(len(compressed))[2:].encode() + b"\r\n" + compressed + b"\r\n"
    headers = (
        b"HTTP/1.1 200 OK\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"Content-Encoding: deflate\r\n"
        b"\r\n"
    )
    data = headers + chunk_data + b"0\r\n\r\n"

    msgs, upgrade, tail = response.feed_data(data)
    payload = msgs[0][-1]
    result = await payload.read()
    assert len(result) == len(original)
    assert result == original


async def test_compressed_chunked_split_chunk_size_line(
    response: HttpResponseParser,
) -> None:
    """First chunk-size line arrives in a feed that carries no body bytes.

    Regression test for an ``IndexError`` in the pure-Python parser: the
    chunked parser fed an empty chunk to ``DeflateBuffer`` before any data
    had been decoded, and the deflate header sniff indexed ``chunk[0]`` on it.
    """
    original = b"Hello, world! " * 4
    compressed = zlib.compress(original)
    size_line = hex(len(compressed))[2:].encode() + b"\r\n"
    headers = (
        b"HTTP/1.1 200 OK\r\n"
        b"Transfer-Encoding: chunked\r\n"
        b"Content-Encoding: deflate\r\n"
        b"\r\n"
    )

    msgs, upgrade, tail = response.feed_data(headers)
    payload = msgs[0][-1]
    # The chunk-size line lands with no body bytes in the same feed, so the
    # parser transitions into the chunk body with an empty buffer.
    response.feed_data(size_line)
    response.feed_data(compressed + b"\r\n0\r\n\r\n")

    result = await payload.read()
    assert result == original
    assert payload.exception() is None


async def test_compressed_until_eof_with_pending(response: HttpResponseParser) -> None:
    """Test read-until-eof + compressed with pause."""
    # Must be large enough to exceed high water mark.
    original = b"B" * 5 * 1024 * 1024
    compressed = zlib.compress(original)
    # No Content-Length or Transfer-Encoding means the parser must parse until EOF.
    headers = b"HTTP/1.1 200 OK\r\nContent-Encoding: deflate\r\n\r\n"

    msgs, upgrade, tail = response.feed_data(headers + compressed)
    response.feed_eof()
    payload = msgs[0][-1]

    # Check that .feed_eof() hasn't decompressed entire payload into memory.
    assert sum(len(b) for b in payload._buffer) <= (1024 * 1024)

    result = await payload.read()
    assert len(result) == len(original)
    assert result == original


async def test_content_length_eof_while_paused(response: HttpResponseParser) -> None:
    """EOF right after a fully received content-length body must complete it.

    Regression test for #13348:
    feeding the final body bytes pauses the parser for flow control before
    the message can complete; a server closing the connection in that state
    raised ContentLengthError despite received == expected.
    """
    # Must be large enough to exceed the high water mark so the parser
    # pauses with the message not yet complete.
    body = b"x" * (1024 * 1024)
    headers = b"HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n" % len(body)

    msgs, upgrade, tail = response.feed_data(headers + body)
    payload = msgs[0][-1]
    # The server has sent everything and closed the connection.
    response.feed_eof()

    result = await payload.read()
    assert result == body
    assert payload.is_eof()
    assert payload.exception() is None


async def test_compressed_content_length_eof_while_paused(
    response: HttpResponseParser,
) -> None:
    """EOF with pending decompressed data on a complete content-length body.

    Like test_content_length_eof_while_paused, but the decompressor still
    holds pending data at EOF, so completion is deferred until the reader
    drains it.
    """
    # Must be large enough to exceed high water mark.
    original = b"B" * 5 * 1024 * 1024
    compressed = zlib.compress(original)
    headers = (
        b"HTTP/1.1 200 OK\r\n"
        b"Content-Length: " + str(len(compressed)).encode() + b"\r\n"
        b"Content-Encoding: deflate\r\n"
        b"\r\n"
    )

    msgs, upgrade, tail = response.feed_data(headers + compressed)
    payload = msgs[0][-1]
    response.feed_eof()

    # Check that .feed_eof() hasn't decompressed entire payload into memory.
    assert sum(len(b) for b in payload._buffer) <= (2 * 1024 * 1024)

    result = await payload.read()
    assert len(result) == len(original)
    assert result == original
    assert payload.is_eof()
    assert payload.exception() is None


async def test_content_length_eof_while_paused_incomplete(
    response: HttpResponseParser,
) -> None:
    """EOF on a paused parser with a genuinely incomplete body still raises."""
    body = b"x" * (1024 * 1024)
    headers = b"HTTP/1.1 200 OK\r\nContent-Length: %d\r\n\r\n" % (len(body) + 1)

    response.feed_data(headers + body)

    with pytest.raises(
        http_exceptions.ContentLengthError,
        match=r"received 1048576 of 1048577 bytes",
    ):
        response.feed_eof()


async def test_compressed_until_eof_high_water(
    response_cls: type[HttpResponseParser],
) -> None:
    """Test read-until-eof + compressed with higher limit."""
    loop = asyncio.get_running_loop()
    protocol = ResponseHandler(loop)
    response = response_cls(
        protocol,
        loop,
        2**19,  # 512 KiB limit
        max_line_size=8190,
        max_headers=128,
        max_field_size=8190,
        read_until_eof=True,
    )
    protocol._parser = response

    # Must be large enough to exceed high water mark.
    original = b"B" * 5 * 1024 * 1024
    compressed = zlib.compress(original)
    # No Content-Length or Transfer-Encoding means the parser must parse until EOF.
    headers = b"HTTP/1.1 200 OK\r\nContent-Encoding: deflate\r\n\r\n"

    msgs, upgrade, tail = response.feed_data(headers + compressed)
    response.feed_eof()
    payload = msgs[0][-1]

    # Check that .feed_eof() hasn't decompressed entire payload into memory.
    assert sum(len(b) for b in payload._buffer) <= (2 * 1024 * 1024)
    # Individual chunks should have been decompressed at limit amount.
    assert all(len(b) == 512 * 1024 for b in payload._buffer)

    result = await payload.read()
    assert len(result) == len(original)
    assert result == original


async def test_compressed_256kb(response: HttpResponseParser) -> None:
    original = b"x" * 256 * 1024
    compressed = zlib.compress(original)
    headers = (
        b"HTTP/1.1 200 OK\r\n"
        b"Content-Length: " + str(len(compressed)).encode() + b"\r\n"
        b"Content-Encoding: deflate\r\n"
        b"\r\n"
    )

    messages, upgrade, tail = response.feed_data(headers + compressed)
    assert len(messages) == 1
    payload = messages[0][-1]
    result = await payload.read()
    assert len(result) == len(original)
    assert result == original


@pytest.mark.parametrize("size", [40965, 8191])
def test_max_header_value_size_continuation(
    response: HttpResponseParser, size: int
) -> None:
    name = b"T" * (size - 5)
    text = b"HTTP/1.1 200 Ok\r\ndata: test\r\n " + name + b"\r\n\r\n"

    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        for i in range(0, len(text), 9000):  # pragma: no branch
            response.feed_data(text[i : i + 9000])


def test_max_header_value_size_continuation_under_limit(
    response: HttpResponseParser,
) -> None:
    value = b"A" * 8179
    text = b"HTTP/1.1 200 Ok\r\ndata: test\r\n " + value + b"\r\n\r\n"

    messages, upgrade, tail = response.feed_data(text)
    msg = messages[0][0]
    assert msg.code == 200
    assert msg.reason == "Ok"
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict({"data": "test " + value.decode()})
    assert msg.raw_headers == ((b"data", b"test " + value),)
    assert msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked


def test_http_request_parser(parser: HttpRequestParser) -> None:
    text = b"GET /path HTTP/1.1\r\nHost: a\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.method == "GET"
    assert msg.path == "/path"
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict({"Host": "a"})
    assert msg.raw_headers == ((b"Host", b"a"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert msg.url == URL("/path")


def test_http_request_bad_status_line(parser: HttpRequestParser) -> None:
    text = b"getpath \r\n\r\n"
    with pytest.raises(http_exceptions.BadStatusLine) as exc_info:
        parser.feed_data(text)
    # Check for accidentally escaped message.
    assert r"\n" not in exc_info.value.message


_num: dict[bytes, str] = {
    # dangerous: accepted by Python int()
    # unicodedata.category("\U0001D7D9") == 'Nd'
    "\N{MATHEMATICAL DOUBLE-STRUCK DIGIT ONE}".encode(): "utf8digit",
    # only added for interop tests, refused by Python int()
    # unicodedata.category("\U000000B9") == 'No'
    "\N{SUPERSCRIPT ONE}".encode(): "utf8number",
    "\N{SUPERSCRIPT ONE}".encode("latin-1"): "latin1number",
}


@pytest.mark.parametrize("nonascii_digit", _num.keys(), ids=_num.values())
def test_http_request_bad_status_line_number(
    parser: HttpRequestParser, nonascii_digit: bytes
) -> None:
    text = b"GET /digit HTTP/1." + nonascii_digit + b"\r\nHost: a\r\n\r\n"
    with pytest.raises(http_exceptions.BadStatusLine):
        parser.feed_data(text)


def test_http_request_bad_status_line_separator(parser: HttpRequestParser) -> None:
    # single code point, old, multibyte NFKC, multibyte NFKD
    utf8sep = "\N{ARABIC LIGATURE SALLALLAHOU ALAYHE WASALLAM}".encode()
    text = b"GET /ligature HTTP/1" + utf8sep + b"1\r\nHost: a\r\n\r\n"
    with pytest.raises(http_exceptions.BadStatusLine):
        parser.feed_data(text)


def test_http_request_bad_status_line_whitespace(parser: HttpRequestParser) -> None:
    text = b"GET\n/path\fHTTP/1.1\r\nHost: a\r\n\r\n"
    with pytest.raises(http_exceptions.BadStatusLine):
        parser.feed_data(text)


def test_http_request_message_after_close(parser: HttpRequestParser) -> None:
    text = b"GET / HTTP/1.1\r\nHost: a\r\nConnection: close\r\n\r\nInvalid\r\n\r\n"
    with pytest.raises(
        http_exceptions.BadHttpMessage, match="Data after `Connection: close`"
    ):
        parser.feed_data(text)


def test_http_request_message_after_close_comma_list(parser: HttpRequestParser) -> None:
    text = b"GET / HTTP/1.1\r\nHost: a\r\nConnection: close, keep-alive\r\n\r\nInvalid\r\n\r\n"
    with pytest.raises(
        http_exceptions.BadHttpMessage, match="Data after `Connection: close`"
    ):
        parser.feed_data(text)


def test_http_request_upgrade(parser: HttpRequestParser) -> None:
    text = (
        b"GET /test HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"connection: upgrade\r\n"
        b"upgrade: websocket\r\n\r\n"
        b"some raw data"
    )
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert not msg.should_close
    assert msg.upgrade
    assert upgrade
    assert tail == b"some raw data"


async def test_http_request_upgrade_unknown(parser: HttpRequestParser) -> None:
    text = (
        b"POST / HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"Connection: Upgrade\r\n"
        b"Content-Length: 2\r\n"
        b"Upgrade: unknown\r\n"
        b"Content-Type: application/json\r\n\r\n"
        b"{}"
    )
    messages, upgrade, tail = parser.feed_data(text)

    msg = messages[0][0]
    assert not msg.should_close
    assert msg.upgrade
    assert not upgrade
    assert not msg.chunked
    assert tail == b""
    assert await messages[0][-1].read() == b"{}"


@pytest.mark.parametrize("chunked", (False, True), ids=("content-length", "chunked"))
async def test_http_request_upgrade_with_body_read(
    parser: HttpRequestParser, chunked: bool
) -> None:
    body_request = b"foobarbaz\r\n\r\n"
    if chunked:
        framing = b"Transfer-Encoding: chunked\r\n"
        body = b"%x\r\n%s\r\n0\r\n\r\n" % (len(body_request), body_request)
    else:
        framing = b"Content-Length: %d\r\n" % len(body_request)
        body = body_request
    after = b"GET /after HTTP/1.1\r\nHost: a\r\n\r\n"
    text = (
        b"GET /ws HTTP/1.1\r\nHost: a\r\n"
        b"Connection: Upgrade\r\nUpgrade: websocket\r\n"
        + framing
        + b"\r\n"
        + body
        + after
    )
    messages, upgrade, tail = parser.feed_data(text)
    assert len(messages) == 1
    msg, payload = messages[0]
    assert msg.method == "GET"
    assert msg.path == "/ws"
    assert msg.upgrade
    assert await payload.read() == body_request
    # The connection switches protocols only after the body is fully read.
    assert upgrade
    assert tail == after


def test_http_request_upgrade_empty_body_allowed(parser: HttpRequestParser) -> None:
    text = (
        b"GET /ws HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"Connection: Upgrade\r\n"
        b"Upgrade: websocket\r\n"
        b"Content-Length: 0\r\n\r\n"
        b"some raw data"
    )
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]
    assert msg.upgrade
    assert upgrade
    assert tail == b"some raw data"


@pytest.fixture
def xfail_c_parser_url(request: pytest.FixtureRequest) -> None:
    if isinstance(request.getfixturevalue("parser"), HttpRequestParserPy):
        return
    request.node.add_marker(
        pytest.mark.xfail(
            reason="Regression test for Py parser. May match C behaviour later.",
            raises=http_exceptions.InvalidURLError,
        )
    )


@pytest.mark.usefixtures("xfail_c_parser_url")
def test_http_request_parser_utf8_request_line(parser: HttpRequestParser) -> None:
    messages, upgrade, tail = parser.feed_data(
        # note the truncated unicode sequence
        b"GET /P\xc3\xbcnktchen\xa0\xef\xb7 HTTP/1.1\r\nHost: a\r\n" +
        # for easier grep: ASCII 0xA0 more commonly known as non-breaking space
        # note the leading and trailing spaces
        "sTeP:  \N{LATIN SMALL LETTER SHARP S}nek\t\N{NO-BREAK SPACE}  "
        "\r\n\r\n".encode()
    )
    msg = messages[0][0]

    assert msg.method == "GET"
    assert msg.path == "/Pünktchen\udca0\udcef\udcb7"
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict([("Host", "a"), ("STEP", "ßnek\t\xa0")])
    assert msg.raw_headers == ((b"Host", b"a"), (b"sTeP", "ßnek\t\xa0".encode()))
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    # python HTTP parser depends on Cython and CPython URL to match
    # .. but yarl.URL("/abs") is not equal to URL.build(path="/abs"), see #6409
    assert msg.url == URL.build(path="/Pünktchen\udca0\udcef\udcb7", encoded=True)


def test_http_request_parser_utf8(parser: HttpRequestParser) -> None:
    text = "GET /path HTTP/1.1\r\nHost: a\r\nx-test:тест\r\n\r\n".encode()
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.method == "GET"
    assert msg.path == "/path"
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict([("Host", "a"), ("X-TEST", "тест")])
    assert msg.raw_headers == ((b"Host", b"a"), (b"x-test", "тест".encode()))
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert msg.url == URL("/path")


def test_http_request_parser_non_utf8(parser: HttpRequestParser) -> None:
    text = "GET /path HTTP/1.1\r\nHost: a\r\nx-test:тест\r\n\r\n".encode("cp1251")
    msg = parser.feed_data(text)[0][0][0]

    assert msg.method == "GET"
    assert msg.path == "/path"
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict(
        [
            ("Host", "a"),
            ("X-TEST", "тест".encode("cp1251").decode("utf8", "surrogateescape")),
        ]
    )
    assert msg.raw_headers == ((b"Host", b"a"), (b"x-test", "тест".encode("cp1251")))
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert msg.url == URL("/path")


def test_http_request_parser_two_slashes(parser: HttpRequestParser) -> None:
    text = b"GET //path HTTP/1.1\r\nHost: a\r\n\r\n"
    msg = parser.feed_data(text)[0][0][0]

    assert msg.method == "GET"
    assert msg.path == "//path"
    assert msg.url.path == "//path"
    assert msg.version == (1, 1)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked


@pytest.mark.parametrize(
    "rfc9110_5_6_2_token_delim",
    [bytes([i]) for i in rb'"(),/:;<=>?@[\]{}'],
)
def test_http_request_parser_bad_method(
    parser: HttpRequestParser, rfc9110_5_6_2_token_delim: bytes
) -> None:
    with pytest.raises(http_exceptions.BadHttpMethod):
        parser.feed_data(
            rfc9110_5_6_2_token_delim + b'ET" /get HTTP/1.1\r\nHost: a\r\n\r\n'
        )


def test_http_request_parser_tls_handshake_on_http_port(
    parser: HttpRequestParser,
) -> None:
    with pytest.raises(http_exceptions.BadHttpMethod) as ctx:
        parser.feed_data(b"\x16\x03\x03\x01F\x01\r\n\r\n")

    assert "Received HTTPS traffic on an HTTP port" in str(ctx.value)


def test_http_request_parser_bad_version(parser: HttpRequestParser) -> None:
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(b"GET //get HT/11\r\nHost: a\r\n\r\n")


def test_http_request_parser_bad_version_number(parser: HttpRequestParser) -> None:
    with pytest.raises(http_exceptions.BadHttpMessage):
        parser.feed_data(b"GET /test HTTP/1.32\r\nHost: a\r\n\r\n")


def test_http_request_parser_bad_ascii_uri(parser: HttpRequestParser) -> None:
    with pytest.raises(http_exceptions.InvalidURLError):
        parser.feed_data(b"GET ! HTTP/1.1\r\n\r\n")


def test_http_request_parser_bad_nonascii_uri(parser: HttpRequestParser) -> None:
    with pytest.raises(http_exceptions.InvalidURLError):
        parser.feed_data(b"GET \xff HTTP/1.1\r\n\r\n")


@pytest.mark.parametrize("size", [40965, 8191])
def test_http_request_max_status_line(parser: HttpRequestParser, size: int) -> None:
    path = b"t" * (size - 5)
    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        parser.feed_data(b"GET /path" + path + b" HTTP/1.1\r\n\r\n")


def test_http_request_max_status_line_under_limit(parser: HttpRequestParser) -> None:
    path = b"t" * 8172
    messages, upgraded, tail = parser.feed_data(
        b"GET /path" + path + b" HTTP/1.1\r\nHost: a\r\n\r\n"
    )
    msg = messages[0][0]

    assert msg.method == "GET"
    assert msg.path == "/path" + path.decode()
    assert msg.version == (1, 1)
    assert msg.headers == CIMultiDict({"Host": "a"})
    assert msg.raw_headers == ((b"Host", b"a"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert msg.url == URL("/path" + path.decode())


def test_http_request_max_status_line_fragmented(
    parser: HttpRequestParser,
) -> None:
    # Split an overlong request target across reads so that each callback
    # fragment is under the limit but the accumulated target is not.
    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        parser.feed_data(b"GET /" + b"a" * 8000)
        parser.feed_data(b"a" * 8000 + b" HTTP/1.1\r\nHost: a\r\n\r\n")


def test_http_response_parser_utf8(response: HttpResponseParser) -> None:
    text = "HTTP/1.1 200 Ok\r\nx-test:тест\r\n\r\n".encode()

    messages, upgraded, tail = response.feed_data(text)
    assert len(messages) == 1
    msg = messages[0][0]

    assert msg.version == (1, 1)
    assert msg.code == 200
    assert msg.reason == "Ok"
    assert msg.headers == CIMultiDict([("X-TEST", "тест")])
    assert msg.raw_headers == ((b"x-test", "тест".encode()),)
    assert not upgraded
    assert not tail


def test_http_response_parser_utf8_without_reason(response: HttpResponseParser) -> None:
    text = "HTTP/1.1 200 \r\nx-test:тест\r\n\r\n".encode()

    messages, upgraded, tail = response.feed_data(text)
    assert len(messages) == 1
    msg = messages[0][0]

    assert msg.version == (1, 1)
    assert msg.code == 200
    assert msg.reason == ""
    assert msg.headers == CIMultiDict([("X-TEST", "тест")])
    assert msg.raw_headers == ((b"x-test", "тест".encode()),)
    assert not upgraded
    assert not tail


def test_http_response_parser_obs_line_folding(response: HttpResponseParser) -> None:
    text = b"HTTP/1.1 200 Ok\r\ntest: line\r\n continue\r\n\r\n"

    messages, upgraded, tail = response.feed_data(text)
    assert len(messages) == 1
    msg = messages[0][0]

    assert msg.version == (1, 1)
    assert msg.code == 200
    assert msg.reason == "Ok"
    assert msg.headers == CIMultiDict([("TEST", "line continue")])
    assert msg.raw_headers == ((b"test", b"line continue"),)
    assert not upgraded
    assert not tail


@pytest.mark.dev_mode
def test_http_response_parser_strict_obs_line_folding(
    response: HttpResponseParser,
) -> None:
    text = b"HTTP/1.1 200 Ok\r\ntest: line\r\n continue\r\n\r\n"

    with pytest.raises(http_exceptions.BadHttpMessage):
        response.feed_data(text)


@pytest.mark.parametrize("size", [40962, 8191])
def test_http_response_parser_bad_status_line_too_long(
    response: HttpResponseParser, size: int
) -> None:
    reason = b"t" * (size - 2)
    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        response.feed_data(b"HTTP/1.1 200 Ok" + reason + b"\r\n\r\n")


def test_http_response_parser_status_line_under_limit(
    response: HttpResponseParser,
) -> None:
    reason = b"O" * 8177
    messages, upgraded, tail = response.feed_data(
        b"HTTP/1.1 200 " + reason + b"\r\n\r\n"
    )
    msg = messages[0][0]
    assert msg.version == (1, 1)
    assert msg.code == 200
    assert msg.reason == reason.decode()


def test_http_response_parser_status_line_too_long_fragmented(
    response: HttpResponseParser,
) -> None:
    # Split an overlong reason phrase across reads so that each callback
    # fragment is under the limit but the accumulated reason is not.
    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(http_exceptions.LineTooLong, match=match):
        response.feed_data(b"HTTP/1.1 200 " + b"a" * 8000)
        response.feed_data(b"a" * 8000 + b"\r\n\r\n")


def test_http_response_parser_bad_version(response: HttpResponseParser) -> None:
    with pytest.raises(http_exceptions.BadHttpMessage):
        response.feed_data(b"HT/11 200 Ok\r\n\r\n")


def test_http_response_parser_bad_version_number(response: HttpResponseParser) -> None:
    with pytest.raises(http_exceptions.BadHttpMessage):
        response.feed_data(b"HTTP/12.3 200 Ok\r\n\r\n")


def test_http_response_parser_no_reason(response: HttpResponseParser) -> None:
    msg = response.feed_data(b"HTTP/1.1 200\r\n\r\n")[0][0][0]

    assert msg.version == (1, 1)
    assert msg.code == 200
    assert msg.reason == ""


def test_http_response_parser_lenient_headers(response: HttpResponseParser) -> None:
    messages, upgrade, tail = response.feed_data(
        b"HTTP/1.1 200 test\r\nFoo: abc\x01def\r\n\r\n"
    )
    msg = messages[0][0]

    assert msg.headers["Foo"] == "abc\x01def"


@pytest.mark.dev_mode
def test_http_response_parser_strict_headers(response: HttpResponseParser) -> None:
    if isinstance(response, HttpResponseParserPy):
        pytest.xfail("Py parser is lenient. May update py-parser later.")
    with pytest.raises(http_exceptions.BadHttpMessage):  # type: ignore[unreachable]
        response.feed_data(b"HTTP/1.1 200 test\r\nFoo: abc\x01def\r\n\r\n")


def test_http_response_parser_null_byte_in_header_value(
    response: HttpResponseParser,
) -> None:
    with pytest.raises(http_exceptions.InvalidHeader):
        response.feed_data(b"HTTP/1.1 200 OK\r\nFoo: abc\x00def\r\n\r\n")


def test_http_response_parser_bad_crlf(response: HttpResponseParser) -> None:
    """Still a lot of dodgy servers sending bad requests like this."""
    messages, upgrade, tail = response.feed_data(
        b"HTTP/1.0 200 OK\nFoo: abc\nBar: def\n\nBODY\n"
    )
    msg = messages[0][0]

    assert msg.headers["Foo"] == "abc"
    assert msg.headers["Bar"] == "def"


async def test_http_response_parser_bad_chunked_lax(
    response: HttpResponseParser,
) -> None:
    text = (
        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5 \r\nabcde\r\n0\r\n\r\n"
    )
    messages, upgrade, tail = response.feed_data(text)

    assert await messages[0][1].read(5) == b"abcde"


@pytest.mark.dev_mode
async def test_http_response_parser_bad_chunked_strict_py() -> None:
    protocol = ResponseHandler(asyncio.get_running_loop())

    response = HttpResponseParserPy(
        protocol,
        asyncio.get_running_loop(),
        DEFAULT_CHUNK_SIZE,
        max_line_size=8190,
        max_field_size=8190,
    )
    protocol._parser = response
    text = (
        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5 \r\nabcde\r\n0\r\n\r\n"
    )
    with pytest.raises(http_exceptions.TransferEncodingError, match="5"):
        response.feed_data(text)


@pytest.mark.dev_mode
@pytest.mark.skipif(
    "HttpRequestParserC" not in dir(aiohttp.http_parser),
    reason="C based HTTP parser not available",
)
async def test_http_response_parser_bad_chunked_strict_c() -> None:
    protocol = ResponseHandler(asyncio.get_running_loop())

    response = HttpResponseParserC(
        protocol,
        asyncio.get_running_loop(),
        2**16,
        max_line_size=8190,
        max_field_size=8190,
    )
    protocol._parser = response
    text = (
        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5 \r\nabcde\r\n0\r\n\r\n"
    )
    with pytest.raises(http_exceptions.BadHttpMessage):
        response.feed_data(text)


# The C parser builds its error message from a snippet of the fed buffer. That
# snippet must be sliced out of the bounded ``data`` bytes object; materializing the
# raw llhttp error-position pointer instead runs strlen past the end of the buffer.
# These tests pin the snippet to bytes that lie inside the fed input, and cover error
# positions right at the buffer end where the over-read used to happen.

_CHUNKED_RESP_HDR = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n"

# (malformed body appended to a chunked response header, expected error snippet,
# llhttp error reason)
_BAD_CHUNKED_RESPONSES = [
    (b"0\rX\r\n\r\n", b"0\rX", "Expected LF after chunk size"),
    (b"5\r\nhello\rX\r\n0\r\n\r\n", b"hello\rX", "Expected LF after chunk data"),
    (b"5\r\nhelloXY\r\n0\r\n\r\n", b"helloXY", "Expected LF after chunk data"),
    (b"1\r\nA\rB\r\n0\r\n\r\n", b"A\rB", "Expected LF after chunk data"),
    (b"0_2e\r\n\r\n", b"0_2e", "Invalid character in chunk size"),
]

# Same, but the buffer ends exactly at the offending byte so the error position sits
# at the end of the input with no trailing CRLF to bound the old strlen-based slice.
_BAD_CHUNKED_RESPONSES_AT_END = [
    (b"0\rX", b"0\rX", "Expected LF after chunk size"),
    (b"5\r\nhello\rX", b"hello\rX", "Expected LF after chunk data"),
    (b"0_", b"0_", "Invalid character in chunk size"),
]

_BAD_REQUESTS_AT_END = [
    (b"GET / HTTP/1.1\r\nHost: x\n", b"Host: x\n"),
    (b"POST / HTTP/1.1\r\nHost: x\r\nA: b\x01", b"A: b\x01"),
]


@pytest.mark.skipif(NO_EXTENSIONS, reason="Only tests C parser.")
@pytest.mark.parametrize(
    ("body", "snippet", "reason"),
    _BAD_CHUNKED_RESPONSES + _BAD_CHUNKED_RESPONSES_AT_END,
)
async def test_c_parser_error_snippet_bounded_response(
    body: bytes, snippet: bytes, reason: str
) -> None:
    loop = asyncio.get_running_loop()
    protocol = ResponseHandler(loop)
    response = HttpResponseParserC(
        protocol, loop, 2**16, max_line_size=8190, max_field_size=8190
    )
    protocol._parser = response
    text = _CHUNKED_RESP_HDR + body
    assert snippet in text
    match = f"{re.escape(reason)}[\\s\\S]*{re.escape(repr(snippet))}"
    with pytest.raises(http_exceptions.BadHttpMessage, match=match):
        response.feed_data(text)


@pytest.mark.skipif(NO_EXTENSIONS, reason="Only tests C parser.")
@pytest.mark.parametrize(("text", "snippet"), _BAD_REQUESTS_AT_END)
def test_c_parser_error_snippet_at_buffer_end_request(
    event_loop: asyncio.AbstractEventLoop,
    server: Server[Request],
    text: bytes,
    snippet: bytes,
) -> None:
    protocol = RequestHandler(server, loop=event_loop)
    parser = HttpRequestParserC(
        protocol, event_loop, 2**16, max_line_size=8190, max_field_size=8190
    )
    protocol._parser = parser
    assert snippet in text
    with pytest.raises(http_exceptions.BadHttpMessage, match=re.escape(repr(snippet))):
        parser.feed_data(text)


@pytest.mark.skipif(NO_EXTENSIONS, reason="Only tests C parser.")
@pytest.mark.parametrize("split", [1, 2, 3, 5])
@pytest.mark.parametrize(("body", "snippet", "reason"), _BAD_CHUNKED_RESPONSES)
async def test_c_parser_error_snippet_fragmented_response(
    body: bytes, snippet: bytes, reason: str, split: int
) -> None:
    # Feed the malformed body in two pieces so the second feed runs through the
    # tail-prepend path before the error is reported; the parser must report a clean
    # error rather than reading past the buffer. The snippet offset shifts with the
    # split, so pin the stable llhttp reason instead.
    loop = asyncio.get_running_loop()
    protocol = ResponseHandler(loop)
    response = HttpResponseParserC(
        protocol, loop, 2**16, max_line_size=8190, max_field_size=8190
    )
    protocol._parser = response
    response.feed_data(_CHUNKED_RESP_HDR)
    with pytest.raises(http_exceptions.BadHttpMessage, match=re.escape(reason)):
        response.feed_data(body[:split])
        response.feed_data(body[split:])


async def test_http_response_parser_notchunked(
    response: HttpResponseParser,
) -> None:
    text = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: notchunked\r\n\r\n1\r\nT\r\n3\r\nest\r\n0\r\n\r\n"
    messages, upgrade, tail = response.feed_data(text)
    response.feed_eof()

    # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.2
    assert await messages[0][1].read() == b"1\r\nT\r\n3\r\nest\r\n0\r\n\r\n"


async def test_http_response_parser_empty_list_te_not_chunked(
    response: HttpResponseParser,
) -> None:
    text = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: ,  \t ,\r\n\r\nbody"
    messages, upgrade, tail = response.feed_data(text)
    response.feed_eof()

    assert await messages[0][1].read() == b"body"


async def test_http_response_parser_last_chunked(
    response: HttpResponseParser,
) -> None:
    text = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: not, chunked\r\n\r\n1\r\nT\r\n3\r\nest\r\n0\r\n\r\n"
    messages, upgrade, tail = response.feed_data(text)

    # https://www.rfc-editor.org/rfc/rfc9112#section-6.3-2.4.2
    assert await messages[0][1].read() == b"Test"


def test_http_response_parser_bad(response: HttpResponseParser) -> None:
    with pytest.raises(http_exceptions.BadHttpMessage):
        response.feed_data(b"HTT/1\r\n\r\n")


def test_http_response_parser_code_under_100(response: HttpResponseParser) -> None:
    with pytest.raises(http_exceptions.BadStatusLine):
        response.feed_data(b"HTTP/1.1 99 test\r\n\r\n")


def test_http_response_parser_code_above_999(response: HttpResponseParser) -> None:
    with pytest.raises(http_exceptions.BadStatusLine):
        response.feed_data(b"HTTP/1.1 9999 test\r\n\r\n")


def test_http_response_parser_code_not_int(response: HttpResponseParser) -> None:
    with pytest.raises(http_exceptions.BadStatusLine):
        response.feed_data(b"HTTP/1.1 ttt test\r\n\r\n")


@pytest.mark.parametrize("nonascii_digit", _num.keys(), ids=_num.values())
def test_http_response_parser_code_not_ascii(
    response: HttpResponseParser, nonascii_digit: bytes
) -> None:
    with pytest.raises(http_exceptions.BadStatusLine):
        response.feed_data(b"HTTP/1.1 20" + nonascii_digit + b" test\r\n\r\n")


def test_http_request_chunked_payload(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntransfer-encoding: chunked\r\n\r\n"
    msg, payload = parser.feed_data(text)[0][0]

    assert msg.chunked
    assert not payload.is_eof()
    assert isinstance(payload, streams.StreamReader)

    parser.feed_data(b"4\r\ndata\r\n4\r\nline\r\n0\r\n\r\n")

    assert b"dataline" == b"".join(d for d in payload._buffer)
    assert payload._http_chunk_splits is not None
    assert [4, 8] == list(payload._http_chunk_splits)
    assert payload.is_eof()


def test_http_request_chunked_payload_and_next_message(
    parser: HttpRequestParser,
) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntransfer-encoding: chunked\r\n\r\n"
    msg, payload = parser.feed_data(text)[0][0]

    messages, upgraded, tail = parser.feed_data(
        b"4\r\ndata\r\n4\r\nline\r\n0\r\n\r\n"
        b"POST /test2 HTTP/1.1\r\n"
        b"Host: a\r\n"
        b"transfer-encoding: chunked\r\n\r\n"
    )

    assert b"dataline" == b"".join(d for d in payload._buffer)
    assert payload._http_chunk_splits is not None
    assert [4, 8] == list(payload._http_chunk_splits)
    assert payload.is_eof()

    assert len(messages) == 1
    msg2, payload2 = messages[0]

    assert msg2.method == "POST"
    assert msg2.chunked
    assert not payload2.is_eof()


def test_http_request_chunked_payload_chunks(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntransfer-encoding: chunked\r\n\r\n"
    msg, payload = parser.feed_data(text)[0][0]

    parser.feed_data(b"4\r\ndata\r")
    parser.feed_data(b"\n4")
    parser.feed_data(b"\r")
    parser.feed_data(b"\n")
    parser.feed_data(b"li")
    parser.feed_data(b"ne\r\n0\r\n")
    parser.feed_data(b"test: test\r\n")

    assert b"dataline" == b"".join(d for d in payload._buffer)
    assert payload._http_chunk_splits is not None
    assert [4, 8] == list(payload._http_chunk_splits)
    assert not payload.is_eof()

    parser.feed_data(b"\r\n")
    assert b"dataline" == b"".join(d for d in payload._buffer)
    assert [4, 8] == list(payload._http_chunk_splits)
    assert payload.is_eof()


def test_parse_chunked_payload_chunk_extension(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\ntransfer-encoding: chunked\r\n\r\n"
    msg, payload = parser.feed_data(text)[0][0]

    parser.feed_data(b"4;test\r\ndata\r\n4\r\nline\r\n0\r\ntest: test\r\n\r\n")

    assert b"dataline" == b"".join(d for d in payload._buffer)
    assert payload._http_chunk_splits is not None
    assert [4, 8] == list(payload._http_chunk_splits)
    assert payload.is_eof()


async def test_request_chunked_with_trailer(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked\r\n\r\n4\r\ntest\r\n0\r\ntest: trailer\r\nsecond: test trailer\r\n\r\n"
    messages, upgraded, tail = parser.feed_data(text)
    assert not tail
    msg, payload = messages[0]
    assert await payload.read() == b"test"

    # TODO: Add assertion of trailers when API added.


async def test_request_chunked_reject_bad_trailer(parser: HttpRequestParser) -> None:
    text = b"GET /test HTTP/1.1\r\nHost: a\r\nTransfer-Encoding: chunked\r\n\r\n0\r\nbad\ntrailer\r\n\r\n"
    with pytest.raises(http_exceptions.BadHttpMessage, match=r"b'bad\\ntrailer'"):
        parser.feed_data(text)


def test_parse_no_length_or_te_on_post(
    event_loop: asyncio.AbstractEventLoop,
    server: Server[Request],
    request_cls: type[HttpRequestParser],
) -> None:
    protocol = RequestHandler(server, loop=event_loop)
    parser = request_cls(protocol, event_loop, limit=DEFAULT_CHUNK_SIZE)
    protocol._parser = parser
    text = b"POST /test HTTP/1.1\r\nHost: a\r\n\r\n"
    msg, payload = parser.feed_data(text)[0][0]

    assert payload.is_eof()


def test_parse_payload_response_without_body(
    event_loop: asyncio.AbstractEventLoop,
    response_cls: type[HttpResponseParser],
) -> None:
    protocol = ResponseHandler(event_loop)
    parser = response_cls(protocol, event_loop, 2**16, response_with_body=False)
    protocol._parser = parser
    text = b"HTTP/1.1 200 Ok\r\ncontent-length: 10\r\n\r\n"
    msg, payload = parser.feed_data(text)[0][0]

    assert payload.is_eof()


async def test_parse_payload_response_with_invalid_body(
    protocol: BaseProtocol,
    response_cls: type[HttpResponseParser],
) -> None:
    loop = asyncio.get_running_loop()
    parser = response_cls(protocol, loop, 2**16, response_with_body=False)
    text = (
        b"HTTP/1.1 200 Ok\r\nTransfer-Encoding: chunked\r\n\r\n"
        b"7\r\nchunked\r\n0\r\n\r\n"
    )
    with pytest.raises(http_exceptions.BadHttpMessage, match="status line"):
        parser.feed_data(text)[0][0]


def test_parse_length_payload(response: HttpResponseParser) -> None:
    text = b"HTTP/1.1 200 Ok\r\ncontent-length: 4\r\n\r\n"
    msg, payload = response.feed_data(text)[0][0]
    assert not payload.is_eof()

    response.feed_data(b"da")
    response.feed_data(b"t")
    response.feed_data(b"aHT")

    assert payload.is_eof()
    assert b"data" == b"".join(d for d in payload._buffer)


def test_parse_no_length_payload(parser: HttpRequestParser) -> None:
    text = b"PUT / HTTP/1.1\r\nHost: a\r\n\r\n"
    msg, payload = parser.feed_data(text)[0][0]
    assert payload.is_eof()


def test_parse_content_length_payload_multiple(response: HttpResponseParser) -> None:
    text = b"HTTP/1.1 200 OK\r\ncontent-length: 5\r\n\r\nfirst"
    msg, payload = response.feed_data(text)[0][0]
    assert msg.version == HttpVersion(major=1, minor=1)
    assert msg.code == 200
    assert msg.reason == "OK"
    assert msg.headers == HeadersDictProxy(CIMultiDict([("Content-Length", "5")]))
    assert msg.raw_headers == ((b"content-length", b"5"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert payload.is_eof()
    assert b"first" == b"".join(d for d in payload._buffer)

    text = b"HTTP/1.1 200 OK\r\ncontent-length: 6\r\n\r\nsecond"
    msg, payload = response.feed_data(text)[0][0]
    assert msg.version == HttpVersion(major=1, minor=1)
    assert msg.code == 200
    assert msg.reason == "OK"
    assert msg.headers == CIMultiDict(
        [
            ("Content-Length", "6"),
        ]
    )
    assert msg.raw_headers == ((b"content-length", b"6"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert payload.is_eof()
    assert b"second" == b"".join(d for d in payload._buffer)


def test_parse_content_length_than_chunked_payload(
    response: HttpResponseParser,
) -> None:
    text = b"HTTP/1.1 200 OK\r\ncontent-length: 5\r\n\r\nfirst"
    msg, payload = response.feed_data(text)[0][0]
    assert msg.version == HttpVersion(major=1, minor=1)
    assert msg.code == 200
    assert msg.reason == "OK"
    assert msg.headers == HeadersDictProxy(CIMultiDict([("Content-Length", "5")]))
    assert msg.raw_headers == ((b"content-length", b"5"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert not msg.chunked
    assert payload.is_eof()
    assert b"first" == b"".join(d for d in payload._buffer)

    text = (
        b"HTTP/1.1 200 OK\r\n"
        b"transfer-encoding: chunked\r\n\r\n"
        b"6\r\nsecond\r\n0\r\n\r\n"
    )
    msg, payload = response.feed_data(text)[0][0]
    assert msg.version == HttpVersion(major=1, minor=1)
    assert msg.code == 200
    assert msg.reason == "OK"
    assert msg.headers == CIMultiDict(
        [
            ("Transfer-Encoding", "chunked"),
        ]
    )
    assert msg.raw_headers == ((b"transfer-encoding", b"chunked"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert msg.chunked
    assert payload.is_eof()
    assert b"second" == b"".join(d for d in payload._buffer)


@pytest.mark.parametrize("code", (204, 304, 101, 102))
def test_parse_chunked_payload_empty_body_than_another_chunked(
    response: HttpResponseParser, code: int
) -> None:
    head = f"HTTP/1.1 {code} OK\r\n".encode()
    text = head + b"transfer-encoding: chunked\r\n\r\n"
    msg, payload = response.feed_data(text)[0][0]
    assert msg.version == HttpVersion(major=1, minor=1)
    assert msg.code == code
    assert msg.reason == "OK"
    assert msg.headers == HeadersDictProxy(
        CIMultiDict([("Transfer-Encoding", "chunked")])
    )
    assert msg.raw_headers == ((b"transfer-encoding", b"chunked"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert msg.chunked
    assert payload.is_eof()

    text = (
        b"HTTP/1.1 200 OK\r\n"
        b"transfer-encoding: chunked\r\n\r\n"
        b"6\r\nsecond\r\n0\r\n\r\n"
    )
    msg, payload = response.feed_data(text)[0][0]
    assert msg.version == HttpVersion(major=1, minor=1)
    assert msg.code == 200
    assert msg.reason == "OK"
    assert msg.headers == CIMultiDict(
        [
            ("Transfer-Encoding", "chunked"),
        ]
    )
    assert msg.raw_headers == ((b"transfer-encoding", b"chunked"),)
    assert not msg.should_close
    assert msg.compression is None
    assert not msg.upgrade
    assert msg.chunked
    assert payload.is_eof()
    assert b"second" == b"".join(d for d in payload._buffer)


async def test_parse_chunked_payload_split_chunks(response: HttpResponseParser) -> None:
    network_chunks = (
        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n",
        b"5\r\nfi",
        b"rst",
        # This simulates a bug in lax mode caused when the \r\n separator, before the
        # next HTTP chunk, appears at the start of the next network chunk.
        b"\r\n",
        b"6",
        b"\r",
        b"\n",
        b"second\r",
        b"\n0\r\n\r\n",
    )
    reader = response.feed_data(network_chunks[0])[0][0][1]
    for c in network_chunks[1:]:
        response.feed_data(c)

    assert response.feed_eof() is None
    assert reader.is_eof()
    assert await reader.read() == b"firstsecond"


async def test_parse_chunked_payload_with_lf_in_extensions(
    parser: HttpRequestParser,
) -> None:
    """Test chunked payload that has a LF in the chunk extensions."""
    payload = (
        b"GET / HTTP/1.1\r\nHost: localhost:5001\r\n"
        b"Transfer-Encoding: chunked\r\n\r\n2;\nxx\r\n4c\r\n0\r\n\r\n"
        b"GET /admin HTTP/1.1\r\nHost: localhost:5001\r\n"
        b"Transfer-Encoding: chunked\r\n\r\n0\r\n\r\n"
    )
    with pytest.raises(http_exceptions.BadHttpMessage, match="\\\\nxx"):
        parser.feed_data(payload)


def test_partial_url(parser: HttpRequestParser) -> None:
    messages, upgrade, tail = parser.feed_data(b"GET /te")
    assert len(messages) == 0
    messages, upgrade, tail = parser.feed_data(b"st HTTP/1.1\r\nHost: a\r\n\r\n")
    assert len(messages) == 1

    msg, payload = messages[0]

    assert msg.method == "GET"
    assert msg.path == "/test"
    assert msg.version == (1, 1)
    assert payload.is_eof()


@pytest.mark.parametrize(
    ("uri", "path", "query", "fragment"),
    [
        ("/path%23frag", "/path#frag", {}, ""),
        ("/path%2523frag", "/path%23frag", {}, ""),
        ("/path?key=value%23frag", "/path", {"key": "value#frag"}, ""),
        ("/path?key=value%2523frag", "/path", {"key": "value%23frag"}, ""),
        ("/path#frag%20", "/path", {}, "frag "),
        ("/path#frag%2520", "/path", {}, "frag%20"),
    ],
)
def test_parse_uri_percent_encoded(
    parser: HttpRequestParser, uri: str, path: str, query: dict[str, str], fragment: str
) -> None:
    text = (f"GET {uri} HTTP/1.1\r\nHost: a\r\n\r\n").encode()
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.path == uri
    assert msg.url == URL(uri)
    assert msg.url.path == path
    assert msg.url.query == query
    assert msg.url.fragment == fragment


def test_parse_uri_utf8(parser: HttpRequestParser) -> None:
    if not isinstance(parser, HttpRequestParserPy):
        pytest.xfail("Not valid HTTP. Maybe update py-parser to reject later.")
    text = ("GET /путь?ключ=знач#фраг HTTP/1.1\r\nHost: a\r\n\r\n").encode()
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.path == "/путь?ключ=знач#фраг"
    assert msg.url.path == "/путь"
    assert msg.url.query == {"ключ": "знач"}
    assert msg.url.fragment == "фраг"


def test_parse_uri_utf8_percent_encoded(parser: HttpRequestParser) -> None:
    text = (
        "GET %s HTTP/1.1\r\nHost: a\r\n\r\n"
        % quote("/путь?ключ=знач#фраг", safe="/?=#")
    ).encode()
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.path == quote("/путь?ключ=знач#фраг", safe="/?=#")
    assert msg.url == URL("/путь?ключ=знач#фраг")
    assert msg.url.path == "/путь"
    assert msg.url.query == {"ключ": "знач"}
    assert msg.url.fragment == "фраг"


def test_parse_uri_empty_query_with_fragment(parser: HttpRequestParser) -> None:
    # Origin-form target with an empty query but a fragment: the ``#`` sits
    # immediately after ``?``. Regression for the C parser folding ``#frag``
    # into the query string instead of the fragment.
    text = b"GET /path?#frag HTTP/1.1\r\nHost: a\r\n\r\n"
    messages, upgrade, tail = parser.feed_data(text)
    msg = messages[0][0]

    assert msg.url.path == "/path"
    assert msg.url.query == {}
    assert msg.url.fragment == "frag"


@pytest.mark.skipif(
    "HttpRequestParserC" not in dir(aiohttp.http_parser),
    reason="C based HTTP parser not available",
)
def test_parse_bad_method_for_c_parser_raises(
    event_loop: asyncio.AbstractEventLoop, server: Server[Request]
) -> None:
    protocol = RequestHandler(server, loop=event_loop)

    payload = b"GET1 /test HTTP/1.1\r\n\r\n"
    parser = HttpRequestParserC(
        protocol,
        event_loop,
        DEFAULT_CHUNK_SIZE,
        max_line_size=8190,
        max_headers=128,
        max_field_size=8190,
    )
    protocol._parser = parser

    with pytest.raises(aiohttp.http_exceptions.BadStatusLine):
        messages, upgrade, tail = parser.feed_data(payload)


class TestParsePayload:
    async def test_parse_eof_payload(self, protocol: BaseProtocol) -> None:
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(out, headers_parser=HeadersParser())
        p.feed_data(b"data")
        p.feed_eof()

        assert out.is_eof()
        assert [bytearray(b"data")] == list(out._buffer)

    async def test_parse_length_payload_eof(self, protocol: BaseProtocol) -> None:
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )

        p = HttpPayloadParser(out, length=4, headers_parser=HeadersParser())
        p.feed_data(b"da")

        with pytest.raises(http_exceptions.ContentLengthError):
            p.feed_eof()

    async def test_parse_length_payload_eof_error_message(
        self, protocol: BaseProtocol
    ) -> None:
        """Test that ContentLengthError includes expected vs received bytes."""
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())

        # Expect 10 bytes, but only send 3
        p = HttpPayloadParser(out, length=10, headers_parser=HeadersParser())
        p.feed_data(b"abc")

        with pytest.raises(
            http_exceptions.ContentLengthError, match=r"received 3 of 10 bytes"
        ):
            p.feed_eof()

    async def test_parse_length_payload_eof_no_data(
        self, protocol: BaseProtocol
    ) -> None:
        """Test ContentLengthError when no data is received."""
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())

        # Expect 20 bytes, but send nothing
        p = HttpPayloadParser(out, length=20, headers_parser=HeadersParser())

        with pytest.raises(
            http_exceptions.ContentLengthError, match=r"received 0 of 20 bytes"
        ):
            p.feed_eof()

    async def test_parse_length_payload_partial_data(
        self, protocol: BaseProtocol
    ) -> None:
        """Test ContentLengthError with various amounts of partial data."""
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())

        # Expect 100 bytes, but only send 45
        p = HttpPayloadParser(out, length=100, headers_parser=HeadersParser())
        p.feed_data(b"a" * 25)
        p.feed_data(b"b" * 20)

        with pytest.raises(
            http_exceptions.ContentLengthError,
            match=r"received 45 of 100 bytes",
        ):
            p.feed_eof()

    async def test_parse_length_payload_eof_completes_after_pause(
        self, protocol: BaseProtocol
    ) -> None:
        """feed_eof() completes a fully received length payload despite a pause.

        Regression test for #13348:
        The parser paused for flow control with pending decompressed data
        when EOF arrived; the fully received body must complete instead of
        raising ContentLengthError.
        """
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        original = b"x" * (1024 * 1024)
        compressed = zlib.compress(original)

        p = HttpPayloadParser(
            out,
            length=len(compressed),
            compression="deflate",
            headers_parser=HeadersParser(),
        )
        p.pause_reading()  # flow control kicked in before the final bytes
        state, tail = p.feed_data(compressed)
        assert state is PayloadState.PAYLOAD_HAS_PENDING_INPUT
        assert not p.done

        # All bytes were received, so EOF drains the pending data and
        # completes the payload.
        p.feed_eof()

        assert p.done
        assert out.is_eof()  # type: ignore[unreachable]
        assert out.exception() is None
        result = await out.read()
        assert result == original

    async def test_parse_chunked_payload_size_error(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        with pytest.raises(http_exceptions.TransferEncodingError):
            p.feed_data(b"blah\r\n")
        assert isinstance(out.exception(), http_exceptions.TransferEncodingError)

    async def test_chunked_chunk_size_line_too_long(
        self, protocol: BaseProtocol
    ) -> None:
        """A complete oversized chunk-size line is rejected with LineTooLong."""
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, chunked=True, headers_parser=HeadersParser(), max_line_size=32
        )
        size_line = b"1;" + b"a" * 4096 + b"\r\n"
        with pytest.raises(http_exceptions.LineTooLong):
            p.feed_data(size_line)

    async def test_chunked_chunk_size_line_within_limit(
        self, protocol: BaseProtocol
    ) -> None:
        """A small chunk-size line still parses when max_line_size is low."""
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, chunked=True, headers_parser=HeadersParser(), max_line_size=32
        )
        p.feed_data(b"1\r\nx\r\n0\r\n\r\n")
        assert out.is_eof()
        assert b"x" == b"".join(out._buffer)

    async def test_chunked_chunk_size_line_at_limit(
        self, protocol: BaseProtocol
    ) -> None:
        """A chunk-size line of exactly max_line_size bytes is accepted (>, not >=)."""
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, chunked=True, headers_parser=HeadersParser(), max_line_size=32
        )
        # "1;" + 30 * "a" is exactly 32 bytes before the CRLF.
        size_line = b"1;" + b"a" * 30
        assert len(size_line) == 32
        p.feed_data(size_line + b"\r\nx\r\n0\r\n\r\n")
        assert out.is_eof()
        assert b"x" == b"".join(out._buffer)

    async def test_chunked_chunk_size_line_one_over_limit(
        self, protocol: BaseProtocol
    ) -> None:
        """A chunk-size line one byte over max_line_size is rejected."""
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, chunked=True, headers_parser=HeadersParser(), max_line_size=32
        )
        # "1;" + 31 * "a" is 33 bytes before the CRLF.
        size_line = b"1;" + b"a" * 31
        assert len(size_line) == 33
        with pytest.raises(http_exceptions.LineTooLong):
            p.feed_data(size_line + b"\r\nx\r\n0\r\n\r\n")

    async def test_parse_chunked_payload_size_data_mismatch(
        self, protocol: BaseProtocol
    ) -> None:
        """Chunk-size does not match actual data: should raise, not hang.

        Regression test for #10596.
        """
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        # Declared chunk-size is 4 but actual data is "Hello" (5 bytes).
        # After consuming 4 bytes, remaining starts with "o" not "\r\n".
        with pytest.raises(http_exceptions.TransferEncodingError):
            p.feed_data(b"4\r\nHello\r\n0\r\n\r\n")
        assert isinstance(out.exception(), http_exceptions.TransferEncodingError)

    async def test_parse_chunked_payload_size_data_mismatch_too_short(
        self, protocol: BaseProtocol
    ) -> None:
        """Chunk-size larger than data: declared 6 but only 5 bytes before CRLF.

        Regression test for #10596.
        """
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        # Declared chunk-size is 6 but actual data before CRLF is "Hello" (5 bytes).
        # Parser reads 6 bytes: "Hello\r", then expects \r\n but sees "\n0\r\n..."
        with pytest.raises(http_exceptions.TransferEncodingError):
            p.feed_data(b"6\r\nHello\r\n0\r\n\r\n")
        assert isinstance(out.exception(), http_exceptions.TransferEncodingError)

    async def test_parse_chunked_payload_split_end(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        p.feed_data(b"4\r\nasdf\r\n0\r\n")
        p.feed_data(b"\r\n")

        assert out.is_eof()
        assert b"asdf" == b"".join(out._buffer)

    async def test_parse_chunked_payload_split_end2(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        p.feed_data(b"4\r\nasdf\r\n0\r\n\r")
        p.feed_data(b"\n")

        assert out.is_eof()
        assert b"asdf" == b"".join(out._buffer)

    async def test_parse_chunked_payload_split_end_trailers(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        p.feed_data(b"4\r\nasdf\r\n0\r\n")
        p.feed_data(b"Content-MD5: 912ec803b2ce49e4a541068d495ab570\r\n")
        p.feed_data(b"\r\n")

        assert out.is_eof()
        assert b"asdf" == b"".join(out._buffer)

    async def test_parse_chunked_payload_split_end_trailers2(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        p.feed_data(b"4\r\nasdf\r\n0\r\n")
        p.feed_data(b"Content-MD5: 912ec803b2ce49e4a541068d495ab570\r\n\r")
        p.feed_data(b"\n")

        assert out.is_eof()
        assert b"asdf" == b"".join(out._buffer)

    async def test_parse_chunked_payload_split_end_trailers3(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        p.feed_data(b"4\r\nasdf\r\n0\r\nContent-MD5: ")
        p.feed_data(b"912ec803b2ce49e4a541068d495ab570\r\n\r\n")

        assert out.is_eof()
        assert b"asdf" == b"".join(out._buffer)

    async def test_parse_chunked_payload_split_end_trailers4(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(out, chunked=True, headers_parser=HeadersParser())
        p.feed_data(b"4\r\nasdf\r\n0\r\nC")
        p.feed_data(b"ontent-MD5: 912ec803b2ce49e4a541068d495ab570\r\n\r\n")

        assert out.is_eof()
        assert b"asdf" == b"".join(out._buffer)

    async def test_http_payload_parser_length(self, protocol: BaseProtocol) -> None:
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(out, length=2, headers_parser=HeadersParser())
        state, tail = p.feed_data(b"1245")
        assert state is PayloadState.PAYLOAD_COMPLETE

        assert b"12" == out._buffer[0]
        assert b"45" == tail

    async def test_http_payload_parser_deflate(self, protocol: BaseProtocol) -> None:
        # c=compressobj(wbits=15); b''.join([c.compress(b'data'), c.flush()])
        COMPRESSED = b"x\x9cKI,I\x04\x00\x04\x00\x01\x9b"

        length = len(COMPRESSED)
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out, length=length, compression="deflate", headers_parser=HeadersParser()
        )
        p.feed_data(COMPRESSED)
        assert b"data" == out._buffer[0]
        assert out.is_eof()

    async def test_http_payload_parser_deflate_no_hdrs(
        self, protocol: BaseProtocol
    ) -> None:
        """Tests incorrectly formed data (no zlib headers)."""
        # c=compressobj(wbits=-15); b''.join([c.compress(b'data'), c.flush()])
        COMPRESSED = b"KI,I\x04\x00"

        length = len(COMPRESSED)
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, length=length, compression="deflate", headers_parser=HeadersParser()
        )
        p.feed_data(COMPRESSED)
        assert b"data" == out._buffer[0]
        assert out.is_eof()

    async def test_http_payload_parser_deflate_light(
        self, protocol: BaseProtocol
    ) -> None:
        # c=compressobj(wbits=9); b''.join([c.compress(b'data'), c.flush()])
        COMPRESSED = b"\x18\x95KI,I\x04\x00\x04\x00\x01\x9b"

        length = len(COMPRESSED)
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, length=length, compression="deflate", headers_parser=HeadersParser()
        )
        p.feed_data(COMPRESSED)

        assert b"data" == out._buffer[0]
        assert out.is_eof()

    async def test_http_payload_parser_deflate_split(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, compression="deflate", headers_parser=HeadersParser()
        )
        # Feeding one correct byte should be enough to choose exact
        # deflate decompressor
        p.feed_data(b"x")
        p.feed_data(b"\x9cKI,I\x04\x00\x04\x00\x01\x9b")
        p.feed_eof()
        assert b"data" == out._buffer[0]

    async def test_http_payload_parser_deflate_split_err(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        p = HttpPayloadParser(
            out, compression="deflate", headers_parser=HeadersParser()
        )
        # Feeding one wrong byte should be enough to choose exact
        # deflate decompressor
        p.feed_data(b"K")
        p.feed_data(b"I,I\x04\x00")
        p.feed_eof()
        assert b"data" == out._buffer[0]

    async def test_http_payload_parser_length_zero(
        self, protocol: BaseProtocol
    ) -> None:
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(out, length=0, headers_parser=HeadersParser())
        assert p.done
        assert out.is_eof()

    @pytest.mark.skipif(brotli is None, reason="brotli is not installed")
    async def test_http_payload_brotli(self, protocol: BaseProtocol) -> None:
        compressed = brotli.compress(b"brotli data")
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(compressed),
            compression="br",
            headers_parser=HeadersParser(),
        )
        p.feed_data(compressed)
        assert b"brotli data" == out._buffer[0]
        assert out.is_eof()

    @pytest.mark.skipif(zstandard is None, reason="zstandard is not installed")
    async def test_http_payload_zstandard(self, protocol: BaseProtocol) -> None:
        compressed = zstandard.compress(b"zstd data")
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(compressed),
            compression="zstd",
            headers_parser=HeadersParser(),
        )
        p.feed_data(compressed)
        assert b"zstd data" == out._buffer[0]
        assert out.is_eof()

    @pytest.mark.skipif(zstandard is None, reason="zstandard is not installed")
    async def test_http_payload_zstandard_multi_frame(
        self, protocol: BaseProtocol
    ) -> None:
        frame1 = zstandard.compress(b"first")
        frame2 = zstandard.compress(b"second")
        payload = frame1 + frame2
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(payload),
            compression="zstd",
            headers_parser=HeadersParser(),
        )
        p.feed_data(payload)
        assert b"firstsecond" == b"".join(out._buffer)
        assert out.is_eof()

    @pytest.mark.skipif(zstandard is None, reason="zstandard is not installed")
    async def test_http_payload_zstandard_multi_frame_chunked(
        self, protocol: BaseProtocol
    ) -> None:
        frame1 = zstandard.compress(b"chunk1")
        frame2 = zstandard.compress(b"chunk2")
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(frame1) + len(frame2),
            compression="zstd",
            headers_parser=HeadersParser(),
        )
        p.feed_data(frame1)
        p.feed_data(frame2)
        assert b"chunk1chunk2" == b"".join(out._buffer)
        assert out.is_eof()

    @pytest.mark.skipif(zstandard is None, reason="zstandard is not installed")
    async def test_http_payload_zstandard_frame_split_mid_chunk(
        self, protocol: BaseProtocol
    ) -> None:
        frame1 = zstandard.compress(b"AAAA")
        frame2 = zstandard.compress(b"BBBB")
        combined = frame1 + frame2
        split_point = len(frame1) + 3  # 3 bytes into frame2
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(combined),
            compression="zstd",
            headers_parser=HeadersParser(),
        )
        p.feed_data(combined[:split_point])
        p.feed_data(combined[split_point:])
        assert b"AAAABBBB" == b"".join(out._buffer)
        assert out.is_eof()

    @pytest.mark.skipif(zstandard is None, reason="zstandard is not installed")
    async def test_http_payload_zstandard_many_small_frames(
        self, protocol: BaseProtocol
    ) -> None:
        parts = [f"part{i}".encode() for i in range(10)]
        payload = b"".join(zstandard.compress(p) for p in parts)
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(payload),
            compression="zstd",
            headers_parser=HeadersParser(),
        )
        p.feed_data(payload)
        assert b"".join(parts) == b"".join(out._buffer)
        assert out.is_eof()

    async def test_http_payload_gzip_multi_member(self, protocol: BaseProtocol) -> None:
        member1 = gzip.compress(b"first")
        member2 = gzip.compress(b"second")
        payload = member1 + member2
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(payload),
            compression="gzip",
            headers_parser=HeadersParser(),
        )
        p.feed_data(payload)
        assert b"firstsecond" == b"".join(out._buffer)
        assert out.is_eof()

    async def test_http_payload_gzip_multi_member_chunked(
        self, protocol: BaseProtocol
    ) -> None:
        member1 = gzip.compress(b"chunk1")
        member2 = gzip.compress(b"chunk2")
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(member1) + len(member2),
            compression="gzip",
            headers_parser=HeadersParser(),
        )
        p.feed_data(member1)
        p.feed_data(member2)
        assert b"chunk1chunk2" == b"".join(out._buffer)
        assert out.is_eof()

    async def test_http_payload_gzip_member_split_mid_chunk(
        self, protocol: BaseProtocol
    ) -> None:
        member1 = gzip.compress(b"AAAA")
        member2 = gzip.compress(b"BBBB")
        combined = member1 + member2
        split_point = len(member1) + 3  # 3 bytes into member2
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(combined),
            compression="gzip",
            headers_parser=HeadersParser(),
        )
        p.feed_data(combined[:split_point])
        p.feed_data(combined[split_point:])
        assert b"AAAABBBB" == b"".join(out._buffer)
        assert out.is_eof()

    async def test_http_payload_gzip_many_small_members(
        self, protocol: BaseProtocol
    ) -> None:
        parts = [f"part{i}".encode() for i in range(10)]
        payload = b"".join(gzip.compress(p) for p in parts)
        out = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        p = HttpPayloadParser(
            out,
            length=len(payload),
            compression="gzip",
            headers_parser=HeadersParser(),
        )
        p.feed_data(payload)
        assert b"".join(parts) == b"".join(out._buffer)
        assert out.is_eof()


class TestDeflateBuffer:
    async def test_feed_data(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        dbuf = DeflateBuffer(buf, "deflate")

        dbuf.decompressor = mock.Mock()
        dbuf.decompressor.decompress_sync.return_value = b"line"

        # First byte should be b'x' in order code not to change the decoder.
        dbuf.feed_data(b"xxxx")
        assert [b"line"] == list(buf._buffer)

    async def test_feed_data_err(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        dbuf = DeflateBuffer(buf, "deflate")

        exc = ValueError()
        dbuf.decompressor = mock.Mock()
        dbuf.decompressor.decompress_sync.side_effect = exc

        with pytest.raises(http_exceptions.ContentEncodingError):
            # Should be more than 4 bytes to trigger deflate FSM error.
            # Should start with b'x', otherwise code switch mocked decoder.
            dbuf.feed_data(b"xsomedata")

    async def test_feed_eof(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        dbuf = DeflateBuffer(buf, "deflate")

        dbuf.decompressor = mock.Mock()
        dbuf.decompressor.data_available = False
        dbuf.decompressor.flush.return_value = b""

        dbuf.feed_eof()
        assert buf._eof

    async def test_feed_eof_err_deflate(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        dbuf = DeflateBuffer(buf, "deflate")

        dbuf.decompressor = mock.Mock()
        dbuf.decompressor.data_available = False
        dbuf.decompressor.flush.return_value = b""
        dbuf.decompressor.eof = False
        dbuf.size = 1  # Simulate that data was previously fed

        with pytest.raises(http_exceptions.ContentEncodingError):
            dbuf.feed_eof()

    async def test_feed_eof_no_err_gzip(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        dbuf = DeflateBuffer(buf, "gzip")

        dbuf.decompressor = mock.Mock()
        dbuf.decompressor.data_available = False
        dbuf.decompressor.flush.return_value = b""
        dbuf.decompressor.eof = False

        dbuf.feed_eof()
        assert buf._eof

    @pytest.mark.skipif(
        sys.platform in ("android", "ios"), reason="brotli not available"
    )
    async def test_feed_eof_no_err_brotli(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        dbuf = DeflateBuffer(buf, "br")

        dbuf.decompressor = mock.Mock()
        dbuf.decompressor.data_available = False
        dbuf.decompressor.flush.return_value = b""
        dbuf.decompressor.eof = False

        dbuf.feed_eof()
        assert buf._eof

    @pytest.mark.skipif(zstandard is None, reason="zstandard is not installed")
    async def test_feed_eof_no_err_zstandard(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
        dbuf = DeflateBuffer(buf, "zstd")

        dbuf.decompressor = mock.Mock()
        dbuf.decompressor.data_available = False
        dbuf.decompressor.flush.return_value = b""
        dbuf.decompressor.eof = False

        dbuf.feed_eof()
        assert buf._eof

    async def test_empty_body(self, protocol: BaseProtocol) -> None:
        buf = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        dbuf = DeflateBuffer(buf, "deflate")
        dbuf.feed_eof()

        assert buf.at_eof()

    @pytest.mark.skipif(platform.python_implementation() == "PyPy", reason="Broken")
    @pytest.mark.parametrize(
        "chunk_size",
        [1024, 2**14, 2**16],  # 1KB, 16KB, 64KB
        ids=["1KB", "16KB", "64KB"],
    )
    async def test_streaming_decompress_large_payload(
        self, protocol: BaseProtocol, chunk_size: int
    ) -> None:
        """Test that large payloads decompress correctly when streamed in chunks.

        This simulates real HTTP streaming where compressed data arrives in
        small network chunks. Each chunk's decompressed output should be within
        the max_decompress_size limit, allowing full recovery of the original data.
        """
        # Create a large payload (3MiB) that compresses well
        original = b"A" * (3 * 2**20)
        compressed = zlib.compress(original)

        buf = aiohttp.StreamReader(
            protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
        )
        dbuf = DeflateBuffer(buf, "deflate")

        # Feed compressed data in chunks (simulating network streaming)
        for i in range(0, len(compressed), chunk_size):  # pragma: no branch
            chunk = compressed[i : i + chunk_size]
            while dbuf.feed_data(chunk):
                chunk = b""

        dbuf.feed_eof()

        # Read all decompressed data
        result = b"".join(buf._buffer)
        assert len(result) == len(original)
        assert result == original


def test_response_parser_incomplete_body_error_message(
    response: HttpResponseParser,
) -> None:
    """Test response parser error message for incomplete body."""
    # Response expects 50 bytes
    response.feed_data(b"HTTP/1.1 200 OK\r\nContent-Length: 50\r\n\r\n")
    # Send only 15 bytes
    response.feed_data(b"partial content")

    with pytest.raises(
        http_exceptions.ContentLengthError, match=r"received 15 of 50 bytes"
    ):
        response.feed_eof()


def test_response_parser_no_body_error_message(response: HttpResponseParser) -> None:
    """Test response parser error when no body is received."""
    # Response expects 25 bytes
    response.feed_data(b"HTTP/1.1 200 OK\r\nContent-Length: 25\r\n\r\n")
    # Send no body data

    with pytest.raises(
        http_exceptions.ContentLengthError, match=r"received 0 of 25 bytes"
    ):
        response.feed_eof()


def test_response_parser_partial_chunks_error_message(
    response: HttpResponseParser,
) -> None:
    """Test error message when body is sent in multiple chunks."""
    # Response expects 100 bytes
    response.feed_data(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n")
    # Send data in chunks totaling 60 bytes
    response.feed_data(b"a" * 20)
    response.feed_data(b"b" * 20)
    response.feed_data(b"c" * 20)

    with pytest.raises(
        http_exceptions.ContentLengthError, match=r"received 60 of 100 bytes"
    ):
        response.feed_eof()


def test_request_parser_incomplete_body_error_message(
    parser: HttpRequestParser,
) -> None:
    """Test request parser error message for incomplete body."""
    # Request with Content-Length but incomplete body
    parser.feed_data(b"POST /test HTTP/1.1\r\nHost: a\r\nContent-Length: 30\r\n\r\n")
    # Send only 10 bytes
    parser.feed_data(b"incomplete")

    with pytest.raises(
        http_exceptions.ContentLengthError, match=r"received 10 of 30 bytes"
    ):
        parser.feed_eof()


def test_response_content_length_zero_no_error(response: HttpResponseParser) -> None:
    """Test that Content-Length: 0 does not raise error on feed_eof."""
    # Response with Content-Length: 0
    response.feed_data(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")

    # This should NOT raise an error
    response.feed_eof()  # Should complete without exception
