text: # type: ignore
  import asyncio
  import socket
  import weakref
  from collections.abc import MutableMapping
  from typing import Any
  from unittest import mock
  
  import pytest
  from multidict import CIMultiDict, CIMultiDictProxy, MultiDict
  from yarl import URL
  
  from aiohttp import HttpVersion, web
  from aiohttp.client_exceptions import ServerDisconnectedError
  from aiohttp.helpers import DEBUG
  from aiohttp.http_parser import RawRequestMessage
  from aiohttp.streams import StreamReader
  from aiohttp.test_utils import make_mocked_request
  from aiohttp.web import HTTPRequestEntityTooLarge, HTTPUnsupportedMediaType
  from aiohttp.web_request import ETag
  
  
  @pytest.fixture
  def protocol():
      return mock.Mock(_reading_paused=False)
  
  
  def test_base_ctor() -> None:
      message = RawRequestMessage(
          "GET",
          "/path/to?a=1&b=2",
          HttpVersion(1, 1),
          CIMultiDictProxy(CIMultiDict()),
          (),
          False,
          False,
          False,
          False,
          URL("/path/to?a=1&b=2"),
      )
  
      req = web.BaseRequest(
          message, mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock()
      )
  
      assert "GET" == req.method
      assert HttpVersion(1, 1) == req.version
      assert req.host == socket.getfqdn()
      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 "__dict__" not in dir(req)
  
      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
      # MacOS may return CamelCased host name, need .lower()
      assert req.host.lower() == socket.getfqdn().lower()
      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
  
      assert "__dict__" not in dir(req)
  
  
  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_content_length() -> None:
      req = make_mocked_request("Get", "/", CIMultiDict([("CONTENT-LENGTH", "123")]))
  
      assert 123 == req.content_length
  
  
  def test_range_to_slice_head() -> None:
      def bytes_gen(size):
          for i in range(size):
              yield i % 256
  
      payload = bytearray(bytes_gen(10000))
      req = make_mocked_request(
          "GET", "/", headers=CIMultiDict([("RANGE", "bytes=0-499")]), payload=payload
      )
      assert isinstance(req.http_range, slice)
      assert req.content[req.http_range] == payload[:500]
  
  
  def test_range_to_slice_mid() -> None:
      def bytes_gen(size):
          for i in range(size):
              yield i % 256
  
      payload = bytearray(bytes_gen(10000))
      req = make_mocked_request(
          "GET", "/", headers=CIMultiDict([("RANGE", "bytes=500-999")]), payload=payload
      )
      assert isinstance(req.http_range, slice)
      assert req.content[req.http_range] == payload[500:1000]
  
  
  def test_range_to_slice_tail_start() -> None:
      def bytes_gen(size):
          for i in range(size):
              yield i % 256
  
      payload = bytearray(bytes_gen(10000))
      req = make_mocked_request(
          "GET", "/", headers=CIMultiDict([("RANGE", "bytes=9500-")]), payload=payload
      )
      assert isinstance(req.http_range, slice)
      assert req.content[req.http_range] == payload[-500:]
  
  
  def test_range_to_slice_tail_stop() -> None:
      def bytes_gen(size):
          for i in range(size):
              yield i % 256
  
      payload = bytearray(bytes_gen(10000))
      req = make_mocked_request(
          "GET", "/", headers=CIMultiDict([("RANGE", "bytes=-500")]), payload=payload
      )
      assert isinstance(req.http_range, slice)
      assert req.content[req.http_range] == payload[-500:]
  
  
  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"
  
  
  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)
      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"
      assert set(req) == {"key", "key2"}
  
  
  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:
      req = make_mocked_request(
          "GET", "/", headers={"Host": "example.com"}, sslcontext=True
      )
      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"
  
  
  @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: Any, forward_for_out: Any) -> 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()
      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()
      headers.add("Forwarded", "for=_1;by=_2")
      headers.add("Forwarded", "invalid value")
      headers.add("Forwarded", "")
      headers.add("Forwarded", "for=_3;by=_4")
      req = make_mocked_request("GET", "/", headers=headers)
      assert len(req.forwarded) == 4
      assert req.forwarded[0]["for"] == "_1"
      assert "for" not in req.forwarded[1]
      assert "for" not in req.forwarded[2]
      assert req.forwarded[3]["by"] == "_4"
  
  
  def test_multiple_forwarded_headers_injection() -> None:
      headers = CIMultiDict()
      # 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_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_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: Any) -> None:
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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: Any) -> None:
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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(HTTPRequestEntityTooLarge) as err:
          await req.read()
  
      assert err.value.status_code == 413
  
  
  async def test_request_with_wrong_content_type_encoding(protocol: Any) -> None:
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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(HTTPUnsupportedMediaType) as err:
          await req.text()
      assert err.value.status_code == 415
  
  
  async def test_make_too_big_request_same_size_to_max(protocol: Any) -> None:
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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: Any) -> None:
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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: Any) -> None:
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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_multipart_formdata_file(protocol: Any) -> None:
      # Make sure file uploads work, even without a content type
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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_make_too_big_request_limit_None(protocol: Any) -> None:
      payload = StreamReader(protocol, 2 ** 16, loop=asyncio.get_event_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 = None
      req = make_mocked_request("POST", "/", payload=payload, client_max_size=max_size)
      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", "/")
      req2 = req.clone(scheme="https")
      assert req2.scheme == "https"
  
  
  def test_clone_host() -> None:
      req = make_mocked_request("GET", "/")
      req2 = req.clone(host="example.com")
      assert req2.host == "example.com"
  
  
  def test_clone_remote() -> None:
      req = make_mocked_request("GET", "/")
      req2 = req.clone(remote="11.11.11.11")
      assert req2.remote == "11.11.11.11"
  
  
  @pytest.mark.skipif(not DEBUG, reason="The check is applied in DEBUG mode only")
  def test_request_custom_attr() -> None:
      req = make_mocked_request("GET", "/")
      with pytest.warns(DeprecationWarning):
          req.custom = None
  
  
  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
      assert req.remote == "10.10.10.10"
  
  
  def test_url_http_with_closed_transport() -> None:
      req = make_mocked_request("GET", "/")
      req._protocol = None
      assert str(req.url).startswith("http://")
  
  
  def test_url_https_with_closed_transport() -> None:
      req = make_mocked_request("GET", "/", sslcontext=True)
      req._protocol = None
      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: Any = None):
          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)
  
      req_extra_info = req.get_extra_info(valid_key, default_value)
      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
  
      req._protocol = 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: Any) -> None:
      async def handler(request):
          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: Any) -> None:
      async def handler(request):
          body_text = await request.text()
          assert body_text == '{"some": "data"}'
          assert request.headers["Content-Type"] == "text/plain"
          await request.json()  # raises HTTP 400
          return web.Response()
  
      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.xfail(
      raises=ServerDisconnectedError,
      reason="see https://github.com/aio-libs/aiohttp/issues/4572",
  )
  async def test_handler_return_type(aiohttp_client: Any) -> None:
      async def invalid_handler_1(request):
          return 1
  
      app = web.Application()
      app.router.add_get("/1", invalid_handler_1)
      client = await aiohttp_client(app)
  
      async with client.get("/1") as resp:
          assert 500 == resp.status
  
  
  @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, header_attr, header_val, expected) -> None:
      req = make_mocked_request("GET", "/", headers={header: header_val})
      assert getattr(req, header_attr) == expected
  