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

Migrate away from testfixtures. (#7793)

This commit is contained in:
Andrey Rakhmatullin 2026-07-28 12:52:52 +05:00 committed by GitHub
parent a5bc43e34c
commit 7e8b58a2b2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 775 additions and 722 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pyproject.toml
View File

@ -225,7 +225,6 @@ module = [
"pyftpdlib.*",
"pytest_twisted",
"robotexclusionrulesparser",
"testfixtures",
"zope.interface.*",
]
ignore_missing_imports = true

24
tests/test_downloader_handler_twisted_http2.py
View File

@ -2,11 +2,11 @@
from __future__ import annotations
import logging
import sys
from typing import TYPE_CHECKING, Any
import pytest
from testfixtures import LogCapture
from twisted.web.http import H2_ENABLED
from scrapy import Spider
@ -130,24 +130,24 @@ class TestHttp2(H2DownloadHandlerMixin, TestHttpsBase):
assert response.text == custom_content_length
@coroutine_test
async def test_custom_content_length_bad(self, mockserver: MockServer) -> None:
async def test_custom_content_length_bad(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
request = Request(mockserver.url("/contentlength", is_secure=self.is_secure))
actual_content_length = str(len(request.body))
bad_content_length = str(len(request.body) + 1)
request.headers["Content-Length"] = bad_content_length
async with self.get_dh() as download_handler:
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
response = await download_handler.download_request(request)
assert response.text == actual_content_length
log.check_present(
(
"scrapy.core.http2.stream",
"WARNING",
f"Ignoring bad Content-Length header "
f"{bad_content_length!r} of request {request}, sending "
f"{actual_content_length!r} instead",
)
)
assert (
"scrapy.core.http2.stream",
logging.WARNING,
f"Ignoring bad Content-Length header "
f"{bad_content_length!r} of request {request}, sending "
f"{actual_content_length!r} instead",
) in caplog.record_tuples
@coroutine_test
async def test_data_loss_handling(self, mockserver: MockServer) -> None:

126
tests/test_downloadermiddleware_cookies.py
View File

@ -2,7 +2,6 @@ import logging
from collections.abc import Iterable
import pytest
from testfixtures import LogCapture
from scrapy.downloadermiddlewares.cookies import CookiesMiddleware
from scrapy.downloadermiddlewares.defaultheaders import DefaultHeadersMiddleware
@ -111,14 +110,15 @@ class TestCookiesMiddleware:
CookiesMiddleware,
)
def test_setting_enabled_cookies_debug(self):
def test_setting_enabled_cookies_debug(
self, caplog: pytest.LogCaptureFixture
) -> None:
crawler = get_crawler(settings_dict={"COOKIES_DEBUG": True})
mw = CookiesMiddleware.from_crawler(crawler)
with LogCapture(
"scrapy.downloadermiddlewares.cookies",
propagate=False,
level=logging.DEBUG,
) as log:
caplog.clear()
with caplog.at_level(
logging.DEBUG, logger="scrapy.downloadermiddlewares.cookies"
):
req = Request("http://scrapytest.org/")
res = Response(
"http://scrapytest.org/", headers={"Set-Cookie": "C1=value1; path=/"}
@ -127,43 +127,44 @@ class TestCookiesMiddleware:
req2 = Request("http://scrapytest.org/sub1/")
mw.process_request(req2)
log.check(
(
"scrapy.downloadermiddlewares.cookies",
"DEBUG",
"Received cookies from: <200 http://scrapytest.org/>\n"
"Set-Cookie: C1=value1; path=/\n",
),
(
"scrapy.downloadermiddlewares.cookies",
"DEBUG",
"Sending cookies to: <GET http://scrapytest.org/sub1/>\n"
"Cookie: C1=value1\n",
),
)
assert caplog.record_tuples == [
(
"scrapy.downloadermiddlewares.cookies",
logging.DEBUG,
"Received cookies from: <200 http://scrapytest.org/>\n"
"Set-Cookie: C1=value1; path=/\n",
),
(
"scrapy.downloadermiddlewares.cookies",
logging.DEBUG,
"Sending cookies to: <GET http://scrapytest.org/sub1/>\n"
"Cookie: C1=value1\n",
),
]
def test_debug_no_cookies(self):
def test_debug_no_cookies(self, caplog: pytest.LogCaptureFixture) -> None:
crawler = get_crawler(settings_dict={"COOKIES_DEBUG": True})
mw = CookiesMiddleware.from_crawler(crawler)
with LogCapture(
"scrapy.downloadermiddlewares.cookies",
propagate=False,
level=logging.DEBUG,
) as log:
caplog.clear()
with caplog.at_level(
logging.DEBUG, logger="scrapy.downloadermiddlewares.cookies"
):
req = Request("http://scrapytest.org/")
res = Response("http://scrapytest.org/") # no Set-Cookie header
mw.process_response(req, res)
mw.process_request(req) # no cookies to send either
log.check() # no log output since cl is empty in both cases
# no log output since cl is empty in both cases
assert caplog.record_tuples == []
def test_setting_disabled_cookies_debug(self):
def test_setting_disabled_cookies_debug(
self, caplog: pytest.LogCaptureFixture
) -> None:
crawler = get_crawler(settings_dict={"COOKIES_DEBUG": False})
mw = CookiesMiddleware.from_crawler(crawler)
with LogCapture(
"scrapy.downloadermiddlewares.cookies",
propagate=False,
level=logging.DEBUG,
) as log:
caplog.clear()
with caplog.at_level(
logging.DEBUG, logger="scrapy.downloadermiddlewares.cookies"
):
req = Request("http://scrapytest.org/")
res = Response(
"http://scrapytest.org/", headers={"Set-Cookie": "C1=value1; path=/"}
@ -172,7 +173,7 @@ class TestCookiesMiddleware:
req2 = Request("http://scrapytest.org/sub1/")
mw.process_request(req2)
log.check()
assert caplog.record_tuples == []
def test_do_not_break_on_non_utf8_header(self):
req = Request("http://scrapytest.org/")
@ -420,44 +421,41 @@ class TestCookiesMiddleware:
assert self.mw.process_request(req3) is None
self.assertCookieValEqual(req3.headers["Cookie"], b"a=\xc3\xa1")
def test_invalid_cookies(self):
def test_invalid_cookies(self, caplog: pytest.LogCaptureFixture) -> None:
"""
Invalid cookies are logged as warnings and discarded
"""
with LogCapture(
"scrapy.utils.request",
propagate=False,
level=logging.INFO,
) as lc:
caplog.clear()
with caplog.at_level(logging.INFO, logger="scrapy.utils.request"):
cookies1 = [{"value": "bar"}, {"name": "key", "value": "value1"}]
req1 = Request("http://example.org/1", cookies=cookies1)
req1 = Request("http://example.org/1", cookies=cookies1) # type: ignore[arg-type]
assert self.mw.process_request(req1) is None
cookies2 = [{"name": "foo"}, {"name": "key", "value": "value2"}]
req2 = Request("http://example.org/2", cookies=cookies2)
req2 = Request("http://example.org/2", cookies=cookies2) # type: ignore[arg-type]
assert self.mw.process_request(req2) is None
cookies3 = [{"name": "foo", "value": None}, {"name": "key", "value": ""}]
req3 = Request("http://example.org/3", cookies=cookies3)
req3 = Request("http://example.org/3", cookies=cookies3) # type: ignore[arg-type]
assert self.mw.process_request(req3) is None
lc.check(
(
"scrapy.utils.request",
"WARNING",
"Invalid cookie found in request <GET http://example.org/1>:"
" {'value': 'bar', 'secure': False} ('name' is missing)",
),
(
"scrapy.utils.request",
"WARNING",
"Invalid cookie found in request <GET http://example.org/2>:"
" {'name': 'foo', 'secure': False} ('value' is missing)",
),
(
"scrapy.utils.request",
"WARNING",
"Invalid cookie found in request <GET http://example.org/3>:"
" {'name': 'foo', 'value': None, 'secure': False} ('value' is missing)",
),
)
assert caplog.record_tuples == [
(
"scrapy.utils.request",
logging.WARNING,
"Invalid cookie found in request <GET http://example.org/1>:"
" {'value': 'bar', 'secure': False} ('name' is missing)",
),
(
"scrapy.utils.request",
logging.WARNING,
"Invalid cookie found in request <GET http://example.org/2>:"
" {'name': 'foo', 'secure': False} ('value' is missing)",
),
(
"scrapy.utils.request",
logging.WARNING,
"Invalid cookie found in request <GET http://example.org/3>:"
" {'name': 'foo', 'value': None, 'secure': False} ('value' is missing)",
),
]
self.assertCookieValEqual(req1.headers["Cookie"], "key=value1")
self.assertCookieValEqual(req2.headers["Cookie"], "key=value2")
self.assertCookieValEqual(req3.headers["Cookie"], "key=")

188
tests/test_downloadermiddleware_httpcompression.py
View File

@ -5,7 +5,6 @@ from logging import WARNING
from pathlib import Path
import pytest
from testfixtures import LogCapture
from w3lib.encoding import resolve_encoding
from scrapy.downloadermiddlewares.httpcompression import (
@ -75,7 +74,7 @@ class TestHttpCompression:
self.mw = HttpCompressionMiddleware.from_crawler(self.crawler)
self.crawler.stats.open_spider()
def _getresponse(self, coding):
def _getresponse(self, coding: str) -> Response:
if coding not in FORMAT:
raise ValueError
@ -169,29 +168,28 @@ class TestHttpCompression:
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
def test_process_response_br_unsupported(self):
def test_process_response_br_unsupported(self, caplog: pytest.LogCaptureFixture):
if find_spec("brotli") is not None or find_spec("brotlicffi") is not None:
pytest.skip("Requires not having brotli support")
response = self._getresponse("br")
request = response.request
assert response.headers["Content-Encoding"] == b"br"
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
caplog.clear()
with caplog.at_level(
WARNING, logger="scrapy.downloadermiddlewares.httpcompression"
):
newresponse = self.mw.process_response(request, response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
WARNING,
(
"HttpCompressionMiddleware cannot decode the response for "
"http://scrapytest.org/ from unsupported encoding(s) 'br'. "
"You need to install brotli or brotlicffi >= 1.2.0 to decode 'br'."
),
),
)
]
assert newresponse is not response
assert newresponse.headers.getlist("Content-Encoding") == [b"br"]
@ -214,29 +212,28 @@ class TestHttpCompression:
assert newresponse.body.startswith(b"<!DOCTYPE")
assert "Content-Encoding" not in newresponse.headers
def test_process_response_zstd_unsupported(self):
def test_process_response_zstd_unsupported(self, caplog: pytest.LogCaptureFixture):
if find_spec("zstandard") is not None:
pytest.skip("Requires not having zstandard support")
response = self._getresponse("zstd-static-content-size")
request = response.request
assert response.headers["Content-Encoding"] == b"zstd"
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
caplog.clear()
with caplog.at_level(
WARNING, logger="scrapy.downloadermiddlewares.httpcompression"
):
newresponse = self.mw.process_response(request, response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
WARNING,
(
"HttpCompressionMiddleware cannot decode the response for"
" http://scrapytest.org/ from unsupported encoding(s) 'zstd'."
" You need to install zstandard to decode 'zstd'."
),
),
)
]
assert newresponse is not response
assert newresponse.headers.getlist("Content-Encoding") == [b"zstd"]
@ -291,26 +288,27 @@ class TestHttpCompression:
assert "Content-Encoding" not in newresponse.headers
assert newresponse.body.startswith(b"<!DOCTYPE")
def test_multi_compression_single_header_invalid_compression(self):
def test_multi_compression_single_header_invalid_compression(
self, caplog: pytest.LogCaptureFixture
) -> None:
response = self._getresponse("gzip-deflate")
response.headers["Content-Encoding"] = [b"gzip, foo, deflate"]
request = response.request
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
caplog.clear()
with caplog.at_level(
WARNING, logger="scrapy.downloadermiddlewares.httpcompression"
):
newresponse = self.mw.process_response(request, response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
WARNING,
(
"HttpCompressionMiddleware cannot decode the response for"
" http://scrapytest.org/ from unsupported encoding(s) 'gzip,foo'."
),
),
)
]
assert newresponse is not response
assert newresponse.headers.getlist("Content-Encoding") == [b"gzip", b"foo"]
@ -629,7 +627,9 @@ class TestHttpCompression:
self._test_compression_bomb_request_meta("zstd")
def _test_download_warnsize_setting(self, compression_id):
def _test_download_warnsize_setting(
self, caplog: pytest.LogCaptureFixture, compression_id: str
) -> None:
settings = {"DOWNLOAD_WARNSIZE": 10_000_000}
crawler = get_crawler(Spider, settings_dict=settings)
spider = crawler._create_spider("scrapytest.org")
@ -637,41 +637,51 @@ class TestHttpCompression:
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
assert response.request
caplog.clear()
with caplog.at_level(
WARNING, logger="scrapy.downloadermiddlewares.httpcompression"
):
mw.process_response(response.request, response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
WARNING,
(
"<200 http://scrapytest.org/> body size after "
"decompression (11511612 B) is larger than the download "
"warning size (10000000 B)."
),
),
)
]
def test_download_warnsize_setting_br(self):
def test_download_warnsize_setting_br(
self, caplog: pytest.LogCaptureFixture
) -> None:
_skip_if_no_br()
self._test_download_warnsize_setting("br")
self._test_download_warnsize_setting(caplog, "br")
def test_download_warnsize_setting_deflate(self):
self._test_download_warnsize_setting("deflate")
def test_download_warnsize_setting_deflate(
self, caplog: pytest.LogCaptureFixture
) -> None:
self._test_download_warnsize_setting(caplog, "deflate")
def test_download_warnsize_setting_gzip(self):
self._test_download_warnsize_setting("gzip")
def test_download_warnsize_setting_gzip(
self, caplog: pytest.LogCaptureFixture
) -> None:
self._test_download_warnsize_setting(caplog, "gzip")
def test_download_warnsize_setting_zstd(self):
def test_download_warnsize_setting_zstd(
self, caplog: pytest.LogCaptureFixture
) -> None:
_skip_if_no_zstd()
self._test_download_warnsize_setting("zstd")
self._test_download_warnsize_setting(caplog, "zstd")
def _test_download_warnsize_spider_attr(self, compression_id):
def _test_download_warnsize_spider_attr(
self, caplog: pytest.LogCaptureFixture, compression_id: str
) -> None:
class DownloadWarnSizeSpider(Spider):
download_warnsize = 10_000_000
@ -681,45 +691,55 @@ class TestHttpCompression:
mw.open_spider(spider)
response = self._getresponse(f"bomb-{compression_id}")
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
assert response.request
caplog.clear()
with caplog.at_level(
WARNING, logger="scrapy.downloadermiddlewares.httpcompression"
):
mw.process_response(response.request, response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
WARNING,
(
"<200 http://scrapytest.org/> body size after "
"decompression (11511612 B) is larger than the download "
"warning size (10000000 B)."
),
),
)
]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_download_warnsize_spider_attr_br(self):
def test_download_warnsize_spider_attr_br(
self, caplog: pytest.LogCaptureFixture
) -> None:
_skip_if_no_br()
self._test_download_warnsize_spider_attr("br")
self._test_download_warnsize_spider_attr(caplog, "br")
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_download_warnsize_spider_attr_deflate(self):
self._test_download_warnsize_spider_attr("deflate")
def test_download_warnsize_spider_attr_deflate(
self, caplog: pytest.LogCaptureFixture
) -> None:
self._test_download_warnsize_spider_attr(caplog, "deflate")
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_download_warnsize_spider_attr_gzip(self):
self._test_download_warnsize_spider_attr("gzip")
def test_download_warnsize_spider_attr_gzip(
self, caplog: pytest.LogCaptureFixture
) -> None:
self._test_download_warnsize_spider_attr(caplog, "gzip")
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_download_warnsize_spider_attr_zstd(self):
def test_download_warnsize_spider_attr_zstd(
self, caplog: pytest.LogCaptureFixture
) -> None:
_skip_if_no_zstd()
self._test_download_warnsize_spider_attr("zstd")
self._test_download_warnsize_spider_attr(caplog, "zstd")
def _test_download_warnsize_request_meta(self, compression_id):
def _test_download_warnsize_request_meta(
self, caplog: pytest.LogCaptureFixture, compression_id: str
) -> None:
crawler = get_crawler(Spider)
spider = crawler._create_spider("scrapytest.org")
mw = HttpCompressionMiddleware.from_crawler(crawler)
@ -727,39 +747,47 @@ class TestHttpCompression:
response = self._getresponse(f"bomb-{compression_id}")
response.meta["download_warnsize"] = 10_000_000
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
level=WARNING,
) as log:
assert response.request
caplog.clear()
with caplog.at_level(
WARNING, logger="scrapy.downloadermiddlewares.httpcompression"
):
mw.process_response(response.request, response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.downloadermiddlewares.httpcompression",
"WARNING",
WARNING,
(
"<200 http://scrapytest.org/> body size after "
"decompression (11511612 B) is larger than the download "
"warning size (10000000 B)."
),
),
)
]
def test_download_warnsize_request_meta_br(self):
def test_download_warnsize_request_meta_br(
self, caplog: pytest.LogCaptureFixture
) -> None:
_skip_if_no_br()
self._test_download_warnsize_request_meta("br")
self._test_download_warnsize_request_meta(caplog, "br")
def test_download_warnsize_request_meta_deflate(self):
self._test_download_warnsize_request_meta("deflate")
def test_download_warnsize_request_meta_deflate(
self, caplog: pytest.LogCaptureFixture
) -> None:
self._test_download_warnsize_request_meta(caplog, "deflate")
def test_download_warnsize_request_meta_gzip(self):
self._test_download_warnsize_request_meta("gzip")
def test_download_warnsize_request_meta_gzip(
self, caplog: pytest.LogCaptureFixture
) -> None:
self._test_download_warnsize_request_meta(caplog, "gzip")
def test_download_warnsize_request_meta_zstd(self):
def test_download_warnsize_request_meta_zstd(
self, caplog: pytest.LogCaptureFixture
) -> None:
_skip_if_no_zstd()
self._test_download_warnsize_request_meta("zstd")
self._test_download_warnsize_request_meta(caplog, "zstd")
def _get_truncated_response(self, compression_id):
crawler = get_crawler(Spider)

