import asyncio
import datetime
import logging
import ssl
import sys
import time
import weakref
from collections.abc import Iterator, MutableMapping
from typing import NoReturn
from unittest import mock

import pytest
from multidict import CIMultiDict, MultiDict
from pytest_aiohttp import AiohttpClient
from yarl import URL

from aiohttp import ETag, HttpVersion, web
from aiohttp.base_protocol import BaseProtocol
from aiohttp.helpers import DEFAULT_CHUNK_SIZE, HeadersDictProxy
from aiohttp.http_exceptions import BadHttpMessage, LineTooLong
from aiohttp.http_parser import RawRequestMessage
from aiohttp.streams import StreamReader
from aiohttp.test_utils import make_mocked_request
from aiohttp.web_request import _FORWARDED_PAIR_RE


@pytest.fixture
def protocol() -> mock.Mock:
    return mock.Mock(_reading_paused=False)


def test_base_ctor() -> None:
    message = RawRequestMessage(
        "GET",
        "/path/to?a=1&b=2",
        HttpVersion(1, 1),
        HeadersDictProxy(CIMultiDict()),
        (),
        False,
        None,
        False,
        False,
        URL("/path/to?a=1&b=2"),
    )

    protocol = mock.Mock()
    protocol.ssl_context = None
    protocol.peername = None
    protocol.sockname = ("127.0.0.1", 80)
    req = web.BaseRequest(
        message, mock.Mock(), protocol, mock.Mock(), mock.Mock(), mock.Mock()
    )

    assert "GET" == req.method
    assert HttpVersion(1, 1) == req.version
    # No Host header in this request, so host falls back to the
    # local socket address the request arrived on.
    assert req.host == "127.0.0.1"
    assert "/path/to?a=1&b=2" == req.path_qs
    assert "/path/to" == req.path
    assert "a=1&b=2" == req.query_string
    assert CIMultiDict() == req.headers
    assert () == req.raw_headers

    get = req.query
    assert MultiDict([("a", "1"), ("b", "2")]) == get
    # second call should return the same object
    assert get is req.query

    assert req.keep_alive

    assert req


def test_ctor() -> None:
    req = make_mocked_request("GET", "/path/to?a=1&b=2")

    assert "GET" == req.method
    assert HttpVersion(1, 1) == req.version
    # No Host header in this request, so host falls back to the
    # local socket address the request arrived on (the default
    # sockname configured by make_mocked_request).
    assert req.host == "127.0.0.1"
    assert "/path/to?a=1&b=2" == req.path_qs
    assert "/path/to" == req.path
    assert "a=1&b=2" == req.query_string
    assert CIMultiDict() == req.headers
    assert () == req.raw_headers

    get = req.query
    assert MultiDict([("a", "1"), ("b", "2")]) == get
    # second call should return the same object
    assert get is req.query

    assert req.keep_alive

    # just make sure that all lines of make_mocked_request covered
    headers = CIMultiDict(FOO="bar")
    payload = mock.Mock()
    protocol = mock.Mock()
    app = mock.Mock()
    req = make_mocked_request(
        "GET",
        "/path/to?a=1&b=2",
        headers=headers,
        protocol=protocol,
        payload=payload,
        app=app,
    )
    assert req.app is app
    assert req.content is payload
    assert req.protocol is protocol
    assert req.transport is protocol.transport
    assert req.headers == headers
    assert req.raw_headers == ((b"FOO", b"bar"),)
    assert req.task is req._task


def test_host_falls_back_to_sockname_not_dns() -> None:
    """Regression: request.host must not call socket.getfqdn().

    socket.getfqdn() does blocking reverse DNS resolution on the
    event loop thread and can stall a worker for many seconds when
    the system resolver is slow. The fallback for a request with no
    Host header is the local socket address the request arrived on,
    not the system FQDN.
    """
    req = make_mocked_request("GET", "/")
    assert req.host == "127.0.0.1"
    assert str(req.url).startswith("http://127.0.0.1")


def test_host_with_ipv6_sockname() -> None:
    """AF_INET6 sockname is bracketed to form a valid URL authority.

    A bare IPv6 string would cause ``URL.build(authority=...)`` to
    raise ``ValueError``.
    """
    transport = mock.Mock()
    transport.get_extra_info.side_effect = lambda key: (
        ("::1", 80, 0, 0) if key == "sockname" else None
    )
    req = make_mocked_request("GET", "/", transport=transport)
    assert req.host == "[::1]"
    assert str(req.url) == "http://[::1]/"


def test_host_with_unix_socket_sockname() -> None:
    """Unix-socket transports expose sockname as a str path."""
    transport = mock.Mock()
    transport.get_extra_info.side_effect = lambda key: (
        "/tmp/aiohttp.sock" if key == "sockname" else None
    )
    req = make_mocked_request("GET", "/", transport=transport)
    assert req.host == "/tmp/aiohttp.sock"


def test_host_with_no_transport_sockname() -> None:
    """An empty string is returned when no sockname is available."""
    transport = mock.Mock()
    transport.get_extra_info.return_value = None
    req = make_mocked_request("GET", "/", transport=transport)
    assert req.host == ""


def test_doubleslashes() -> None:
    # NB: //foo/bar is an absolute URL with foo netloc and /bar path
    req = make_mocked_request("GET", "/bar//foo/")
    assert "/bar//foo/" == req.path


def test_content_type_not_specified() -> None:
    req = make_mocked_request("Get", "/")
    assert "application/octet-stream" == req.content_type


def test_content_type_from_spec() -> None:
    req = make_mocked_request(
        "Get", "/", CIMultiDict([("CONTENT-TYPE", "application/json")])
    )
    assert "application/json" == req.content_type


def test_content_type_from_spec_with_charset() -> None:
    req = make_mocked_request(
        "Get", "/", CIMultiDict([("CONTENT-TYPE", "text/html; charset=UTF-8")])
    )
    assert "text/html" == req.content_type
    assert "UTF-8" == req.charset


