Repository: scrapy/scrapy
Pull request #6363 (2024-05-14 14:05:35)

Merge 2.11.2

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy) Report
Attention: Patch coverage is `95.83333%` with `4 lines` in your changes are missing coverage. Please review.
> Project coverage is 84.93%. Comparing base [(`180bc9b`)](https://app.codecov.io/gh/scrapy/scrapy/commit/180bc9bad7aceb3a9e10c1411212914bc32fb721?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy) to head [(`566cb38`)](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy).
> Report is 10 commits behind head on master.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #6363      +/-   ##
==========================================
- Coverage   85.00%   84.93%   -0.07%     
==========================================
  Files         161      162       +1     
  Lines       11962    12101     +139     
  Branches     1872     1835      -37     
==========================================
+ Hits        10168    10278     +110     
- Misses       1512     1534      +22     
- Partials      282      289       +7     
```

| [Files](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy) | Coverage Δ | |
|---|---|---|
| [scrapy/core/engine.py](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?src=pr&el=tree&filepath=scrapy%2Fcore%2Fengine.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy#diff-c2NyYXB5L2NvcmUvZW5naW5lLnB5) | `86.94% <100.00%> (+0.24%)` | :arrow_up: |
| [scrapy/downloadermiddlewares/httpproxy.py](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?src=pr&el=tree&filepath=scrapy%2Fdownloadermiddlewares%2Fhttpproxy.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy#diff-c2NyYXB5L2Rvd25sb2FkZXJtaWRkbGV3YXJlcy9odHRwcHJveHkucHk=) | `97.05% <100.00%> (+0.13%)` | :arrow_up: |
| [scrapy/downloadermiddlewares/redirect.py](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?src=pr&el=tree&filepath=scrapy%2Fdownloadermiddlewares%2Fredirect.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy#diff-c2NyYXB5L2Rvd25sb2FkZXJtaWRkbGV3YXJlcy9yZWRpcmVjdC5weQ==) | `98.05% <100.00%> (+1.50%)` | :arrow_up: |
| [scrapy/settings/default\_settings.py](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?src=pr&el=tree&filepath=scrapy%2Fsettings%2Fdefault_settings.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy#diff-c2NyYXB5L3NldHRpbmdzL2RlZmF1bHRfc2V0dGluZ3MucHk=) | `98.80% <ø> (ø)` | |
| [scrapy/spidermiddlewares/offsite.py](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?src=pr&el=tree&filepath=scrapy%2Fspidermiddlewares%2Foffsite.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy#diff-c2NyYXB5L3NwaWRlcm1pZGRsZXdhcmVzL29mZnNpdGUucHk=) | `92.85% <100.00%> (-5.68%)` | :arrow_down: |
| [scrapy/utils/python.py](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?src=pr&el=tree&filepath=scrapy%2Futils%2Fpython.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy#diff-c2NyYXB5L3V0aWxzL3B5dGhvbi5weQ==) | `86.87% <100.00%> (-1.03%)` | :arrow_down: |
| [scrapy/downloadermiddlewares/offsite.py](https://app.codecov.io/gh/scrapy/scrapy/pull/6363?src=pr&el=tree&filepath=scrapy%2Fdownloadermiddlewares%2Foffsite.py&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy#diff-c2NyYXB5L2Rvd25sb2FkZXJtaWRkbGV3YXJlcy9vZmZzaXRlLnB5) | `92.59% <92.59%> (ø)` | |

... and [9 files with indirect coverage changes](https://app.codecov.io/gh/scrapy/scrapy/pull/6363/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=scrapy)

</details>

Patch:
diff --git a/.bumpversion.cfg b/.bumpversion.cfg
index 968a34d96..599cd0cff 100644
--- a/.bumpversion.cfg
+++ b/.bumpversion.cfg
@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 2.11.1
+current_version = 2.11.2
 commit = True
 tag = True
 tag_name = {new_version}
diff --git a/docs/faq.rst b/docs/faq.rst
index 7090f0bcd..d394406e8 100644
--- a/docs/faq.rst
+++ b/docs/faq.rst
@@ -138,39 +138,37 @@ See previous question.
 How can I prevent memory errors due to many allowed domains?
 ------------------------------------------------------------
 
-If you have a spider with a long list of
-:attr:`~scrapy.Spider.allowed_domains` (e.g. 50,000+), consider
-replacing the default
-:class:`~scrapy.spidermiddlewares.offsite.OffsiteMiddleware` spider middleware
-with a :ref:`custom spider middleware <custom-spider-middleware>` that requires
-less memory. For example:
+If you have a spider with a long list of :attr:`~scrapy.Spider.allowed_domains`
+(e.g. 50,000+), consider replacing the default
+:class:`~scrapy.downloadermiddlewares.offsite.OffsiteMiddleware` downloader
+middleware with a :ref:`custom downloader middleware
+<topics-downloader-middleware-custom>` that requires less memory. For example:
 
 -   If your domain names are similar enough, use your own regular expression
-    instead joining the strings in
-    :attr:`~scrapy.Spider.allowed_domains` into a complex regular
-    expression.
+    instead joining the strings in :attr:`~scrapy.Spider.allowed_domains` into
+    a complex regular expression.
 
 -   If you can `meet the installation requirements`_, use pyre2_ instead of
     Python’s re_ to compile your URL-filtering regular expression. See
     :issue:`1908`.
 
-See also other suggestions at `StackOverflow`_.
+See also `other suggestions at StackOverflow
+<https://stackoverflow.com/q/36440681>`__.
 
 .. note:: Remember to disable
-   :class:`scrapy.spidermiddlewares.offsite.OffsiteMiddleware` when you enable
-   your custom implementation:
+   :class:`scrapy.downloadermiddlewares.offsite.OffsiteMiddleware` when you
+   enable your custom implementation:
 
    .. code-block:: python
 
-       SPIDER_MIDDLEWARES = {
-           "scrapy.spidermiddlewares.offsite.OffsiteMiddleware": None,
-           "myproject.middlewares.CustomOffsiteMiddleware": 500,
+       DOWNLOADER_MIDDLEWARES = {
+           "scrapy.downloadermiddlewares.offsite.OffsiteMiddleware": None,
+           "myproject.middlewares.CustomOffsiteMiddleware": 50,
        }
 
 .. _meet the installation requirements: https://github.com/andreasvc/pyre2#installation
 .. _pyre2: https://github.com/andreasvc/pyre2
 .. _re: https://docs.python.org/library/re.html
-.. _StackOverflow: https://stackoverflow.com/q/36440681/939364
 
 Can I use Basic HTTP Authentication in my spiders?
 --------------------------------------------------
@@ -206,12 +204,10 @@ I get "Filtered offsite request" messages. How can I fix them?
 Those messages (logged with ``DEBUG`` level) don't necessarily mean there is a
 problem, so you may not need to fix them.
 
-Those messages are thrown by the Offsite Spider Middleware, which is a spider
-middleware (enabled by default) whose purpose is to filter out requests to
-domains outside the ones covered by the spider.
-
-For more info see:
-:class:`~scrapy.spidermiddlewares.offsite.OffsiteMiddleware`.
+Those messages are thrown by
+:class:`~scrapy.downloadermiddlewares.offsite.OffsiteMiddleware`, which is a
+downloader middleware (enabled by default) whose purpose is to filter out
+requests to domains outside the ones covered by the spider.
 
 What is the recommended way to deploy a Scrapy crawler in production?
 ---------------------------------------------------------------------
diff --git a/docs/news.rst b/docs/news.rst
index 7db4e59a1..758b22d80 100644
--- a/docs/news.rst
+++ b/docs/news.rst
@@ -3,7 +3,6 @@
 Release notes
 =============
 
-
 .. _release-VERSION:
 
 Scrapy VERSION (YYYY-MM-DD)
@@ -12,11 +11,122 @@ Scrapy VERSION (YYYY-MM-DD)
 Deprecations
 ~~~~~~~~~~~~
 
--   :func:`scrapy.core.downloader.Downloader._get_slot_key` is now deprecated.
-    Consider using its corresponding public method get_slot_key() instead.
+-   :meth:`scrapy.core.downloader.Downloader._get_slot_key` is deprecated, use
+    :meth:`scrapy.core.downloader.Downloader.get_slot_key` instead.
     (:issue:`6340`)
 
 
+.. _release-2.11.2:
+
+Scrapy 2.11.2 (2024-05-14)
+--------------------------
+
+Security bug fixes
+~~~~~~~~~~~~~~~~~~
+
+-   Redirects to non-HTTP protocols are no longer followed. Please, see the
+    `23j4-mw76-5v7h security advisory`_ for more information. (:issue:`457`)
+
+    .. _23j4-mw76-5v7h security advisory: https://github.com/scrapy/scrapy/security/advisories/GHSA-23j4-mw76-5v7h
+
+-   The ``Authorization`` header is now dropped on redirects to a different
+    scheme (``http://`` or ``https://``) or port, even if the domain is the
+    same. Please, see the `4qqq-9vqf-3h3f security advisory`_ for more
+    information.
+
+    .. _4qqq-9vqf-3h3f security advisory: https://github.com/scrapy/scrapy/security/advisories/GHSA-4qqq-9vqf-3h3f
+
+-   When using system proxy settings that are different for ``http://`` and
+    ``https://``, redirects to a different URL scheme will now also trigger the
+    corresponding change in proxy settings for the redirected request. Please,
+    see the `jm3v-qxmh-hxwv security advisory`_ for more information.
+    (:issue:`767`)
+
+    .. _jm3v-qxmh-hxwv security advisory: https://github.com/scrapy/scrapy/security/advisories/GHSA-jm3v-qxmh-hxwv
+
+-   :attr:`Spider.allowed_domains <scrapy.Spider.allowed_domains>` is now
+    enforced for all requests, and not only requests from spider callbacks.
+    (:issue:`1042`, :issue:`2241`, :issue:`6358`)
+
+-   :func:`~scrapy.utils.iterators.xmliter_lxml` no longer resolves XML
+    entities. (:issue:`6265`)
+
+-   defusedxml_ is now used to make
+    :class:`scrapy.http.request.rpc.XmlRpcRequest` more secure.
+    (:issue:`6250`, :issue:`6251`)
+
+    .. _defusedxml: https://github.com/tiran/defusedxml
+
+Bug fixes
+~~~~~~~~~
+
+-   Restored support for brotlipy_, which had been dropped in Scrapy 2.11.1 in
+    favor of brotli_. (:issue:`6261`)
+
+    .. _brotli: https://github.com/google/brotli
+
+    .. note:: brotlipy is deprecated, both in Scrapy and upstream. Use brotli
+        instead if you can.
+
+-   Make :setting:`METAREFRESH_IGNORE_TAGS` ``["noscript"]`` by default. This
+    prevents
+    :class:`~scrapy.downloadermiddlewares.redirect.MetaRefreshMiddleware` from
+    following redirects that would not be followed by web browsers with
+    JavaScript enabled. (:issue:`6342`, :issue:`6347`)
+
+-   During :ref:`feed export <topics-feed-exports>`, do not close the
+    underlying file from :ref:`built-in post-processing plugins
+    <builtin-plugins>`.
+    (:issue:`5932`, :issue:`6178`, :issue:`6239`)
+
+-   :class:`LinkExtractor <scrapy.linkextractors.lxmlhtml.LxmlLinkExtractor>`
+    now properly applies the ``unique`` and ``canonicalize`` parameters.
+    (:issue:`3273`, :issue:`6221`)
+
+-   Do not initialize the scheduler disk queue if :setting:`JOBDIR` is an empty
+    string. (:issue:`6121`, :issue:`6124`)
+
+-   Fix :attr:`Spider.logger <scrapy.Spider.logger>` not logging custom extra
+    information. (:issue:`6323`, :issue:`6324`)
+
+-   ``robots.txt`` files with a non-UTF-8 encoding no longer prevent parsing
+    the UTF-8-compatible (e.g. ASCII) parts of the document.
+    (:issue:`6292`, :issue:`6298`)
+
+-   :meth:`scrapy.http.cookies.WrappedRequest.get_header` no longer raises an
+    exception if ``default`` is ``None``.
+    (:issue:`6308`, :issue:`6310`)
+
+-   :class:`~scrapy.selector.Selector` now uses
+    :func:`scrapy.utils.response.get_base_url` to determine the base URL of a
+    given :class:`~scrapy.http.Response`. (:issue:`6265`)
+
+-   The :meth:`media_to_download` method of :ref:`media pipelines
+    <topics-media-pipeline>` now logs exceptions before stripping them.
+    (:issue:`5067`, :issue:`5068`)
+
+-   When passing a callback to the :command:`parse` command, build the callback
+    callable with the right signature.
+    (:issue:`6182`)
+
+Documentation
+~~~~~~~~~~~~~
+
+-   Add a FAQ entry about :ref:`creating blank requests <faq-blank-request>`.
+    (:issue:`6203`, :issue:`6208`)
+
+-   Document that :attr:`scrapy.selector.Selector.type` can be ``"json"``.
+    (:issue:`6328`, :issue:`6334`)
+
+Quality assurance
+~~~~~~~~~~~~~~~~~
+
+-   Make builds reproducible. (:issue:`5019`, :issue:`6322`)
+
+-   Packaging and test fixes.
+    (:issue:`6286`, :issue:`6290`, :issue:`6312`, :issue:`6316`, :issue:`6344`)
+
+
 .. _release-2.11.1:
 
 Scrapy 2.11.1 (2024-02-14)
diff --git a/docs/topics/benchmarking.rst b/docs/topics/benchmarking.rst
index 0643df6a6..b704e54ed 100644
--- a/docs/topics/benchmarking.rst
+++ b/docs/topics/benchmarking.rst
@@ -24,7 +24,8 @@ You should see an output like this::
      'scrapy.extensions.telnet.TelnetConsole',
      'scrapy.extensions.corestats.CoreStats']
     2016-12-16 21:18:49 [scrapy.middleware] INFO: Enabled downloader middlewares:
-    ['scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware',
+    ['scrapy.downloadermiddlewares.offsite.OffsiteMiddleware',
+     'scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware',
      'scrapy.downloadermiddlewares.httpauth.HttpAuthMiddleware',
      'scrapy.downloadermiddlewares.downloadtimeout.DownloadTimeoutMiddleware',
      'scrapy.downloadermiddlewares.defaultheaders.DefaultHeadersMiddleware',
@@ -37,7 +38,6 @@ You should see an output like this::
      'scrapy.downloadermiddlewares.stats.DownloaderStats']
     2016-12-16 21:18:49 [scrapy.middleware] INFO: Enabled spider middlewares:
     ['scrapy.spidermiddlewares.httperror.HttpErrorMiddleware',
-     'scrapy.spidermiddlewares.offsite.OffsiteMiddleware',
      'scrapy.spidermiddlewares.referer.RefererMiddleware',
      'scrapy.spidermiddlewares.urllength.UrlLengthMiddleware',
      'scrapy.spidermiddlewares.depth.DepthMiddleware']
diff --git a/docs/topics/downloader-middleware.rst b/docs/topics/downloader-middleware.rst
index d4cd062fe..c31f7fe43 100644
--- a/docs/topics/downloader-middleware.rst
+++ b/docs/topics/downloader-middleware.rst
@@ -763,6 +763,44 @@ HttpProxyMiddleware
    Keep in mind this value will take precedence over ``http_proxy``/``https_proxy``
    environment variables, and it will also ignore ``no_proxy`` environment variable.
 
+OffsiteMiddleware
+-----------------
+
+.. module:: scrapy.downloadermiddlewares.offsite
+   :synopsis: Offsite Middleware
+
+.. class:: OffsiteMiddleware
+
+   .. versionadded:: 2.11.2
+
+   Filters out Requests for URLs outside the domains covered by the spider.
+
+   This middleware filters out every request whose host names aren't in the
+   spider's :attr:`~scrapy.Spider.allowed_domains` attribute.
+   All subdomains of any domain in the list are also allowed.
+   E.g. the rule ``www.example.org`` will also allow ``bob.www.example.org``
+   but not ``www2.example.com`` nor ``example.com``.
+
+   When your spider returns a request for a domain not belonging to those
+   covered by the spider, this middleware will log a debug message similar to
+   this one::
+
+      DEBUG: Filtered offsite request to 'offsite.example': <GET http://offsite.example/some/page.html>
+
+   To avoid filling the log with too much noise, it will only print one of
+   these messages for each new domain filtered. So, for example, if another
+   request for ``offsite.example`` is filtered, no log message will be
+   printed. But if a request for ``other.example`` is filtered, a message
+   will be printed (but only for the first request filtered).
+
+   If the spider doesn't define an
+   :attr:`~scrapy.Spider.allowed_domains` attribute, or the
+   attribute is empty, the offsite middleware will allow all requests.
+
+   If the request has the :attr:`~scrapy.Request.dont_filter` attribute
+   set, the offsite middleware will allow the request even if its domain is not
+   listed in allowed domains.
+
 RedirectMiddleware
 ------------------
 
@@ -882,7 +920,11 @@ Meta tags within these tags are ignored.
 
 .. versionchanged:: 2.0
    The default value of :setting:`METAREFRESH_IGNORE_TAGS` changed from
-   ``['script', 'noscript']`` to ``[]``.
+   ``["script", "noscript"]`` to ``[]``.
+
+.. versionchanged:: 2.11.2
+   The default value of :setting:`METAREFRESH_IGNORE_TAGS` changed from
+   ``[]`` to ``["noscript"]``.
 
 .. versionchanged:: VERSION
    The default value of :setting:`METAREFRESH_IGNORE_TAGS` changed from
diff --git a/docs/topics/settings.rst b/docs/topics/settings.rst
index 2bd9cf1ed..904bd7ecc 100644
--- a/docs/topics/settings.rst
+++ b/docs/topics/settings.rst
@@ -674,6 +674,7 @@ Default:
 .. code-block:: python
 
     {
+        "scrapy.downloadermiddlewares.offsite.OffsiteMiddleware": 50,
         "scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware": 100,
         "scrapy.downloadermiddlewares.httpauth.HttpAuthMiddleware": 300,
         "scrapy.downloadermiddlewares.downloadtimeout.DownloadTimeoutMiddleware": 350,
@@ -1613,7 +1614,6 @@ Default:
 
     {
         "scrapy.spidermiddlewares.httperror.HttpErrorMiddleware": 50,
-        "scrapy.spidermiddlewares.offsite.OffsiteMiddleware": 500,
         "scrapy.spidermiddlewares.referer.RefererMiddleware": 700,
         "scrapy.spidermiddlewares.urllength.UrlLengthMiddleware": 800,
         "scrapy.spidermiddlewares.depth.DepthMiddleware": 900,
diff --git a/docs/topics/signals.rst b/docs/topics/signals.rst
index 9bfd1761c..13e636055 100644
--- a/docs/topics/signals.rst
+++ b/docs/topics/signals.rst
@@ -343,11 +343,18 @@ request_scheduled
 .. signal:: request_scheduled
 .. function:: request_scheduled(request, spider)
 
-    Sent when the engine schedules a :class:`~scrapy.Request`, to be
-    downloaded later.
+    Sent when the engine is asked to schedule a :class:`~scrapy.Request`, to be
+    downloaded later, before the request reaches the :ref:`scheduler
+    <topics-scheduler>`.
+
+    Raise :exc:`~scrapy.exceptions.IgnoreRequest` to drop a request before it
+    reaches the scheduler.
 
     This signal does not support returning deferreds from its handlers.
 
+    .. versionadded:: 2.11.2
+        Allow dropping requests with :exc:`~scrapy.exceptions.IgnoreRequest`.
+
     :param request: the request that reached the scheduler
     :type request: :class:`~scrapy.Request` object
 
diff --git a/docs/topics/spider-middleware.rst b/docs/topics/spider-middleware.rst
index 3f16efea5..8ddf17a14 100644
--- a/docs/topics/spider-middleware.rst
+++ b/docs/topics/spider-middleware.rst
@@ -51,8 +51,8 @@ value.  For example, if you want to disable the off-site middleware:
 .. code-block:: python
 
     SPIDER_MIDDLEWARES = {
-        "myproject.middlewares.CustomSpiderMiddleware": 543,
-        "scrapy.spidermiddlewares.offsite.OffsiteMiddleware": None,
+        "scrapy.spidermiddlewares.referer.RefererMiddleware": None,
+        "myproject.middlewares.CustomRefererSpiderMiddleware": 700,
     }
 
 Finally, keep in mind that some middlewares may need to be enabled through a
@@ -313,42 +313,6 @@ Default: ``False``
 
 Pass all responses, regardless of its status code.
 
-OffsiteMiddleware
------------------
-
-.. module:: scrapy.spidermiddlewares.offsite
-   :synopsis: Offsite Spider Middleware
-
-.. class:: OffsiteMiddleware
-
-   Filters out Requests for URLs outside the domains covered by the spider.
-
-   This middleware filters out every request whose host names aren't in the
-   spider's :attr:`~scrapy.Spider.allowed_domains` attribute.
-   All subdomains of any domain in the list are also allowed.
-   E.g. the rule ``www.example.org`` will also allow ``bob.www.example.org``
-   but not ``www2.example.com`` nor ``example.com``.
-
-   When your spider returns a request for a domain not belonging to those
-   covered by the spider, this middleware will log a debug message similar to
-   this one::
-
-      DEBUG: Filtered offsite request to 'www.othersite.com': <GET http://www.othersite.com/some/page.html>
-
-   To avoid filling the log with too much noise, it will only print one of
-   these messages for each new domain filtered. So, for example, if another
-   request for ``www.othersite.com`` is filtered, no log message will be
-   printed. But if a request for ``someothersite.com`` is filtered, a message
-   will be printed (but only for the first request filtered).
-
-   If the spider doesn't define an
-   :attr:`~scrapy.Spider.allowed_domains` attribute, or the
-   attribute is empty, the offsite middleware will allow all requests.
-
-   If the request has the :attr:`~scrapy.Request.dont_filter` attribute
-   set, the offsite middleware will allow the request even if its domain is not
-   listed in allowed domains.
-
 
 RefererMiddleware
 -----------------
diff --git a/docs/topics/spiders.rst b/docs/topics/spiders.rst
index 30677fe74..8a0102a51 100644
--- a/docs/topics/spiders.rst
+++ b/docs/topics/spiders.rst
@@ -75,7 +75,8 @@ scrapy.Spider
        An optional list of strings containing domains that this spider is
        allowed to crawl. Requests for URLs not belonging to the domain names
        specified in this list (or their subdomains) won't be followed if
-       :class:`~scrapy.spidermiddlewares.offsite.OffsiteMiddleware` is enabled.
+       :class:`~scrapy.downloadermiddlewares.offsite.OffsiteMiddleware` is
+       enabled.
 
        Let's say your target url is ``https://www.example.com/1.html``,
        then add ``'example.com'`` to the list.
diff --git a/scrapy/VERSION b/scrapy/VERSION
index 6ceb272ee..9e5bb77a3 100644
--- a/scrapy/VERSION
+++ b/scrapy/VERSION
@@ -1,1 +1,1 @@
-2.11.1
+2.11.2
diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py
index 6bf3f3e26..4eca03800 100644
--- a/scrapy/core/engine.py
+++ b/scrapy/core/engine.py
@@ -28,7 +28,7 @@ from twisted.python.failure import Failure
 from scrapy import signals
 from scrapy.core.downloader import Downloader
 from scrapy.core.scraper import Scraper
-from scrapy.exceptions import CloseSpider, DontCloseSpider
+from scrapy.exceptions import CloseSpider, DontCloseSpider, IgnoreRequest
 from scrapy.http import Request, Response
 from scrapy.logformatter import LogFormatter
 from scrapy.settings import BaseSettings, Settings
@@ -36,6 +36,7 @@ from scrapy.signalmanager import SignalManager
 from scrapy.spiders import Spider
 from scrapy.utils.log import failure_to_exc_info, logformatter_adapter
 from scrapy.utils.misc import build_from_crawler, load_object
+from scrapy.utils.python import global_object_name
 from scrapy.utils.reactor import CallLaterOnce
 
 if TYPE_CHECKING:
@@ -292,9 +293,19 @@ class ExecutionEngine:
         self.slot.nextcall.schedule()  # type: ignore[union-attr]
 
     def _schedule_request(self, request: Request, spider: Spider) -> None:
-        self.signals.send_catch_log(
-            signals.request_scheduled, request=request, spider=spider
+        request_scheduled_result = self.signals.send_catch_log(
+            signals.request_scheduled,
+            request=request,
+            spider=spider,
+            dont_log=IgnoreRequest,
         )
+        for handler, result in request_scheduled_result:
+            if isinstance(result, Failure) and isinstance(result.value, IgnoreRequest):
+                logger.debug(
+                    f"Signal handler {global_object_name(handler)} dropped "
+                    f"request {request} before it reached the scheduler."
+                )
+                return
         if not self.slot.scheduler.enqueue_request(request):  # type: ignore[union-attr]
             self.signals.send_catch_log(
                 signals.request_dropped, request=request, spider=spider
diff --git a/scrapy/downloadermiddlewares/httpproxy.py b/scrapy/downloadermiddlewares/httpproxy.py
index 335896ac1..5b56ad449 100644
--- a/scrapy/downloadermiddlewares/httpproxy.py
+++ b/scrapy/downloadermiddlewares/httpproxy.py
@@ -60,26 +60,33 @@ class HttpProxyMiddleware:
     def process_request(
         self, request: Request, spider: Spider
     ) -> Union[Request, Response, None]:
-        creds, proxy_url = None, None
+        creds, proxy_url, scheme = None, None, None
         if "proxy" in request.meta:
             if request.meta["proxy"] is not None:
                 creds, proxy_url = self._get_proxy(request.meta["proxy"], "")
         elif self.proxies:
             parsed = urlparse_cached(request)
-            scheme = parsed.scheme
+            _scheme = parsed.scheme
             if (
                 # 'no_proxy' is only supported by http schemes
-                scheme not in ("http", "https")
+                _scheme not in ("http", "https")
                 or (parsed.hostname and not proxy_bypass(parsed.hostname))
-            ) and scheme in self.proxies:
+            ) and _scheme in self.proxies:
+                scheme = _scheme
                 creds, proxy_url = self.proxies[scheme]
 
-        self._set_proxy_and_creds(request, proxy_url, creds)
+        self._set_proxy_and_creds(request, proxy_url, creds, scheme)
         return None
 
     def _set_proxy_and_creds(
-        self, request: Request, proxy_url: Optional[str], creds: Optional[bytes]
+        self,
+        request: Request,
+        proxy_url: Optional[str],
+        creds: Optional[bytes],
+        scheme: Optional[str],
     ) -> None:
+        if scheme:
+            request.meta["_scheme_proxy"] = True
         if proxy_url:
             request.meta["proxy"] = proxy_url
         elif request.meta.get("proxy") is not None:
diff --git a/scrapy/downloadermiddlewares/offsite.py b/scrapy/downloadermiddlewares/offsite.py
new file mode 100644
index 000000000..1e5026925
--- /dev/null
+++ b/scrapy/downloadermiddlewares/offsite.py
@@ -0,0 +1,77 @@
+import logging
+import re
+import warnings
+
+from scrapy import signals
+from scrapy.exceptions import IgnoreRequest
+from scrapy.utils.httpobj import urlparse_cached
+
+logger = logging.getLogger(__name__)
+
+
+class OffsiteMiddleware:
+    @classmethod
+    def from_crawler(cls, crawler):
+        o = cls(crawler.stats)
+        crawler.signals.connect(o.spider_opened, signal=signals.spider_opened)
+        crawler.signals.connect(o.request_scheduled, signal=signals.request_scheduled)
+        return o
+
+    def __init__(self, stats):
+        self.stats = stats
+        self.domains_seen = set()
+
+    def spider_opened(self, spider):
+        self.host_regex = self.get_host_regex(spider)
+
+    def request_scheduled(self, request, spider):
+        self.process_request(request, spider)
+
+    def process_request(self, request, spider):
+        if request.dont_filter or self.should_follow(request, spider):
+            return None
+        domain = urlparse_cached(request).hostname
+        if domain and domain not in self.domains_seen:
+            self.domains_seen.add(domain)
+            logger.debug(
+                "Filtered offsite request to %(domain)r: %(request)s",
+                {"domain": domain, "request": request},
+                extra={"spider": spider},
+            )
+            self.stats.inc_value("offsite/domains", spider=spider)
+        self.stats.inc_value("offsite/filtered", spider=spider)
+        raise IgnoreRequest
+
+    def should_follow(self, request, spider):
+        regex = self.host_regex
+        # hostname can be None for wrong urls (like javascript links)
+        host = urlparse_cached(request).hostname or ""
+        return bool(regex.search(host))
+
+    def get_host_regex(self, spider):
+        """Override this method to implement a different offsite policy"""
+        allowed_domains = getattr(spider, "allowed_domains", None)
+        if not allowed_domains:
+            return re.compile("")  # allow all by default
+        url_pattern = re.compile(r"^https?://.*$")
+        port_pattern = re.compile(r":\d+$")
+        domains = []
+        for domain in allowed_domains:
+            if domain is None:
+                continue
+            if url_pattern.match(domain):
+                message = (
+                    "allowed_domains accepts only domains, not URLs. "
+                    f"Ignoring URL entry {domain} in allowed_domains."
+                )
+                warnings.warn(message)
+            elif port_pattern.search(domain):
+                message = (
+                    "allowed_domains accepts only domains without ports. "
+                    f"Ignoring entry {domain} in allowed_domains."
+                )
+                warnings.warn(message)
+            else:
+                domains.append(re.escape(domain))
+        regex = rf'^(.*\.)?({"|".join(domains)})$'
+        return re.compile(regex)
diff --git a/scrapy/downloadermiddlewares/redirect.py b/scrapy/downloadermiddlewares/redirect.py
index 24089afea..aa08827c4 100644
--- a/scrapy/downloadermiddlewares/redirect.py
+++ b/scrapy/downloadermiddlewares/redirect.py
@@ -29,17 +29,49 @@ def _build_redirect_request(
         **kwargs,
         cookies=None,
     )
+    if "_scheme_proxy" in redirect_request.meta:
+        source_request_scheme = urlparse_cached(source_request).scheme
+        redirect_request_scheme = urlparse_cached(redirect_request).scheme
+        if source_request_scheme != redirect_request_scheme:
+            redirect_request.meta.pop("_scheme_proxy")
+            redirect_request.meta.pop("proxy", None)
+            redirect_request.meta.pop("_auth_proxy", None)
+            redirect_request.headers.pop(b"Proxy-Authorization", None)
     has_cookie_header = "Cookie" in redirect_request.headers
     has_authorization_header = "Authorization" in redirect_request.headers
     if has_cookie_header or has_authorization_header:
-        source_request_netloc = urlparse_cached(source_request).netloc
-        redirect_request_netloc = urlparse_cached(redirect_request).netloc
-        if source_request_netloc != redirect_request_netloc:
-            if has_cookie_header:
-                del redirect_request.headers["Cookie"]
-            # https://fetch.spec.whatwg.org/#ref-for-cors-non-wildcard-request-header-name
-            if has_authorization_header:
-                del redirect_request.headers["Authorization"]
+        default_ports = {"http": 80, "https": 443}
+
+        parsed_source_request = urlparse_cached(source_request)
+        source_scheme, source_host, source_port = (
+            parsed_source_request.scheme,
+            parsed_source_request.hostname,
+            parsed_source_request.port
+            or default_ports.get(parsed_source_request.scheme),
+        )
+
+        parsed_redirect_request = urlparse_cached(redirect_request)
+        redirect_scheme, redirect_host, redirect_port = (
+            parsed_redirect_request.scheme,
+            parsed_redirect_request.hostname,
+            parsed_redirect_request.port
+            or default_ports.get(parsed_redirect_request.scheme),
+        )
+
+        if has_cookie_header and (
+            redirect_scheme not in {source_scheme, "https"}
+            or source_host != redirect_host
+        ):
+            del redirect_request.headers["Cookie"]
+
+        # https://fetch.spec.whatwg.org/#ref-for-cors-non-wildcard-request-header-name
+        if has_authorization_header and (
+            source_scheme != redirect_scheme
+            or source_host != redirect_host
+            or source_port != redirect_port
+        ):
+            del redirect_request.headers["Authorization"]
+
     return redirect_request
 
 
@@ -129,9 +161,11 @@ class RedirectMiddleware(BaseRedirectMiddleware):
             location = request_scheme + "://" + location.lstrip("/")
 
         redirected_url = urljoin(request.url, location)
+        redirected = _build_redirect_request(request, url=redirected_url)
+        if urlparse_cached(redirected).scheme not in {"http", "https"}:
+            return response
 
         if response.status in (301, 307, 308) or request.method == "HEAD":
-            redirected = _build_redirect_request(request, url=redirected_url)
             return self._redirect(redirected, request, spider, response.status)
 
         redirected = self._redirect_request_using_get(request, redirected_url)
@@ -153,12 +187,16 @@ class MetaRefreshMiddleware(BaseRedirectMiddleware):
             request.meta.get("dont_redirect", False)
             or request.method == "HEAD"
             or not isinstance(response, HtmlResponse)
+            or urlparse_cached(request).scheme not in {"http", "https"}
         ):
             return response
 
         interval, url = get_meta_refresh(response, ignore_tags=self._ignore_tags)
-        if url and cast(float, interval) < self._maxdelay:
-            redirected = self._redirect_request_using_get(request, url)
+        if not url:
+            return response
+        redirected = self._redirect_request_using_get(request, url)
+        if urlparse_cached(redirected).scheme not in {"http", "https"}:
+            return response
+        if cast(float, interval) < self._maxdelay:
             return self._redirect(redirected, request, spider, "meta refresh")
-
         return response
diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py
index d7ac7ec35..932475fb5 100644
--- a/scrapy/settings/default_settings.py
+++ b/scrapy/settings/default_settings.py
@@ -101,6 +101,7 @@ DOWNLOADER_MIDDLEWARES = {}
 
 DOWNLOADER_MIDDLEWARES_BASE = {
     # Engine side
+    "scrapy.downloadermiddlewares.offsite.OffsiteMiddleware": 50,
     "scrapy.downloadermiddlewares.robotstxt.RobotsTxtMiddleware": 100,
     "scrapy.downloadermiddlewares.httpauth.HttpAuthMiddleware": 300,
     "scrapy.downloadermiddlewares.downloadtimeout.DownloadTimeoutMiddleware": 350,
@@ -301,7 +302,6 @@ SPIDER_MIDDLEWARES = {}
 SPIDER_MIDDLEWARES_BASE = {
     # Engine side
     "scrapy.spidermiddlewares.httperror.HttpErrorMiddleware": 50,
-    "scrapy.spidermiddlewares.offsite.OffsiteMiddleware": 500,
     "scrapy.spidermiddlewares.referer.RefererMiddleware": 700,
     "scrapy.spidermiddlewares.urllength.UrlLengthMiddleware": 800,
     "scrapy.spidermiddlewares.depth.DepthMiddleware": 900,
diff --git a/scrapy/spidermiddlewares/offsite.py b/scrapy/spidermiddlewares/offsite.py
index dd2fccfcb..50c93ac9f 100644
--- a/scrapy/spidermiddlewares/offsite.py
+++ b/scrapy/spidermiddlewares/offsite.py
@@ -13,15 +13,21 @@ from typing import TYPE_CHECKING, Any, AsyncIterable, Iterable, Set
 
 from scrapy import Spider, signals
 from scrapy.crawler import Crawler
+from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.http import Request, Response
 from scrapy.statscollectors import StatsCollector
 from scrapy.utils.httpobj import urlparse_cached
 
+warnings.warn(
+    "The scrapy.spidermiddlewares.offsite module is deprecated, use "
+    "scrapy.downloadermiddlewares.offsite instead.",
+    ScrapyDeprecationWarning,
+)
+
 if TYPE_CHECKING:
     # typing.Self requires Python 3.11
     from typing_extensions import Self
 
-
 logger = logging.getLogger(__name__)
 
 
diff --git a/scrapy/utils/python.py b/scrapy/utils/python.py
index 5d2d490b2..578cde2ac 100644
--- a/scrapy/utils/python.py
+++ b/scrapy/utils/python.py
@@ -331,7 +331,7 @@ def global_object_name(obj: Any) -> str:
     >>> global_object_name(Request)
     'scrapy.http.request.Request'
     """
-    return f"{obj.__module__}.{obj.__name__}"
+    return f"{obj.__module__}.{obj.__qualname__}"
 
 
 if hasattr(sys, "pypy_version_info"):
diff --git a/tox.ini b/tox.ini
index ede139756..cde4243f3 100644
--- a/tox.ini
+++ b/tox.ini
@@ -26,6 +26,9 @@ deps =
 
     # mitmproxy does not support PyPy
     mitmproxy; implementation_name != 'pypy'
+    # https://github.com/pallets/werkzeug/pull/2768 breaks flask, required by
+    # mitmproxy.
+    werkzeug < 3; python_version < '3.9' and implementation_name != 'pypy'
 passenv =
     S3_TEST_FILE_URI
     AWS_ACCESS_KEY_ID
@@ -90,6 +93,7 @@ commands =
     twine check dist/*
 
 [pinned]
+basepython = python3.8
 deps =
     cryptography==36.0.0
     cssselect==0.9.1
@@ -116,7 +120,7 @@ commands =
     pytest --cov=scrapy --cov-report=xml --cov-report= {posargs:--durations=10 scrapy tests}
 
 [testenv:pinned]
-basepython = python3.8
+basepython = {[pinned]basepython}
 deps =
     {[pinned]deps}
     PyDispatcher==2.0.5
@@ -126,7 +130,7 @@ setenv =
 commands = {[pinned]commands}
 
 [testenv:windows-pinned]
-basepython = python3
+basepython = {[pinned]basepython}
 deps =
     {[pinned]deps}
     PyDispatcher==2.0.5
@@ -155,7 +159,7 @@ deps =
     ipython
 
 [testenv:extra-deps-pinned]
-basepython = python3.8
+basepython = {[pinned]basepython}
 deps =
     {[pinned]deps}
     boto3==1.20.0
@@ -179,6 +183,7 @@ commands =
     {[testenv]commands} --reactor=asyncio
 
 [testenv:asyncio-pinned]
+basepython = {[pinned]basepython}
 deps = {[testenv:pinned]deps}
 commands = {[pinned]commands} --reactor=asyncio
 install_command = {[pinned]install_command}
@@ -191,12 +196,12 @@ commands =
     pytest {posargs:--durations=10 docs scrapy tests}
 
 [testenv:pypy3-pinned]
-basepython = {[testenv:pypy3]basepython}
+basepython = pypy3.8
 deps =
     {[pinned]deps}
     PyPyDispatcher==2.1.0
 commands =
-    pytest --durations=10 scrapy tests
+    pytest {posargs:--durations=10 scrapy tests}
 install_command = {[pinned]install_command}
 setenv =
     {[pinned]setenv}
@@ -244,7 +249,7 @@ commands =
     pytest --cov=scrapy --cov-report=xml --cov-report= {posargs:tests -k s3}
 
 [testenv:botocore-pinned]
-basepython = python3.8
+basepython = {[pinned]basepython}
 deps =
     {[pinned]deps}
     botocore==1.4.87


Test patch:
diff --git a/tests/test_downloadermiddleware.py b/tests/test_downloadermiddleware.py
index 062e8a8b4..0155c62eb 100644
--- a/tests/test_downloadermiddleware.py
+++ b/tests/test_downloadermiddleware.py
@@ -22,13 +22,11 @@ class ManagerTestCase(TestCase):
         self.crawler = get_crawler(Spider, self.settings_dict)
         self.spider = self.crawler._create_spider("foo")
         self.mwman = DownloaderMiddlewareManager.from_crawler(self.crawler)
-        # some mw depends on stats collector
-        self.crawler.stats.open_spider(self.spider)
-        return self.mwman.open_spider(self.spider)
+        self.crawler.engine = self.crawler._create_engine()
+        return self.crawler.engine.open_spider(self.spider, start_requests=())
 
     def tearDown(self):
-        self.crawler.stats.close_spider(self.spider, "")
-        return self.mwman.close_spider(self.spider)
+        return self.crawler.engine.close_spider(self.spider)
 
     def _download(self, request, response=None):
         """Executes downloader mw manager's download method and returns
diff --git a/tests/test_downloadermiddleware_offsite.py b/tests/test_downloadermiddleware_offsite.py
new file mode 100644
index 000000000..d4669f450
--- /dev/null
+++ b/tests/test_downloadermiddleware_offsite.py
@@ -0,0 +1,184 @@
+import pytest
+
+from scrapy import Request, Spider
+from scrapy.downloadermiddlewares.offsite import OffsiteMiddleware
+from scrapy.exceptions import IgnoreRequest
+from scrapy.utils.test import get_crawler
+
+UNSET = object()
+
+
+@pytest.mark.parametrize(
+    ("allowed_domain", "url", "allowed"),
+    (
+        ("example.com", "http://example.com/1", True),
+        ("example.com", "http://example.org/1", False),
+        ("example.com", "http://sub.example.com/1", True),
+        ("sub.example.com", "http://sub.example.com/1", True),
+        ("sub.example.com", "http://example.com/1", False),
+        ("example.com", "http://example.com:8000/1", True),
+        ("example.com", "http://example.org/example.com", False),
+        ("example.com", "http://example.org/foo.example.com", False),
+        ("example.com", "http://example.com.example", False),
+        ("a.example", "http://nota.example", False),
+        ("b.a.example", "http://notb.a.example", False),
+    ),
+)
+def test_process_request_domain_filtering(allowed_domain, url, allowed):
+    crawler = get_crawler(Spider)
+    spider = crawler._create_spider(name="a", allowed_domains=[allowed_domain])
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    request = Request(url)
+    if allowed:
+        assert mw.process_request(request, spider) is None
+    else:
+        with pytest.raises(IgnoreRequest):
+            mw.process_request(request, spider)
+
+
+@pytest.mark.parametrize(
+    ("value", "filtered"),
+    (
+        (UNSET, True),
+        (None, True),
+        (False, True),
+        (True, False),
+    ),
+)
+def test_process_request_dont_filter(value, filtered):
+    crawler = get_crawler(Spider)
+    spider = crawler._create_spider(name="a", allowed_domains=["a.example"])
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    kwargs = {}
+    if value is not UNSET:
+        kwargs["dont_filter"] = value
+    request = Request("https://b.example", **kwargs)
+    if filtered:
+        with pytest.raises(IgnoreRequest):
+            mw.process_request(request, spider)
+    else:
+        assert mw.process_request(request, spider) is None
+
+
+@pytest.mark.parametrize(
+    "value",
+    (
+        UNSET,
+        None,
+        [],
+    ),
+)
+def test_process_request_no_allowed_domains(value):
+    crawler = get_crawler(Spider)
+    kwargs = {}
+    if value is not UNSET:
+        kwargs["allowed_domains"] = value
+    spider = crawler._create_spider(name="a", **kwargs)
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    request = Request("https://example.com")
+    assert mw.process_request(request, spider) is None
+
+
+def test_process_request_invalid_domains():
+    crawler = get_crawler(Spider)
+    allowed_domains = ["a.example", None, "http:////b.example", "//c.example"]
+    spider = crawler._create_spider(name="a", allowed_domains=allowed_domains)
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    request = Request("https://a.example")
+    assert mw.process_request(request, spider) is None
+    for letter in ("b", "c"):
+        request = Request(f"https://{letter}.example")
+        with pytest.raises(IgnoreRequest):
+            mw.process_request(request, spider)
+
+
+@pytest.mark.parametrize(
+    ("allowed_domain", "url", "allowed"),
+    (
+        ("example.com", "http://example.com/1", True),
+        ("example.com", "http://example.org/1", False),
+        ("example.com", "http://sub.example.com/1", True),
+        ("sub.example.com", "http://sub.example.com/1", True),
+        ("sub.example.com", "http://example.com/1", False),
+        ("example.com", "http://example.com:8000/1", True),
+        ("example.com", "http://example.org/example.com", False),
+        ("example.com", "http://example.org/foo.example.com", False),
+        ("example.com", "http://example.com.example", False),
+        ("a.example", "http://nota.example", False),
+        ("b.a.example", "http://notb.a.example", False),
+    ),
+)
+def test_request_scheduled_domain_filtering(allowed_domain, url, allowed):
+    crawler = get_crawler(Spider)
+    spider = crawler._create_spider(name="a", allowed_domains=[allowed_domain])
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    request = Request(url)
+    if allowed:
+        assert mw.request_scheduled(request, spider) is None
+    else:
+        with pytest.raises(IgnoreRequest):
+            mw.request_scheduled(request, spider)
+
+
+@pytest.mark.parametrize(
+    ("value", "filtered"),
+    (
+        (UNSET, True),
+        (None, True),
+        (False, True),
+        (True, False),
+    ),
+)
+def test_request_scheduled_dont_filter(value, filtered):
+    crawler = get_crawler(Spider)
+    spider = crawler._create_spider(name="a", allowed_domains=["a.example"])
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    kwargs = {}
+    if value is not UNSET:
+        kwargs["dont_filter"] = value
+    request = Request("https://b.example", **kwargs)
+    if filtered:
+        with pytest.raises(IgnoreRequest):
+            mw.request_scheduled(request, spider)
+    else:
+        assert mw.request_scheduled(request, spider) is None
+
+
+@pytest.mark.parametrize(
+    "value",
+    (
+        UNSET,
+        None,
+        [],
+    ),
+)
+def test_request_scheduled_no_allowed_domains(value):
+    crawler = get_crawler(Spider)
+    kwargs = {}
+    if value is not UNSET:
+        kwargs["allowed_domains"] = value
+    spider = crawler._create_spider(name="a", **kwargs)
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    request = Request("https://example.com")
+    assert mw.request_scheduled(request, spider) is None
+
+
+def test_request_scheduled_invalid_domains():
+    crawler = get_crawler(Spider)
+    allowed_domains = ["a.example", None, "http:////b.example", "//c.example"]
+    spider = crawler._create_spider(name="a", allowed_domains=allowed_domains)
+    mw = OffsiteMiddleware.from_crawler(crawler)
+    mw.spider_opened(spider)
+    request = Request("https://a.example")
+    assert mw.request_scheduled(request, spider) is None
+    for letter in ("b", "c"):
+        request = Request(f"https://{letter}.example")
+        with pytest.raises(IgnoreRequest):
+            mw.request_scheduled(request, spider)
diff --git a/tests/test_downloadermiddleware_redirect.py b/tests/test_downloadermiddleware_redirect.py
index 83ff25982..4bfd34fe2 100644
--- a/tests/test_downloadermiddleware_redirect.py
+++ b/tests/test_downloadermiddleware_redirect.py
@@ -1,5 +1,9 @@
 import unittest
+from itertools import chain, product
 
+import pytest
+
+from scrapy.downloadermiddlewares.httpproxy import HttpProxyMiddleware
 from scrapy.downloadermiddlewares.redirect import (
     MetaRefreshMiddleware,
     RedirectMiddleware,
@@ -7,22 +11,1030 @@ from scrapy.downloadermiddlewares.redirect import (
 from scrapy.exceptions import IgnoreRequest
 from scrapy.http import HtmlResponse, Request, Response
 from scrapy.spiders import Spider
+from scrapy.utils.misc import set_environ
 from scrapy.utils.test import get_crawler
 
 
-class RedirectMiddlewareTest(unittest.TestCase):
+class Base:
+    class Test(unittest.TestCase):
+        def test_priority_adjust(self):
+            req = Request("http://a.com")
+            rsp = self.get_response(req, "http://a.com/redirected")
+            req2 = self.mw.process_response(req, rsp, self.spider)
+            self.assertGreater(req2.priority, req.priority)
+
+        def test_dont_redirect(self):
+            url = "http://www.example.com/301"
+            url2 = "http://www.example.com/redirected"
+            req = Request(url, meta={"dont_redirect": True})
+            rsp = self.get_response(req, url2)
+
+            r = self.mw.process_response(req, rsp, self.spider)
+            assert isinstance(r, Response)
+            assert r is rsp
+
+            # Test that it redirects when dont_redirect is False
+            req = Request(url, meta={"dont_redirect": False})
+            rsp = self.get_response(req, url2)
+
+            r = self.mw.process_response(req, rsp, self.spider)
+            assert isinstance(r, Request)
+
+        def test_post(self):
+            url = "http://www.example.com/302"
+            url2 = "http://www.example.com/redirected2"
+            req = Request(
+                url,
+                method="POST",
+                body="test",
+                headers={"Content-Type": "text/plain", "Content-length": "4"},
+            )
+            rsp = self.get_response(req, url2)
+
+            req2 = self.mw.process_response(req, rsp, self.spider)
+            assert isinstance(req2, Request)
+            self.assertEqual(req2.url, url2)
+            self.assertEqual(req2.method, "GET")
+            assert (
+                "Content-Type" not in req2.headers
+            ), "Content-Type header must not be present in redirected request"
+            assert (
+                "Content-Length" not in req2.headers
+            ), "Content-Length header must not be present in redirected request"
+            assert not req2.body, f"Redirected body must be empty, not '{req2.body}'"
+
+        def test_max_redirect_times(self):
+            self.mw.max_redirect_times = 1
+            req = Request("http://scrapytest.org/302")
+            rsp = self.get_response(req, "/redirected")
+
+            req = self.mw.process_response(req, rsp, self.spider)
+            assert isinstance(req, Request)
+            assert "redirect_times" in req.meta
+            self.assertEqual(req.meta["redirect_times"], 1)
+            self.assertRaises(
+                IgnoreRequest, self.mw.process_response, req, rsp, self.spider
+            )
+
+        def test_ttl(self):
+            self.mw.max_redirect_times = 100
+            req = Request("http://scrapytest.org/302", meta={"redirect_ttl": 1})
+            rsp = self.get_response(req, "/a")
+
+            req = self.mw.process_response(req, rsp, self.spider)
+            assert isinstance(req, Request)
+            self.assertRaises(
+                IgnoreRequest, self.mw.process_response, req, rsp, self.spider
+            )
+
+        def test_redirect_urls(self):
+            req1 = Request("http://scrapytest.org/first")
+            rsp1 = self.get_response(req1, "/redirected")
+            req2 = self.mw.process_response(req1, rsp1, self.spider)
+            rsp2 = self.get_response(req1, "/redirected2")
+            req3 = self.mw.process_response(req2, rsp2, self.spider)
+
+            self.assertEqual(req2.url, "http://scrapytest.org/redirected")
+            self.assertEqual(
+                req2.meta["redirect_urls"], ["http://scrapytest.org/first"]
+            )
+            self.assertEqual(req3.url, "http://scrapytest.org/redirected2")
+            self.assertEqual(
+                req3.meta["redirect_urls"],
+                ["http://scrapytest.org/first", "http://scrapytest.org/redirected"],
+            )
+
+        def test_redirect_reasons(self):
+            req1 = Request("http://scrapytest.org/first")
+            rsp1 = self.get_response(req1, "/redirected1")
+            req2 = self.mw.process_response(req1, rsp1, self.spider)
+            rsp2 = self.get_response(req2, "/redirected2")
+            req3 = self.mw.process_response(req2, rsp2, self.spider)
+            self.assertEqual(req2.meta["redirect_reasons"], [self.reason])
+            self.assertEqual(req3.meta["redirect_reasons"], [self.reason, self.reason])
+
+        def test_cross_origin_header_dropping(self):
+            safe_headers = {"A": "B"}
+            cookie_header = {"Cookie": "a=b"}
+            authorization_header = {"Authorization": "Bearer 123456"}
+
+            original_request = Request(
+                "https://example.com",
+                headers={**safe_headers, **cookie_header, **authorization_header},
+            )
+
+            # Redirects to the same origin (same scheme, same domain, same port)
+            # keep all headers.
+            internal_response = self.get_response(
+                original_request, "https://example.com/a"
+            )
+            internal_redirect_request = self.mw.process_response(
+                original_request, internal_response, self.spider
+            )
+            self.assertIsInstance(internal_redirect_request, Request)
+            self.assertEqual(
+                original_request.headers, internal_redirect_request.headers
+            )
+
+            # Redirects to the same origin (same scheme, same domain, same port)
+            # keep all headers also when the scheme is http.
+            http_request = Request(
+                "http://example.com",
+                headers={**safe_headers, **cookie_header, **authorization_header},
+            )
+            http_response = self.get_response(http_request, "http://example.com/a")
+            http_redirect_request = self.mw.process_response(
+                http_request, http_response, self.spider
+            )
+            self.assertIsInstance(http_redirect_request, Request)
+            self.assertEqual(http_request.headers, http_redirect_request.headers)
+
+            # For default ports, whether the port is explicit or implicit does not
+            # affect the outcome, it is still the same origin.
+            to_explicit_port_response = self.get_response(
+                original_request, "https://example.com:443/a"
+            )
+            to_explicit_port_redirect_request = self.mw.process_response(
+                original_request, to_explicit_port_response, self.spider
+            )
+            self.assertIsInstance(to_explicit_port_redirect_request, Request)
+            self.assertEqual(
+                original_request.headers, to_explicit_port_redirect_request.headers
+            )
+
+            # For default ports, whether the port is explicit or implicit does not
+            # affect the outcome, it is still the same origin.
+            to_implicit_port_response = self.get_response(
+                original_request, "https://example.com/a"
+            )
+            to_implicit_port_redirect_request = self.mw.process_response(
+                original_request, to_implicit_port_response, self.spider
+            )
+            self.assertIsInstance(to_implicit_port_redirect_request, Request)
+            self.assertEqual(
+                original_request.headers, to_implicit_port_redirect_request.headers
+            )
+
+            # A port change drops the Authorization header because the origin
+            # changes, but keeps the Cookie header because the domain remains the
+            # same.
+            different_port_response = self.get_response(
+                original_request, "https://example.com:8080/a"
+            )
+            different_port_redirect_request = self.mw.process_response(
+                original_request, different_port_response, self.spider
+            )
+            self.assertIsInstance(different_port_redirect_request, Request)
+            self.assertEqual(
+                {**safe_headers, **cookie_header},
+                different_port_redirect_request.headers.to_unicode_dict(),
+            )
+
+            # A domain change drops both the Authorization and the Cookie header.
+            external_response = self.get_response(
+                original_request, "https://example.org/a"
+            )
+            external_redirect_request = self.mw.process_response(
+                original_request, external_response, self.spider
+            )
+            self.assertIsInstance(external_redirect_request, Request)
+            self.assertEqual(
+                safe_headers, external_redirect_request.headers.to_unicode_dict()
+            )
+
+            # A scheme upgrade (http → https) drops the Authorization header
+            # because the origin changes, but keeps the Cookie header because the
+            # domain remains the same.
+            upgrade_response = self.get_response(http_request, "https://example.com/a")
+            upgrade_redirect_request = self.mw.process_response(
+                http_request, upgrade_response, self.spider
+            )
+            self.assertIsInstance(upgrade_redirect_request, Request)
+            self.assertEqual(
+                {**safe_headers, **cookie_header},
+                upgrade_redirect_request.headers.to_unicode_dict(),
+            )
+
+            # A scheme downgrade (https → http) drops the Authorization header
+            # because the origin changes, and the Cookie header because its value
+            # cannot indicate whether the cookies were secure (HTTPS-only) or not.
+            #
+            # Note: If the Cookie header is set by the cookie management
+            # middleware, as recommended in the docs, the dropping of Cookie on
+            # scheme downgrade is not an issue, because the cookie management
+            # middleware will add again the Cookie header to the new request if
+            # appropriate.
+            downgrade_response = self.get_response(
+                original_request, "http://example.com/a"
+            )
+            downgrade_redirect_request = self.mw.process_response(
+                original_request, downgrade_response, self.spider
+            )
+            self.assertIsInstance(downgrade_redirect_request, Request)
+            self.assertEqual(
+                safe_headers,
+                downgrade_redirect_request.headers.to_unicode_dict(),
+            )
+
+        def test_meta_proxy_http_absolute(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            meta = {"proxy": "https://a:@a.example"}
+            request1 = Request("http://example.com", meta=meta)
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "http://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "http://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_meta_proxy_http_relative(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            meta = {"proxy": "https://a:@a.example"}
+            request1 = Request("http://example.com", meta=meta)
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "/a")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "/a")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_meta_proxy_https_absolute(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            meta = {"proxy": "https://a:@a.example"}
+            request1 = Request("https://example.com", meta=meta)
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "https://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "https://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_meta_proxy_https_relative(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            meta = {"proxy": "https://a:@a.example"}
+            request1 = Request("https://example.com", meta=meta)
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "/a")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "/a")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_meta_proxy_http_to_https(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            meta = {"proxy": "https://a:@a.example"}
+            request1 = Request("http://example.com", meta=meta)
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "https://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "http://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_meta_proxy_https_to_http(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            meta = {"proxy": "https://a:@a.example"}
+            request1 = Request("https://example.com", meta=meta)
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "http://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "https://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_system_proxy_http_absolute(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "http_proxy": "https://a:@a.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("http://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "http://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "http://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_system_proxy_http_relative(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "http_proxy": "https://a:@a.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("http://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "/a")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "/a")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_system_proxy_https_absolute(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "https_proxy": "https://a:@a.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("https://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "https://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "https://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_system_proxy_https_relative(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "https_proxy": "https://a:@a.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("https://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "/a")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "/a")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_system_proxy_proxied_http_to_proxied_https(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "http_proxy": "https://a:@a.example",
+                "https_proxy": "https://b:@b.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("http://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "https://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic Yjo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://b.example")
+            self.assertEqual(request2.meta["proxy"], "https://b.example")
+
+            response2 = self.get_response(request2, "http://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_system_proxy_proxied_http_to_unproxied_https(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "http_proxy": "https://a:@a.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("http://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request1.meta["proxy"], "https://a.example")
+
+            response1 = self.get_response(request1, "https://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            response2 = self.get_response(request2, "http://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request3.meta["proxy"], "https://a.example")
+
+        def test_system_proxy_unproxied_http_to_proxied_https(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "https_proxy": "https://b:@b.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("http://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertNotIn("Proxy-Authorization", request1.headers)
+            self.assertNotIn("_auth_proxy", request1.meta)
+            self.assertNotIn("proxy", request1.meta)
+
+            response1 = self.get_response(request1, "https://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic Yjo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://b.example")
+            self.assertEqual(request2.meta["proxy"], "https://b.example")
+
+            response2 = self.get_response(request2, "http://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+        def test_system_proxy_unproxied_http_to_unproxied_https(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("http://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertNotIn("Proxy-Authorization", request1.headers)
+            self.assertNotIn("_auth_proxy", request1.meta)
+            self.assertNotIn("proxy", request1.meta)
+
+            response1 = self.get_response(request1, "https://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            response2 = self.get_response(request2, "http://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+        def test_system_proxy_proxied_https_to_proxied_http(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "http_proxy": "https://a:@a.example",
+                "https_proxy": "https://b:@b.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("https://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic Yjo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://b.example")
+            self.assertEqual(request1.meta["proxy"], "https://b.example")
+
+            response1 = self.get_response(request1, "http://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "https://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic Yjo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://b.example")
+            self.assertEqual(request3.meta["proxy"], "https://b.example")
+
+        def test_system_proxy_proxied_https_to_unproxied_http(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "https_proxy": "https://b:@b.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("https://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertEqual(request1.headers["Proxy-Authorization"], b"Basic Yjo=")
+            self.assertEqual(request1.meta["_auth_proxy"], "https://b.example")
+            self.assertEqual(request1.meta["proxy"], "https://b.example")
+
+            response1 = self.get_response(request1, "http://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            response2 = self.get_response(request2, "https://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertEqual(request3.headers["Proxy-Authorization"], b"Basic Yjo=")
+            self.assertEqual(request3.meta["_auth_proxy"], "https://b.example")
+            self.assertEqual(request3.meta["proxy"], "https://b.example")
+
+        def test_system_proxy_unproxied_https_to_proxied_http(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            env = {
+                "http_proxy": "https://a:@a.example",
+            }
+            with set_environ(**env):
+                proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("https://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertNotIn("Proxy-Authorization", request1.headers)
+            self.assertNotIn("_auth_proxy", request1.meta)
+            self.assertNotIn("proxy", request1.meta)
+
+            response1 = self.get_response(request1, "http://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertEqual(request2.headers["Proxy-Authorization"], b"Basic YTo=")
+            self.assertEqual(request2.meta["_auth_proxy"], "https://a.example")
+            self.assertEqual(request2.meta["proxy"], "https://a.example")
+
+            response2 = self.get_response(request2, "https://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+        def test_system_proxy_unproxied_https_to_unproxied_http(self):
+            crawler = get_crawler()
+            redirect_mw = self.mwcls.from_crawler(crawler)
+            proxy_mw = HttpProxyMiddleware.from_crawler(crawler)
+
+            request1 = Request("https://example.com")
+            spider = None
+            proxy_mw.process_request(request1, spider)
+
+            self.assertNotIn("Proxy-Authorization", request1.headers)
+            self.assertNotIn("_auth_proxy", request1.meta)
+            self.assertNotIn("proxy", request1.meta)
+
+            response1 = self.get_response(request1, "http://example.com")
+            request2 = redirect_mw.process_response(request1, response1, spider)
+
+            self.assertIsInstance(request2, Request)
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            proxy_mw.process_request(request2, spider)
+
+            self.assertNotIn("Proxy-Authorization", request2.headers)
+            self.assertNotIn("_auth_proxy", request2.meta)
+            self.assertNotIn("proxy", request2.meta)
+
+            response2 = self.get_response(request2, "https://example.com")
+            request3 = redirect_mw.process_response(request2, response2, spider)
+
+            self.assertIsInstance(request3, Request)
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+            proxy_mw.process_request(request3, spider)
+
+            self.assertNotIn("Proxy-Authorization", request3.headers)
+            self.assertNotIn("_auth_proxy", request3.meta)
+            self.assertNotIn("proxy", request3.meta)
+
+
+class RedirectMiddlewareTest(Base.Test):
+    mwcls = RedirectMiddleware
+    reason = 302
+
     def setUp(self):
         self.crawler = get_crawler(Spider)
         self.spider = self.crawler._create_spider("foo")
-        self.mw = RedirectMiddleware.from_crawler(self.crawler)
+        self.mw = self.mwcls.from_crawler(self.crawler)
 
-    def test_priority_adjust(self):
-        req = Request("http://a.com")
-        rsp = Response(
-            "http://a.com", headers={"Location": "http://a.com/redirected"}, status=301
-        )
-        req2 = self.mw.process_response(req, rsp, self.spider)
-        assert req2.priority > req.priority
+    def get_response(self, request, location, status=302):
+        headers = {"Location": location}
+        return Response(request.url, status=status, headers=headers)
 
     def test_redirect_3xx_permanent(self):
         def _test(method, status=301):
@@ -52,51 +1064,6 @@ class RedirectMiddlewareTest(unittest.TestCase):
         _test("POST", status=308)
         _test("HEAD", status=308)
 
-    def test_dont_redirect(self):
-        url = "http://www.example.com/301"
-        url2 = "http://www.example.com/redirected"
-        req = Request(url, meta={"dont_redirect": True})
-        rsp = Response(url, headers={"Location": url2}, status=301)
-
-        r = self.mw.process_response(req, rsp, self.spider)
-        assert isinstance(r, Response)
-        assert r is rsp
-
-        # Test that it redirects when dont_redirect is False
-        req = Request(url, meta={"dont_redirect": False})
-        rsp = Response(url2, status=200)
-
-        r = self.mw.process_response(req, rsp, self.spider)
-        assert isinstance(r, Response)
-        assert r is rsp
-
-    def test_redirect_302(self):
-        url = "http://www.example.com/302"
-        url2 = "http://www.example.com/redirected2"
-        req = Request(
-            url,
-            method="POST",
-            body="test",
-            headers={"Content-Type": "text/plain", "Content-length": "4"},
-        )
-        rsp = Response(url, headers={"Location": url2}, status=302)
-
-        req2 = self.mw.process_response(req, rsp, self.spider)
-        assert isinstance(req2, Request)
-        self.assertEqual(req2.url, url2)
-        self.assertEqual(req2.method, "GET")
-        assert (
-            "Content-Type" not in req2.headers
-        ), "Content-Type header must not be present in redirected request"
-        assert (
-            "Content-Length" not in req2.headers
-        ), "Content-Length header must not be present in redirected request"
-        assert not req2.body, f"Redirected body must be empty, not '{req2.body}'"
-
-        # response without Location header but with status code is 3XX should be ignored
-        del rsp.headers["Location"]
-        assert self.mw.process_response(req, rsp, self.spider) is rsp
-
     def test_redirect_302_head(self):
         url = "http://www.example.com/302"
         url2 = "http://www.example.com/redirected2"
@@ -108,10 +1075,6 @@ class RedirectMiddlewareTest(unittest.TestCase):
         self.assertEqual(req2.url, url2)
         self.assertEqual(req2.method, "HEAD")
 
-        # response without Location header but with status code is 3XX should be ignored
-        del rsp.headers["Location"]
-        assert self.mw.process_response(req, rsp, self.spider) is rsp
-
     def test_redirect_302_relative(self):
         url = "http://www.example.com/302"
         url2 = "///i8n.example2.com/302"
@@ -124,81 +1087,6 @@ class RedirectMiddlewareTest(unittest.TestCase):
         self.assertEqual(req2.url, url3)
         self.assertEqual(req2.method, "HEAD")
 
-        # response without Location header but with status code is 3XX should be ignored
-        del rsp.headers["Location"]
-        assert self.mw.process_response(req, rsp, self.spider) is rsp
-
-    def test_max_redirect_times(self):
-        self.mw.max_redirect_times = 1
-        req = Request("http://scrapytest.org/302")
-        rsp = Response(
-            "http://scrapytest.org/302", headers={"Location": "/redirected"}, status=302
-        )
-
-        req = self.mw.process_response(req, rsp, self.spider)
-        assert isinstance(req, Request)
-        assert "redirect_times" in req.meta
-        self.assertEqual(req.meta["redirect_times"], 1)
-        self.assertRaises(
-            IgnoreRequest, self.mw.process_response, req, rsp, self.spider
-        )
-
-    def test_ttl(self):
-        self.mw.max_redirect_times = 100
-        req = Request("http://scrapytest.org/302", meta={"redirect_ttl": 1})
-        rsp = Response(
-            "http://www.scrapytest.org/302",
-            headers={"Location": "/redirected"},
-            status=302,
-        )
-
-        req = self.mw.process_response(req, rsp, self.spider)
-        assert isinstance(req, Request)
-        self.assertRaises(
-            IgnoreRequest, self.mw.process_response, req, rsp, self.spider
-        )
-
-    def test_redirect_urls(self):
-        req1 = Request("http://scrapytest.org/first")
-        rsp1 = Response(
-            "http://scrapytest.org/first",
-            headers={"Location": "/redirected"},
-            status=302,
-        )
-        req2 = self.mw.process_response(req1, rsp1, self.spider)
-        rsp2 = Response(
-            "http://scrapytest.org/redirected",
-            headers={"Location": "/redirected2"},
-            status=302,
-        )
-        req3 = self.mw.process_response(req2, rsp2, self.spider)
-
-        self.assertEqual(req2.url, "http://scrapytest.org/redirected")
-        self.assertEqual(req2.meta["redirect_urls"], ["http://scrapytest.org/first"])
-        self.assertEqual(req3.url, "http://scrapytest.org/redirected2")
-        self.assertEqual(
-            req3.meta["redirect_urls"],
-            ["http://scrapytest.org/first", "http://scrapytest.org/redirected"],
-        )
-
-    def test_redirect_reasons(self):
-        req1 = Request("http://scrapytest.org/first")
-        rsp1 = Response(
-            "http://scrapytest.org/first",
-            headers={"Location": "/redirected1"},
-            status=301,
-        )
-        req2 = self.mw.process_response(req1, rsp1, self.spider)
-        rsp2 = Response(
-            "http://scrapytest.org/redirected1",
-            headers={"Location": "/redirected2"},
-            status=301,
-        )
-        req3 = self.mw.process_response(req2, rsp2, self.spider)
-
-        self.assertEqual(req2.meta["redirect_reasons"], [301])
-        self.assertEqual(req3.meta["redirect_reasons"], [301, 301])
-
     def test_spider_handling(self):
         smartspider = self.crawler._create_spider("smarty")
         smartspider.handle_httpstatus_list = [404, 301, 302]
@@ -247,53 +1135,84 @@ class RedirectMiddlewareTest(unittest.TestCase):
         perc_encoded_utf8_url = "http://scrapytest.org/a%C3%A7%C3%A3o"
         self.assertEqual(perc_encoded_utf8_url, req_result.url)
 
-    def test_cross_domain_header_dropping(self):
-        safe_headers = {"A": "B"}
-        original_request = Request(
-            "https://example.com",
-            headers={"Cookie": "a=b", "Authorization": "a", **safe_headers},
-        )
+    def test_no_location(self):
+        request = Request("https://example.com")
+        response = Response(request.url, status=302)
+        assert self.mw.process_response(request, response, self.spider) is response
 
-        internal_response = Response(
-            "https://example.com",
-            headers={"Location": "https://example.com/a"},
-            status=301,
-        )
-        internal_redirect_request = self.mw.process_response(
-            original_request, internal_response, self.spider
-        )
-        self.assertIsInstance(internal_redirect_request, Request)
-        self.assertEqual(original_request.headers, internal_redirect_request.headers)
 
-        external_response = Response(
-            "https://example.com",
-            headers={"Location": "https://example.org/a"},
-            status=301,
+SCHEME_PARAMS = ("url", "location", "target")
+HTTP_SCHEMES = ("http", "https")
+NON_HTTP_SCHEMES = ("data", "file", "ftp", "s3", "foo")
+REDIRECT_SCHEME_CASES = (
+    # http/https → http/https redirects
+    *(
+        (
+            f"{input_scheme}://example.com/a",
+            f"{output_scheme}://example.com/b",
+            f"{output_scheme}://example.com/b",
         )
-        external_redirect_request = self.mw.process_response(
-            original_request, external_response, self.spider
+        for input_scheme, output_scheme in product(HTTP_SCHEMES, repeat=2)
+    ),
+    # http/https → data/file/ftp/s3/foo does not redirect
+    *(
+        (
+            f"{input_scheme}://example.com/a",
+            f"{output_scheme}://example.com/b",
+            None,
         )
-        self.assertIsInstance(external_redirect_request, Request)
-        self.assertEqual(
-            safe_headers, external_redirect_request.headers.to_unicode_dict()
+        for input_scheme in HTTP_SCHEMES
+        for output_scheme in NON_HTTP_SCHEMES
+    ),
+    # http/https → relative redirects
+    *(
+        (
+            f"{scheme}://example.com/a",
+            location,
+            f"{scheme}://example.com/b",
         )
+        for scheme in HTTP_SCHEMES
+        for location in ("//example.com/b", "/b")
+    ),
+    # Note: We do not test data/file/ftp/s3 schemes for the initial URL
+    # because their download handlers cannot return a status code of 3xx.
+)
+
 
+@pytest.mark.parametrize(SCHEME_PARAMS, REDIRECT_SCHEME_CASES)
+def test_redirect_schemes(url, location, target):
+    crawler = get_crawler(Spider)
+    spider = crawler._create_spider("foo")
+    mw = RedirectMiddleware.from_crawler(crawler)
+    request = Request(url)
+    response = Response(url, headers={"Location": location}, status=301)
+    redirect = mw.process_response(request, response, spider)
+    if target is None:
+        assert redirect == response
+    else:
+        assert isinstance(redirect, Request)
+        assert redirect.url == target
+
+
+def meta_refresh_body(url, interval=5):
+    html = f"""<html><head><meta http-equiv="refresh" content="{interval};url={url}"/></head></html>"""
+    return html.encode("utf-8")
+
+
+class MetaRefreshMiddlewareTest(Base.Test):
+    mwcls = MetaRefreshMiddleware
+    reason = "meta refresh"
 
-class MetaRefreshMiddlewareTest(unittest.TestCase):
     def setUp(self):
         crawler = get_crawler(Spider)
         self.spider = crawler._create_spider("foo")
-        self.mw = MetaRefreshMiddleware.from_crawler(crawler)
+        self.mw = self.mwcls.from_crawler(crawler)
 
     def _body(self, interval=5, url="http://example.org/newpage"):
-        html = f"""<html><head><meta http-equiv="refresh" content="{interval};url={url}"/></head></html>"""
-        return html.encode("utf-8")
+        return meta_refresh_body(url, interval)
 
-    def test_priority_adjust(self):
-        req = Request("http://a.com")
-        rsp = HtmlResponse(req.url, body=self._body())
-        req2 = self.mw.process_response(req, rsp, self.spider)
-        assert req2.priority > req.priority
+    def get_response(self, request, location):
+        return HtmlResponse(request.url, body=self._body(url=location))
 
     def test_meta_refresh(self):
         req = Request(url="http://example.org")
@@ -332,62 +1251,6 @@ class MetaRefreshMiddlewareTest(unittest.TestCase):
         ), "Content-Length header must not be present in redirected request"
         assert not req2.body, f"Redirected body must be empty, not '{req2.body}'"
 
-    def test_max_redirect_times(self):
-        self.mw.max_redirect_times = 1
-        req = Request("http://scrapytest.org/max")
-        rsp = HtmlResponse(req.url, body=self._body())
-
-        req = self.mw.process_response(req, rsp, self.spider)
-        assert isinstance(req, Request)
-        assert "redirect_times" in req.meta
-        self.assertEqual(req.meta["redirect_times"], 1)
-        self.assertRaises(
-            IgnoreRequest, self.mw.process_response, req, rsp, self.spider
-        )
-
-    def test_ttl(self):
-        self.mw.max_redirect_times = 100
-        req = Request("http://scrapytest.org/302", meta={"redirect_ttl": 1})
-        rsp = HtmlResponse(req.url, body=self._body())
-
-        req = self.mw.process_response(req, rsp, self.spider)
-        assert isinstance(req, Request)
-        self.assertRaises(
-            IgnoreRequest, self.mw.process_response, req, rsp, self.spider
-        )
-
-    def test_redirect_urls(self):
-        req1 = Request("http://scrapytest.org/first")
-        rsp1 = HtmlResponse(req1.url, body=self._body(url="/redirected"))
-        req2 = self.mw.process_response(req1, rsp1, self.spider)
-        assert isinstance(req2, Request), req2
-        rsp2 = HtmlResponse(req2.url, body=self._body(url="/redirected2"))
-        req3 = self.mw.process_response(req2, rsp2, self.spider)
-        assert isinstance(req3, Request), req3
-        self.assertEqual(req2.url, "http://scrapytest.org/redirected")
-        self.assertEqual(req2.meta["redirect_urls"], ["http://scrapytest.org/first"])
-        self.assertEqual(req3.url, "http://scrapytest.org/redirected2")
-        self.assertEqual(
-            req3.meta["redirect_urls"],
-            ["http://scrapytest.org/first", "http://scrapytest.org/redirected"],
-        )
-
-    def test_redirect_reasons(self):
-        req1 = Request("http://scrapytest.org/first")
-        rsp1 = HtmlResponse(
-            "http://scrapytest.org/first", body=self._body(url="/redirected")
-        )
-        req2 = self.mw.process_response(req1, rsp1, self.spider)
-        rsp2 = HtmlResponse(
-            "http://scrapytest.org/redirected", body=self._body(url="/redirected1")
-        )
-        req3 = self.mw.process_response(req2, rsp2, self.spider)
-
-        self.assertEqual(req2.meta["redirect_reasons"], ["meta refresh"])
-        self.assertEqual(
-            req3.meta["redirect_reasons"], ["meta refresh", "meta refresh"]
-        )
-
     def test_ignore_tags_default(self):
         req = Request(url="http://example.org")
         body = (
@@ -413,5 +1276,45 @@ class MetaRefreshMiddlewareTest(unittest.TestCase):
         assert isinstance(response, Response)
 
 
+@pytest.mark.parametrize(
+    SCHEME_PARAMS,
+    (
+        *REDIRECT_SCHEME_CASES,
+        # data/file/ftp/s3/foo → * does not redirect
+        *(
+            (
+                f"{input_scheme}://example.com/a",
+                f"{output_scheme}://example.com/b",
+                None,
+            )
+            for input_scheme in NON_HTTP_SCHEMES
+            for output_scheme in chain(HTTP_SCHEMES, NON_HTTP_SCHEMES)
+        ),
+        # data/file/ftp/s3/foo → relative does not redirect
+        *(
+            (
+                f"{scheme}://example.com/a",
+                location,
+                None,
+            )
+            for scheme in NON_HTTP_SCHEMES
+            for location in ("//example.com/b", "/b")
+        ),
+    ),
+)
+def test_meta_refresh_schemes(url, location, target):
+    crawler = get_crawler(Spider)
+    spider = crawler._create_spider("foo")
+    mw = MetaRefreshMiddleware.from_crawler(crawler)
+    request = Request(url)
+    response = HtmlResponse(url, body=meta_refresh_body(location))
+    redirect = mw.process_response(request, response, spider)
+    if target is None:
+        assert redirect == response
+    else:
+        assert isinstance(redirect, Request)
+        assert redirect.url == target
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/test_engine.py b/tests/test_engine.py
index 8d7afb6a1..33544e8db 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -15,8 +15,10 @@ import subprocess
 import sys
 from collections import defaultdict
 from dataclasses import dataclass
+from logging import DEBUG
 from pathlib import Path
 from threading import Timer
+from unittest.mock import Mock
 from urllib.parse import urlparse
 
 import attr
@@ -27,11 +29,13 @@ from twisted.trial import unittest
 from twisted.web import server, static, util
 
 from scrapy import signals
-from scrapy.core.engine import ExecutionEngine
-from scrapy.exceptions import CloseSpider
+from scrapy.core.engine import ExecutionEngine, Slot
+from scrapy.core.scheduler import BaseScheduler
+from scrapy.exceptions import CloseSpider, IgnoreRequest
 from scrapy.http import Request
 from scrapy.item import Field, Item
 from scrapy.linkextractors import LinkExtractor
+from scrapy.signals import request_scheduled
 from scrapy.spiders import Spider
 from scrapy.utils.signal import disconnect_all
 from scrapy.utils.test import get_crawler
@@ -467,6 +471,38 @@ class EngineTest(unittest.TestCase):
         self.assertNotIn(b"Traceback", stderr)
 
 
+def test_request_scheduled_signal(caplog):
+    class TestScheduler(BaseScheduler):
+        def __init__(self):
+            self.enqueued = []
+
+        def enqueue_request(self, request: Request) -> bool:
+            self.enqueued.append(request)
+            return True
+
+    def signal_handler(request: Request, spider: Spider) -> None:
+        if "drop" in request.url:
+            raise IgnoreRequest
+
+    spider = TestSpider()
+    crawler = get_crawler(spider.__class__)
+    engine = ExecutionEngine(crawler, lambda _: None)
+    engine.downloader._slot_gc_loop.stop()
+    scheduler = TestScheduler()
+    engine.slot = Slot((), None, Mock(), scheduler)
+    crawler.signals.connect(signal_handler, request_scheduled)
+    keep_request = Request("https://keep.example")
+    engine._schedule_request(keep_request, spider)
+    drop_request = Request("https://drop.example")
+    caplog.set_level(DEBUG)
+    engine._schedule_request(drop_request, spider)
+    assert scheduler.enqueued == [
+        keep_request
+    ], f"{scheduler.enqueued!r} != [{keep_request!r}]"
+    assert "dropped request <GET https://drop.example>" in caplog.text
+    crawler.signals.disconnect(signal_handler, request_scheduled)
+
+
 if __name__ == "__main__":
     if len(sys.argv) > 1 and sys.argv[1] == "runserver":
         start_test_site(debug=True)
diff --git a/tests/test_utils_project.py b/tests/test_utils_project.py
index 90bd350a5..3831f4c21 100644
--- a/tests/test_utils_project.py
+++ b/tests/test_utils_project.py
@@ -6,6 +6,7 @@ import unittest
 import warnings
 from pathlib import Path
 
+from scrapy.utils.misc import set_environ
 from scrapy.utils.project import data_path, get_project_settings
 
 
@@ -38,20 +39,6 @@ class ProjectUtilsTest(unittest.TestCase):
             self.assertEqual(abspath, data_path(abspath))
 
 
-@contextlib.contextmanager
-def set_env(**update):
-    modified = set(update.keys()) & set(os.environ.keys())
-    update_after = {k: os.environ[k] for k in modified}
-    remove_after = frozenset(k for k in update if k not in os.environ)
-    try:
-        os.environ.update(update)
-        yield
-    finally:
-        os.environ.update(update_after)
-        for k in remove_after:
-            os.environ.pop(k)
-
-
 class GetProjectSettingsTestCase(unittest.TestCase):
     def test_valid_envvar(self):
         value = "tests.test_cmdline.settings"
@@ -60,7 +47,7 @@ class GetProjectSettingsTestCase(unittest.TestCase):
         }
         with warnings.catch_warnings():
             warnings.simplefilter("error")
-            with set_env(**envvars):
+            with set_environ(**envvars):
                 settings = get_project_settings()
 
         assert settings.get("SETTINGS_MODULE") == value
@@ -69,7 +56,7 @@ class GetProjectSettingsTestCase(unittest.TestCase):
         envvars = {
             "SCRAPY_FOO": "bar",
         }
-        with set_env(**envvars):
+        with set_environ(**envvars):
             settings = get_project_settings()
 
         assert settings.get("SCRAPY_FOO") is None
@@ -80,7 +67,7 @@ class GetProjectSettingsTestCase(unittest.TestCase):
             "SCRAPY_FOO": "bar",
             "SCRAPY_SETTINGS_MODULE": value,
         }
-        with set_env(**envvars):
+        with set_environ(**envvars):
             settings = get_project_settings()
         assert settings.get("SETTINGS_MODULE") == value
         assert settings.get("SCRAPY_FOO") is None