"""Base classes for HTTP download handler tests."""

from __future__ import annotations

import gzip
import json
import sys
from abc import ABC, abstractmethod
from contextlib import asynccontextmanager
from http import HTTPStatus
from typing import TYPE_CHECKING, Any
from unittest import mock

import pytest
from testfixtures import LogCapture
from twisted.internet import defer, error
from twisted.web._newclient import ResponseFailed
from twisted.web.http import _DataLoss

from scrapy.http import Headers, HtmlResponse, Request, Response, TextResponse
from scrapy.utils.asyncio import call_later
from scrapy.utils.defer import (
    deferred_f_from_coro_f,
    deferred_from_coro,
    maybe_deferred_to_future,
)
from scrapy.utils.misc import build_from_crawler
from scrapy.utils.spider import DefaultSpider
from scrapy.utils.test import get_crawler
from tests import NON_EXISTING_RESOLVABLE
from tests.mockserver.proxy_echo import ProxyEchoMockServer
from tests.mockserver.simple_https import SimpleMockServer
from tests.spiders import SingleRequestSpider

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator, Generator

    from scrapy.core.downloader.handlers import DownloadHandlerProtocol
    from tests.mockserver.http import MockServer


class TestHttpBase(ABC):
    is_secure = False

    @property
    @abstractmethod
    def download_handler_cls(self) -> type[DownloadHandlerProtocol]:
        raise NotImplementedError

    @asynccontextmanager
    async def get_dh(
        self, settings_dict: dict[str, Any] | None = None
    ) -> AsyncGenerator[DownloadHandlerProtocol]:
        crawler = get_crawler(DefaultSpider, settings_dict)
        crawler.spider = crawler._create_spider()
        dh = build_from_crawler(self.download_handler_cls, crawler)
        try:
            yield dh
        finally:
            await dh.close()

    @deferred_f_from_coro_f
    async def test_download(self, mockserver: MockServer) -> None:
        request = Request(mockserver.url("/text", is_secure=self.is_secure))
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b"Works"

    @deferred_f_from_coro_f
    async def test_download_head(self, mockserver: MockServer) -> None:
        request = Request(
            mockserver.url("/text", is_secure=self.is_secure), method="HEAD"
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b""

    @pytest.mark.parametrize(
        "http_status",
        [
            pytest.param(http_status, id=f"status={http_status.value}")
            for http_status in HTTPStatus
            if http_status.value == 200 or http_status.value // 100 in (4, 5)
        ],
    )
    @deferred_f_from_coro_f
    async def test_download_has_correct_http_status_code(
        self, mockserver: MockServer, http_status: HTTPStatus
    ) -> None:
        request = Request(
            mockserver.url(f"/status?n={http_status.value}", is_secure=self.is_secure)
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == http_status.value

    @deferred_f_from_coro_f
    async def test_server_receives_correct_request_headers(
        self, mockserver: MockServer
    ) -> None:
        request_headers = {
            # common request headers
            "Accept": "text/html",
            "Accept-Charset": "utf-8",
            "Accept-Datetime": "Thu, 31 May 2007 20:35:00 GMT",
            "Accept-Encoding": "gzip, deflate",
            # custom headers
            "X-Custom-Header": "Custom Value",
        }

        request = Request(
            mockserver.url("/echo", is_secure=self.is_secure),
            headers=request_headers,
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == HTTPStatus.OK
        body = json.loads(response.body.decode("utf-8"))
        assert "headers" in body
        for header_name, header_value in request_headers.items():
            assert header_name in body["headers"]
            assert body["headers"][header_name] == [header_value]

    @deferred_f_from_coro_f
    async def test_server_receives_correct_request_body(
        self, mockserver: MockServer
    ) -> None:
        request_body = {
            "message": "It works!",
        }
        request = Request(
            mockserver.url("/echo", is_secure=self.is_secure),
            body=json.dumps(request_body),
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == HTTPStatus.OK
        body = json.loads(response.body.decode("utf-8"))
        assert json.loads(body["body"]) == request_body

    @deferred_f_from_coro_f
    async def test_download_has_correct_response_headers(
        self, mockserver: MockServer
    ) -> None:
        # these headers will be set on the response in the resource and returned
        response_headers = {
            # common response headers
            "Access-Control-Allow-Origin": "*",
            "Allow": "Get, Head",
            "Age": "12",
            "Cache-Control": "max-age=3600",
            "Content-Encoding": "gzip",
            "Content-MD5": "Q2hlY2sgSW50ZWdyaXR5IQ==",
            "Content-Type": "text/html; charset=utf-8",
            "Date": "Date: Tue, 15 Nov 1994 08:12:31 GMT",
            "Pragma": "no-cache",
            "Retry-After": "120",
            "Set-Cookie": "CookieName=CookieValue; Max-Age=3600; Version=1",
            "WWW-Authenticate": "Basic",
            # custom headers
            "X-Custom-Header": "Custom Header Value",
        }

        request = Request(
            mockserver.url("/response-headers", is_secure=self.is_secure),
            headers={"content-type": "application/json"},
            body=json.dumps(response_headers),
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == 200
        for header_name, header_value in response_headers.items():
            assert header_name in response.headers, (
                f"Response was missing expected header {header_name}"
            )
            assert response.headers[header_name] == bytes(
                header_value, encoding="utf-8"
            )

    @deferred_f_from_coro_f
    async def test_redirect_status(self, mockserver: MockServer) -> None:
        request = Request(mockserver.url("/redirect", is_secure=self.is_secure))
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == 302

    @deferred_f_from_coro_f
    async def test_redirect_status_head(self, mockserver: MockServer) -> None:
        request = Request(
            mockserver.url("/redirect", is_secure=self.is_secure), method="HEAD"
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == 302

    @deferred_f_from_coro_f
    async def test_timeout_download_from_spider_nodata_rcvd(
        self, mockserver: MockServer, reactor_pytest: str
    ) -> None:
        if reactor_pytest == "asyncio" and sys.platform == "win32":
            # https://twistedmatrix.com/trac/ticket/10279
            pytest.skip(
                "This test produces DirtyReactorAggregateError on Windows with asyncio"
            )

        # client connects but no data is received
        meta = {"download_timeout": 0.5}
        request = Request(mockserver.url("/wait", is_secure=self.is_secure), meta=meta)
        async with self.get_dh() as download_handler:
            d = deferred_from_coro(download_handler.download_request(request))
            with pytest.raises((defer.TimeoutError, error.TimeoutError)):
                await maybe_deferred_to_future(d)

    @deferred_f_from_coro_f
    async def test_timeout_download_from_spider_server_hangs(
        self,
        mockserver: MockServer,
        reactor_pytest: str,
    ) -> None:
        if reactor_pytest == "asyncio" and sys.platform == "win32":
            # https://twistedmatrix.com/trac/ticket/10279
            pytest.skip(
                "This test produces DirtyReactorAggregateError on Windows with asyncio"
            )
        # client connects, server send headers and some body bytes but hangs
        meta = {"download_timeout": 0.5}
        request = Request(
            mockserver.url("/hang-after-headers", is_secure=self.is_secure), meta=meta
        )
        async with self.get_dh() as download_handler:
            d = deferred_from_coro(download_handler.download_request(request))
            with pytest.raises((defer.TimeoutError, error.TimeoutError)):
                await maybe_deferred_to_future(d)

    @pytest.mark.parametrize("send_header", [True, False])
    @deferred_f_from_coro_f
    async def test_host_header(self, send_header: bool, mockserver: MockServer) -> None:
        host_port = f"{mockserver.host}:{mockserver.port(is_secure=self.is_secure)}"
        request = Request(
            mockserver.url("/host", is_secure=self.is_secure),
            headers={"Host": host_port} if send_header else {},
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == host_port.encode()
        if send_header:
            assert request.headers.get("Host") == host_port.encode()
        else:
            assert not request.headers

    @deferred_f_from_coro_f
    async def test_content_length_zero_bodyless_post_request_headers(
        self, mockserver: MockServer
    ) -> None:
        """Tests if "Content-Length: 0" is sent for bodyless POST requests.

        This is not strictly required by HTTP RFCs but can cause trouble
        for some web servers.
        See:
        https://github.com/scrapy/scrapy/issues/823
        https://issues.apache.org/jira/browse/TS-2902
        https://github.com/kennethreitz/requests/issues/405
        https://bugs.python.org/issue14721
        """
        request = Request(
            mockserver.url("/contentlength", is_secure=self.is_secure), method="POST"
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b"0"

    @deferred_f_from_coro_f
    async def test_content_length_zero_bodyless_post_only_one(
        self, mockserver: MockServer
    ) -> None:
        request = Request(
            mockserver.url("/echo", is_secure=self.is_secure), method="POST"
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        headers = Headers(json.loads(response.text)["headers"])
        contentlengths = headers.getlist("Content-Length")
        assert len(contentlengths) == 1
        assert contentlengths == [b"0"]

    @deferred_f_from_coro_f
    async def test_payload(self, mockserver: MockServer) -> None:
        body = b"1" * 100  # PayloadResource requires body length to be 100
        request = Request(
            mockserver.url("/payload", is_secure=self.is_secure),
            method="POST",
            body=body,
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == body

    @deferred_f_from_coro_f
    async def test_response_header_content_length(self, mockserver: MockServer) -> None:
        request = Request(
            mockserver.url("/text", is_secure=self.is_secure), method="GET"
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.headers[b"content-length"] == b"5"

    @pytest.mark.parametrize(
        ("filename", "body", "response_class"),
        [
            ("foo.html", b"", HtmlResponse),
            ("foo", b"<!DOCTYPE html>\n<title>.</title>", HtmlResponse),
        ],
    )
    @deferred_f_from_coro_f
    async def test_response_class(
        self,
        filename: str,
        body: bytes,
        response_class: type[Response],
        mockserver: MockServer,
    ) -> None:
        request = Request(
            mockserver.url(f"/{filename}", is_secure=self.is_secure), body=body
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert type(response) is response_class  # pylint: disable=unidiomatic-typecheck

    @deferred_f_from_coro_f
    async def test_get_duplicate_header(self, mockserver: MockServer) -> None:
        request = Request(mockserver.url("/duplicate-header", is_secure=self.is_secure))
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.headers.getlist(b"Set-Cookie") == [b"a=b", b"c=d"]

    @deferred_f_from_coro_f
    async def test_download_is_not_automatically_gzip_decoded(
        self, mockserver: MockServer
    ) -> None:
        """Test download handler does not automatically decode content using the scheme provided in Content-Encoding header"""

        data = "compress-me"

        # send a request to mock resource that gzip encodes the "data" url parameter
        request = Request(
            mockserver.url(f"/compress?data={data}", is_secure=self.is_secure),
            headers={
                "accept-encoding": "gzip",
            },
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)

        assert response.status == 200

        # check that the Content-Encoding header is gzip
        content_encoding = response.headers[b"Content-Encoding"]
        assert content_encoding == b"gzip"

        # check that the response is still encoded
        # by checking for the magic number that is always included at the start of a gzip encoding
        # see https://datatracker.ietf.org/doc/html/rfc1952#page-5 section 2.3.1
        GZIP_MAGIC = b"\x1f\x8b"
        assert response.body[:2] == GZIP_MAGIC, "Response body was not in gzip format"

        # check that a gzip decoding matches the data sent in the request
        expected_decoding = bytes(data, encoding="utf-8")
        assert gzip.decompress(response.body) == expected_decoding

    @deferred_f_from_coro_f
    async def test_no_cookie_processing_or_persistence(
        self, mockserver: MockServer
    ) -> None:
        cookie_name = "foo"
        cookie_value = "bar"

        # check that cookies are not modified
        request = Request(
            mockserver.url(
                f"/set-cookie?{cookie_name}={cookie_value}", is_secure=self.is_secure
            )
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
            assert response.status == 200
            set_cookie = response.headers.get(b"Set-Cookie")
            assert set_cookie == f"{cookie_name}={cookie_value}".encode()

            # check that cookies are not sent in the next request
            request = Request(mockserver.url("/echo", is_secure=self.is_secure))
            response = await download_handler.download_request(request)
            assert response.status == 200
            headers = Headers(json.loads(response.text)["headers"])
            assert "Cookie" not in headers
            assert "cookie" not in headers


class TestHttp11Base(TestHttpBase):
    """HTTP 1.1 test case"""

    @deferred_f_from_coro_f
    async def test_download_without_maxsize_limit(self, mockserver: MockServer) -> None:
        request = Request(mockserver.url("/text", is_secure=self.is_secure))
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b"Works"

    @deferred_f_from_coro_f
    async def test_response_class_choosing_request(
        self, mockserver: MockServer
    ) -> None:
        """Tests choosing of correct response type
        in case of Content-Type is empty but body contains text.
        """
        body = b"Some plain text\ndata with tabs\t and null bytes\0"
        request = Request(
            mockserver.url("/nocontenttype", is_secure=self.is_secure), body=body
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert type(response) is TextResponse  # pylint: disable=unidiomatic-typecheck

    @deferred_f_from_coro_f
    async def test_download_with_maxsize(self, mockserver: MockServer) -> None:
        request = Request(mockserver.url("/text", is_secure=self.is_secure))

        # 10 is minimal size for this request and the limit is only counted on
        # response body. (regardless of headers)
        async with self.get_dh({"DOWNLOAD_MAXSIZE": 5}) as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b"Works"

        async with self.get_dh({"DOWNLOAD_MAXSIZE": 4}) as download_handler:
            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
                await download_handler.download_request(request)

    @deferred_f_from_coro_f
    async def test_download_with_maxsize_very_large_file(
        self, mockserver: MockServer
    ) -> None:
        # TODO: the logger check is specific to scrapy.core.downloader.handlers.http11
        with mock.patch("scrapy.core.downloader.handlers.http11.logger") as logger:
            request = Request(
                mockserver.url("/largechunkedfile", is_secure=self.is_secure)
            )

            def check(logger: mock.Mock) -> None:
                logger.warning.assert_called_once_with(mock.ANY, mock.ANY)

            async with self.get_dh({"DOWNLOAD_MAXSIZE": 1_500}) as download_handler:
                with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
                    await download_handler.download_request(request)

            # As the error message is logged in the dataReceived callback, we
            # have to give a bit of time to the reactor to process the queue
            # after closing the connection.
            d: defer.Deferred[mock.Mock] = defer.Deferred()
            d.addCallback(check)
            call_later(0.1, d.callback, logger)
            await maybe_deferred_to_future(d)

    @deferred_f_from_coro_f
    async def test_download_with_maxsize_per_req(self, mockserver: MockServer) -> None:
        meta = {"download_maxsize": 2}
        request = Request(mockserver.url("/text", is_secure=self.is_secure), meta=meta)
        async with self.get_dh() as download_handler:
            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
                await download_handler.download_request(request)

    @deferred_f_from_coro_f
    async def test_download_with_small_maxsize_via_setting(
        self, mockserver: MockServer
    ) -> None:
        request = Request(mockserver.url("/text", is_secure=self.is_secure))
        async with self.get_dh({"DOWNLOAD_MAXSIZE": 2}) as download_handler:
            with pytest.raises((defer.CancelledError, error.ConnectionAborted)):
                await download_handler.download_request(request)

    @deferred_f_from_coro_f
    async def test_download_with_large_maxsize_via_setting(
        self, mockserver: MockServer
    ) -> None:
        request = Request(mockserver.url("/text", is_secure=self.is_secure))
        async with self.get_dh({"DOWNLOAD_MAXSIZE": 100}) as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b"Works"

    @deferred_f_from_coro_f
    async def test_download_chunked_content(self, mockserver: MockServer) -> None:
        request = Request(mockserver.url("/chunked", is_secure=self.is_secure))
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b"chunked content\n"

    @pytest.mark.parametrize("url", ["broken", "broken-chunked"])
    @deferred_f_from_coro_f
    async def test_download_cause_data_loss(
        self, url: str, mockserver: MockServer
    ) -> None:
        # TODO: this one checks for Twisted-specific exceptions
        request = Request(mockserver.url(f"/{url}", is_secure=self.is_secure))
        async with self.get_dh() as download_handler:
            with pytest.raises(ResponseFailed) as exc_info:
                await download_handler.download_request(request)
        assert any(r.check(_DataLoss) for r in exc_info.value.reasons)

    @pytest.mark.parametrize("url", ["broken", "broken-chunked"])
    @deferred_f_from_coro_f
    async def test_download_allow_data_loss(
        self, url: str, mockserver: MockServer
    ) -> None:
        request = Request(
            mockserver.url(f"/{url}", is_secure=self.is_secure),
            meta={"download_fail_on_dataloss": False},
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.flags == ["dataloss"]

    @pytest.mark.parametrize("url", ["broken", "broken-chunked"])
    @deferred_f_from_coro_f
    async def test_download_allow_data_loss_via_setting(
        self, url: str, mockserver: MockServer
    ) -> None:
        request = Request(mockserver.url(f"/{url}", is_secure=self.is_secure))
        async with self.get_dh(
            {"DOWNLOAD_FAIL_ON_DATALOSS": False}
        ) as download_handler:
            response = await download_handler.download_request(request)
        assert response.flags == ["dataloss"]

    @deferred_f_from_coro_f
    async def test_protocol(self, mockserver: MockServer) -> None:
        request = Request(
            mockserver.url("/host", is_secure=self.is_secure), method="GET"
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.protocol == "HTTP/1.1"


class TestHttps11Base(TestHttp11Base):
    is_secure = True

    tls_log_message = (
        'SSL connection certificate: issuer "/C=IE/O=Scrapy/CN=localhost", '
        'subject "/C=IE/O=Scrapy/CN=localhost"'
    )

    @deferred_f_from_coro_f
    async def test_tls_logging(self, mockserver: MockServer) -> None:
        request = Request(mockserver.url("/text", is_secure=self.is_secure))
        async with self.get_dh(
            {"DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING": True}
        ) as download_handler:
            with LogCapture() as log_capture:
                response = await download_handler.download_request(request)
        assert response.body == b"Works"
        log_capture.check_present(
            ("scrapy.core.downloader.tls", "DEBUG", self.tls_log_message)
        )


class TestSimpleHttpsBase(ABC):
    """Base class for special cases tested with just one simple request"""

    keyfile = "keys/localhost.key"
    certfile = "keys/localhost.crt"
    host = "localhost"
    cipher_string: str | None = None

    @pytest.fixture(scope="class")
    def simple_mockserver(self) -> Generator[SimpleMockServer]:
        with SimpleMockServer(
            self.keyfile, self.certfile, self.cipher_string
        ) as simple_mockserver:
            yield simple_mockserver

    @pytest.fixture(scope="class")
    def url(self, simple_mockserver: SimpleMockServer) -> str:
        # need to use self.host instead of what mockserver returns
        return f"https://{self.host}:{simple_mockserver.port(is_secure=True)}/file"

    @property
    @abstractmethod
    def download_handler_cls(self) -> type[DownloadHandlerProtocol]:
        raise NotImplementedError

    @asynccontextmanager
    async def get_dh(self) -> AsyncGenerator[DownloadHandlerProtocol]:
        if self.cipher_string is not None:
            settings_dict = {"DOWNLOADER_CLIENT_TLS_CIPHERS": self.cipher_string}
        else:
            settings_dict = None
        crawler = get_crawler(DefaultSpider, settings_dict=settings_dict)
        crawler.spider = crawler._create_spider()
        dh = build_from_crawler(self.download_handler_cls, crawler)
        try:
            yield dh
        finally:
            await dh.close()

    @deferred_f_from_coro_f
    async def test_download(self, url: str) -> None:
        request = Request(url)
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.body == b"0123456789"


class TestHttpsWrongHostnameBase(TestSimpleHttpsBase):
    # above tests use a server certificate for "localhost",
    # client connection to "localhost" too.
    # here we test that even if the server certificate is for another domain,
    # "www.example.com" in this case,
    # the tests still pass
    keyfile = "keys/example-com.key.pem"
    certfile = "keys/example-com.cert.pem"


class TestHttpsInvalidDNSIdBase(TestSimpleHttpsBase):
    """Connect to HTTPS hosts with IP while certificate uses domain names IDs."""

    host = "127.0.0.1"


class TestHttpsInvalidDNSPatternBase(TestSimpleHttpsBase):
    """Connect to HTTPS hosts where the certificate are issued to an ip instead of a domain."""

    keyfile = "keys/localhost.ip.key"
    certfile = "keys/localhost.ip.crt"


class TestHttpsCustomCiphersBase(TestSimpleHttpsBase):
    cipher_string = "CAMELLIA256-SHA"


class TestHttpWithCrawlerBase(ABC):
    @property
    @abstractmethod
    def settings_dict(self) -> dict[str, Any] | None:
        raise NotImplementedError

    is_secure = False

    @deferred_f_from_coro_f
    async def test_download_with_content_length(self, mockserver: MockServer) -> None:
        crawler = get_crawler(SingleRequestSpider, self.settings_dict)
        # http://localhost:8998/partial set Content-Length to 1024, use download_maxsize= 1000 to avoid
        # download it
        await maybe_deferred_to_future(
            crawler.crawl(
                seed=Request(
                    url=mockserver.url("/partial", is_secure=self.is_secure),
                    meta={"download_maxsize": 1000},
                )
            )
        )
        assert crawler.spider
        failure = crawler.spider.meta["failure"]  # type: ignore[attr-defined]
        assert isinstance(failure.value, defer.CancelledError)

    @deferred_f_from_coro_f
    async def test_download(self, mockserver: MockServer) -> None:
        crawler = get_crawler(SingleRequestSpider, self.settings_dict)
        await maybe_deferred_to_future(
            crawler.crawl(
                seed=Request(url=mockserver.url("", is_secure=self.is_secure))
            )
        )
        assert crawler.spider
        failure = crawler.spider.meta.get("failure")  # type: ignore[attr-defined]
        assert failure is None
        reason = crawler.spider.meta["close_reason"]  # type: ignore[attr-defined]
        assert reason == "finished"


class TestHttpProxyBase(ABC):
    is_secure = False
    expected_http_proxy_request_body = b"http://example.com"

    @property
    @abstractmethod
    def download_handler_cls(self) -> type[DownloadHandlerProtocol]:
        raise NotImplementedError

    @pytest.fixture(scope="session")
    def proxy_mockserver(self) -> Generator[ProxyEchoMockServer]:
        with ProxyEchoMockServer() as proxy:
            yield proxy

    @asynccontextmanager
    async def get_dh(self) -> AsyncGenerator[DownloadHandlerProtocol]:
        crawler = get_crawler(DefaultSpider)
        crawler.spider = crawler._create_spider()
        dh = build_from_crawler(self.download_handler_cls, crawler)
        try:
            yield dh
        finally:
            await dh.close()

    @deferred_f_from_coro_f
    async def test_download_with_proxy(
        self, proxy_mockserver: ProxyEchoMockServer
    ) -> None:
        http_proxy = proxy_mockserver.url("", is_secure=self.is_secure)
        request = Request("http://example.com", meta={"proxy": http_proxy})
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == 200
        assert response.url == request.url
        assert response.body == self.expected_http_proxy_request_body

    @deferred_f_from_coro_f
    async def test_download_without_proxy(
        self, proxy_mockserver: ProxyEchoMockServer
    ) -> None:
        request = Request(
            proxy_mockserver.url("/path/to/resource", is_secure=self.is_secure)
        )
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == 200
        assert response.url == request.url
        assert response.body == b"/path/to/resource"

    @deferred_f_from_coro_f
    async def test_download_with_proxy_https_timeout(
        self, proxy_mockserver: ProxyEchoMockServer
    ) -> None:
        if NON_EXISTING_RESOLVABLE:
            pytest.skip("Non-existing hosts are resolvable")
        http_proxy = proxy_mockserver.url("", is_secure=self.is_secure)
        domain = "https://no-such-domain.nosuch"
        request = Request(domain, meta={"proxy": http_proxy, "download_timeout": 0.2})
        async with self.get_dh() as download_handler:
            with pytest.raises(error.TimeoutError) as exc_info:
                await download_handler.download_request(request)
        assert domain in exc_info.value.osError

    @deferred_f_from_coro_f
    async def test_download_with_proxy_without_http_scheme(
        self, proxy_mockserver: ProxyEchoMockServer
    ) -> None:
        http_proxy = f"{proxy_mockserver.host}:{proxy_mockserver.port()}"
        request = Request("http://example.com", meta={"proxy": http_proxy})
        async with self.get_dh() as download_handler:
            response = await download_handler.download_request(request)
        assert response.status == 200
        assert response.url == request.url
        assert response.body == self.expected_http_proxy_request_body