def test_calc_content_type_on_getting_charset() -> None:
    req = make_mocked_request(
        "Get", "/", CIMultiDict([("CONTENT-TYPE", "text/html; charset=UTF-8")])
    )
    assert "UTF-8" == req.charset
    assert "text/html" == req.content_type


def test_urlencoded_querystring() -> None:
    req = make_mocked_request("GET", "/yandsearch?text=%D1%82%D0%B5%D0%BA%D1%81%D1%82")
    assert {"text": "текст"} == req.query


def test_non_ascii_path() -> None:
    req = make_mocked_request("GET", "/путь")
    assert "/путь" == req.path


def test_non_ascii_raw_path() -> None:
    req = make_mocked_request("GET", "/путь")
    assert "/путь" == req.raw_path


def test_absolute_url() -> None:
    req = make_mocked_request("GET", "https://example.com/path/to?a=1")
    assert req.url == URL("https://example.com/path/to?a=1")
    assert req.scheme == "https"
    assert req.host == "example.com"
    assert req.rel_url == URL.build(path="/path/to", query={"a": "1"})


def test_absolute_form_raw_path() -> None:
    # An absolute-form target (RFC 9112 3.2.2) must not leak the scheme/host
    # into raw_path. The path, query and fragment are kept byte-for-byte, the
    # same raw form an origin-form target yields.
    req = make_mocked_request("GET", "https://example.com/path/to?a=1#frag")
    assert req.raw_path == "/path/to?a=1#frag"
    assert req.raw_path == make_mocked_request("GET", "/path/to?a=1#frag").raw_path


def test_connect_authority_form_raw_path() -> None:
    # Authority-form is only used by CONNECT (RFC 9112 3.2.3); its target is a
    # bare host:port with no scheme prefix, so raw_path returns it unchanged.
    message = RawRequestMessage(
        "CONNECT",
        "example.com:443",
        HttpVersion(1, 1),
        HeadersDictProxy(CIMultiDict()),
        (),
        False,
        None,
        False,
        False,
        URL.build(authority="example.com:443", encoded=True),
    )
    protocol = mock.Mock()
    protocol.ssl_context = None
    protocol.peername = None
    protocol.sockname = ("127.0.0.1", 80)
    req = web.BaseRequest(
        message, mock.Mock(), protocol, mock.Mock(), mock.Mock(), mock.Mock()
    )
    assert req._message.url.absolute
    assert req.raw_path == "example.com:443"


def test_clone_absolute_scheme() -> None:
    req = make_mocked_request("GET", "https://example.com/path/to?a=1")
    assert req.scheme == "https"
    req2 = req.clone(scheme="http")
    assert req2.scheme == "http"
    assert req2.url.scheme == "http"


def test_clone_absolute_host() -> None:
    req = make_mocked_request("GET", "https://example.com/path/to?a=1")
    assert req.host == "example.com"
    req2 = req.clone(host="foo.test")
    assert req2.host == "foo.test"
    assert req2.url.host == "foo.test"


def test_content_length() -> None:
    req = make_mocked_request("Get", "/", CIMultiDict([("CONTENT-LENGTH", "123")]))

    assert 123 == req.content_length


def test_range_to_slice_head() -> None:
    req = make_mocked_request(
        "GET", "/", headers=CIMultiDict([("RANGE", "bytes=0-499")])
    )
    assert isinstance(req.http_range, slice)
    assert req.http_range.start == 0 and req.http_range.stop == 500


def test_range_to_slice_mid() -> None:
    req = make_mocked_request(
        "GET", "/", headers=CIMultiDict([("RANGE", "bytes=500-999")])
    )
    assert isinstance(req.http_range, slice)
    assert req.http_range.start == 500 and req.http_range.stop == 1000


def test_range_to_slice_tail_start() -> None:
    req = make_mocked_request(
        "GET", "/", headers=CIMultiDict([("RANGE", "bytes=9500-")])
    )
    assert isinstance(req.http_range, slice)
    assert req.http_range.start == 9500 and req.http_range.stop is None


def test_range_to_slice_tail_stop() -> None:
    req = make_mocked_request(
        "GET", "/", headers=CIMultiDict([("RANGE", "bytes=-500")])
    )
    assert isinstance(req.http_range, slice)
    assert req.http_range.start == -500 and req.http_range.stop is None


def test_range_non_ascii() -> None:
    # ५ = DEVANAGARI DIGIT FIVE
    req = make_mocked_request("GET", "/", headers=CIMultiDict([("RANGE", "bytes=4-५")]))
    with pytest.raises(ValueError, match="range not in acceptable format"):
        req.http_range


def test_non_keepalive_on_http10() -> None:
    req = make_mocked_request("GET", "/", version=HttpVersion(1, 0))
    assert not req.keep_alive


def test_non_keepalive_on_closing() -> None:
    req = make_mocked_request("GET", "/", closing=True)
    assert not req.keep_alive


async def test_call_POST_on_GET_request() -> None:
    req = make_mocked_request("GET", "/")

    ret = await req.post()
    assert CIMultiDict() == ret


async def test_call_POST_on_weird_content_type() -> None:
    req = make_mocked_request(
        "POST", "/", headers=CIMultiDict({"CONTENT-TYPE": "something/weird"})
    )

    ret = await req.post()
    assert CIMultiDict() == ret


async def test_call_POST_twice() -> None:
    req = make_mocked_request("GET", "/")

    ret1 = await req.post()
    ret2 = await req.post()
    assert ret1 is ret2


def test_no_request_cookies() -> None:
    req = make_mocked_request("GET", "/")

    assert req.cookies == {}

    cookies = req.cookies
    assert cookies is req.cookies