350
tests/test_downloadermiddleware_retry.py
View File

@ -1,7 +1,9 @@
from __future__ import annotations
import logging
from typing import Any, cast
import pytest
from testfixtures import LogCapture
from twisted.internet.error import ConnectError, ConnectionDone, ConnectionLost
from scrapy.downloadermiddlewares.retry import RetryMiddleware, get_retry_request
@ -85,7 +87,7 @@ class TestRetry:
)
assert self.crawler.stats.get_value("retry/count") == 2
def test_give_up_log_level_setting(self):
def test_give_up_log_level_setting(self, caplog: pytest.LogCaptureFixture) -> None:
crawler = get_crawler(
DefaultSpider, settings_dict={"RETRY_GIVE_UP_LOG_LEVEL": "WARNING"}
)
@ -94,29 +96,25 @@ class TestRetry:
mw.max_retry_times = 0
req = Request("http://example.com/503")
rsp = Response("http://example.com/503", body=b"", status=503)
with LogCapture() as log:
with caplog.at_level(logging.WARNING):
assert mw.process_response(req, rsp) is rsp
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"WARNING",
f"Gave up retrying {req} (failed 1 times): 503 Service Unavailable",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.WARNING,
f"Gave up retrying {req} (failed 1 times): 503 Service Unavailable",
) in caplog.record_tuples
def test_give_up_log_level_meta(self):
def test_give_up_log_level_meta(self, caplog: pytest.LogCaptureFixture) -> None:
self.mw.max_retry_times = 0
req = Request("http://example.com/503", meta={"give_up_log_level": "WARNING"})
rsp = Response("http://example.com/503", body=b"", status=503)
with LogCapture() as log:
with caplog.at_level(logging.WARNING):
assert self.mw.process_response(req, rsp) is rsp
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"WARNING",
f"Gave up retrying {req} (failed 1 times): 503 Service Unavailable",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.WARNING,
f"Gave up retrying {req} (failed 1 times): 503 Service Unavailable",
) in caplog.record_tuples
def test_twistederrors(self):
exceptions = [
@ -294,14 +292,15 @@ class TestMaxRetryTimes:
class TestGetRetryRequest:
def get_spider(self, settings=None):
@staticmethod
def get_spider(settings: dict[str, Any] | None = None) -> Spider:
crawler = get_crawler(Spider, settings or {})
return crawler._create_spider("foo")
def test_basic_usage(self):
def test_basic_usage(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
new_request = get_retry_request(
request,
spider=spider,
@ -313,44 +312,42 @@ class TestGetRetryRequest:
assert new_request.meta["retry_times"] == expected_retry_times
assert new_request.priority == -1
expected_reason = "unspecified"
assert spider.crawler.stats
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
assert spider.crawler.stats.get_value(stat) == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_max_retries_reached(self):
def test_max_retries_reached(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
max_retry_times = 0
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
new_request = get_retry_request(
request,
spider=spider,
max_retry_times=max_retry_times,
)
assert new_request is None
assert spider.crawler.stats
assert spider.crawler.stats.get_value("retry/max_reached") == 1
failure_count = max_retry_times + 1
expected_reason = "unspecified"
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"ERROR",
f"Gave up retrying {request} (failed {failure_count} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.ERROR,
f"Gave up retrying {request} (failed {failure_count} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_one_retry(self):
def test_one_retry(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
new_request = get_retry_request(
request,
spider=spider,
@ -363,28 +360,31 @@ class TestGetRetryRequest:
assert new_request.meta["retry_times"] == expected_retry_times
assert new_request.priority == -1
expected_reason = "unspecified"
assert spider.crawler.stats
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
assert spider.crawler.stats.get_value(stat) == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_two_retries(self):
def test_two_retries(self, caplog: pytest.LogCaptureFixture) -> None:
spider = self.get_spider()
request = Request("https://example.com")
new_request = request
max_retry_times = 2
for index in range(max_retry_times):
with LogCapture() as log:
new_request = get_retry_request(
new_request,
spider=spider,
max_retry_times=max_retry_times,
caplog.clear()
with caplog.at_level(logging.DEBUG):
new_request = cast(
"Request",
get_retry_request(
new_request,
spider=spider,
max_retry_times=max_retry_times,
),
)
assert isinstance(new_request, Request)
assert new_request != request
@ -393,36 +393,37 @@ class TestGetRetryRequest:
assert new_request.meta["retry_times"] == expected_retry_times
assert new_request.priority == -expected_retry_times
expected_reason = "unspecified"
assert spider.crawler.stats
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
value = spider.crawler.stats.get_value(stat)
assert value == expected_retry_times
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
with LogCapture() as log:
new_request = get_retry_request(
new_request,
spider=spider,
max_retry_times=max_retry_times,
caplog.clear()
with caplog.at_level(logging.DEBUG):
new_request = cast(
"Request",
get_retry_request(
new_request,
spider=spider,
max_retry_times=max_retry_times,
),
)
assert new_request is None
assert spider.crawler.stats.get_value("retry/max_reached") == 1
failure_count = max_retry_times + 1
expected_reason = "unspecified"
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"ERROR",
f"Gave up retrying {request} (failed {failure_count} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.ERROR,
f"Gave up retrying {request} (failed {failure_count} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_no_spider(self):
request = Request("https://example.com")
@ -483,238 +484,231 @@ class TestGetRetryRequest:
)
assert new_request.priority == priority_adjust
def test_log_extra_retry_success(self):
def test_log_extra_retry_success(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
with LogCapture(attributes=("spider",)) as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
)
log.check_present(spider)
assert any(getattr(r, "spider", None) is spider for r in caplog.records)
def test_log_extra_retries_exceeded(self):
def test_log_extra_retries_exceeded(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
with LogCapture(attributes=("spider",)) as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
max_retry_times=0,
)
log.check_present(spider)
assert any(getattr(r, "spider", None) is spider for r in caplog.records)
def test_reason_string(self):
def test_reason_string(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
expected_reason = "because"
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
reason=expected_reason,
)
expected_retry_times = 1
assert spider.crawler.stats
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
assert spider.crawler.stats.get_value(stat) == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_reason_builtin_exception(self):
def test_reason_builtin_exception(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
expected_reason = NotImplementedError()
expected_reason_string = "builtins.NotImplementedError"
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
reason=expected_reason,
)
expected_retry_times = 1
assert spider.crawler.stats
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_reason_builtin_exception_class(self):
def test_reason_builtin_exception_class(
self, caplog: pytest.LogCaptureFixture
) -> None:
request = Request("https://example.com")
spider = self.get_spider()
expected_reason = NotImplementedError
expected_reason_string = "builtins.NotImplementedError"
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
reason=expected_reason,
)
expected_retry_times = 1
assert spider.crawler.stats
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_reason_custom_exception(self):
def test_reason_custom_exception(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
expected_reason = IgnoreRequest()
expected_reason_string = "scrapy.exceptions.IgnoreRequest"
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
reason=expected_reason,
)
expected_retry_times = 1
assert spider.crawler.stats
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_reason_custom_exception_class(self):
def test_reason_custom_exception_class(
self, caplog: pytest.LogCaptureFixture
) -> None:
request = Request("https://example.com")
spider = self.get_spider()
expected_reason = IgnoreRequest
expected_reason_string = "scrapy.exceptions.IgnoreRequest"
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
reason=expected_reason,
)
expected_retry_times = 1
assert spider.crawler.stats
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"DEBUG",
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.DEBUG,
f"Retrying {request} (failed {expected_retry_times} times): "
f"{expected_reason}",
) in caplog.record_tuples
def test_custom_logger(self):
def test_custom_logger(self, caplog: pytest.LogCaptureFixture) -> None:
logger = logging.getLogger("custom-logger")
request = Request("https://example.com")
spider = self.get_spider()
expected_reason = "because"
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
get_retry_request(
request,
spider=spider,
reason=expected_reason,
logger=logger,
)
log.check_present(
(
"custom-logger",
"DEBUG",
f"Retrying {request} (failed 1 times): {expected_reason}",
)
)
assert (
"custom-logger",
logging.DEBUG,
f"Retrying {request} (failed 1 times): {expected_reason}",
) in caplog.record_tuples
def test_give_up_log_level_default(self):
def test_give_up_log_level_default(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider()
with LogCapture() as log:
with caplog.at_level(logging.ERROR):
get_retry_request(
request,
spider=spider,
max_retry_times=0,
)
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"ERROR",
f"Gave up retrying {request} (failed 1 times): unspecified",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.ERROR,
f"Gave up retrying {request} (failed 1 times): unspecified",
) in caplog.record_tuples
def test_give_up_log_level_argument_name(self):
def test_give_up_log_level_argument_name(
self, caplog: pytest.LogCaptureFixture
) -> None:
request = Request("https://example.com")
spider = self.get_spider()
with LogCapture() as log:
with caplog.at_level(logging.WARNING):
get_retry_request(
request,
spider=spider,
max_retry_times=0,
give_up_log_level="WARNING",
)
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"WARNING",
f"Gave up retrying {request} (failed 1 times): unspecified",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.WARNING,
f"Gave up retrying {request} (failed 1 times): unspecified",
) in caplog.record_tuples
def test_give_up_log_level_argument_number(self):
def test_give_up_log_level_argument_number(
self, caplog: pytest.LogCaptureFixture
) -> None:
request = Request("https://example.com")
spider = self.get_spider()
with LogCapture() as log:
with caplog.at_level(logging.WARNING):
get_retry_request(
request,
spider=spider,
max_retry_times=0,
give_up_log_level=logging.WARNING,
)
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"WARNING",
f"Gave up retrying {request} (failed 1 times): unspecified",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.WARNING,
f"Gave up retrying {request} (failed 1 times): unspecified",
) in caplog.record_tuples
def test_give_up_log_level_setting(self):
def test_give_up_log_level_setting(self, caplog: pytest.LogCaptureFixture) -> None:
request = Request("https://example.com")
spider = self.get_spider({"RETRY_GIVE_UP_LOG_LEVEL": "WARNING"})
with LogCapture() as log:
with caplog.at_level(logging.WARNING):
get_retry_request(
request,
spider=spider,
max_retry_times=0,
)
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
"WARNING",
f"Gave up retrying {request} (failed 1 times): unspecified",
)
)
assert (
"scrapy.downloadermiddlewares.retry",
logging.WARNING,
f"Gave up retrying {request} (failed 1 times): unspecified",
) in caplog.record_tuples
def test_give_up_log_level_invalid(self):
request = Request("https://example.com")

