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