def test_request_cookie() -> None:
    headers = CIMultiDict(COOKIE="cookie1=value1; cookie2=value2")
    req = make_mocked_request("GET", "/", headers=headers)

    assert req.cookies == {"cookie1": "value1", "cookie2": "value2"}


def test_request_cookie__set_item() -> None:
    headers = CIMultiDict(COOKIE="name=value")
    req = make_mocked_request("GET", "/", headers=headers)

    assert req.cookies == {"name": "value"}

    with pytest.raises(TypeError):
        req.cookies["my"] = "value"  # type: ignore[index]


def test_request_cookies_with_special_characters() -> None:
    """Test that cookies with special characters in names are accepted.

    This tests the fix for issue #2683 where cookies with special characters
    like {, }, / in their names would cause a 500 error. The fix makes the
    cookie parser more tolerant to handle real-world cookies.
    """
    # Test cookie names with curly braces (e.g., ISAWPLB{DB45DF86-F806-407C-932C-D52A60E4019E})
    headers = CIMultiDict(COOKIE="{test}=value1; normal=value2")
    req = make_mocked_request("GET", "/", headers=headers)
    # Both cookies should be parsed successfully
    assert req.cookies == {"{test}": "value1", "normal": "value2"}

    # Test cookie names with forward slash
    headers = CIMultiDict(COOKIE="test/name=value1; valid=value2")
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"test/name": "value1", "valid": "value2"}

    # Test cookie names with various special characters
    headers = CIMultiDict(
        COOKIE="test{foo}bar=value1; test/path=value2; normal_cookie=value3"
    )
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {
        "test{foo}bar": "value1",
        "test/path": "value2",
        "normal_cookie": "value3",
    }


def test_request_cookies_real_world_examples() -> None:
    """Test handling of real-world cookie examples from issue #2683."""
    # Example from the issue: ISAWPLB{DB45DF86-F806-407C-932C-D52A60E4019E}
    headers = CIMultiDict(
        COOKIE="ISAWPLB{DB45DF86-F806-407C-932C-D52A60E4019E}=val1; normal_cookie=val2"
    )
    req = make_mocked_request("GET", "/", headers=headers)
    # All cookies should be parsed successfully
    assert req.cookies == {
        "ISAWPLB{DB45DF86-F806-407C-932C-D52A60E4019E}": "val1",
        "normal_cookie": "val2",
    }

    # Multiple cookies with special characters
    headers = CIMultiDict(
        COOKIE="{cookie1}=val1; cookie/2=val2; cookie[3]=val3; cookie(4)=val4"
    )
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {
        "{cookie1}": "val1",
        "cookie/2": "val2",
        "cookie[3]": "val3",
        "cookie(4)": "val4",
    }


def test_request_cookies_edge_cases() -> None:
    """Test edge cases for cookie parsing."""
    # Empty cookie value
    headers = CIMultiDict(COOKIE="test=; normal=value")
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"test": "", "normal": "value"}

    # Cookie with quoted value
    headers = CIMultiDict(COOKIE='test="quoted value"; normal=unquoted')
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"test": "quoted value", "normal": "unquoted"}


def test_request_cookies_many_invalid(caplog: pytest.LogCaptureFixture) -> None:
    """Test many invalid cookies doesn't cause too many logs."""
    bad = "bad" + chr(1) + "name"
    cookie = "; ".join(f"{bad}{i}=1" for i in range(3000))
    req = make_mocked_request("GET", "/", headers=CIMultiDict(COOKIE=cookie))

    with caplog.at_level(logging.DEBUG):
        cookies = req.cookies

    assert len(caplog.record_tuples) == 1
    _, level, msg = caplog.record_tuples[0]
    assert level is logging.DEBUG
    assert "Cannot load cookie" in msg
    assert cookies == {}


def test_request_cookies_no_500_error() -> None:
    """Test that cookies with special characters don't cause 500 errors.

    This specifically tests that issue #2683 is fixed - previously cookies
    with characters like { } would cause CookieError and 500 responses.
    """
    # This cookie format previously caused 500 errors
    headers = CIMultiDict(COOKIE="ISAWPLB{DB45DF86-F806-407C-932C-D52A60E4019E}=test")

    # Should not raise any exception when accessing cookies
    req = make_mocked_request("GET", "/", headers=headers)
    cookies = req.cookies  # This used to raise CookieError

    # Verify the cookie was parsed successfully
    assert "ISAWPLB{DB45DF86-F806-407C-932C-D52A60E4019E}" in cookies
    assert cookies["ISAWPLB{DB45DF86-F806-407C-932C-D52A60E4019E}"] == "test"


def test_request_cookies_quoted_values() -> None:
    """Test that quoted cookie values are handled consistently.

    This tests the fix for issue #5397 where quoted cookie values were
    handled inconsistently based on whether domain attributes were present.
    The new parser should always unquote cookie values consistently.
    """
    # Test simple quoted cookie value
    headers = CIMultiDict(COOKIE='sess="quoted_value"')
    req = make_mocked_request("GET", "/", headers=headers)
    # Quotes should be removed consistently
    assert req.cookies == {"sess": "quoted_value"}

    # Test quoted cookie with semicolon in value
    headers = CIMultiDict(COOKIE='data="value;with;semicolons"')
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"data": "value;with;semicolons"}

    # Test mixed quoted and unquoted cookies
    headers = CIMultiDict(
        COOKIE='quoted="value1"; unquoted=value2; also_quoted="value3"'
    )
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {
        "quoted": "value1",
        "unquoted": "value2",
        "also_quoted": "value3",
    }

    # Test escaped quotes in cookie value
    headers = CIMultiDict(COOKIE=r'escaped="value with \" quote"')
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"escaped": 'value with " quote'}

    # Test empty quoted value
    headers = CIMultiDict(COOKIE='empty=""')
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"empty": ""}