156
tests/test_dupefilters.py
View File

@ -1,11 +1,14 @@
from __future__ import annotations
import hashlib
import logging
import shutil
import sys
import tempfile
from pathlib import Path
from typing import TYPE_CHECKING, Any
import pytest
from testfixtures import LogCapture
from scrapy.core.scheduler import Scheduler
from scrapy.dupefilters import BaseDupeFilter, RFPDupeFilter
@ -15,8 +18,16 @@ from scrapy.utils.python import to_bytes
from scrapy.utils.test import get_crawler
from tests.spiders import SimpleSpider
if TYPE_CHECKING:
from scrapy.crawler import Crawler
def _get_dupefilter(*, crawler=None, settings=None, open_=True):
def _get_dupefilter(
*,
crawler: Crawler | None = None,
settings: dict[str, Any] | None = None,
open_: bool = True,
) -> BaseDupeFilter:
if crawler is None:
crawler = get_crawler(settings_dict=settings)
scheduler = Scheduler.from_crawler(crawler)
@ -151,108 +162,71 @@ class TestRFPDupeFilter:
finally:
shutil.rmtree(path)
def test_log(self):
with LogCapture() as log:
settings = {
"DUPEFILTER_DEBUG": False,
"DUPEFILTER_CLASS": FromCrawlerRFPDupeFilter,
}
crawler = get_crawler(SimpleSpider, settings_dict=settings)
spider = SimpleSpider.from_crawler(crawler)
dupefilter = _get_dupefilter(crawler=crawler)
def test_log(self, caplog: pytest.LogCaptureFixture) -> None:
settings = {
"DUPEFILTER_DEBUG": False,
"DUPEFILTER_CLASS": FromCrawlerRFPDupeFilter,
}
crawler = get_crawler(SimpleSpider, settings_dict=settings)
spider = SimpleSpider.from_crawler(crawler)
dupefilter = _get_dupefilter(crawler=crawler)
r1 = Request("http://scrapytest.org/index.html")
r2 = Request("http://scrapytest.org/index.html")
r1 = Request("http://scrapytest.org/index.html")
r2 = Request("http://scrapytest.org/index.html")
with caplog.at_level(logging.DEBUG):
dupefilter.log(r1, spider)
dupefilter.log(r2, spider)
assert crawler.stats.get_value("dupefilter/filtered") == 2
log.check_present(
(
"scrapy.dupefilters",
"DEBUG",
"Filtered duplicate request: <GET http://scrapytest.org/index.html> - no more"
" duplicates will be shown (see DUPEFILTER_DEBUG to show all duplicates)",
)
)
assert crawler.stats
assert crawler.stats.get_value("dupefilter/filtered") == 2
assert (
"scrapy.dupefilters",
logging.DEBUG,
"Filtered duplicate request: <GET http://scrapytest.org/index.html> - no more"
" duplicates will be shown (see DUPEFILTER_DEBUG to show all duplicates)",
) in caplog.record_tuples
dupefilter.close("finished")
dupefilter.close("finished")
def test_log_debug(self):
with LogCapture() as log:
settings = {
"DUPEFILTER_DEBUG": True,
"DUPEFILTER_CLASS": FromCrawlerRFPDupeFilter,
}
crawler = get_crawler(SimpleSpider, settings_dict=settings)
spider = SimpleSpider.from_crawler(crawler)
dupefilter = _get_dupefilter(crawler=crawler)
@pytest.mark.parametrize("df", [None, FromCrawlerRFPDupeFilter])
def test_log_debug(
self, caplog: pytest.LogCaptureFixture, df: type[BaseDupeFilter] | None
) -> None:
settings: dict[str, Any] = {
"DUPEFILTER_DEBUG": True,
}
if df:
settings["DUPEFILTER_CLASS"] = df
crawler = get_crawler(SimpleSpider, settings_dict=settings)
spider = SimpleSpider.from_crawler(crawler)
dupefilter = _get_dupefilter(crawler=crawler)
r1 = Request("http://scrapytest.org/index.html")
r2 = Request(
"http://scrapytest.org/index.html",
headers={"Referer": "http://scrapytest.org/INDEX.html"},
)
r1 = Request("http://scrapytest.org/index.html")
r2 = Request(
"http://scrapytest.org/index.html",
headers={"Referer": "http://scrapytest.org/INDEX.html"},
)
with caplog.at_level(logging.DEBUG):
dupefilter.log(r1, spider)
dupefilter.log(r2, spider)
assert crawler.stats.get_value("dupefilter/filtered") == 2
log.check_present(
(
"scrapy.dupefilters",
"DEBUG",
"Filtered duplicate request: <GET http://scrapytest.org/index.html> (referer: None)",
)
)
log.check_present(
(
"scrapy.dupefilters",
"DEBUG",
"Filtered duplicate request: <GET http://scrapytest.org/index.html>"
" (referer: http://scrapytest.org/INDEX.html)",
)
)
assert crawler.stats
assert crawler.stats.get_value("dupefilter/filtered") == 2
assert (
"scrapy.dupefilters",
logging.DEBUG,
"Filtered duplicate request: <GET http://scrapytest.org/index.html> (referer: None)",
) in caplog.record_tuples
assert (
"scrapy.dupefilters",
logging.DEBUG,
"Filtered duplicate request: <GET http://scrapytest.org/index.html>"
" (referer: http://scrapytest.org/INDEX.html)",
) in caplog.record_tuples
dupefilter.close("finished")
def test_log_debug_default_dupefilter(self):
with LogCapture() as log:
settings = {
"DUPEFILTER_DEBUG": True,
}
crawler = get_crawler(SimpleSpider, settings_dict=settings)
spider = SimpleSpider.from_crawler(crawler)
dupefilter = _get_dupefilter(crawler=crawler)
r1 = Request("http://scrapytest.org/index.html")
r2 = Request(
"http://scrapytest.org/index.html",
headers={"Referer": "http://scrapytest.org/INDEX.html"},
)
dupefilter.log(r1, spider)
dupefilter.log(r2, spider)
assert crawler.stats.get_value("dupefilter/filtered") == 2
log.check_present(
(
"scrapy.dupefilters",
"DEBUG",
"Filtered duplicate request: <GET http://scrapytest.org/index.html> (referer: None)",
)
)
log.check_present(
(
"scrapy.dupefilters",
"DEBUG",
"Filtered duplicate request: <GET http://scrapytest.org/index.html>"
" (referer: http://scrapytest.org/INDEX.html)",
)
)
dupefilter.close("finished")
dupefilter.close("finished")
class TestBaseDupeFilter:

16
tests/test_engine.py
View File

@ -1,13 +1,13 @@
from __future__ import annotations
import asyncio
import logging
import subprocess
import sys
from typing import TYPE_CHECKING, Any
from unittest.mock import Mock
import pytest
from testfixtures import LogCapture
from scrapy import signals
from scrapy.core.engine import ExecutionEngine, _Slot
@ -145,8 +145,10 @@ class TestEngine(TestEngineBase):
await asyncio.gather(e.start_async(), e.start_async())
await e.stop_async()
@inline_callbacks_test
def test_start_request_processing_exception(self):
@coroutine_test
async def test_start_request_processing_exception(
self, caplog: pytest.LogCaptureFixture
) -> None:
class BadRequestFingerprinter:
def fingerprint(self, request):
raise ValueError # to make Scheduler.enqueue_request() fail
@ -160,10 +162,10 @@ class TestEngine(TestEngineBase):
crawler = get_crawler(
SimpleSpider, {"REQUEST_FINGERPRINTER_CLASS": BadRequestFingerprinter}
)
with LogCapture() as log:
yield crawler.crawl()
assert "Error while processing requests from start()" in str(log)
assert "Spider closed (shutdown)" in str(log)
with caplog.at_level(logging.DEBUG):
await crawler.crawl_async()
assert "Error while processing requests from start()" in caplog.text
assert "Spider closed (shutdown)" in caplog.text
def test_short_timeout(self):
args = (

26
tests/test_feedexport.py
View File

@ -2,6 +2,7 @@ from __future__ import annotations
import csv
import json
import logging
import marshal
import pickle
import tempfile
@ -12,7 +13,6 @@ from unittest import mock
import lxml.etree
import pytest
from testfixtures import LogCapture
from w3lib.url import file_uri_to_path
import scrapy
@ -563,7 +563,9 @@ class TestFeedExport(TestFeedExportBase):
assert expctd == data[fmt]
@coroutine_test
async def test_export_no_items_multiple_feeds(self):
async def test_export_no_items_multiple_feeds(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""Make sure that `storage.store` is not called."""
settings = {
"FEEDS": {
@ -575,10 +577,10 @@ class TestFeedExport(TestFeedExportBase):
"FEED_STORE_EMPTY": False,
}
with LogCapture() as log:
with caplog.at_level(logging.INFO):
await self.exported_no_data(settings)
assert str(log).count("Storage.store is called") == 0
assert caplog.text.count("Storage.store is called") == 0
@coroutine_test
async def test_export_multiple_item_classes(self):
@ -1080,7 +1082,9 @@ class TestFeedExport(TestFeedExportBase):
assert data["csv"] == b""
@coroutine_test
async def test_multiple_feeds_success_logs_blocking_feed_storage(self):
async def test_multiple_feeds_success_logs_blocking_feed_storage(
self, caplog: pytest.LogCaptureFixture
):
settings = {
"FEEDS": {
self._random_temp_filename(): {"format": "json"},
@ -1093,14 +1097,16 @@ class TestFeedExport(TestFeedExportBase):
{"foo": "bar1", "baz": ""},
{"foo": "bar2", "baz": "quux"},
]
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
await self.exported_data(items, settings)
for fmt in ["json", "xml", "csv"]:
assert f"Stored {fmt} feed (2 items)" in str(log)
assert f"Stored {fmt} feed (2 items)" in caplog.text
@coroutine_test
async def test_multiple_feeds_failing_logs_blocking_feed_storage(self):
async def test_multiple_feeds_failing_logs_blocking_feed_storage(
self, caplog: pytest.LogCaptureFixture
):
settings = {
"FEEDS": {
self._random_temp_filename(): {"format": "json"},
@ -1113,11 +1119,11 @@ class TestFeedExport(TestFeedExportBase):
{"foo": "bar1", "baz": ""},
{"foo": "bar2", "baz": "quux"},
]
with LogCapture() as log:
with caplog.at_level(logging.DEBUG):
await self.exported_data(items, settings)
for fmt in ["json", "xml", "csv"]:
assert f"Error storing {fmt} feed (2 items)" in str(log)
assert f"Error storing {fmt} feed (2 items)" in caplog.text
@coroutine_test
async def test_extend_kwargs(self):

59
tests/test_feedexport_storages.py
View File

@ -1,5 +1,6 @@
from __future__ import annotations
import logging
import os
import string
import tempfile
@ -10,7 +11,6 @@ from unittest import mock
from urllib.parse import quote
import pytest
from testfixtures import LogCapture
from w3lib.url import path_to_file_uri
import scrapy
@ -415,23 +415,21 @@ class TestS3FeedStorage:
acl = storage.s3_client.upload_fileobj.call_args[1]["ExtraArgs"]["ACL"]
assert acl == "custom-acl"
def test_overwrite_default(self):
with LogCapture() as log:
S3FeedStorage(
"s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl"
)
assert "S3 does not support appending to files" not in str(log)
def test_overwrite_default(self, caplog: pytest.LogCaptureFixture) -> None:
S3FeedStorage(
"s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl"
)
assert "S3 does not support appending to files" not in caplog.text
def test_overwrite_false(self):
with LogCapture() as log:
S3FeedStorage(
"s3://mybucket/export.csv",
"access_key",
"secret_key",
"custom-acl",
feed_options={"overwrite": False},
)
assert "S3 does not support appending to files" in str(log)
def test_overwrite_false(self, caplog: pytest.LogCaptureFixture) -> None:
S3FeedStorage(
"s3://mybucket/export.csv",
"access_key",
"secret_key",
"custom-acl",
feed_options={"overwrite": False},
)
assert "S3 does not support appending to files" in caplog.text
class TestGCSFeedStorage:
@ -505,20 +503,20 @@ class TestGCSFeedStorage:
blob_mock.upload_from_file.assert_called_once_with(f, predefined_acl=acl)
f.close.assert_called_once_with()
def test_overwrite_default(self):
with LogCapture() as log:
def test_overwrite_default(self, caplog: pytest.LogCaptureFixture):
with caplog.at_level(logging.DEBUG):
GCSFeedStorage("gs://mybucket/export.csv", "myproject-123", "custom-acl")
assert "GCS does not support appending to files" not in str(log)
assert "GCS does not support appending to files" not in caplog.text
def test_overwrite_false(self):
with LogCapture() as log:
def test_overwrite_false(self, caplog: pytest.LogCaptureFixture):
with caplog.at_level(logging.DEBUG):
GCSFeedStorage(
"gs://mybucket/export.csv",
"myproject-123",
"custom-acl",
feed_options={"overwrite": False},
)
assert "GCS does not support appending to files" in str(log)
assert "GCS does not support appending to files" in caplog.text
class TestStdoutFeedStorage:
@ -530,17 +528,18 @@ class TestStdoutFeedStorage:
storage.store(file)
assert out.getvalue() == b"content"
def test_overwrite_default(self):
with LogCapture() as log:
def test_overwrite_default(self, caplog: pytest.LogCaptureFixture):
with caplog.at_level(logging.DEBUG):
StdoutFeedStorage("stdout:")
assert (
"Standard output (stdout) storage does not support overwriting"
not in str(log)
not in caplog.text
)
def test_overwrite_true(self):
with LogCapture() as log:
def test_overwrite_true(self, caplog: pytest.LogCaptureFixture):
with caplog.at_level(logging.DEBUG):
StdoutFeedStorage("stdout:", feed_options={"overwrite": True})
assert "Standard output (stdout) storage does not support overwriting" in str(
log
assert (
"Standard output (stdout) storage does not support overwriting"
in caplog.text
)

51
tests/test_logformatter.py
View File

@ -1,7 +1,9 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
import pytest
from testfixtures import LogCapture
from twisted.python.failure import Failure
from scrapy.exceptions import DropItem
@ -10,9 +12,11 @@ from scrapy.item import Field, Item
from scrapy.logformatter import LogFormatter
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.spiders import ItemSpider
from tests.utils.decorators import inline_callbacks_test
from tests.utils.decorators import coroutine_test
if TYPE_CHECKING:
from tests.mockserver.http import MockServer
class CustomItem(Item):
@ -254,15 +258,6 @@ class DropSomeItemsPipeline:
class TestShowOrSkipMessages:
@classmethod
def setup_class(cls):
cls.mockserver = MockServer()
cls.mockserver.__enter__()
@classmethod
def teardown_class(cls):
cls.mockserver.__exit__(None, None, None)
def setup_method(self):
self.base_settings = {
"LOG_LEVEL": "DEBUG",
@ -271,22 +266,26 @@ class TestShowOrSkipMessages:
},
}
@inline_callbacks_test
def test_show_messages(self):
@coroutine_test
async def test_show_messages(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
crawler = get_crawler(ItemSpider, self.base_settings)
with LogCapture() as lc:
yield crawler.crawl(mockserver=self.mockserver)
assert "Scraped from <200 http://127.0.0.1:" in str(lc)
assert "Crawled (200) <GET http://127.0.0.1:" in str(lc)
assert "Dropped: Ignoring item" in str(lc)
with caplog.at_level(logging.DEBUG):
await crawler.crawl_async(mockserver=mockserver)
assert "Scraped from <200 http://127.0.0.1:" in caplog.text
assert "Crawled (200) <GET http://127.0.0.1:" in caplog.text
assert "Dropped: Ignoring item" in caplog.text
@inline_callbacks_test
def test_skip_messages(self):
@coroutine_test
async def test_skip_messages(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
settings = self.base_settings.copy()
settings["LOG_FORMATTER"] = SkipMessagesLogFormatter
crawler = get_crawler(ItemSpider, settings)
with LogCapture() as lc:
yield crawler.crawl(mockserver=self.mockserver)
assert "Scraped from <200 http://127.0.0.1:" not in str(lc)
assert "Crawled (200) <GET http://127.0.0.1:" not in str(lc)
assert "Dropped: Ignoring item" not in str(lc)
with caplog.at_level(logging.DEBUG):
await crawler.crawl_async(mockserver=mockserver)
assert "Scraped from <200 http://127.0.0.1:" not in caplog.text
assert "Crawled (200) <GET http://127.0.0.1:" not in caplog.text
assert "Dropped: Ignoring item" not in caplog.text

74
tests/test_pipeline_crawl.py
View File

@ -1,12 +1,12 @@
from __future__ import annotations
import logging
import shutil
from pathlib import Path
from tempfile import mkdtemp
from typing import TYPE_CHECKING, Any
import pytest
from testfixtures import LogCapture
from w3lib.url import add_or_replace_parameter
from scrapy import Spider, signals
@ -14,7 +14,7 @@ from scrapy.utils.misc import load_object
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.spiders import SimpleSpider
from tests.utils.decorators import inline_callbacks_test
from tests.utils.decorators import coroutine_test
if TYPE_CHECKING:
from scrapy.crawler import Crawler
@ -58,6 +58,8 @@ class RedirectedMediaDownloadSpider(MediaDownloadSpider):
class TestFileDownloadCrawl:
mockserver: MockServer
pipeline_class = "scrapy.pipelines.files.FilesPipeline"
store_setting_key = "FILES_STORE"
media_key = "files"
@ -102,7 +104,7 @@ class TestFileDownloadCrawl:
crawler.signals.connect(self._on_item_scraped, signals.item_scraped)
return crawler
def _assert_files_downloaded(self, items, logs):
def _assert_files_downloaded(self, items: list[Any], logs: str) -> None:
assert len(items) == 1
assert self.media_key in items[0]
@ -125,13 +127,16 @@ class TestFileDownloadCrawl:
for i in item[self.media_key]:
assert (self.tmpmediastore / i["path"]).exists()
def _assert_files_download_failure(self, crawler, items, code, logs):
def _assert_files_download_failure(
self, crawler: Crawler, items: list[Any], code: int, logs: str
) -> None:
# check that the item does NOT have the "images/files" field populated
assert len(items) == 1
assert self.media_key in items[0]
assert not items[0][self.media_key]
# check that there was 1 successful fetch and 3 other responses with non-200 code
assert crawler.stats
assert crawler.stats.get_value("downloader/request_method_count/GET") == 4
assert crawler.stats.get_value("downloader/response_count") == 4
assert crawler.stats.get_value("downloader/response_status_count/200") == 1
@ -144,62 +149,71 @@ class TestFileDownloadCrawl:
# check that no files were written to the media store
assert not list(self.tmpmediastore.iterdir())
@inline_callbacks_test
def test_download_media(self):
@coroutine_test
async def test_download_media(self, caplog: pytest.LogCaptureFixture) -> None:
crawler = self._create_crawler(MediaDownloadSpider)
with LogCapture() as log:
yield crawler.crawl(
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,
)
self._assert_files_downloaded(self.items, str(log))
self._assert_files_downloaded(self.items, caplog.text)
@inline_callbacks_test
def test_download_media_wrong_urls(self):
@coroutine_test
async def test_download_media_wrong_urls(
self, caplog: pytest.LogCaptureFixture
) -> None:
crawler = self._create_crawler(BrokenLinksMediaDownloadSpider)
with LogCapture() as log:
yield crawler.crawl(
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,
)
self._assert_files_download_failure(crawler, self.items, 404, str(log))
self._assert_files_download_failure(crawler, self.items, 404, caplog.text)
@inline_callbacks_test
def test_download_media_redirected_default_failure(self):
@coroutine_test
async def test_download_media_redirected_default_failure(
self, caplog: pytest.LogCaptureFixture
):
crawler = self._create_crawler(RedirectedMediaDownloadSpider)
with LogCapture() as log:
yield crawler.crawl(
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_download_failure(crawler, self.items, 302, str(log))
self._assert_files_download_failure(crawler, self.items, 302, caplog.text)
@inline_callbacks_test
def test_download_media_redirected_allowed(self):
@coroutine_test
async def test_download_media_redirected_allowed(
self, caplog: pytest.LogCaptureFixture
) -> None:
settings = {
**self.settings,
"MEDIA_ALLOW_REDIRECTS": True,
}
crawler = self._create_crawler(RedirectedMediaDownloadSpider, settings)
with LogCapture() as log:
yield crawler.crawl(
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, str(log))
self._assert_files_downloaded(self.items, caplog.text)
assert crawler.stats
assert crawler.stats.get_value("downloader/response_status_count/302") == 3
@inline_callbacks_test
def test_download_media_file_path_error(self):
@coroutine_test
async def test_download_media_file_path_error(
self, caplog: pytest.LogCaptureFixture
) -> None:
cls = load_object(self.pipeline_class)
class ExceptionRaisingMediaPipeline(cls):
class ExceptionRaisingMediaPipeline(cls): # type: ignore[misc,valid-type]
def file_path(self, request, response=None, info=None, *, item=None):
return 1 / 0
@ -208,14 +222,14 @@ class TestFileDownloadCrawl:
"ITEM_PIPELINES": {ExceptionRaisingMediaPipeline: 1},
}
crawler = self._create_crawler(MediaDownloadSpider, settings)
with LogCapture() as log:
yield crawler.crawl(
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 "ZeroDivisionError" in str(log)
assert "ZeroDivisionError" in caplog.text
pillow_available: bool

18
tests/test_pipeline_media.py
View File

@ -4,7 +4,6 @@ import logging
from unittest.mock import MagicMock
import pytest
from testfixtures import LogCapture
from twisted.python.failure import Failure
from scrapy import signals
@ -129,7 +128,7 @@ class TestBaseMediaPipeline:
context = getattr(info.downloaded[fp].value, "__context__", None)
assert context is None
def test_default_item_completed(self):
def test_default_item_completed(self, caplog: pytest.LogCaptureFixture) -> None:
item = {"name": "name"}
assert self.pipe.item_completed([], item, self.info) is item
@ -137,21 +136,20 @@ class TestBaseMediaPipeline:
fail = Failure(Exception())
results = [(True, 1), (False, fail)]
with LogCapture() as log:
new_item = self.pipe.item_completed(results, item, self.info)
caplog.clear()
new_item = self.pipe.item_completed(results, item, self.info)
assert new_item is item
assert len(log.records) == 1
record = log.records[0]
assert len(caplog.records) == 1
record = caplog.records[0]
assert record.levelname == "ERROR"
assert record.exc_info == failure_to_exc_info(fail)
# disable failure logging and check again
caplog.clear()
self.pipe.LOG_FAILED_RESULTS = False
with LogCapture() as log:
new_item = self.pipe.item_completed(results, item, self.info)
new_item = self.pipe.item_completed(results, item, self.info)
assert new_item is item
assert len(log.records) == 0
assert len(caplog.records) == 0
def test_item_completed_filtered_request_not_logged(
self, caplog: pytest.LogCaptureFixture

60
tests/test_request_attribute_binding.py
View File

@ -1,11 +1,17 @@
from testfixtures import LogCapture
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from scrapy import Request, signals
from scrapy.http.response import Response
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.spiders import SingleRequestSpider
from tests.utils.decorators import inline_callbacks_test
from tests.utils.decorators import coroutine_test, inline_callbacks_test
if TYPE_CHECKING:
import pytest
OVERRIDDEN_URL = "https://example.org"
@ -63,6 +69,8 @@ class AlternativeCallbacksMiddleware:
class TestCrawl:
mockserver: MockServer
@classmethod
def setup_class(cls):
cls.mockserver = MockServer()
@ -107,8 +115,10 @@ class TestCrawl:
assert failure.request.url == url
assert isinstance(failure.value, ZeroDivisionError)
@inline_callbacks_test
def test_downloader_middleware_override_request_in_process_response(self):
@coroutine_test
async def test_downloader_middleware_override_request_in_process_response(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
Downloader middleware which returns a response with an specific 'request' attribute.
@ -133,22 +143,21 @@ class TestCrawl:
)
crawler.signals.connect(signal_handler, signal=signals.response_received)
with LogCapture() as log:
yield crawler.crawl(seed=url, mockserver=self.mockserver)
with caplog.at_level(logging.DEBUG):
await crawler.crawl_async(seed=url, mockserver=self.mockserver)
assert isinstance(crawler.spider, SingleRequestSpider)
response = crawler.spider.meta["responses"][0]
assert response.request.url == OVERRIDDEN_URL
assert signal_params["response"].url == url
assert signal_params["request"].url == OVERRIDDEN_URL
log.check_present(
(
"scrapy.core.engine",
"DEBUG",
f"Crawled (200) <GET {OVERRIDDEN_URL}> (referer: None)",
),
)
assert (
"scrapy.core.engine",
logging.DEBUG,
f"Crawled (200) <GET {OVERRIDDEN_URL}> (referer: None)",
) in caplog.record_tuples
@inline_callbacks_test
def test_downloader_middleware_override_in_process_exception(self):
@ -196,8 +205,10 @@ class TestCrawl:
assert response.body == b"Caught ZeroDivisionError"
assert response.request.url == url
@inline_callbacks_test
def test_downloader_middleware_alternative_callback(self):
@coroutine_test
async def test_downloader_middleware_alternative_callback(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
Downloader middleware which returns a response with a
specific 'request' attribute, with an alternative callback
@ -211,14 +222,11 @@ class TestCrawl:
},
)
with LogCapture() as log:
url = self.mockserver.url("/status?n=200")
yield crawler.crawl(seed=url, mockserver=self.mockserver)
log.check_present(
(
"alternative_callbacks_spider",
"INFO",
"alt_callback was invoked with foo=bar",
),
)
url = self.mockserver.url("/status?n=200")
with caplog.at_level(logging.INFO):
await crawler.crawl_async(seed=url, mockserver=self.mockserver)
assert (
"alternative_callbacks_spider",
logging.INFO,
"alt_callback was invoked with foo=bar",
) in caplog.record_tuples

52
tests/test_request_cb_kwargs.py
View File

@ -1,10 +1,17 @@
from testfixtures import LogCapture
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from scrapy.http import Request
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.spiders import MockServerSpider
from tests.utils.decorators import inline_callbacks_test
from tests.utils.decorators import coroutine_test
if TYPE_CHECKING:
import pytest
from tests.mockserver.http import MockServer
class InjectArgumentsDownloaderMiddleware:
@ -147,33 +154,32 @@ class KeywordArgumentsSpider(MockServerSpider):
class TestCallbackKeywordArguments:
@classmethod
def setup_class(cls):
cls.mockserver = MockServer()
cls.mockserver.__enter__()
@classmethod
def teardown_class(cls):
cls.mockserver.__exit__(None, None, None)
@inline_callbacks_test
def test_callback_kwargs(self):
@coroutine_test
async def test_callback_kwargs(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
crawler = get_crawler(KeywordArgumentsSpider)
with LogCapture() as log:
yield crawler.crawl(mockserver=self.mockserver)
with caplog.at_level(logging.ERROR):
await crawler.crawl_async(mockserver=mockserver)
assert isinstance(crawler.spider, KeywordArgumentsSpider)
assert all(crawler.spider.checks)
assert crawler.stats
assert len(crawler.spider.checks) == crawler.stats.get_value("boolean_checks")
# check exceptions for argument mismatch
exceptions = {}
for line in log.records:
for line in caplog.records:
for key in ("takes_less", "takes_more"):
if key in line.getMessage():
exceptions[key] = line
assert exceptions["takes_less"].exc_info[0] is TypeError
assert str(exceptions["takes_less"].exc_info[1]).endswith(
takes_less_exc_info = exceptions["takes_less"].exc_info
assert takes_less_exc_info is not None
assert takes_less_exc_info[0] is TypeError
assert str(takes_less_exc_info[1]).endswith(
"parse_takes_less() got an unexpected keyword argument 'number'"
), "Exception message: " + str(exceptions["takes_less"].exc_info[1])
assert exceptions["takes_more"].exc_info[0] is TypeError
assert str(exceptions["takes_more"].exc_info[1]).endswith(
)
takes_more_exc_info = exceptions["takes_more"].exc_info
assert takes_more_exc_info is not None
assert takes_more_exc_info[0] is TypeError
assert str(takes_more_exc_info[1]).endswith(
"parse_takes_more() missing 1 required positional argument: 'other'"
), "Exception message: " + str(exceptions["takes_more"].exc_info[1])
)

34
tests/test_scheduler_base.py
View File

@ -1,9 +1,10 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from urllib.parse import urljoin
import pytest
from testfixtures import LogCapture
from twisted.internet import defer
from scrapy.core.scheduler import BaseScheduler
@ -12,8 +13,10 @@ from scrapy.spiders import Spider
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.request import fingerprint
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.utils.decorators import inline_callbacks_test
from tests.utils.decorators import coroutine_test, inline_callbacks_test
if TYPE_CHECKING:
from tests.mockserver.http import MockServer
PATHS = ["/a", "/b", "/c"]
URLS = [urljoin("https://example.org", p) for p in PATHS]
@ -147,18 +150,19 @@ class TestSimpleScheduler(InterfaceCheckMixin):
class TestMinimalSchedulerCrawl:
scheduler_cls = MinimalScheduler
@inline_callbacks_test
def test_crawl(self):
with MockServer() as mockserver:
settings = {
"SCHEDULER": self.scheduler_cls,
}
with LogCapture() as log:
crawler = get_crawler(PathsSpider, settings)
yield crawler.crawl(mockserver)
for path in PATHS:
assert f"{{'path': '{path}'}}" in str(log)
assert f"'item_scraped_count': {len(PATHS)}" in str(log)
@coroutine_test
async def test_crawl(
self, caplog: pytest.LogCaptureFixture, mockserver: MockServer
) -> None:
settings = {
"SCHEDULER": self.scheduler_cls,
}
with caplog.at_level(logging.DEBUG):
crawler = get_crawler(PathsSpider, settings)
await crawler.crawl_async(mockserver)
for path in PATHS:
assert f"{{'path': '{path}'}}" in caplog.text
assert f"'item_scraped_count': {len(PATHS)}" in caplog.text
class TestSimpleSchedulerCrawl(TestMinimalSchedulerCrawl):

57
tests/test_spider_sitemap.py
View File

@ -8,7 +8,6 @@ from logging import WARNING
from pathlib import Path
import pytest
from testfixtures import LogCapture
from scrapy.http import HtmlResponse, Request, Response, TextResponse, XmlResponse
from scrapy.spiders import SitemapSpider
@ -254,22 +253,23 @@ Sitemap: /sitemap-relative-url.xml
urls = [req.url for req in spider._parse_sitemap(r)]
assert urls == result
def test_parse_sitemap_empty_body(self):
def test_parse_sitemap_empty_body(self, caplog: pytest.LogCaptureFixture) -> None:
r = XmlResponse(url="http://www.example.com/sitemap.xml", body=b"")
spider = self.spider_class("example.com")
with LogCapture() as lc:
caplog.clear()
with caplog.at_level(WARNING):
results = list(spider._parse_sitemap(r))
assert not results
lc.check(
assert caplog.record_tuples == [
(
"scrapy.spiders.sitemap",
"WARNING",
WARNING,
"Ignoring invalid sitemap: <200 http://www.example.com/sitemap.xml>",
)
)
]
def test_parse_sitemap_not_sitemap(self):
body = b"""<?xml version="1.0" encoding="UTF-8"?>
@ -342,7 +342,7 @@ Sitemap: /sitemap-relative-url.xml
response = Response(url="https://example.com", body=body, request=request)
assert spider._get_sitemap_body(response) is None
def test_download_warnsize_setting(self):
def test_download_warnsize_setting(self, caplog: pytest.LogCaptureFixture) -> None:
settings = {"DOWNLOAD_WARNSIZE": 10_000_000}
crawler = get_crawler(settings_dict=settings)
spider = self.spider_class.from_crawler(crawler, "example.com")
@ -350,25 +350,26 @@ Sitemap: /sitemap-relative-url.xml
body = body_path.read_bytes()
request = Request(url="https://example.com")
response = Response(url="https://example.com", body=body, request=request)
with LogCapture(
"scrapy.spiders.sitemap", propagate=False, level=WARNING
) as log:
caplog.clear()
with caplog.at_level(WARNING, logger="scrapy.spiders.sitemap"):
spider._get_sitemap_body(response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.spiders.sitemap",
"WARNING",
WARNING,
(
"<200 https://example.com> body size after decompression "
"(11511612 B) is larger than the download warning size "
"(10000000 B)."
),
),
)
]
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
def test_download_warnsize_spider_attr(self):
class DownloadWarnSizeSpider(self.spider_class):
def test_download_warnsize_spider_attr(
self, caplog: pytest.LogCaptureFixture
) -> None:
class DownloadWarnSizeSpider(self.spider_class): # type: ignore[name-defined,misc]
download_warnsize = 10_000_000
crawler = get_crawler()
@ -379,23 +380,24 @@ Sitemap: /sitemap-relative-url.xml
url="https://example.com", meta={"download_warnsize": 10_000_000}
)
response = Response(url="https://example.com", body=body, request=request)
with LogCapture(
"scrapy.spiders.sitemap", propagate=False, level=WARNING
) as log:
caplog.clear()
with caplog.at_level(WARNING, logger="scrapy.spiders.sitemap"):
spider._get_sitemap_body(response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.spiders.sitemap",
"WARNING",
WARNING,
(
"<200 https://example.com> body size after decompression "
"(11511612 B) is larger than the download warning size "
"(10000000 B)."
),
),
)
]
def test_download_warnsize_request_meta(self):
def test_download_warnsize_request_meta(
self, caplog: pytest.LogCaptureFixture
) -> None:
crawler = get_crawler()
spider = self.spider_class.from_crawler(crawler, "example.com")
body_path = Path(tests_datadir, "compressed", "bomb-gzip.bin")
@ -404,21 +406,20 @@ Sitemap: /sitemap-relative-url.xml
url="https://example.com", meta={"download_warnsize": 10_000_000}
)
response = Response(url="https://example.com", body=body, request=request)
with LogCapture(
"scrapy.spiders.sitemap", propagate=False, level=WARNING
) as log:
caplog.clear()
with caplog.at_level(WARNING, logger="scrapy.spiders.sitemap"):
spider._get_sitemap_body(response)
log.check(
assert caplog.record_tuples == [
(
"scrapy.spiders.sitemap",
"WARNING",
WARNING,
(
"<200 https://example.com> body size after decompression "
"(11511612 B) is larger than the download warning size "
"(10000000 B)."
),
),
)
]
@coroutine_test
async def test_sitemap_urls(self):

