RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add REDIRECT_HTTP_CODES for 201 redirect support and Response.follow_redirect()

This commit is contained in:
Adrian Chaves 2026-07-28 17:29:05 +02:00
parent e7d8b34e73
commit 6d2f728377
17 changed files with 510 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
docs/topics/downloader-middleware.rst
View File

@ -882,6 +882,9 @@ RedirectMiddleware
This middleware handles redirection of requests based on response status.
Requests are redirected to the URL from the ``Location`` header of responses
whose status is in :setting:`REDIRECT_HTTP_CODES`.
.. reqmeta:: redirect_urls
The urls which the request goes through (while being redirected) can be found
@ -902,6 +905,7 @@ The :class:`RedirectMiddleware` can be configured through the following
settings (see the settings documentation for more info):
* :setting:`REDIRECT_ENABLED`
* :setting:`REDIRECT_HTTP_CODES`
* :setting:`REDIRECT_MAX_TIMES`
.. reqmeta:: dont_redirect
@ -909,6 +913,13 @@ settings (see the settings documentation for more info):
If :attr:`Request.meta <scrapy.Request.meta>` has ``dont_redirect``
key set to True, the request will be ignored by this middleware.
.. reqmeta:: redirect_http_codes
.. versionadded:: VERSION
The ``redirect_http_codes`` key of :attr:`Request.meta <scrapy.Request.meta>`
overrides the :setting:`REDIRECT_HTTP_CODES` setting for a single request.
If you want to handle some redirect status codes in your spider, you can
specify these in the ``handle_httpstatus_list`` spider attribute.
@ -939,6 +950,53 @@ Default: ``True``
Whether the Redirect middleware will be enabled.
.. setting:: REDIRECT_HTTP_CODES
REDIRECT_HTTP_CODES
^^^^^^^^^^^^^^^^^^^
.. versionadded:: VERSION
Default: ``[301, 302, 303, 307, 308]``
Response status codes whose ``Location`` header is followed.
The default value contains the status codes for which the HTTP standard defines
redirection handling. Add other status codes that may report a ``Location``
header to have it followed as well.
For example, an HTTP 201 (Created) response reports that the request created a
resource, and its ``Location`` header identifies that resource. Because that is
not a redirection target, it is not followed by default, matching the behavior
of web browsers. However, servers that create resources on the fly, e.g. to
serve an image that they generate on demand, sometimes send an empty 201
response and expect a separate request for the ``Location`` URL:
.. code-block:: python
REDIRECT_HTTP_CODES = [201, 301, 302, 303, 307, 308]
Status codes other than the ones in the default value get conservative
handling, because their redirection semantics are unknown:
- Their ``Location`` header is only followed if the response has an empty
body, as a response that carries a resource would otherwise be discarded.
- They are followed like 303 responses, i.e. with a GET request without a
body, unless the original request used the GET or HEAD method, which is
kept.
Use the :reqmeta:`redirect_http_codes` :attr:`Request.meta
<scrapy.Request.meta>` key to override this setting for a single request, and
:meth:`Response.follow_redirect() <scrapy.http.Response.follow_redirect>` to
follow a ``Location`` header from a spider callback instead.
To stop following a status code, you can also use the ``handle_httpstatus_list``
spider attribute or :attr:`Request.meta <scrapy.Request.meta>` key, which
additionally lets your spider callbacks receive the affected responses.
For media pipelines, see also :setting:`MEDIA_ALLOW_REDIRECTS`.
.. setting:: REDIRECT_MAX_TIMES
REDIRECT_MAX_TIMES

22
docs/topics/media-pipeline.rst
View File

@ -497,6 +497,28 @@ To handle media redirections, set this setting to ``True``:
MEDIA_ALLOW_REDIRECTS = True
Media files created on the fly
------------------------------
.. versionchanged:: VERSION
Added support for HTTP 201 responses.
Some servers create media files on the fly and report that with an HTTP 201
(Created) response instead of an HTTP 200 (OK) one. Media pipelines treat both
as successful downloads.
If such a response has an empty body, the media file must be requested
separately, from the URL in its ``Location`` header. To do that, add ``201`` to
:setting:`REDIRECT_HTTP_CODES`, so that
:class:`~scrapy.downloadermiddlewares.redirect.RedirectMiddleware` follows that
header, and enable :setting:`MEDIA_ALLOW_REDIRECTS`, so that media requests may
be redirected:
.. code-block:: python
REDIRECT_HTTP_CODES = [201, 301, 302, 303, 307, 308]
MEDIA_ALLOW_REDIRECTS = True
.. _topics-media-pipeline-override:
Extending the Media Pipelines