def test_request_cookies_with_attributes() -> None:
    """Test that cookie attributes are parsed as cookies per RFC 6265.

    Per RFC 6265 Section 5.4, Cookie headers contain only name-value pairs.
    Names that match attribute names (Domain, Path, etc.) should be treated
    as regular cookies, not as attributes.
    """
    # Cookie with domain - both should be parsed as cookies
    headers = CIMultiDict(COOKIE='sess="quoted_value"; Domain=.example.com')
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"sess": "quoted_value", "Domain": ".example.com"}

    # Cookie with multiple attribute names - all parsed as cookies
    headers = CIMultiDict(COOKIE='token="abc123"; Path=/; Secure; HttpOnly')
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {"token": "abc123", "Path": "/", "Secure": "", "HttpOnly": ""}

    # Multiple cookies with attribute names mixed in
    headers = CIMultiDict(
        COOKIE='c1="v1"; Domain=.example.com; c2="v2"; Path=/api; c3=v3; Secure'
    )
    req = make_mocked_request("GET", "/", headers=headers)
    assert req.cookies == {
        "c1": "v1",
        "Domain": ".example.com",
        "c2": "v2",
        "Path": "/api",
        "c3": "v3",
        "Secure": "",
    }


def test_match_info() -> None:
    req = make_mocked_request("GET", "/")
    assert req._match_info is req.match_info


def test_request_is_mutable_mapping() -> None:
    req = make_mocked_request("GET", "/")
    assert isinstance(req, MutableMapping)
    assert req  # even when the MutableMapping is empty, request should always be True
    req["key"] = "value"
    assert "value" == req["key"]


def test_request_delitem() -> None:
    req = make_mocked_request("GET", "/")
    req["key"] = "value"
    assert "value" == req["key"]
    del req["key"]
    assert "key" not in req


def test_request_len() -> None:
    req = make_mocked_request("GET", "/")
    assert len(req) == 0
    req["key"] = "value"
    assert len(req) == 1


def test_request_iter() -> None:
    req = make_mocked_request("GET", "/")
    req["key"] = "value"
    req["key2"] = "value2"
    key3 = web.RequestKey("key3", str)
    req[key3] = "value3"
    assert set(req) == {"key", "key2", key3}


def test_requestkey() -> None:
    req = make_mocked_request("GET", "/")
    key = web.RequestKey("key", str)
    req[key] = "value"
    assert req[key] == "value"
    assert len(req) == 1
    del req[key]
    assert len(req) == 0


def test_request_get_requestkey() -> None:
    req = make_mocked_request("GET", "/")
    key = web.RequestKey("key", int)
    assert req.get(key, "foo") == "foo"
    req[key] = 5
    assert req.get(key, "foo") == 5


def test_requestkey_repr_concrete() -> None:
    key = web.RequestKey("key", int)
    assert repr(key) in (
        "<RequestKey(__channelexec__.key, type=int)>",  # pytest-xdist
        "<RequestKey(__main__.key, type=int)>",
    )
    key2 = web.RequestKey("key", web.Request)
    assert repr(key2) in (
        # pytest-xdist:
        "<RequestKey(__channelexec__.key, type=aiohttp.web_request.Request)>",
        "<RequestKey(__main__.key, type=aiohttp.web_request.Request)>",
    )


def test_requestkey_repr_nonconcrete() -> None:
    key = web.RequestKey("key", Iterator[int])
    if sys.version_info < (3, 11):
        assert repr(key) in (
            # pytest-xdist:
            "<RequestKey(__channelexec__.key, type=collections.abc.Iterator)>",
            "<RequestKey(__main__.key, type=collections.abc.Iterator)>",
        )
    else:
        assert repr(key) in (
            # pytest-xdist:
            "<RequestKey(__channelexec__.key, type=collections.abc.Iterator[int])>",
            "<RequestKey(__main__.key, type=collections.abc.Iterator[int])>",
        )


def test_requestkey_repr_annotated() -> None:
    key = web.RequestKey[Iterator[int]]("key")
    if sys.version_info < (3, 11):
        assert repr(key) in (
            # pytest-xdist:
            "<RequestKey(__channelexec__.key, type=collections.abc.Iterator)>",
            "<RequestKey(__main__.key, type=collections.abc.Iterator)>",
        )
    else:
        assert repr(key) in (
            # pytest-xdist:
            "<RequestKey(__channelexec__.key, type=collections.abc.Iterator[int])>",
            "<RequestKey(__main__.key, type=collections.abc.Iterator[int])>",
        )


def test___repr__() -> None:
    req = make_mocked_request("GET", "/path/to")
    assert "<Request GET /path/to >" == repr(req)


def test___repr___non_ascii_path() -> None:
    req = make_mocked_request("GET", "/path/\U0001f415\U0001f308")
    assert "<Request GET /path/\\U0001f415\\U0001f308 >" == repr(req)


def test_http_scheme() -> None:
    req = make_mocked_request("GET", "/", headers={"Host": "example.com"})
    assert "http" == req.scheme
    assert req.secure is False


def test_https_scheme_by_ssl_transport() -> None:
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    req = make_mocked_request(
        "GET", "/", headers={"Host": "example.com"}, sslcontext=context
    )
    assert "https" == req.scheme
    assert req.secure is True


def test_single_forwarded_header() -> None:
    header = "by=identifier;for=identifier;host=identifier;proto=identifier"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert req.forwarded[0]["by"] == "identifier"
    assert req.forwarded[0]["for"] == "identifier"
    assert req.forwarded[0]["host"] == "identifier"
    assert req.forwarded[0]["proto"] == "identifier"


