mirror of https://github.com/scrapy/scrapy.git
234 lines
8.0 KiB
Python
234 lines
8.0 KiB
Python
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)
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=[allowed_domain])
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
request = Request(url)
|
|
if allowed:
|
|
assert mw.process_request(request) is None
|
|
else:
|
|
with pytest.raises(IgnoreRequest):
|
|
mw.process_request(request)
|
|
|
|
|
|
@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)
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=["a.example"])
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.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)
|
|
else:
|
|
assert mw.process_request(request) is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("allow_offsite", "dont_filter", "filtered"),
|
|
[
|
|
(True, UNSET, False),
|
|
(True, None, False),
|
|
(True, False, False),
|
|
(True, True, False),
|
|
(False, UNSET, True),
|
|
(False, None, True),
|
|
(False, False, True),
|
|
(False, True, False),
|
|
],
|
|
)
|
|
def test_process_request_allow_offsite(allow_offsite, dont_filter, filtered):
|
|
crawler = get_crawler(Spider)
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=["a.example"])
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
kwargs = {"meta": {}}
|
|
if allow_offsite is not UNSET:
|
|
kwargs["meta"]["allow_offsite"] = allow_offsite
|
|
if dont_filter is not UNSET:
|
|
kwargs["dont_filter"] = dont_filter
|
|
request = Request("https://b.example", **kwargs)
|
|
if filtered:
|
|
with pytest.raises(IgnoreRequest):
|
|
mw.process_request(request)
|
|
else:
|
|
assert mw.process_request(request) 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
|
|
crawler.spider = crawler._create_spider(name="a", **kwargs)
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
request = Request("https://example.com")
|
|
assert mw.process_request(request) is None
|
|
|
|
|
|
def test_process_request_invalid_domains():
|
|
crawler = get_crawler(Spider)
|
|
allowed_domains = ["a.example", None, "http:////b.example", "//c.example"]
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=allowed_domains)
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
request = Request("https://a.example")
|
|
assert mw.process_request(request) is None
|
|
for letter in ("b", "c"):
|
|
request = Request(f"https://{letter}.example")
|
|
with pytest.raises(IgnoreRequest):
|
|
mw.process_request(request)
|
|
|
|
|
|
@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)
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=[allowed_domain])
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
request = Request(url)
|
|
if allowed:
|
|
assert mw.request_scheduled(request, crawler.spider) is None
|
|
else:
|
|
with pytest.raises(IgnoreRequest):
|
|
mw.request_scheduled(request, crawler.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)
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=["a.example"])
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.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, crawler.spider)
|
|
else:
|
|
assert mw.request_scheduled(request, crawler.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
|
|
crawler.spider = crawler._create_spider(name="a", **kwargs)
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
request = Request("https://example.com")
|
|
assert mw.request_scheduled(request, crawler.spider) is None
|
|
|
|
|
|
def test_request_scheduled_invalid_domains():
|
|
crawler = get_crawler(Spider)
|
|
allowed_domains = ["a.example", None, "http:////b.example", "//c.example"]
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=allowed_domains)
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
request = Request("https://a.example")
|
|
assert mw.request_scheduled(request, crawler.spider) is None
|
|
for letter in ("b", "c"):
|
|
request = Request(f"https://{letter}.example")
|
|
with pytest.raises(IgnoreRequest):
|
|
mw.request_scheduled(request, crawler.spider)
|
|
|
|
|
|
def test_repeated_offsite_domain():
|
|
crawler = get_crawler(Spider)
|
|
crawler.spider = crawler._create_spider(name="a", allowed_domains=["example.com"])
|
|
mw = OffsiteMiddleware.from_crawler(crawler)
|
|
mw.spider_opened(crawler.spider)
|
|
req1 = Request("http://other.org/1")
|
|
req2 = Request("http://other.org/2")
|
|
with pytest.raises(IgnoreRequest):
|
|
mw.process_request(req1)
|
|
assert "other.org" in mw.domains_seen
|
|
assert crawler.stats.get_value("offsite/domains") == 1
|
|
assert crawler.stats.get_value("offsite/filtered") == 1
|
|
with pytest.raises(IgnoreRequest):
|
|
mw.process_request(req2)
|
|
assert crawler.stats.get_value("offsite/domains") == 1 # not incremented again
|
|
assert crawler.stats.get_value("offsite/filtered") == 2
|