137
tests/test_spidermiddleware_output_chain.py
View File

@ -1,10 +1,16 @@
from testfixtures import LogCapture
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from scrapy import Request, Spider
from scrapy.utils.test import get_crawler
from tests.mockserver.http import MockServer
from tests.utils.decorators import coroutine_test
if TYPE_CHECKING:
import pytest
class _BaseSpiderMiddleware:
def __init__(self, crawler):
@ -250,113 +256,132 @@ class TestSpiderMiddleware:
def teardown_class(cls):
cls.mockserver.__exit__(None, None, None)
async def crawl_log(self, spider: type[Spider]) -> LogCapture:
async def crawl_log(
self, spider: type[Spider], caplog: pytest.LogCaptureFixture
) -> str:
crawler = get_crawler(spider)
with LogCapture() as log:
caplog.clear()
with caplog.at_level(logging.DEBUG):
await crawler.crawl_async(mockserver=self.mockserver)
return log
return caplog.text
@coroutine_test
async def test_recovery(self):
async def test_recovery(self, caplog: pytest.LogCaptureFixture) -> None:
"""
(0) Recover from an exception in a spider callback. The final item count should be 3
(one yielded from the callback method before the exception is raised, one directly
from the recovery middleware and one from the spider when processing the request that
was enqueued from the recovery middleware)
"""
log = await self.crawl_log(RecoverySpider)
assert "Middleware: TabError exception caught" in str(log)
assert str(log).count("Middleware: TabError exception caught") == 1
assert "'item_scraped_count': 3" in str(log)
log = await self.crawl_log(RecoverySpider, caplog)
assert "Middleware: TabError exception caught" in log
assert log.count("Middleware: TabError exception caught") == 1
assert "'item_scraped_count': 3" in log
@coroutine_test
async def test_recovery_asyncgen(self):
async def test_recovery_asyncgen(self, caplog: pytest.LogCaptureFixture) -> None:
"""
Same as test_recovery but with an async callback.
"""
log = await self.crawl_log(RecoveryAsyncGenSpider)
assert "Middleware: TabError exception caught" in str(log)
assert str(log).count("Middleware: TabError exception caught") == 1
assert "'item_scraped_count': 3" in str(log)
log = await self.crawl_log(RecoveryAsyncGenSpider, caplog)
assert "Middleware: TabError exception caught" in log
assert log.count("Middleware: TabError exception caught") == 1
assert "'item_scraped_count': 3" in log
@coroutine_test
async def test_process_spider_input_without_errback(self):
async def test_process_spider_input_without_errback(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
(1.1) An exception from the process_spider_input chain should be caught by the
process_spider_exception chain from the start if the Request has no errback
"""
log1 = await self.crawl_log(ProcessSpiderInputSpiderWithoutErrback)
assert "Middleware: will raise IndexError" in str(log1)
assert "Middleware: IndexError exception caught" in str(log1)
log1 = await self.crawl_log(ProcessSpiderInputSpiderWithoutErrback, caplog)
assert "Middleware: will raise IndexError" in log1
assert "Middleware: IndexError exception caught" in log1
@coroutine_test
async def test_process_spider_input_with_errback(self):
async def test_process_spider_input_with_errback(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
(1.2) An exception from the process_spider_input chain should not be caught by the
process_spider_exception chain if the Request has an errback
"""
log1 = await self.crawl_log(ProcessSpiderInputSpiderWithErrback)
assert "Middleware: IndexError exception caught" not in str(log1)
assert "Middleware: will raise IndexError" in str(log1)
assert "Got a Failure on the Request errback" in str(log1)
assert "{'from': 'errback'}" in str(log1)
assert "{'from': 'callback'}" not in str(log1)
assert "'item_scraped_count': 1" in str(log1)
log1 = await self.crawl_log(ProcessSpiderInputSpiderWithErrback, caplog)
assert "Middleware: IndexError exception caught" not in log1
assert "Middleware: will raise IndexError" in log1
assert "Got a Failure on the Request errback" in log1
assert "{'from': 'errback'}" in log1
assert "{'from': 'callback'}" not in log1
assert "'item_scraped_count': 1" in log1
@coroutine_test
async def test_generator_callback(self):
async def test_generator_callback(self, caplog: pytest.LogCaptureFixture) -> None:
"""
(2) An exception from a spider callback (returning a generator) should
be caught by the process_spider_exception chain. Items yielded before the
exception is raised should be processed normally.
"""
log2 = await self.crawl_log(GeneratorCallbackSpider)
assert "Middleware: ImportError exception caught" in str(log2)
assert "'item_scraped_count': 2" in str(log2)
log2 = await self.crawl_log(GeneratorCallbackSpider, caplog)
assert "Middleware: ImportError exception caught" in log2
assert "'item_scraped_count': 2" in log2
@coroutine_test
async def test_async_generator_callback(self):
async def test_async_generator_callback(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
Same as test_generator_callback but with an async callback.
"""
log2 = await self.crawl_log(AsyncGeneratorCallbackSpider)
assert "Middleware: ImportError exception caught" in str(log2)
assert "'item_scraped_count': 2" in str(log2)
log2 = await self.crawl_log(AsyncGeneratorCallbackSpider, caplog)
assert "Middleware: ImportError exception caught" in log2
assert "'item_scraped_count': 2" in log2
@coroutine_test
async def test_generator_callback_right_after_callback(self):
async def test_generator_callback_right_after_callback(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
(2.1) Special case of (2): Exceptions should be caught
even if the middleware is placed right after the spider
"""
log21 = await self.crawl_log(GeneratorCallbackSpiderMiddlewareRightAfterSpider)
assert "Middleware: ImportError exception caught" in str(log21)
assert "'item_scraped_count': 2" in str(log21)
log21 = await self.crawl_log(
GeneratorCallbackSpiderMiddlewareRightAfterSpider, caplog
)
assert "Middleware: ImportError exception caught" in log21
assert "'item_scraped_count': 2" in log21
@coroutine_test
async def test_not_a_generator_callback(self):
async def test_not_a_generator_callback(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
(3) An exception from a spider callback (returning a list) should
be caught by the process_spider_exception chain. No items should be processed.
"""
log3 = await self.crawl_log(NotGeneratorCallbackSpider)
assert "Middleware: ZeroDivisionError exception caught" in str(log3)
assert "item_scraped_count" not in str(log3)
log3 = await self.crawl_log(NotGeneratorCallbackSpider, caplog)
assert "Middleware: ZeroDivisionError exception caught" in log3
assert "item_scraped_count" not in log3
@coroutine_test
async def test_not_a_generator_callback_right_after_callback(self):
async def test_not_a_generator_callback_right_after_callback(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
(3.1) Special case of (3): Exceptions should be caught
even if the middleware is placed right after the spider
"""
log31 = await self.crawl_log(
NotGeneratorCallbackSpiderMiddlewareRightAfterSpider
NotGeneratorCallbackSpiderMiddlewareRightAfterSpider, caplog
)
assert "Middleware: ZeroDivisionError exception caught" in str(log31)
assert "item_scraped_count" not in str(log31)
assert "Middleware: ZeroDivisionError exception caught" in log31
assert "item_scraped_count" not in log31
@coroutine_test
async def test_generator_output_chain(self):
async def test_generator_output_chain(
self, caplog: pytest.LogCaptureFixture
) -> None:
"""
(4) An exception from a middleware's process_spider_output method should be sent
to the process_spider_exception method from the next middleware in the chain.
@ -365,23 +390,23 @@ class TestSpiderMiddleware:
The final item count should be 2 (one from the spider callback and one from the
process_spider_exception chain)
"""
log4 = await self.crawl_log(GeneratorOutputChainSpider)
assert "'item_scraped_count': 2" in str(log4)
log4 = await self.crawl_log(GeneratorOutputChainSpider, caplog)
assert "'item_scraped_count': 2" in log4
assert (
"GeneratorRecoverMiddleware.process_spider_exception: LookupError caught"
in str(log4)
in log4
)
assert (
"GeneratorDoNothingAfterFailureMiddleware.process_spider_exception: LookupError caught"
in str(log4)
in log4
)
assert (
"GeneratorFailMiddleware.process_spider_exception: LookupError caught"
not in str(log4)
not in log4
)
assert (
"GeneratorDoNothingAfterRecoveryMiddleware.process_spider_exception: LookupError caught"
not in str(log4)
not in log4
)
item_from_callback = {
"processed": [
@ -398,6 +423,6 @@ class TestSpiderMiddleware:
"GeneratorDoNothingAfterRecoveryMiddleware.process_spider_output",
]
}
assert str(item_from_callback) in str(log4)
assert str(item_recovered) in str(log4)
assert "parse-second-item" not in str(log4)
assert str(item_from_callback) in log4
assert str(item_recovered) in log4
assert "parse-second-item" not in log4

9
tests/test_utils_log.py
View File

@ -8,7 +8,6 @@ from io import StringIO
from typing import TYPE_CHECKING, Any
import pytest
from testfixtures import LogCapture
from twisted.python.failure import Failure
from scrapy.utils.log import (
@ -107,15 +106,15 @@ class TestLogCounterHandler:
class TestStreamLogger:
def test_redirect(self):
def test_redirect(self, caplog: pytest.LogCaptureFixture) -> None:
logger = logging.getLogger("test")
logger.setLevel(logging.WARNING)
old_stdout = sys.stdout
sys.stdout = StreamLogger(logger, logging.ERROR)
with LogCapture() as log:
print("test log msg")
log.check(("test", "ERROR", "test log msg"))
caplog.clear()
print("test log msg")
assert caplog.record_tuples == [("test", logging.ERROR, "test log msg")]
sys.stdout = old_stdout

44
tests/test_utils_signal.py
View File

@ -1,23 +1,25 @@
from __future__ import annotations
import asyncio
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
import pytest
from pydispatch import dispatcher
from testfixtures import LogCapture
from twisted.internet import defer
from twisted.python.failure import Failure
from scrapy.utils.asyncio import call_later
from scrapy.utils.defer import deferred_from_coro
from scrapy.utils.defer import deferred_from_coro, ensure_awaitable
from scrapy.utils.signal import (
send_catch_log,
send_catch_log_async,
send_catch_log_deferred,
)
from scrapy.utils.test import get_from_asyncio_queue
from tests.utils.decorators import inline_callbacks_test
from tests.utils.decorators import coroutine_test
if TYPE_CHECKING:
from collections.abc import Callable
if TYPE_CHECKING:
from collections.abc import Callable
@ -27,25 +29,22 @@ class TestSendCatchLog:
# whether the function being tested returns exceptions or failures
returns_exceptions: bool = False
@inline_callbacks_test
def test_send_catch_log(self):
@coroutine_test
async def test_send_catch_log(self, caplog: pytest.LogCaptureFixture) -> None:
test_signal = object()
handlers_called: set[Callable[..., None]] = set()
dispatcher.connect(self.error_handler, signal=test_signal)
dispatcher.connect(self.ok_handler, signal=test_signal)
with LogCapture() as log:
result = yield defer.maybeDeferred(
self._get_result,
test_signal,
arg="test",
handlers_called=handlers_called,
)
caplog.clear()
result = await ensure_awaitable(
self._get_result(test_signal, arg="test", handlers_called=handlers_called)
)
assert self.error_handler in handlers_called
assert self.ok_handler in handlers_called
assert len(log.records) == 1
record = log.records[0]
assert len(caplog.records) == 1
record = caplog.records[0]
assert "error_handler" in record.getMessage()
assert record.levelname == "ERROR"
assert result[0][0] == self.error_handler # pylint: disable=comparison-with-callable
@ -57,7 +56,7 @@ class TestSendCatchLog:
dispatcher.disconnect(self.error_handler, signal=test_signal)
dispatcher.disconnect(self.ok_handler, signal=test_signal)
def _get_result(self, signal, *a, **kw):
def _get_result(self, signal: Any, *a: Any, **kw: Any) -> Any:
return send_catch_log(signal, *a, **kw)
def error_handler(self, arg, handlers_called):
@ -72,7 +71,7 @@ class TestSendCatchLog:
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
class TestSendCatchLogDeferred(TestSendCatchLog):
def _get_result(self, signal, *a, **kw):
def _get_result(self, signal: Any, *a: Any, **kw: Any) -> Any:
return send_catch_log_deferred(signal, *a, **kw)
@ -137,14 +136,15 @@ class TestSendCatchLogAsyncAsyncio(TestSendCatchLogAsync):
class TestSendCatchLog2:
def test_error_logged_if_deferred_not_supported(self):
def test_error_logged_if_deferred_not_supported(
self, caplog: pytest.LogCaptureFixture
) -> None:
def test_handler():
return defer.Deferred()
test_signal = object()
dispatcher.connect(test_handler, test_signal)
with LogCapture() as log:
send_catch_log(test_signal)
assert len(log.records) == 1
assert "Cannot return deferreds from signal handler" in str(log)
send_catch_log(test_signal)
assert len(caplog.records) == 1
assert "Cannot return deferreds from signal handler" in caplog.text
dispatcher.disconnect(test_handler, test_signal)

14
tests/utils/bases/spider.py
View File

@ -1,11 +1,11 @@
from __future__ import annotations
import logging
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
from unittest import mock
import pytest
from testfixtures import LogCapture
from scrapy import signals
from scrapy.crawler import Crawler
@ -109,15 +109,15 @@ class TestSpiderBase(ABC):
yield crawler.crawl()
assert crawler.settings.get("TEST1") == "spider_instance"
def test_logger(self):
def test_logger(self, caplog: pytest.LogCaptureFixture) -> None:
spider = self.spider_class("example.com")
with LogCapture() as lc:
caplog.clear()
with caplog.at_level(logging.INFO):
spider.logger.info("test log msg")
lc.check(("example.com", "INFO", "test log msg"))
assert caplog.record_tuples == [("example.com", logging.INFO, "test log msg")]
record = lc.records[0]
assert "spider" in record.__dict__
assert record.spider is spider
record = caplog.records[0]
assert getattr(record, "spider", None) is spider
def test_log(self):
spider = self.spider_class("example.com")

1
tox.ini
View File

@ -44,7 +44,6 @@ deps =
pytest-cov >= 7.0.0
pytest-xdist
sybil >= 1.3.0 # https://github.com/cjw296/sybil/issues/20#issuecomment-605433422
testfixtures
pytest-twisted >= 1.14.3
[testenv]