def test_forwarded_re_performance() -> None:
    FORWARDED_RE_TIME_THRESHOLD_SECONDS = 0.08
    value = "{" + "f" * 54773 + "z\x00a=v"
    start = time.perf_counter()
    match = _FORWARDED_PAIR_RE.match(value)
    elapsed = time.perf_counter() - start

    # If this is taking more time, there's probably a performance/ReDoS issue.
    assert elapsed < FORWARDED_RE_TIME_THRESHOLD_SECONDS, (
        f"Regex took {elapsed * 1000:.1f}ms, "
        f"expected <{FORWARDED_RE_TIME_THRESHOLD_SECONDS * 1000:.0f}ms - potential ReDoS issue"
    )
    # This example shouldn't produce a match either.
    assert match is None


@pytest.mark.parametrize(
    "forward_for_in, forward_for_out",
    [
        ("1.2.3.4:1234", "1.2.3.4:1234"),
        ("1.2.3.4", "1.2.3.4"),
        ('"[2001:db8:cafe::17]:1234"', "[2001:db8:cafe::17]:1234"),
        ('"[2001:db8:cafe::17]"', "[2001:db8:cafe::17]"),
    ],
)
def test_forwarded_node_identifier(forward_for_in: str, forward_for_out: str) -> None:
    header = f"for={forward_for_in}"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert req.forwarded == ({"for": forward_for_out},)


def test_single_forwarded_header_camelcase() -> None:
    header = "bY=identifier;fOr=identifier;HOst=identifier;pRoTO=identifier"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert req.forwarded[0]["by"] == "identifier"
    assert req.forwarded[0]["for"] == "identifier"
    assert req.forwarded[0]["host"] == "identifier"
    assert req.forwarded[0]["proto"] == "identifier"


def test_single_forwarded_header_single_param() -> None:
    header = "BY=identifier"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert req.forwarded[0]["by"] == "identifier"


def test_single_forwarded_header_multiple_param() -> None:
    header = "By=identifier1,BY=identifier2,  By=identifier3 ,  BY=identifier4"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert len(req.forwarded) == 4
    assert req.forwarded[0]["by"] == "identifier1"
    assert req.forwarded[1]["by"] == "identifier2"
    assert req.forwarded[2]["by"] == "identifier3"
    assert req.forwarded[3]["by"] == "identifier4"


def test_single_forwarded_header_quoted_escaped() -> None:
    header = r'BY=identifier;pROTO="\lala lan\d\~ 123\!&"'
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert req.forwarded[0]["by"] == "identifier"
    assert req.forwarded[0]["proto"] == "lala land~ 123!&"


def test_single_forwarded_header_custom_param() -> None:
    header = r'BY=identifier;PROTO=https;SOME="other, \"value\""'
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert len(req.forwarded) == 1
    assert req.forwarded[0]["by"] == "identifier"
    assert req.forwarded[0]["proto"] == "https"
    assert req.forwarded[0]["some"] == 'other, "value"'


def test_single_forwarded_header_empty_params() -> None:
    # This is allowed by the grammar given in RFC 7239
    header = ";For=identifier;;PROTO=https;;;"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert req.forwarded[0]["for"] == "identifier"
    assert req.forwarded[0]["proto"] == "https"


def test_single_forwarded_header_bad_separator() -> None:
    header = "BY=identifier PROTO=https"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert "proto" not in req.forwarded[0]


def test_single_forwarded_header_injection1() -> None:
    # We might receive a header like this if we're sitting behind a reverse
    # proxy that blindly appends a forwarded-element without checking
    # the syntax of existing field-values. We should be able to recover
    # the appended element anyway.
    header = 'for=_injected;by=", for=_real'
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert len(req.forwarded) == 2
    assert "by" not in req.forwarded[0]
    assert req.forwarded[1]["for"] == "_real"


def test_single_forwarded_header_injection2() -> None:
    header = "very bad syntax, for=_real"
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert len(req.forwarded) == 2
    assert "for" not in req.forwarded[0]
    assert req.forwarded[1]["for"] == "_real"


def test_single_forwarded_header_long_quoted_string() -> None:
    header = 'for="' + "\\\\" * 5000 + '"'
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Forwarded": header}))
    assert req.forwarded[0]["for"] == "\\" * 5000


def test_multiple_forwarded_headers() -> None:
    headers = CIMultiDict[str]()
    headers.add("Forwarded", "By=identifier1;for=identifier2, BY=identifier3")
    headers.add("Forwarded", "By=identifier4;fOr=identifier5")
    req = make_mocked_request("GET", "/", headers=headers)
    assert len(req.forwarded) == 3
    assert req.forwarded[0]["by"] == "identifier1"
    assert req.forwarded[0]["for"] == "identifier2"
    assert req.forwarded[1]["by"] == "identifier3"
    assert req.forwarded[2]["by"] == "identifier4"
    assert req.forwarded[2]["for"] == "identifier5"


def test_multiple_forwarded_headers_bad_syntax() -> None:
    headers = CIMultiDict[str]()
    headers.add("Forwarded", "for=_1;by=_2")
    headers.add("Forwarded", "invalid value")
    headers.add("Forwarded", "for=_3;by=_4")
    req = make_mocked_request("GET", "/", headers=headers)
    assert len(req.forwarded) == 3
    assert req.forwarded[0]["for"] == "_1"
    assert "for" not in req.forwarded[1]
    assert req.forwarded[2]["by"] == "_4"


def test_multiple_forwarded_headers_injection() -> None:
    headers = CIMultiDict[str]()
    # This could be sent by an attacker, hoping to "shadow" the second header.
    headers.add("Forwarded", 'for=_injected;by="')
    # This is added by our trusted reverse proxy.
    headers.add("Forwarded", "for=_real;by=_actual_proxy")
    req = make_mocked_request("GET", "/", headers=headers)
    assert len(req.forwarded) == 2
    assert "by" not in req.forwarded[0]
    assert req.forwarded[1]["for"] == "_real"
    assert req.forwarded[1]["by"] == "_actual_proxy"