3
docs/topics/request-response.rst
View File

@ -736,6 +736,7 @@ Those are:
* :reqmeta:`is_start_request`
* :reqmeta:`max_retry_times`
* :reqmeta:`proxy`
* :reqmeta:`redirect_http_codes`
* :reqmeta:`redirect_reasons`
* :reqmeta:`redirect_urls`
* :reqmeta:`referrer_policy`
@ -1153,6 +1154,8 @@ Response objects
.. automethod:: Response.follow_all
.. automethod:: Response.follow_redirect
.. _topics-request-response-ref-response-subclasses:

62
scrapy/downloadermiddlewares/redirect.py
View File

@ -2,9 +2,6 @@ from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import urljoin, urlparse
from w3lib.url import safe_url_string
from scrapy import signals
from scrapy.exceptions import IgnoreRequest, NotConfigured
@ -14,6 +11,7 @@ from scrapy.utils.decorators import _warn_spider_arg
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.python import global_object_name
from scrapy.utils.response import get_meta_refresh
from scrapy.utils.url import _redirect_url
if TYPE_CHECKING:
# typing.Self requires Python 3.11
@ -27,6 +25,12 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
# Status codes for which the HTTP standard defines redirection handling. Other
# status codes from REDIRECT_HTTP_CODES get conservative handling, see
# RedirectMiddleware.
_STANDARD_HTTP_CODES = frozenset({301, 302, 303, 307, 308})
class BaseRedirectMiddleware:
crawler: Crawler
enabled_setting: str = "REDIRECT_ENABLED"
@ -198,6 +202,28 @@ class BaseRedirectMiddleware:
class RedirectMiddleware(BaseRedirectMiddleware):
"""Handle redirection of requests based on response status."""
def __init__(self, settings: BaseSettings):
super().__init__(settings)
self.redirect_http_codes: set[int] = {
int(x) for x in settings.getlist("REDIRECT_HTTP_CODES")
}
def _is_redirect(self, request: Request, response: Response) -> bool:
meta_http_codes = request.meta.get("redirect_http_codes")
http_codes = (
{int(x) for x in meta_http_codes}
if meta_http_codes is not None
else self.redirect_http_codes
)
if response.status not in http_codes:
return False
if response.status in _STANDARD_HTTP_CODES:
return True
# For other status codes, e.g. 201, Location identifies a resource
# rather than a redirection target, and the response body may be that
# resource, which following Location would discard.
return not response.body
@_warn_spider_arg
def process_response(
self, request: Request, response: Response, spider: Spider | None = None
@ -211,34 +237,26 @@ class RedirectMiddleware(BaseRedirectMiddleware):
):
return response
if "Location" not in response.headers or response.status not in {
301,
302,
303,
307,
308,
}:
if "Location" not in response.headers or not self._is_redirect(
request, response
):
return response
assert response.headers["Location"] is not None
location = safe_url_string(response.headers["Location"])
if response.headers["Location"].startswith(b"//"):
request_scheme = urlparse_cached(request).scheme
location = request_scheme + "://" + location.lstrip("/")
redirected_url = urljoin(request.url, location)
if not urlparse(redirected_url).fragment:
fragment = urlparse_cached(request).fragment
if fragment:
redirected_url = urljoin(redirected_url, f"#{fragment}")
redirected_url = _redirect_url(request.url, response.headers["Location"])
redirected = self._build_redirect_request(request, response, url=redirected_url)
if urlparse_cached(redirected).scheme not in {"http", "https"}:
return response
# 307 and 308 responses keep the method and the body of the original
# request as is, and so do 301 and 302 responses, except for POST
# requests, for historical reasons. Any other status code gets the
# handling of 303 responses, a bodyless GET request, which is the safest
# choice for status codes with no defined redirection semantics.
if (response.status in {301, 302} and request.method == "POST") or (
response.status == 303 and request.method not in {"GET", "HEAD"}
response.status not in {301, 302, 307, 308}
and request.method not in {"GET", "HEAD"}
):
redirected = self._redirect_request_using_get(
request, response, redirected_url

26
scrapy/http/response/__init__.py
View File

@ -15,6 +15,7 @@ from scrapy.http.headers import Headers
from scrapy.http.request import Request
from scrapy.link import Link
from scrapy.utils.trackref import object_ref
from scrapy.utils.url import _redirect_url
if TYPE_CHECKING:
from collections.abc import Callable, Iterable, Mapping
@ -277,6 +278,31 @@ class Response(object_ref):
flags=flags,
)
def follow_redirect(self, **kwargs: Any) -> Request:
"""Return a :class:`~.Request` instance to follow the ``Location``
header of this response.
.. versionadded:: VERSION
It accepts the same arguments as :meth:`follow`, except ``url``, which
comes from the ``Location`` header. Unlike in :meth:`follow`, that URL is
always resolved against :attr:`url`, never against the base URL of an
HTML document, as the HTTP standard mandates for the ``Location``
header.
It raises :exc:`ValueError` if this response has no ``Location`` header.
Note that the resulting request uses the GET method and has no body,
unless you specify otherwise. To reproduce instead how
:class:`~scrapy.downloadermiddlewares.redirect.RedirectMiddleware`
follows the ``Location`` header of a given response status, see
:setting:`REDIRECT_HTTP_CODES`.
"""
location = self.headers.get("Location")
if location is None:
raise ValueError(f"{self} has no Location header")
return self.follow(_redirect_url(self.url, location), **kwargs)
def follow_all(
self,
urls: Iterable[str | Link],

2
scrapy/pipelines/files.py
View File

@ -627,7 +627,7 @@ class FilesPipeline(MediaPipeline):
) -> FileInfo:
referer = referer_str(request)
if response.status != 200:
if response.status not in {200, 201}:
logger.warning(
"File (code: %(status)s): Error downloading file from "
"%(request)s referred in <%(referer)s>",

9
scrapy/pipelines/media.py
View File

@ -109,6 +109,9 @@ class MediaPipeline(ABC):
resolve = functools.partial(
self._key_for_pipe, base_class_name="MediaPipeline", settings=settings
)
self._redirect_http_codes: set[int] = {
int(x) for x in settings.getlist("REDIRECT_HTTP_CODES")
}
self.allow_redirects: bool = settings.getbool(
resolve("MEDIA_ALLOW_REDIRECTS"), False
)
@ -117,7 +120,11 @@ class MediaPipeline(ABC):
def _handle_statuses(self, allow_redirects: bool) -> None:
self.handle_httpstatus_list = None
if allow_redirects:
self.handle_httpstatus_list = SequenceExclude(range(300, 400))
# Statuses that RedirectMiddleware may redirect, i.e. those from
# REDIRECT_HTTP_CODES, are left for it to handle.
self.handle_httpstatus_list = SequenceExclude(
{*range(300, 400), *self._redirect_http_codes}
)
def _key_for_pipe(
self,

2
scrapy/settings/default_settings.py
View File

@ -169,6 +169,7 @@ __all__ = [
"RANDOMIZE_DOWNLOAD_DELAY",
"REACTOR_THREADPOOL_MAXSIZE",
"REDIRECT_ENABLED",
"REDIRECT_HTTP_CODES",
"REDIRECT_MAX_TIMES",
"REDIRECT_PRIORITY_ADJUST",
"REFERER_ENABLED",
@ -491,6 +492,7 @@ RANDOMIZE_DOWNLOAD_DELAY = True
REACTOR_THREADPOOL_MAXSIZE = 10
REDIRECT_ENABLED = True
REDIRECT_HTTP_CODES = [301, 302, 303, 307, 308]
REDIRECT_MAX_TIMES = 20 # uses Firefox default setting
REDIRECT_PRIORITY_ADJUST = +2

21
scrapy/utils/url.py
View File

@ -7,9 +7,11 @@ from __future__ import annotations
import re
from typing import TYPE_CHECKING, TypeAlias
from urllib.parse import ParseResult, urlparse, urlunparse
from urllib.parse import ParseResult, urljoin, urlparse, urlunparse
from w3lib.url import any_to_uri, parse_url
from w3lib.url import any_to_uri, parse_url, safe_url_string
from scrapy.utils.python import to_bytes
if TYPE_CHECKING:
from collections.abc import Iterable
@ -152,3 +154,18 @@ def strip_url(
"" if strip_fragment else parsed_url.fragment,
)
)
def _redirect_url(url: str, location: str | bytes) -> str:
"""Return the absolute URL that the *location* value of the ``Location``
header of a response to a request to *url* points to."""
target = safe_url_string(location)
parsed_url = urlparse(url)
if to_bytes(location).startswith(b"//"):
# safe_url_string() may drop leading slashes of a scheme-relative URL,
# so build the absolute URL without relying on urljoin().
target = f"{parsed_url.scheme}://{target.lstrip('/')}"
redirect_url = urljoin(url, target)
if not urlparse(redirect_url).fragment and parsed_url.fragment:
redirect_url = urljoin(redirect_url, f"#{parsed_url.fragment}")
return redirect_url

2
tests/mockserver/http.py
View File

@ -17,6 +17,7 @@ from .http_resources import (
ClientIPResource,
Compress,
ContentLengthHeaderResource,
Created,
Delay,
Drop,
DuplicateHeaderResource,
@ -52,6 +53,7 @@ class Root(resource.Resource):
self.putChild(b"alpayload", ArbitraryLengthPayloadResource())
self.putChild(b"static", File(str(Path(tests_datadir, "test_site/"))))
self.putChild(b"redirect-to", RedirectTo())
self.putChild(b"created", Created())
self.putChild(b"text", Data(b"Works", "text/plain"))
self.putChild(
b"html",

10
tests/mockserver/http_resources.py
View File

@ -206,6 +206,16 @@ class RedirectTo(LeafResource):
return b"redirecting..."
class Created(LeafResource):
"""Emulate a server that creates a resource on the fly, reporting it with an
empty 201 response that points at the created resource through Location."""
def render(self, request):
request.setResponseCode(201)
request.setHeader(b"Location", getarg(request, b"goto", b"/"))
return b""
class Partial(LeafResource):
def render_GET(self, request):
request.setHeader(b"Content-Length", b"1024")

136
tests/test_downloadermiddleware_redirect.py
View File

@ -1,4 +1,7 @@
from __future__ import annotations
import logging
from abc import ABC
from unittest.mock import MagicMock
import pytest
@ -348,6 +351,139 @@ class TestRedirectMiddleware(TestRedirectBase):
response = Response(request.url, status=302)
assert self.mw.process_response(request, response) is response
@pytest.mark.parametrize("status", [201, 300])
def test_status_not_in_redirect_http_codes(self, status):
"""Status codes outside REDIRECT_HTTP_CODES are not followed, even if they
report a Location header, as 201 (Created) or 300 (Multiple Choices) may
do."""
url = f"https://example.com/{status}"
request = Request(url)
response = Response(url, status=status, headers={"Location": "/target"})
assert self.mw.process_response(request, response) is response
def test_standard_status_not_in_redirect_http_codes(self):
crawler = get_crawler(DefaultSpider, {"REDIRECT_HTTP_CODES": [301]})
crawler.spider = crawler._create_spider()
mw = RedirectMiddleware.from_crawler(crawler)
url = "https://example.com/302"
request = Request(url)
response = Response(url, status=302, headers={"Location": "/redirected"})
assert mw.process_response(request, response) is response
def test_meta_http_codes(self):
"""The redirect_http_codes request metadata key overrides the
REDIRECT_HTTP_CODES setting."""
url = "https://example.com/201"
request = Request(url, meta={"redirect_http_codes": [201]})
response = Response(url, status=201, headers={"Location": "/created"})
redirect_request = self.mw.process_response(request, response)
assert isinstance(redirect_request, Request)
assert redirect_request.url == "https://example.com/created"
def test_meta_http_codes_excluding_standard_status(self):
url = "https://example.com/302"
request = Request(url, meta={"redirect_http_codes": [201]})
response = Response(url, status=302, headers={"Location": "/redirected"})
assert self.mw.process_response(request, response) is response
class NonStandardStatusMixin(ABC):
"""Tests of status codes added to REDIRECT_HTTP_CODES for which the HTTP
standard does not define redirection handling, and which therefore get
conservative handling."""
status: int
def setup_method(self):
crawler = get_crawler(
DefaultSpider,
{"REDIRECT_HTTP_CODES": [self.status, 301, 302, 303, 307, 308]},
)
crawler.spider = crawler._create_spider()
self.mw = RedirectMiddleware.from_crawler(crawler)
self.url = f"https://example.com/{self.status}"
def _response(
self, body: bytes = b"", location: str | None = "/target"
) -> Response:
headers = {"Location": location} if location else {}
return Response(self.url, status=self.status, headers=headers, body=body)
def test_empty_body(self):
redirect_request = self.mw.process_response(Request(self.url), self._response())
assert isinstance(redirect_request, Request)
assert redirect_request.url == "https://example.com/target"
assert redirect_request.method == "GET"
assert redirect_request.meta["redirect_urls"] == [self.url]
assert redirect_request.meta["redirect_reasons"] == [self.status]
def test_non_empty_body(self):
"""A response that carries a resource in its body is not followed, doing so
would discard that resource."""
response = self._response(body=b"file content")
assert self.mw.process_response(Request(self.url), response) is response
def test_no_location(self):
response = self._response(location=None)
assert self.mw.process_response(Request(self.url), response) is response
@pytest.mark.parametrize("method", ["POST", "PUT"])
def test_method_becomes_get(self, method):
request = Request(
self.url,
method=method,
body=b"payload",
headers={"Content-Type": "text/plain", "Content-Length": "7"},
)
redirect_request = self.mw.process_response(request, self._response())
assert isinstance(redirect_request, Request)
assert redirect_request.method == "GET"
assert not redirect_request.body
assert "Content-Type" not in redirect_request.headers
assert "Content-Length" not in redirect_request.headers
def test_head_method_preserved(self):
request = Request(self.url, method="HEAD")
redirect_request = self.mw.process_response(request, self._response())
assert isinstance(redirect_request, Request)
assert redirect_request.method == "HEAD"
@pytest.mark.parametrize(
"meta",
[
{"dont_redirect": True},
{"handle_httpstatus_all": True},
],
)
def test_request_meta_handling(self, meta):
request = Request(self.url, meta=meta)
response = self._response()
assert self.mw.process_response(request, response) is response
def test_request_meta_status_handling(self):
request = Request(self.url, meta={"handle_httpstatus_list": [self.status]})
response = self._response()
assert self.mw.process_response(request, response) is response
def test_spider_handling(self):
self.mw.crawler.spider.handle_httpstatus_list = [self.status]
response = self._response()
assert self.mw.process_response(Request(self.url), response) is response
class TestRedirect201(NonStandardStatusMixin):
"""201 (Created) reports the created resource through Location."""
status = 201
class TestRedirect300(NonStandardStatusMixin):
"""300 (Multiple Choices) may report a preferred choice through Location. It
is a 3xx status code, but the standard defines no redirection handling for
it, so it gets the same conservative handling as 201."""
status = 300
@pytest.mark.parametrize(SCHEME_PARAMS, REDIRECT_SCHEME_CASES)
def test_redirect_schemes(url, location, target):

12
tests/test_http_response_text.py
View File

@ -305,6 +305,18 @@ class TestTextResponse(TestResponseBase):
absolute = "http://www.example.com/elsewhere/test"
assert joined == absolute
def test_follow_redirect_ignores_base_url(self):
"""The Location header is resolved against the response URL, not against
the base URL of the document, which only applies to its own links."""
body = b'<html><body><base href="https://example.net"></body></html>'
response = self.response_class(
"http://www.example.com/dir/index",
body=body,
headers={"Location": "/test"},
)
assert response.urljoin("/test") == "https://example.net/test"
assert response.follow_redirect().url == "http://www.example.com/test"
def test_follow_selector(self):
resp = self._links_response()
urls = [

64
tests/test_pipeline_crawl.py
View File

@ -57,6 +57,17 @@ class RedirectedMediaDownloadSpider(MediaDownloadSpider):
)
class CreatedMediaDownloadSpider(MediaDownloadSpider):
"""Requests media files from a server that creates them on the fly, reports
that with an empty 201 response, and points at the created file through the
Location header."""
name = "createdmedia"
def _process_url(self, url):
return add_or_replace_parameter(self.mockserver.url("/created"), "goto", url)
class TestFileDownloadCrawl:
mockserver: MockServer
@ -207,6 +218,59 @@ class TestFileDownloadCrawl:
assert crawler.stats
assert crawler.stats.get_value("downloader/response_status_count/302") == 3
@pytest.mark.parametrize(
"settings",
[
{},
# Following the Location header of a 201 response requires both
# settings.
{"REDIRECT_HTTP_CODES": [201, 301, 302, 303, 307, 308]},
{"MEDIA_ALLOW_REDIRECTS": True},
],
)
@coroutine_test
async def test_download_media_created_default_failure(
self, settings: dict[str, Any], caplog: pytest.LogCaptureFixture
) -> None:
crawler = self._create_crawler(
CreatedMediaDownloadSpider, {**self.settings, **settings}
)
with caplog.at_level(logging.DEBUG):
await crawler.crawl_async(
self.mockserver.url("/static/files/images/"),
media_key=self.media_key,
media_urls_key=self.media_urls_key,
mockserver=self.mockserver,
)
assert len(self.items) == 1
assert not self.items[0][self.media_key]
assert crawler.stats
assert crawler.stats.get_value("downloader/response_status_count/201") == 3
assert caplog.text.count("File (empty-content): Empty file from") == 3
assert not list(self.tmpmediastore.iterdir())
@coroutine_test
async def test_download_media_created_allowed(
self, caplog: pytest.LogCaptureFixture
) -> None:
settings = {
**self.settings,
"REDIRECT_HTTP_CODES": [201, 301, 302, 303, 307, 308],
"MEDIA_ALLOW_REDIRECTS": True,
}
crawler = self._create_crawler(CreatedMediaDownloadSpider, settings)
with caplog.at_level(logging.DEBUG):
await crawler.crawl_async(
self.mockserver.url("/static/files/images/"),
media_key=self.media_key,
media_urls_key=self.media_urls_key,
mockserver=self.mockserver,
)
self._assert_files_downloaded(self.items, caplog.text)
assert crawler.stats
assert crawler.stats.get_value("downloader/response_status_count/201") == 3
assert caplog.text.count("Redirecting (201)") == 3
@coroutine_test
async def test_download_media_file_path_error(
self, caplog: pytest.LogCaptureFixture

42
tests/test_pipeline_files.py
View File

@ -277,6 +277,39 @@ class TestFilesPipeline:
assert path.exists()
assert path.read_bytes() == b"data"
@coroutine_test
async def test_file_created(self) -> None:
"""A 201 response with the file in its body is a successful download."""
item_url = "http://example.com/created.pdf"
item = _create_item_with_files(item_url)
with mock.patch.object(
FilesPipeline,
"get_media_requests",
return_value=[_prepare_request_object(item_url, status=201)],
):
result = await self.pipeline.process_item(item)
assert result["files"][0]["status"] == "downloaded"
path = Path(self.tempdir) / result["files"][0]["path"]
assert path.read_bytes() == b"data"
@coroutine_test
async def test_file_created_empty(self, caplog: pytest.LogCaptureFixture) -> None:
"""A 201 response without a body is an empty download, e.g. because its
Location header was not followed, 201 not being in REDIRECT_HTTP_CODES."""
item_url = "http://example.com/created-empty.pdf"
item = _create_item_with_files(item_url)
with (
caplog.at_level(logging.WARNING),
mock.patch.object(
FilesPipeline,
"get_media_requests",
return_value=[_prepare_request_object(item_url, status=201, body=b"")],
),
):
result = await self.pipeline.process_item(item)
assert result["files"] == []
assert "File (empty-content)" in caplog.text
def test_file_path_from_item(self):
"""
Custom file path based on item data, overriding default implementation
@ -913,10 +946,15 @@ def _create_item_with_files(*files: str) -> ItemWithFiles:
return item
def _prepare_request_object(item_url: str, flags: list[str] | None = None) -> Request:
def _prepare_request_object(
item_url: str,
flags: list[str] | None = None,
status: int = 200,
body: bytes = b"data",
) -> Request:
return Request(
item_url,
meta={"response": Response(item_url, status=200, body=b"data", flags=flags)},
meta={"response": Response(item_url, status=status, body=body, flags=flags)},
)

27
tests/test_pipeline_media.py
View File

@ -386,7 +386,7 @@ class TestAsyncMediaDownloaded(TestMediaPipeline):
class TestMediaPipelineAllowRedirectSettings:
def _assert_request_no3xx(self, pipeline_class, settings):
def _assert_redirect_statuses_not_handled(self, pipeline_class, settings):
pipe = pipeline_class(crawler=get_crawler(None, settings))
request = Request("http://url")
pipe._modify_media_request(request)
@ -394,6 +394,8 @@ class TestMediaPipelineAllowRedirectSettings:
assert "handle_httpstatus_list" in request.meta
for status, check in [
(200, True),
# 201 is not in REDIRECT_HTTP_CODES by default
(201, True),
# These are the status codes we want
# the downloader to handle itself
(301, False),
@ -412,13 +414,32 @@ class TestMediaPipelineAllowRedirectSettings:
assert status not in request.meta["handle_httpstatus_list"]
def test_subclass_standard_setting(self):
self._assert_request_no3xx(UserDefinedPipeline, {"MEDIA_ALLOW_REDIRECTS": True})
self._assert_redirect_statuses_not_handled(
UserDefinedPipeline, {"MEDIA_ALLOW_REDIRECTS": True}
)
def test_subclass_specific_setting(self):
self._assert_request_no3xx(
self._assert_redirect_statuses_not_handled(
UserDefinedPipeline, {"USERDEFINEDPIPELINE_MEDIA_ALLOW_REDIRECTS": True}
)
def test_custom_redirect_http_codes(self):
"""Statuses added to REDIRECT_HTTP_CODES are also left to
RedirectMiddleware."""
crawler = get_crawler(
None,
{
"MEDIA_ALLOW_REDIRECTS": True,
"REDIRECT_HTTP_CODES": [201, 301, 302, 303, 307, 308],
},
)
request = Request("http://url")
UserDefinedPipeline(crawler=crawler)._modify_media_request(request)
assert 201 not in request.meta["handle_httpstatus_list"]
assert 302 not in request.meta["handle_httpstatus_list"]
assert 200 in request.meta["handle_httpstatus_list"]
class TestBuildFromCrawler:
def setup_method(self):

43
tests/utils/bases/http_response.py
View File

@ -286,6 +286,49 @@ class TestResponseBase(ABC):
fol = res.follow("http://example.com/", flags=["cached", "allowed"])
assert fol.flags == ["cached", "allowed"]
# Response.follow_redirect
@pytest.mark.parametrize(
("location", "target_url"),
[
("http://foo.example.com/x", "http://foo.example.com/x"),
("/foo", "http://example.com/foo"),
("foo", "http://example.com/dir/foo"),
(b"//foo.example.com/x", "http://foo.example.com/x"),
(b"/a\xe7\xe3o", "http://example.com/a%E7%E3o"),
],
)
def test_follow_redirect(self, location, target_url):
response = self.response_class(
"http://example.com/dir/index", headers={"Location": location}
)
request = response.follow_redirect()
assert request.url == target_url
assert request.method == "GET"
def test_follow_redirect_keeps_fragment(self):
response = self.response_class(
"http://example.com/index#frag", headers={"Location": "/foo"}
)
assert response.follow_redirect().url == "http://example.com/foo#frag"
def test_follow_redirect_no_location(self):
response = self.response_class("http://example.com")
with pytest.raises(ValueError, match="has no Location header"):
response.follow_redirect()
def test_follow_redirect_kwargs(self):
response = self.response_class(
"http://example.com/index", headers={"Location": "/foo"}
)
request = response.follow_redirect(
method="HEAD", meta={"foo": "bar"}, flags=["cached"]
)
assert request.url == "http://example.com/foo"
assert request.method == "HEAD"
assert request.meta["foo"] == "bar"
assert request.flags == ["cached"]
# Response.follow_all
def test_follow_all_absolute(self):