Add regression tests for chain-aware redirect dupefilter fix

Adds test_redirect_chain_loop_bypasses_dupefilter, covering an
A -> B -> C -> A redirect loop, plus the single-hop
test_self_redirect_bypasses_dupefilter and the
test_cross_request_duplicate_redirect_still_filtered regression
guard.

Refs #1225
This commit is contained in:
Dev Kumar 2026-07-21 07:56:41 -05:00
parent 901ec3f08c
commit 6122180869
1 changed files with 79 additions and 0 deletions

View File

@ -4,6 +4,7 @@ from unittest.mock import MagicMock
import pytest
from scrapy.downloadermiddlewares.redirect import RedirectMiddleware
from scrapy.dupefilters import RFPDupeFilter
from scrapy.exceptions import NotConfigured
from scrapy.http import Request, Response
from scrapy.spidermiddlewares.referer import (
@ -296,6 +297,84 @@ class TestRedirectMiddleware(Base.Test):
assert req2.url == url3
assert req2.method == "HEAD"
def test_self_redirect_bypasses_dupefilter(self):
# A redirect target that fingerprints identically to the request
# that produced it (e.g. a literal self-redirect, or a redirect
# whose query-string parameters are just reordered, which the
# fingerprinter canonicalizes away) must not be dropped by the
# dupe filter, since the filter already marked that fingerprint
# as seen when the source request was scheduled (see #1225).
url = "http://www.example.com/302?a=1&b=2"
redirected_url = "http://www.example.com/302?b=2&a=1"
req = Request(url)
rsp = Response(url, headers={"Location": redirected_url}, status=302)
req2 = self.mw.process_response(req, rsp)
assert isinstance(req2, Request)
assert req2.url == redirected_url
assert req2.dont_filter is True
fingerprint = self.mw.crawler.request_fingerprinter.fingerprint
assert fingerprint(req2) == fingerprint(req)
dupefilter = RFPDupeFilter.from_crawler(self.mw.crawler)
assert dupefilter.request_seen(req) is False
# Without the dont_filter bypass, the scheduler would treat req2 as
# an already-seen duplicate of req and silently drop it.
assert dupefilter.request_seen(req2) is True
def test_redirect_chain_loop_bypasses_dupefilter(self):
# A redirect chain that loops back to an *earlier* request in the
# same chain (A -> B -> C -> A) must also bypass the dupe filter for
# the final hop, even though that hop's immediate predecessor (C)
# fingerprints differently from A. A fix that only compares a
# redirect against its direct predecessor would miss this case.
dupefilter = RFPDupeFilter.from_crawler(self.mw.crawler)
fingerprint = self.mw.crawler.request_fingerprinter.fingerprint
req_a = Request("http://www.example.com/a")
assert dupefilter.request_seen(req_a) is False
rsp_a = Response(
req_a.url, headers={"Location": "http://www.example.com/b"}, status=302
)
req_b = self.mw.process_response(req_a, rsp_a)
assert isinstance(req_b, Request)
assert req_b.dont_filter is False
assert dupefilter.request_seen(req_b) is False
rsp_b = Response(
req_b.url, headers={"Location": "http://www.example.com/c"}, status=302
)
req_c = self.mw.process_response(req_b, rsp_b)
assert isinstance(req_c, Request)
assert req_c.dont_filter is False
assert dupefilter.request_seen(req_c) is False
rsp_c = Response(
req_c.url, headers={"Location": "http://www.example.com/a"}, status=302
)
req_a_again = self.mw.process_response(req_c, rsp_c)
assert isinstance(req_a_again, Request)
assert fingerprint(req_a_again) == fingerprint(req_a)
assert req_a_again.dont_filter is True
assert dupefilter.request_seen(req_a_again) is True
def test_cross_request_duplicate_redirect_still_filtered(self):
# The dupefilter bypass must stay narrowly scoped: a redirect to a
# URL that genuinely differs from every request in its own chain
# should not be exempted, so normal cross-request dedup (e.g. two
# different pages redirecting to the same canonical URL) keeps
# working as before.
url = "http://www.example.com/302"
redirected_url = "http://www.example.com/redirected"
req = Request(url)
rsp = Response(url, headers={"Location": redirected_url}, status=302)
req2 = self.mw.process_response(req, rsp)
assert isinstance(req2, Request)
assert req2.dont_filter is False
def test_spider_handling(self):
self.mw.crawler.spider.handle_httpstatus_list = [404, 301, 302]
url = "http://www.example.com/301"