def test_host_by_host_header() -> None:
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"Host": "example.com"}))
    assert req.host == "example.com"


def test_raw_headers() -> None:
    req = make_mocked_request("GET", "/", headers=CIMultiDict({"X-HEADER": "aaa"}))
    assert req.raw_headers == ((b"X-HEADER", b"aaa"),)


def test_rel_url() -> None:
    req = make_mocked_request("GET", "/path")
    assert URL("/path") == req.rel_url


def test_url_url() -> None:
    req = make_mocked_request("GET", "/path", headers={"HOST": "example.com"})
    assert URL("http://example.com/path") == req.url


def test_url_non_default_port() -> None:
    req = make_mocked_request("GET", "/path", headers={"HOST": "example.com:8123"})
    assert req.url == URL("http://example.com:8123/path")


def test_url_ipv6() -> None:
    req = make_mocked_request("GET", "/path", headers={"HOST": "[::1]:8123"})
    assert req.url == URL("http://[::1]:8123/path")


def test_clone() -> None:
    req = make_mocked_request("GET", "/path")
    req2 = req.clone()
    assert req2.method == "GET"
    assert req2.rel_url == URL("/path")


def test_clone_client_max_size() -> None:
    req = make_mocked_request("GET", "/path", client_max_size=1024)
    req2 = req.clone()
    assert req._client_max_size == req2._client_max_size
    assert req2._client_max_size == 1024


def test_clone_override_client_max_size() -> None:
    req = make_mocked_request("GET", "/path", client_max_size=1024)
    req2 = req.clone(client_max_size=2048)
    assert req2.client_max_size == 2048


def test_clone_preserves_pre_handler_error() -> None:
    req = make_mocked_request("GET", "/path")
    err = web.HTTPBadRequest(text="bad")
    req._pre_handler_error = err
    assert req.clone().pre_handler_error is err


def test_clone_method() -> None:
    req = make_mocked_request("GET", "/path")
    req2 = req.clone(method="POST")
    assert req2.method == "POST"
    assert req2.rel_url == URL("/path")


def test_clone_rel_url() -> None:
    req = make_mocked_request("GET", "/path")
    req2 = req.clone(rel_url=URL("/path2"))
    assert req2.rel_url == URL("/path2")


def test_clone_rel_url_str() -> None:
    req = make_mocked_request("GET", "/path")
    req2 = req.clone(rel_url="/path2")
    assert req2.rel_url == URL("/path2")


def test_clone_headers() -> None:
    req = make_mocked_request("GET", "/path", headers={"A": "B"})
    req2 = req.clone(headers=CIMultiDict({"B": "C"}))
    assert req2.headers == CIMultiDict({"B": "C"})
    assert req2.raw_headers == ((b"B", b"C"),)


def test_clone_headers_dict() -> None:
    req = make_mocked_request("GET", "/path", headers={"A": "B"})
    req2 = req.clone(headers={"B": "C"})
    assert req2.headers == CIMultiDict({"B": "C"})
    assert req2.raw_headers == ((b"B", b"C"),)


async def test_cannot_clone_after_read(protocol: BaseProtocol) -> None:
    payload = StreamReader(
        protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
    )
    payload.feed_data(b"data")
    payload.feed_eof()
    req = make_mocked_request("GET", "/path", payload=payload)
    await req.read()
    with pytest.raises(RuntimeError):
        req.clone()


async def test_make_too_big_request(protocol: BaseProtocol) -> None:
    payload = StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
    large_file = 1024**2 * b"x"
    too_large_file = large_file + b"x"
    payload.feed_data(too_large_file)
    payload.feed_eof()
    req = make_mocked_request("POST", "/", payload=payload)
    with pytest.raises(web.HTTPRequestEntityTooLarge) as err:
        await req.read()

    assert err.value.status_code == 413


async def test_request_with_wrong_content_type_encoding(protocol: BaseProtocol) -> None:
    payload = StreamReader(
        protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
    )
    payload.feed_data(b"{}")
    payload.feed_eof()
    headers = {"Content-Type": "text/html; charset=test"}
    req = make_mocked_request("POST", "/", payload=payload, headers=headers)

    with pytest.raises(web.HTTPUnsupportedMediaType) as err:
        await req.text()
    assert err.value.status_code == 415


async def test_request_text_with_invalid_default_encoding(
    protocol: BaseProtocol,
) -> None:
    payload = StreamReader(
        protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
    )
    payload.feed_data(b"\xff")
    payload.feed_eof()
    headers = {"Content-Type": "text/html"}
    req = make_mocked_request("POST", "/", payload=payload, headers=headers)

    with pytest.raises(web.HTTPUnsupportedMediaType) as err:
        await req.text()
    assert err.value.status_code == 415


async def test_make_too_big_request_same_size_to_max(protocol: BaseProtocol) -> None:
    payload = StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
    large_file = 1024**2 * b"x"
    payload.feed_data(large_file)
    payload.feed_eof()
    req = make_mocked_request("POST", "/", payload=payload)
    resp_text = await req.read()

    assert resp_text == large_file


async def test_make_too_big_request_adjust_limit(protocol: BaseProtocol) -> None:
    payload = StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
    large_file = 1024**2 * b"x"
    too_large_file = large_file + b"x"
    payload.feed_data(too_large_file)
    payload.feed_eof()
    max_size = 1024**2 + 2
    req = make_mocked_request("POST", "/", payload=payload, client_max_size=max_size)
    txt = await req.read()
    assert len(txt) == 1024**2 + 1


async def test_multipart_formdata(protocol: BaseProtocol) -> None:
    payload = StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
    payload.feed_data(
        b"-----------------------------326931944431359\r\n"
        b'Content-Disposition: form-data; name="a"\r\n'
        b"\r\n"
        b"b\r\n"
        b"-----------------------------326931944431359\r\n"
        b'Content-Disposition: form-data; name="c"\r\n'
        b"\r\n"
        b"d\r\n"
        b"-----------------------------326931944431359--\r\n"
    )
    content_type = (
        "multipart/form-data; boundary=---------------------------326931944431359"
    )
    payload.feed_eof()
    req = make_mocked_request(
        "POST", "/", headers={"CONTENT-TYPE": content_type}, payload=payload
    )
    result = await req.post()
    assert dict(result) == {"a": "b", "c": "d"}


async def test_urlencoded_form_with_invalid_default_encoding(
    protocol: BaseProtocol,
) -> None:
    payload = StreamReader(
        protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
    )
    payload.feed_data(b"a=1&b=\xff")
    payload.feed_eof()
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
    }
    req = make_mocked_request("POST", "/", payload=payload, headers=headers)

    with pytest.raises(web.HTTPUnsupportedMediaType) as err:
        await req.post()
    assert err.value.status_code == 415


async def test_multipart_formdata_field_missing_name(protocol: BaseProtocol) -> None:
    # Ensure ValueError is raised when Content-Disposition has no name
    payload = StreamReader(
        protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
    )
    payload.feed_data(
        b"-----------------------------326931944431359\r\n"
        b"Content-Disposition: form-data\r\n"  # Missing name!
        b"\r\n"
        b"value\r\n"
        b"-----------------------------326931944431359--\r\n"
    )
    content_type = (
        "multipart/form-data; boundary=---------------------------326931944431359"
    )
    payload.feed_eof()
    req = make_mocked_request(
        "POST", "/", headers={"CONTENT-TYPE": content_type}, payload=payload
    )
    with pytest.raises(ValueError, match="Multipart field missing name"):
        await req.post()


async def test_multipart_formdata_file(protocol: BaseProtocol) -> None:
    # Make sure file uploads work, even without a content type
    payload = StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
    payload.feed_data(
        b"-----------------------------326931944431359\r\n"
        b'Content-Disposition: form-data; name="a_file"; filename="binary"\r\n'
        b"\r\n"
        b"\ff\r\n"
        b"-----------------------------326931944431359--\r\n"
    )
    content_type = (
        "multipart/form-data; boundary=---------------------------326931944431359"
    )
    payload.feed_eof()
    req = make_mocked_request(
        "POST", "/", headers={"CONTENT-TYPE": content_type}, payload=payload
    )
    result = await req.post()
    assert hasattr(result["a_file"], "file")
    content = result["a_file"].file.read()
    assert content == b"\ff"

    req._finish()


async def test_multipart_formdata_headers_too_many(protocol: BaseProtocol) -> None:
    many = b"".join(f"X-{i}: a\r\n".encode() for i in range(130))
    body = (
        b"--b\r\n"
        b'Content-Disposition: form-data; name="a"\r\n' + many + b"\r\n1\r\n"
        b"--b--\r\n"
    )
    content_type = "multipart/form-data; boundary=b"
    payload = StreamReader(
        protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
    )
    payload.feed_data(body)
    payload.feed_eof()
    req = make_mocked_request(
        "POST",
        "/",
        headers={"CONTENT-TYPE": content_type},
        payload=payload,
    )

    with pytest.raises(BadHttpMessage, match="Too many headers received"):
        await req.post()


async def test_multipart_formdata_header_too_long(protocol: BaseProtocol) -> None:
    k = b"t" * 4100
    body = (
        b"--b\r\n"
        b'Content-Disposition: form-data; name="a"\r\n'
        + k
        + b":"
        + k
        + b"\r\n"
        + b"\r\n1\r\n"
        b"--b--\r\n"
    )
    content_type = "multipart/form-data; boundary=b"
    payload = StreamReader(
        protocol, DEFAULT_CHUNK_SIZE, loop=asyncio.get_running_loop()
    )
    payload.feed_data(body)
    payload.feed_eof()
    req = make_mocked_request(
        "POST",
        "/",
        headers={"CONTENT-TYPE": content_type},
        payload=payload,
    )

    match = "400, message:\n  Got more than 8190 bytes when reading"
    with pytest.raises(LineTooLong, match=match):
        await req.post()


async def test_make_too_big_request_limit_None(protocol: BaseProtocol) -> None:
    payload = StreamReader(protocol, 2**16, loop=asyncio.get_running_loop())
    large_file = 1024**2 * b"x"
    too_large_file = large_file + b"x"
    payload.feed_data(too_large_file)
    payload.feed_eof()
    req = make_mocked_request("POST", "/", payload=payload, client_max_size=0)
    txt = await req.read()
    assert len(txt) == 1024**2 + 1


def test_remote_peername_tcp() -> None:
    transp = mock.Mock()
    transp.get_extra_info.return_value = ("10.10.10.10", 1234)
    req = make_mocked_request("GET", "/", transport=transp)
    assert req.remote == "10.10.10.10"


def test_remote_peername_unix() -> None:
    transp = mock.Mock()
    transp.get_extra_info.return_value = "/path/to/sock"
    req = make_mocked_request("GET", "/", transport=transp)
    assert req.remote == "/path/to/sock"


def test_save_state_on_clone() -> None:
    req = make_mocked_request("GET", "/")
    req["key"] = "val"
    req2 = req.clone()
    req2["key"] = "val2"
    assert req["key"] == "val"
    assert req2["key"] == "val2"


def test_clone_scheme() -> None:
    req = make_mocked_request("GET", "/")
    assert req.scheme == "http"
    req2 = req.clone(scheme="https")
    assert req2.scheme == "https"
    assert req2.url.scheme == "https"


def test_clone_host() -> None:
    req = make_mocked_request("GET", "/")
    assert req.host != "example.com"
    req2 = req.clone(host="example.com")
    assert req2.host == "example.com"
    assert req2.url.host == "example.com"


def test_clone_remote() -> None:
    req = make_mocked_request("GET", "/")
    assert req.remote != "11.11.11.11"
    req2 = req.clone(remote="11.11.11.11")
    assert req2.remote == "11.11.11.11"


def test_remote_with_closed_transport() -> None:
    transp = mock.Mock()
    transp.get_extra_info.return_value = ("10.10.10.10", 1234)
    req = make_mocked_request("GET", "/", transport=transp)
    req._protocol = None  # type: ignore[assignment]
    assert req.remote == "10.10.10.10"


def test_url_http_with_closed_transport() -> None:
    req = make_mocked_request("GET", "/")
    req._protocol = None  # type: ignore[assignment]
    assert str(req.url).startswith("http://")


def test_url_https_with_closed_transport() -> None:
    c = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    req = make_mocked_request("GET", "/", sslcontext=c)
    req._protocol = None  # type: ignore[assignment]
    assert str(req.url).startswith("https://")


async def test_get_extra_info() -> None:
    valid_key = "test"
    valid_value = "existent"
    default_value = "default"

    def get_extra_info(name: str, default: object = None) -> object:
        return {valid_key: valid_value}.get(name, default)

    transp = mock.Mock()
    transp.get_extra_info.side_effect = get_extra_info
    req = make_mocked_request("GET", "/", transport=transp)

    assert req is not None
    req_extra_info = req.get_extra_info(valid_key, default_value)
    assert req._protocol.transport is not None
    transp_extra_info = req._protocol.transport.get_extra_info(valid_key, default_value)
    assert req_extra_info == transp_extra_info

    req._protocol.transport = None
    extra_info = req.get_extra_info(valid_key, default_value)
    assert extra_info == default_value


def test_eq() -> None:
    req1 = make_mocked_request("GET", "/path/to?a=1&b=2")
    req2 = make_mocked_request("GET", "/path/to?a=1&b=2")
    assert req1 != req2
    assert req1 == req1


async def test_json(aiohttp_client: AiohttpClient) -> None:
    async def handler(request: web.Request) -> web.Response:
        body_text = await request.text()
        assert body_text == '{"some": "data"}'
        assert request.headers["Content-Type"] == "application/json"
        body_json = await request.json()
        assert body_json == {"some": "data"}
        return web.Response()

    app = web.Application()
    app.router.add_post("/", handler)
    client = await aiohttp_client(app)

    json_data = {"some": "data"}
    async with client.post("/", json=json_data) as resp:
        assert 200 == resp.status


async def test_json_invalid_content_type(aiohttp_client: AiohttpClient) -> None:
    async def handler(request: web.Request) -> NoReturn:
        body_text = await request.text()
        assert body_text == '{"some": "data"}'
        assert request.headers["Content-Type"] == "text/plain"
        await request.json()  # raises HTTP 400
        assert False

    app = web.Application()
    app.router.add_post("/", handler)
    client = await aiohttp_client(app)

    json_data = {"some": "data"}
    headers = {"Content-Type": "text/plain"}
    async with client.post("/", json=json_data, headers=headers) as resp:
        assert 400 == resp.status
        resp_text = await resp.text()
        assert resp_text == (
            "Attempt to decode JSON with unexpected mimetype: text/plain"
        )


def test_weakref_creation() -> None:
    req = make_mocked_request("GET", "/")
    weakref.ref(req)


@pytest.mark.parametrize(
    ("header", "header_attr"),
    (
        pytest.param("If-Match", "if_match"),
        pytest.param("If-None-Match", "if_none_match"),
    ),
)
@pytest.mark.parametrize(
    ("header_val", "expected"),
    (
        pytest.param(
            '"67ab43", W/"54ed21", "7892,dd"',
            (
                ETag(is_weak=False, value="67ab43"),
                ETag(is_weak=True, value="54ed21"),
                ETag(is_weak=False, value="7892,dd"),
            ),
        ),
        pytest.param(
            '"bfc1ef-5b2c2730249c88ca92d82d"',
            (ETag(is_weak=False, value="bfc1ef-5b2c2730249c88ca92d82d"),),
        ),
        pytest.param(
            '"valid-tag", "also-valid-tag",somegarbage"last-tag"',
            (
                ETag(is_weak=False, value="valid-tag"),
                ETag(is_weak=False, value="also-valid-tag"),
            ),
        ),
        pytest.param(
            '"ascii", "это точно не ascii", "ascii again"',
            (ETag(is_weak=False, value="ascii"),),
        ),
        pytest.param(
            "*",
            (ETag(is_weak=False, value="*"),),
        ),
    ),
)
def test_etag_headers(
    header: str, header_attr: str, header_val: str, expected: tuple[ETag, ...]
) -> None:
    req = make_mocked_request("GET", "/", headers={header: header_val})
    assert getattr(req, header_attr) == expected


@pytest.mark.parametrize(
    ("header", "header_attr"),
    (
        pytest.param("If-Modified-Since", "if_modified_since"),
        pytest.param("If-Unmodified-Since", "if_unmodified_since"),
        pytest.param("If-Range", "if_range"),
    ),
)
@pytest.mark.parametrize(
    ("header_val", "expected"),
    (
        pytest.param("xxyyzz", None),
        pytest.param("Tue, 08 Oct 4446413 00:56:40 GMT", None),
        pytest.param("Tue, 08 Oct 2000 00:56:80 GMT", None),
        pytest.param(
            "Tue, 08 Oct 2000 00:56:40 GMT",
            datetime.datetime(2000, 10, 8, 0, 56, 40, tzinfo=datetime.timezone.utc),
        ),
    ),
)
def test_datetime_headers(
    header: str,
    header_attr: str,
    header_val: str,
    expected: datetime.datetime | None,
) -> None:
    req = make_mocked_request("GET", "/", headers={header: header_val})
    assert getattr(req, header_attr) == expected
