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

Merge commit 'c9f952c25'

This commit is contained in:
aliowka 2026-06-21 20:16:46 +03:00
parent 7d2f108630 c9f952c258
commit e6b785cd04
45 changed files with 664 additions and 410 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pyproject.toml
View File

@ -231,6 +231,7 @@ disable = [
"undefined-variable",
"unused-argument",
"unused-variable",
"use-implicit-booleaness-not-comparison",
"useless-import-alias", # used as a hint to mypy
"useless-return", # https://github.com/pylint-dev/pylint/issues/6530
"wrong-import-position",

1
scrapy/__main__.py
View File

@ -1,3 +1,4 @@
# pragma: no file cover
from scrapy.cmdline import execute
if __name__ == "__main__":

19
scrapy/commands/__init__.py
View File

@ -16,6 +16,8 @@ from twisted.python import failure
from scrapy.exceptions import ScrapyDeprecationWarning, UsageError
from scrapy.utils.conf import arglist_to_dict, feed_process_params_from_cli
from scrapy.utils.deprecate import method_is_overridden
from scrapy.utils.python import global_object_name
if TYPE_CHECKING:
from collections.abc import Iterable
@ -36,6 +38,14 @@ class ScrapyCommand(ABC):
def __init__(self) -> None:
self.settings: Settings | None = None # set in scrapy.cmdline
if method_is_overridden(self.__class__, ScrapyCommand, "help"):
warnings.warn(
"The ScrapyCommand.help() method is deprecated and overriding "
f"it, as the {global_object_name(self.__class__)} class does, "
"has no effect; override long_desc() instead.",
ScrapyDeprecationWarning,
stacklevel=2,
)
def set_crawler(self, crawler: Crawler) -> None: # pragma: no cover
warnings.warn(
@ -68,10 +78,11 @@ class ScrapyCommand(ABC):
return self.short_desc()
def help(self) -> str:
"""An extensive help for the command. It will be shown when using the
"help" command. It can contain newlines since no post-formatting will
be applied to its contents.
"""
warnings.warn(
"ScrapyCommand.help() is deprecated, use long_desc() instead.",
ScrapyDeprecationWarning,
stacklevel=2,
)
return self.long_desc()
def add_options(self, parser: argparse.ArgumentParser) -> None:

4
scrapy/contracts/__init__.py
View File

@ -51,6 +51,8 @@ class Contract:
results.addSuccess(self.testcase_pre)
cb_result = cb(response, **cb_kwargs)
if isinstance(cb_result, (AsyncGenerator, CoroutineType)):
if isinstance(cb_result, CoroutineType):
cb_result.close()
raise TypeError("Contracts don't support async callbacks")
return list(cast("Iterable[Any]", iterate_spider_output(cb_result)))
@ -67,6 +69,8 @@ class Contract:
def wrapper(response: Response, **cb_kwargs: Any) -> list[Any]:
cb_result = cb(response, **cb_kwargs)
if isinstance(cb_result, (AsyncGenerator, CoroutineType)):
if isinstance(cb_result, CoroutineType):
cb_result.close()
raise TypeError("Contracts don't support async callbacks")
output = list(cast("Iterable[Any]", iterate_spider_output(cb_result)))
try:

6
scrapy/extensions/memusage.py
View File

@ -113,7 +113,7 @@ class MemoryUsage:
{"memusage": mem},
extra={"crawler": self.crawler},
)
if self.notify_mails:
if self.notify_mails: # pragma: no cover
subj = (
f"{self.crawler.settings['BOT_NAME']} terminated: "
f"memory usage exceeded {mem}MiB at {socket.gethostname()}"
@ -146,7 +146,7 @@ class MemoryUsage:
{"memusage": mem},
extra={"crawler": self.crawler},
)
if self.notify_mails:
if self.notify_mails: # pragma: no cover
subj = (
f"{self.crawler.settings['BOT_NAME']} warning: "
f"memory usage reached {mem}MiB at {socket.gethostname()}"
@ -155,7 +155,7 @@ class MemoryUsage:
self.crawler.stats.set_value("memusage/warning_notified", 1)
self.warned = True
def _send_report(self, rcpts: list[str], subject: str) -> None:
def _send_report(self, rcpts: list[str], subject: str) -> None: # pragma: no cover
"""send notification mail with some additional useful info"""
assert self.crawler.engine
assert self.crawler.stats

2
scrapy/resolver.py
View File

@ -75,7 +75,7 @@ class HostResolution:
def __init__(self, name: str):
self.name: str = name
def cancel(self) -> None:
def cancel(self) -> None: # pragma: no cover
raise NotImplementedError

6
scrapy/spidermiddlewares/referer.py
View File

@ -11,7 +11,7 @@ from typing import TYPE_CHECKING, cast
from urllib.parse import urlparse
from warnings import warn
from scrapy.exceptions import NotConfigured
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.http import Request, Response
from scrapy.spidermiddlewares.base import BaseSpiderMiddleware
from scrapy.utils.misc import load_object
@ -349,7 +349,7 @@ class RefererMiddleware(BaseSpiderMiddleware):
response = kwargs.pop("resp_or_url")
warn(
"Passing 'resp_or_url' is deprecated, use 'response' instead.",
DeprecationWarning,
ScrapyDeprecationWarning,
stacklevel=2,
)
if response is None:
@ -360,7 +360,7 @@ class RefererMiddleware(BaseSpiderMiddleware):
warn(
"Passing a response URL to RefererMiddleware.policy() instead "
"of a Response object is deprecated.",
DeprecationWarning,
ScrapyDeprecationWarning,
stacklevel=2,
)
allow_import_path = True

2
scrapy/spiders/crawl.py
View File

@ -110,6 +110,7 @@ class CrawlSpider(Spider):
"deprecated: it will be removed in future Scrapy releases. "
"Please override the CrawlSpider.parse_with_rules method "
"instead.",
ScrapyDeprecationWarning,
stacklevel=2,
)
@ -199,6 +200,7 @@ class CrawlSpider(Spider):
"The CrawlSpider._parse_response method is deprecated: "
"it will be removed in future Scrapy releases. "
"Please use the CrawlSpider.parse_with_rules method instead.",
ScrapyDeprecationWarning,
stacklevel=2,
)
return self.parse_with_rules(response, callback, cb_kwargs, follow)

9
scrapy/utils/misc.py
View File

@ -135,14 +135,9 @@ def walk_modules(path: str) -> list[ModuleType]: # pragma: no cover
return list(walk_modules_iter(path))
def md5sum(file: IO[bytes]) -> str:
def md5sum(file: IO[bytes]) -> str: # pragma: no cover
"""Calculate the md5 checksum of a file-like object without reading its
whole content in memory.
>>> from io import BytesIO
>>> md5sum(BytesIO(b'file content to hash'))
'784406af91dd5a54fbb9c84c2236595a'
"""
whole content in memory."""
warnings.warn(
(
"The scrapy.utils.misc.md5sum function is deprecated and will be "

2
tests/ignores.txt
View File

@ -1 +1,3 @@
scrapy/core/downloader/handlers/http.py
scrapy/extensions/statsmailer.py
scrapy/mail.py

45
tests/test_commands.py
View File

@ -12,6 +12,7 @@ import pytest
import scrapy
from scrapy.cmdline import _pop_command_name, _print_unknown_command_msg
from scrapy.commands import ScrapyCommand, ScrapyHelpFormatter, view
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.settings import Settings
from scrapy.utils.reactor import _asyncio_reactor_path
from tests.utils.cmdline import call, proc
@ -28,6 +29,50 @@ class EmptyCommand(ScrapyCommand):
pass
class TestHelpDeprecation:
def test_calling_help_is_deprecated(self) -> None:
command = EmptyCommand()
with pytest.warns(
ScrapyDeprecationWarning,
match=r"ScrapyCommand\.help\(\) is deprecated, use long_desc\(\) instead\.",
):
result = command.help()
# help() still delegates to long_desc() for backward compatibility.
assert result == command.long_desc()
def test_overriding_help_is_deprecated(self) -> None:
class HelpCommand(ScrapyCommand):
def short_desc(self) -> str:
return ""
def run(self, args: list[str], opts: argparse.Namespace) -> None:
pass
def help(self) -> str:
return "custom help"
with pytest.warns(
ScrapyDeprecationWarning,
match=r"The ScrapyCommand\.help\(\) method is deprecated and "
r"overriding it, as the .*HelpCommand class does, has no effect; "
r"override long_desc\(\) instead\.",
):
HelpCommand()
def test_not_overriding_help_does_not_warn(self, recwarn) -> None:
# Commands that do not override help() must not emit the
# override-deprecation warning when instantiated, including subclasses
# several levels below ScrapyCommand (as the built-in commands are).
class SubCommand(EmptyCommand):
pass
EmptyCommand()
SubCommand()
assert not [
w for w in recwarn.list if issubclass(w.category, ScrapyDeprecationWarning)
]
class TestCommandSettings:
def setup_method(self):
self.command = EmptyCommand()

3
tests/test_crawl.py
View File

@ -715,6 +715,9 @@ class TestCrawlSpider:
assert isinstance(crawler.spider, SingleRequestSpider)
assert crawler.spider.meta["responses"][0].certificate is None
@pytest.mark.filterwarnings(
r"ignore:.*You should use cryptography's X\.509 APIs:DeprecationWarning"
)
@pytest.mark.parametrize(
"url",
[

5
tests/test_crawler.py
View File

@ -3,7 +3,6 @@ from __future__ import annotations
import asyncio
import logging
import re
import warnings
from pathlib import Path
from typing import Any, ClassVar
@ -88,9 +87,7 @@ class TestCrawler(TestBaseCrawler):
self.assertOptionIsDefault(crawler.settings, "RETRY_ENABLED")
def test_crawler_accepts_None(self) -> None:
with warnings.catch_warnings():
warnings.simplefilter("ignore", ScrapyDeprecationWarning)
crawler = Crawler(DefaultSpider)
crawler = Crawler(DefaultSpider)
self.assertOptionIsDefault(crawler.settings, "RETRY_ENABLED")
def test_crawler_rejects_spider_objects(self) -> None:

3
tests/test_downloader_handlers_http_base.py
View File

@ -1148,6 +1148,9 @@ class TestHttpWithCrawlerBase(ABC):
reason = crawler.spider.meta["close_reason"] # type: ignore[attr-defined]
assert reason == "finished"
@pytest.mark.filterwarnings(
r"ignore:.*You should use cryptography's X\.509 APIs:DeprecationWarning"
)
@coroutine_test
async def test_response_ssl_certificate(self, mockserver: MockServer) -> None:
if not self.is_secure:

13
tests/test_dupefilters.py
View File

@ -3,8 +3,8 @@ import shutil
import sys
import tempfile
from pathlib import Path
from warnings import catch_warnings
import pytest
from testfixtures import LogCapture
from scrapy.core.scheduler import Scheduler
@ -260,11 +260,8 @@ class TestBaseDupeFilter:
dupefilter = _get_dupefilter(
settings={"DUPEFILTER_CLASS": BaseDupeFilter},
)
with catch_warnings(record=True) as warning_list:
with pytest.warns(
ScrapyDeprecationWarning,
match=r"Calling BaseDupeFilter\.log\(\) is deprecated.",
):
dupefilter.log(None, None)
assert len(warning_list) == 1
assert (
str(warning_list[0].message)
== "Calling BaseDupeFilter.log() is deprecated."
)
assert warning_list[0].category == ScrapyDeprecationWarning

115
tests/test_extension_memusage.py Normal file
View File

@ -0,0 +1,115 @@
from __future__ import annotations
import logging
import sys
import pytest
from scrapy import signals
from scrapy.core import engine as engine_mod
from scrapy.exceptions import NotConfigured
from scrapy.extensions import memusage as memusage_mod
from scrapy.extensions.memusage import MemoryUsage
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
from tests.utils import OneShotLoop
from tests.utils.decorators import coroutine_test
# MemoryUsage relies on the stdlib 'resource' module (not available on Windows)
pytestmark = pytest.mark.skipif(
sys.platform.startswith("win"),
reason="MemoryUsage extension not available on Windows",
)
MB = 1024 * 1024
class _LoopSpider(Spider):
name = "loop-data-spider"
def __init__(self, url: str, loops: int = 60, **kw):
super().__init__(**kw)
self.url = url
self.loops = loops
self.start_urls = [url]
def parse(self, response):
count = response.meta.get("count", 0)
if count + 1 < self.loops:
yield response.follow(
self.url, callback=self.parse, meta={"count": count + 1}
)
def test_memusage_disabled() -> None:
settings = {
"MEMUSAGE_ENABLED": False,
}
with pytest.raises(NotConfigured):
MemoryUsage.from_crawler(get_crawler(settings_dict=settings))
@coroutine_test
async def test_memusage_limit_closes_spider_with_reason_and_error_log(
caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
settings = {
"MEMUSAGE_LIMIT_MB": 10,
"MEMUSAGE_CHECK_INTERVAL_SECONDS": 0.01,
"TELNETCONSOLE_ENABLED": False,
"LOG_LEVEL": "INFO",
}
# Avoid background LoopingCall that can log after the test finishes.
monkeypatch.setattr(memusage_mod, "create_looping_call", OneShotLoop)
# Avoid engine start/stop races (the extension stops the engine in engine_started).
monkeypatch.setattr(engine_mod, "create_looping_call", OneShotLoop)
monkeypatch.setattr(MemoryUsage, "get_virtual_size", lambda _: 250 * MB)
crawler = get_crawler(spidercls=_LoopSpider, settings_dict=settings)
with caplog.at_level(logging.ERROR, logger="scrapy.extensions.memusage"):
await crawler.crawl_async(url="data:,", loops=100)
assert crawler.stats
assert crawler.stats.get_value("memusage/limit_reached") == 1
assert crawler.stats.get_value("finish_reason") == "memusage_exceeded"
assert any(
"memory usage exceeded" in r.getMessage().lower() for r in caplog.records
)
@coroutine_test
async def test_memusage_warning_logs_but_allows_normal_finish(
caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
settings = {
"MEMUSAGE_WARNING_MB": 50,
"MEMUSAGE_LIMIT_MB": 0, # no hard limit
"MEMUSAGE_CHECK_INTERVAL_SECONDS": 0.01,
"TELNETCONSOLE_ENABLED": False,
"LOG_LEVEL": "INFO",
}
# Avoid background LoopingCall that can log after the test finishes.
monkeypatch.setattr(memusage_mod, "create_looping_call", OneShotLoop)
monkeypatch.setattr(MemoryUsage, "get_virtual_size", lambda self: 75 * MB)
crawler = get_crawler(spidercls=_LoopSpider, settings_dict=settings)
warning_signals: list[int] = []
def on_warning_reached() -> None:
warning_signals.append(1)
crawler.signals.connect(on_warning_reached, signal=signals.memusage_warning_reached)
with caplog.at_level(logging.WARNING, logger="scrapy.extensions.memusage"):
await crawler.crawl_async(url="data:,", loops=60)
assert warning_signals == [1]
assert crawler.stats
assert crawler.stats.get_value("memusage/warning_reached") == 1
assert crawler.stats.get_value("finish_reason") == "finished"
assert any("memory usage reached" in r.getMessage().lower() for r in caplog.records)

23
tests/test_extension_statsmailer.py
View File

@ -1,20 +1,27 @@
import warnings
from unittest.mock import MagicMock
import pytest
from scrapy import signals
from scrapy.exceptions import NotConfigured
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.signalmanager import SignalManager
from scrapy.statscollectors import StatsCollector
from scrapy.utils.spider import DefaultSpider
pytestmark = pytest.mark.filterwarnings(
"ignore:The scrapy.extensions.statsmailer module is deprecated:scrapy.exceptions.ScrapyDeprecationWarning",
"ignore:The scrapy.mail module is deprecated:scrapy.exceptions.ScrapyDeprecationWarning",
)
from scrapy.extensions import statsmailer # noqa: E402
from scrapy.mail import MailSender # noqa: E402
with warnings.catch_warnings():
warnings.filterwarnings(
"ignore",
r"The scrapy\.extensions\.statsmailer module is deprecated",
ScrapyDeprecationWarning,
)
warnings.filterwarnings(
"ignore",
r"The scrapy\.mail module is deprecated",
ScrapyDeprecationWarning,
)
from scrapy.extensions import statsmailer
from scrapy.mail import MailSender
@pytest.fixture

3
tests/test_http2_client_protocol.py
View File

@ -669,6 +669,9 @@ class TestHttps2ClientProtocol:
response = await make_request(client, request)
assert response.status == status
@pytest.mark.filterwarnings(
r"ignore:.*You should use cryptography's X\.509 APIs:DeprecationWarning"
)
@deferred_f_from_coro_f
async def test_response_has_correct_certificate_ip_address(
self,

6
tests/test_http_request.py
View File

@ -460,11 +460,13 @@ class TestRequest:
def test_from_curl_ignore_unknown_options(self):
# By default: it works and ignores the unknown options: --foo and -z
with warnings.catch_warnings(): # avoid warning when executing tests
warnings.simplefilter("ignore")
warnings.filterwarnings(
"ignore", category=UserWarning, message="Unrecognized options:"
)
r = self.request_class.from_curl(
'curl -X DELETE "http://example.org" --foo -z',
)
assert r.method == "DELETE"
assert r.method == "DELETE"
# If `ignore_unknown_options` is set to `False` it raises an error with
# the unknown options: --foo and -z

41
tests/test_http_request_json.py
View File

@ -4,6 +4,8 @@ import json
import warnings
from unittest import mock
import pytest
from scrapy.http import JsonRequest
from scrapy.utils.python import to_bytes
from tests.test_http_request import TestRequest
@ -63,40 +65,40 @@ class TestJsonRequest(TestRequest):
data = {
"name": "value",
}
with warnings.catch_warnings(record=True) as _warnings:
with pytest.warns(UserWarning, match="data will be ignored"):
r5 = self.request_class(url="http://www.example.com/", body=body, data=data)
assert r5.body == body
assert r5.method == "GET"
assert len(_warnings) == 1
assert "data will be ignored" in str(_warnings[0].message)
assert r5.body == body
assert r5.method == "GET"
def test_empty_body_data(self):
"""passing any body value and data should result a warning"""
data = {
"name": "value",
}
with warnings.catch_warnings(record=True) as _warnings:
with pytest.warns(UserWarning, match="data will be ignored"):
r6 = self.request_class(url="http://www.example.com/", body=b"", data=data)
assert r6.body == b""
assert r6.method == "GET"
assert len(_warnings) == 1
assert "data will be ignored" in str(_warnings[0].message)
assert r6.body == b""
assert r6.method == "GET"
def test_body_none_data(self):
data = {
"name": "value",
}
with warnings.catch_warnings(record=True) as _warnings:
with warnings.catch_warnings():
warnings.filterwarnings(
"error", category=UserWarning, message="Both body and data passed"
)
r7 = self.request_class(url="http://www.example.com/", body=None, data=data)
assert r7.body == to_bytes(json.dumps(data))
assert r7.method == "POST"
assert len(_warnings) == 0
assert r7.body == to_bytes(json.dumps(data))
assert r7.method == "POST"
def test_body_data_none(self):
with warnings.catch_warnings(record=True) as _warnings:
with warnings.catch_warnings():
warnings.filterwarnings(
"error", category=UserWarning, message="Both body and data passed"
)
r8 = self.request_class(url="http://www.example.com/", body=None, data=None)
assert r8.method == "GET"
assert len(_warnings) == 0
assert r8.method == "GET"
def test_dumps_sort_keys(self):
"""Test that sort_keys=True is passed to json.dumps by default"""
@ -183,8 +185,5 @@ class TestJsonRequest(TestRequest):
}
r1 = self.request_class(url="http://www.example.com/", data=data1, body=body1)
with warnings.catch_warnings(record=True) as _warnings:
with pytest.warns(UserWarning, match="data will be ignored"):
r1.replace(data=data2, body=body2)
assert "Both body and data passed. data will be ignored" in str(
_warnings[0].message
)

35
tests/test_linkextractors.py
View File

@ -9,7 +9,7 @@ from w3lib import __version__ as w3lib_version
from scrapy.http import HtmlResponse, XmlResponse
from scrapy.link import Link
from scrapy.linkextractors.lxmlhtml import LxmlLinkExtractor
from scrapy.linkextractors.lxmlhtml import LxmlLinkExtractor, LxmlParserLinkExtractor
from tests import get_testdata
@ -837,3 +837,36 @@ class TestLxmlLinkExtractor(Base.TestLinkExtractorBase):
def test_link_allowed_is_false_with_missing_url_prefix(self):
bad_link = Link("should_have_prefix.example")
assert not LxmlLinkExtractor()._link_allowed(bad_link)
class TestLxmlParserLinkExtractor:
def test_extract_links(self):
html = b'<a href="http://example.com/page.html">Link</a>'
response = HtmlResponse("http://example.com/", body=html)
lx = LxmlParserLinkExtractor()
assert lx.extract_links(response) == [
Link(url="http://example.com/page.html", text="Link", nofollow=False),
]
def test_strip_false(self):
# With strip=False, trailing whitespace on a relative href survives urljoin
# and is visible to process_value (safe_url_string cleans it up afterward).
# Here process_value rejects URLs that still carry trailing whitespace,
# demonstrating the difference from strip=True.
def reject_trailing_whitespace(url):
return None if url != url.rstrip() else url
html = b'<a href="page.html ">Link</a>'
response = HtmlResponse("http://example.com/", body=html)
lx_strip = LxmlParserLinkExtractor(
strip=True, process=reject_trailing_whitespace
)
assert lx_strip.extract_links(response) == [
Link(url="http://example.com/page.html", text="Link", nofollow=False),
]
lx_no_strip = LxmlParserLinkExtractor(
strip=False, process=reject_trailing_whitespace
)
assert lx_no_strip.extract_links(response) == []

23
tests/test_pipeline_files.py
View File

@ -2,7 +2,6 @@ import dataclasses
import os
import random
import time
import warnings
from abc import ABC, abstractmethod
from datetime import datetime
from ftplib import FTP
@ -786,12 +785,10 @@ class TestBuildFromCrawler:
class Pipeline(FilesPipeline):
pass
with warnings.catch_warnings(record=True) as w:
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert len(w) == 0
assert pipe.store
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert pipe.store
def test_has_from_crawler_and_init(self):
class Pipeline(FilesPipeline):
@ -805,13 +802,11 @@ class TestBuildFromCrawler:
o._from_crawler_called = True
return o
with warnings.catch_warnings(record=True) as w:
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert len(w) == 0
assert pipe.store
assert pipe._from_crawler_called
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert pipe.store
assert pipe._from_crawler_called
@pytest.mark.parametrize("store", [None, ""])

32
tests/test_pipeline_media.py
View File

@ -1,6 +1,5 @@
from __future__ import annotations
import warnings
from unittest.mock import MagicMock
import pytest
@ -425,11 +424,9 @@ class TestBuildFromCrawler:
class Pipeline(UserDefinedPipeline):
pass
with warnings.catch_warnings(record=True) as w:
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert len(w) == 0
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
def test_has_from_crawler_and_init(self):
class Pipeline(UserDefinedPipeline):
@ -447,13 +444,11 @@ class TestBuildFromCrawler:
o._from_crawler_called = True
return o
with warnings.catch_warnings(record=True) as w:
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert len(w) == 0
assert pipe._from_crawler_called
assert pipe._init_called
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert pipe._from_crawler_called
assert pipe._init_called
def test_has_from_crawler(self):
class Pipeline(UserDefinedPipeline):
@ -467,13 +462,10 @@ class TestBuildFromCrawler:
o.store_uri = settings["FILES_STORE"]
return o
with warnings.catch_warnings(record=True) as w:
pipe = Pipeline.from_crawler(self.crawler)
# this and the next assert will fail as MediaPipeline.from_crawler() wasn't called
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert len(w) == 0
assert pipe._from_crawler_called
pipe = Pipeline.from_crawler(self.crawler)
assert pipe.crawler == self.crawler
assert pipe._fingerprinter
assert pipe._from_crawler_called
class MediaFailedFailurePipeline(MockedMediaPipeline):

52
tests/test_pqueues.py
View File

@ -8,7 +8,7 @@ from scrapy.core.downloader import Downloader
from scrapy.http.request import Request
from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue
from scrapy.spiders import Spider
from scrapy.squeues import FifoMemoryQueue
from scrapy.squeues import FifoMemoryQueue, PickleFifoDiskQueue
from scrapy.utils.misc import build_from_crawler, load_object
from scrapy.utils.test import get_crawler
from tests.test_scheduler import MockDownloader
@ -76,6 +76,29 @@ class TestPriorityQueue:
assert queue.pop().url == req3.url
assert not queue.close()
def test_init_prios_with_start_queue(self):
temp_dir = tempfile.mkdtemp()
queue = ScrapyPriorityQueue.from_crawler(
self.crawler,
PickleFifoDiskQueue,
temp_dir,
start_queue_cls=PickleFifoDiskQueue,
)
req = Request("https://example.org/", meta={"is_start_request": True})
queue.push(req)
startprios = queue.close()
queue2 = ScrapyPriorityQueue.from_crawler(
self.crawler,
PickleFifoDiskQueue,
temp_dir,
startprios,
start_queue_cls=PickleFifoDiskQueue,
)
assert len(queue2) == 1
assert queue2.pop().url == req.url
queue2.close()
def test_queue_push_pop_priorities(self):
temp_dir = tempfile.mkdtemp()
queue = ScrapyPriorityQueue.from_crawler(
@ -207,6 +230,33 @@ class TestDownloaderAwarePriorityQueue:
assert slots == ["slot-a", "slot-b", "slot-c", "slot-a"]
def test_pop_prefers_slot_with_fewer_active_downloads(self):
downloader = self.queue._downloader_interface.downloader
req_a = Request("https://example.org/a")
req_a.meta[Downloader.DOWNLOAD_SLOT] = "slot-a"
req_b = Request("https://example.org/b")
req_b.meta[Downloader.DOWNLOAD_SLOT] = "slot-b"
req_c = Request("https://example.org/c")
req_c.meta[Downloader.DOWNLOAD_SLOT] = "slot-c"
for req in (req_a, req_b, req_c):
self.queue.push(req)
downloader.increment("slot-a")
downloader.increment("slot-c")
popped = self.queue.pop()
assert popped.url == req_b.url
def test_contains(self):
req = Request("https://example.org/")
req.meta[Downloader.DOWNLOAD_SLOT] = "example-slot"
assert "example-slot" not in self.queue
self.queue.push(req)
assert "example-slot" in self.queue
assert "other-slot" not in self.queue
@pytest.mark.parametrize(
("input_", "output"),

4
tests/test_request_cb_kwargs.py
View File

@ -102,9 +102,7 @@ class KeywordArgumentsSpider(MockServerSpider):
self.checks.append(kwargs["callback"] == "some_callback")
self.crawler.stats.inc_value("boolean_checks", 3)
elif response.url.endswith("/general_without"):
self.checks.append(
kwargs == {} # pylint: disable=use-implicit-booleaness-not-comparison
)
self.checks.append(kwargs == {})
self.crawler.stats.inc_value("boolean_checks")
def parse_no_kwargs(self, response):

55
tests/test_resolver.py Normal file
View File

@ -0,0 +1,55 @@
from __future__ import annotations
from unittest.mock import Mock
import pytest
from scrapy.resolver import CachingHostnameResolver, CachingThreadedResolver, dnscache
from scrapy.utils.defer import maybe_deferred_to_future
from scrapy.utils.test import get_crawler
from tests.utils.decorators import coroutine_test
@pytest.fixture(autouse=True)
def reset_dnscache():
original_limit = dnscache.limit
dnscache.clear()
yield
dnscache.clear()
dnscache.limit = original_limit
def test_caching_threaded_resolver_dnscache_disabled():
crawler = get_crawler(settings_dict={"DNSCACHE_ENABLED": False})
CachingThreadedResolver.from_crawler(crawler, Mock())
assert dnscache.limit == 0
@coroutine_test
async def test_caching_threaded_resolver_getHostByName_cache_hit():
resolver = CachingThreadedResolver(Mock(), cache_size=10, timeout=5.0)
dnscache["example.com"] = "1.2.3.4"
result = await maybe_deferred_to_future(resolver.getHostByName("example.com"))
assert result == "1.2.3.4"
def test_caching_hostname_resolver_dnscache_disabled():
crawler = get_crawler(settings_dict={"DNSCACHE_ENABLED": False})
CachingHostnameResolver.from_crawler(crawler, Mock())
assert dnscache.limit == 0
def test_caching_hostname_resolver_no_addresses_not_cached():
def fake_resolve(receiver, *_):
receiver.resolutionBegan(Mock())
receiver.resolutionComplete()
return receiver
reactor = Mock()
reactor.nameResolver.resolveHostName.side_effect = fake_resolve
resolver = CachingHostnameResolver(reactor, cache_size=10)
resolver.resolveHostName(Mock(), "example.com")
assert "example.com" not in dnscache

7
tests/test_scheduler.py
View File

@ -12,6 +12,7 @@ import pytest
from scrapy.core.downloader import Downloader
from scrapy.core.scheduler import BaseScheduler, Scheduler
from scrapy.crawler import Crawler
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import Request
from scrapy.spiders import Spider
from scrapy.utils.defer import ensure_awaitable
@ -396,7 +397,11 @@ class TestIncompatibility:
def test_incompatibility(self):
with warnings.catch_warnings():
warnings.filterwarnings("ignore")
warnings.filterwarnings(
"ignore",
category=ScrapyDeprecationWarning,
message="The CONCURRENT_REQUESTS_PER_IP setting is deprecated",
)
with pytest.raises(
ValueError, match="does not support CONCURRENT_REQUESTS_PER_IP"
):

21
tests/test_scrapy__getattr__.py
View File

@ -1,15 +1,18 @@
import warnings
from __future__ import annotations
import pytest
from scrapy.exceptions import ScrapyDeprecationWarning
def test_deprecated_concurrent_requests_per_ip_attribute():
with warnings.catch_warnings(record=True) as warns:
def test_deprecated_concurrent_requests_per_ip_attribute() -> None:
with pytest.warns(
ScrapyDeprecationWarning,
match=r"scrapy\.settings\.default_settings\.CONCURRENT_REQUESTS_PER_IP attribute is deprecated",
):
from scrapy.settings.default_settings import ( # noqa: PLC0415
CONCURRENT_REQUESTS_PER_IP,
)
assert CONCURRENT_REQUESTS_PER_IP is not None
assert isinstance(CONCURRENT_REQUESTS_PER_IP, int)
assert (
"The scrapy.settings.default_settings.CONCURRENT_REQUESTS_PER_IP attribute is deprecated, use scrapy.settings.default_settings.CONCURRENT_REQUESTS_PER_DOMAIN instead."
in warns[0].message.args
)
assert CONCURRENT_REQUESTS_PER_IP is not None
assert isinstance(CONCURRENT_REQUESTS_PER_IP, int)

16
tests/test_settings/__init__.py
View File

@ -1,13 +1,13 @@
# pylint: disable=unsubscriptable-object,unsupported-membership-test,use-implicit-booleaness-not-comparison
# pylint: disable=unsubscriptable-object,unsupported-membership-test
# (too many false positives)
import logging
import warnings
from unittest import mock
import pytest
from scrapy.core.downloader.handlers.file import FileDownloadHandler
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.settings import (
SETTINGS_PRIORITIES,
BaseSettings,
@ -711,15 +711,13 @@ def test_remove_from_list(before, name, item, after):
def test_deprecated_concurrent_requests_per_ip_setting():
with warnings.catch_warnings(record=True) as warns:
settings = Settings({"CONCURRENT_REQUESTS_PER_IP": 1})
settings = Settings({"CONCURRENT_REQUESTS_PER_IP": 1})
with pytest.warns(
ScrapyDeprecationWarning,
match="The CONCURRENT_REQUESTS_PER_IP setting is deprecated",
):
settings.get("CONCURRENT_REQUESTS_PER_IP")
assert (
str(warns[0].message)
== "The CONCURRENT_REQUESTS_PER_IP setting is deprecated, use CONCURRENT_REQUESTS_PER_DOMAIN instead."
)
class Component1:
pass

2
tests/test_spider.py
View File

@ -22,7 +22,7 @@ class TestSpider:
def test_base_spider(self):
spider = self.spider_class("example.com")
assert spider.name == "example.com"
assert spider.start_urls == [] # pylint: disable=use-implicit-booleaness-not-comparison
assert spider.start_urls == []
def test_spider_args(self):
"""``__init__`` method arguments are assigned to spider attributes"""

24
tests/test_spider_crawl.py
View File

@ -8,6 +8,7 @@ import pytest
from testfixtures import LogCapture
from w3lib.url import safe_url_string
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import HtmlResponse, Request, TextResponse
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule, Spider
@ -264,13 +265,16 @@ class TestCrawlSpider(TestSpider):
start_urls = "https://www.example.com"
_follow_links = False
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.simplefilter("error", category=ScrapyDeprecationWarning)
spider = _CrawlSpider()
assert len(w) == 0
with pytest.warns(
ScrapyDeprecationWarning,
match=r"CrawlSpider\._parse_response method is deprecated",
):
spider._parse_response(
TextResponse(spider.start_urls, body=b""), None, None
)
assert len(w) == 1
def test_parse_response_override(self):
class _CrawlSpider(CrawlSpider):
@ -281,26 +285,28 @@ class TestCrawlSpider(TestSpider):
start_urls = "https://www.example.com"
_follow_links = False
with warnings.catch_warnings(record=True) as w:
assert len(w) == 0
with pytest.warns(
ScrapyDeprecationWarning,
match=r"CrawlSpider\._parse_response method, which the",
):
spider = _CrawlSpider()
assert len(w) == 1
with warnings.catch_warnings():
warnings.simplefilter("error", category=ScrapyDeprecationWarning)
spider._parse_response(
TextResponse(spider.start_urls, body=b""), None, None
)
assert len(w) == 1
def test_parse_with_rules(self):
class _CrawlSpider(CrawlSpider):
name = "test"
start_urls = "https://www.example.com"
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.simplefilter("error", category=ScrapyDeprecationWarning)
spider = _CrawlSpider()
spider.parse_with_rules(
TextResponse(spider.start_urls, body=b""), None, None
)
assert len(w) == 0
class TestDeprecation:

5
tests/test_spider_sitemap.py
View File

@ -2,7 +2,6 @@ from __future__ import annotations
import gzip
import re
import warnings
from datetime import datetime
from io import BytesIO
from logging import WARNING
@ -429,9 +428,7 @@ Sitemap: /sitemap-relative-url.xml
crawler = get_crawler(TestSpider)
spider = TestSpider.from_crawler(crawler)
with warnings.catch_warnings():
warnings.simplefilter("error")
requests = [request async for request in spider.start()]
requests = [request async for request in spider.start()]
assert len(requests) == 1
request = requests[0]

13
tests/test_spider_start.py
View File

@ -1,6 +1,5 @@
from __future__ import annotations
import warnings
from asyncio import sleep
from typing import Any
@ -45,9 +44,7 @@ class TestMain:
async def parse(self, response):
yield ITEM_A
with warnings.catch_warnings():
warnings.simplefilter("error")
await self._test_spider(TestSpider, [ITEM_A])
await self._test_spider(TestSpider, [ITEM_A])
@coroutine_test
async def test_start(self):
@ -57,9 +54,7 @@ class TestMain:
async def start(self):
yield ITEM_A
with warnings.catch_warnings():
warnings.simplefilter("error")
await self._test_spider(TestSpider, [ITEM_A])
await self._test_spider(TestSpider, [ITEM_A])
@coroutine_test
async def test_start_subclass(self):
@ -70,9 +65,7 @@ class TestMain:
class TestSpider(BaseSpider):
name = "test"
with warnings.catch_warnings():
warnings.simplefilter("error")
await self._test_spider(TestSpider, [ITEM_A])
await self._test_spider(TestSpider, [ITEM_A])
async def _test_start(self, start_, expected_items=None):
class TestSpider(Spider):

105
tests/test_spiderloader/__init__.py
View File

@ -1,7 +1,6 @@
import contextlib
import shutil
import sys
import warnings
from pathlib import Path
from unittest import mock
@ -137,50 +136,38 @@ class TestSpiderLoader:
SpiderLoader.from_settings(settings)
def test_bad_spider_modules_warning(self):
with warnings.catch_warnings(record=True) as w:
module = "tests.test_spiderloader.test_spiders.doesnotexist"
settings = Settings(
{"SPIDER_MODULES": [module], "SPIDER_LOADER_WARN_ONLY": True}
)
module = "tests.test_spiderloader.test_spiders.doesnotexist"
settings = Settings(
{"SPIDER_MODULES": [module], "SPIDER_LOADER_WARN_ONLY": True}
)
with pytest.warns(RuntimeWarning, match="Could not load spiders from module"):
spider_loader = SpiderLoader.from_settings(settings)
if str(w[0].message).startswith("_SixMetaPathImporter"):
# needed on 3.10 because of https://github.com/benjaminp/six/issues/349,
# at least until all six versions we can import (including botocore.vendored.six)
# are updated to 1.16.0+
w.pop(0)
assert "Could not load spiders from module" in str(w[0].message)
spiders = spider_loader.list()
assert not spiders
spiders = spider_loader.list()
assert not spiders
def test_syntax_error_exception(self):
module = "tests.test_spiderloader.test_spiders.spider1"
settings = Settings({"SPIDER_MODULES": [module]})
with mock.patch.object(SpiderLoader, "_load_spiders") as m:
m.side_effect = SyntaxError
settings = Settings({"SPIDER_MODULES": [module]})
with pytest.raises(SyntaxError):
SpiderLoader.from_settings(settings)
def test_syntax_error_warning(self):
with (
warnings.catch_warnings(record=True) as w,
mock.patch.object(SpiderLoader, "_load_spiders") as m,
):
module = "tests.test_spiderloader.test_spiders.spider1"
settings = Settings(
{"SPIDER_MODULES": [module], "SPIDER_LOADER_WARN_ONLY": True}
)
with mock.patch.object(SpiderLoader, "_load_spiders") as m:
m.side_effect = SyntaxError
module = "tests.test_spiderloader.test_spiders.spider1"
settings = Settings(
{"SPIDER_MODULES": [module], "SPIDER_LOADER_WARN_ONLY": True}
)
spider_loader = SpiderLoader.from_settings(settings)
if str(w[0].message).startswith("_SixMetaPathImporter"):
# needed on 3.10 because of https://github.com/benjaminp/six/issues/349,
# at least until all six versions we can import (including botocore.vendored.six)
# are updated to 1.16.0+
w.pop(0)
assert "Could not load spiders from module" in str(w[0].message)
with pytest.warns(
RuntimeWarning, match="Could not load spiders from module"
):
spider_loader = SpiderLoader.from_settings(settings)
spiders = spider_loader.list()
assert not spiders
spiders = spider_loader.list()
assert not spiders
class TestDuplicateSpiderNameLoader:
@ -190,21 +177,17 @@ class TestDuplicateSpiderNameLoader:
# copy 1 spider module so as to have duplicate spider name
shutil.copyfile(spiders_dir / "spider3.py", spiders_dir / "spider3dupe.py")
with warnings.catch_warnings(record=True) as w:
msg = r"""There are several spiders with the same name:
Spider3 named 'spider3' \(in test_spiders_xxx\.spider3\)
Spider3 named 'spider3' \(in test_spiders_xxx\.spider3dupe\)
This can cause unexpected behavior\."""
with pytest.warns(UserWarning, match=msg):
spider_loader = SpiderLoader.from_settings(settings)
assert len(w) == 1
msg = str(w[0].message)
assert "several spiders with the same name" in msg
assert "'spider3'" in msg
assert msg.count("'spider3'") == 2
assert "'spider1'" not in msg
assert "'spider2'" not in msg
assert "'spider4'" not in msg
spiders = set(spider_loader.list())
assert spiders == {"spider1", "spider2", "spider3", "spider4"}
spiders = set(spider_loader.list())
assert spiders == {"spider1", "spider2", "spider3", "spider4"}
def test_multiple_dupename_warning(self, spider_loader_env):
settings, spiders_dir = spider_loader_env
@ -213,23 +196,21 @@ class TestDuplicateSpiderNameLoader:
shutil.copyfile(spiders_dir / "spider1.py", spiders_dir / "spider1dupe.py")
shutil.copyfile(spiders_dir / "spider2.py", spiders_dir / "spider2dupe.py")
with warnings.catch_warnings(record=True) as w:
msg = r"""There are several spiders with the same name:
Spider1 named 'spider1' \(in test_spiders_xxx\.spider1\)
Spider1 named 'spider1' \(in test_spiders_xxx\.spider1dupe\)
Spider2 named 'spider2' \(in test_spiders_xxx\.spider2\)
Spider2 named 'spider2' \(in test_spiders_xxx\.spider2dupe\)
This can cause unexpected behavior\."""
with pytest.warns(UserWarning, match=msg):
spider_loader = SpiderLoader.from_settings(settings)
assert len(w) == 1
msg = str(w[0].message)
assert "several spiders with the same name" in msg
assert "'spider1'" in msg
assert msg.count("'spider1'") == 2
assert "'spider2'" in msg
assert msg.count("'spider2'") == 2
assert "'spider3'" not in msg
assert "'spider4'" not in msg
spiders = set(spider_loader.list())
assert spiders == {"spider1", "spider2", "spider3", "spider4"}
spiders = set(spider_loader.list())
assert spiders == {"spider1", "spider2", "spider3", "spider4"}
class CustomSpiderLoader(SpiderLoader):

5
tests/test_spidermiddleware_process_start.py
View File

@ -1,4 +1,3 @@
import warnings
from asyncio import sleep
import pytest
@ -76,9 +75,7 @@ class TestMain:
@coroutine_test
async def test_modern_mw_modern_spider(self):
with warnings.catch_warnings():
warnings.simplefilter("error")
await self._test_wrap(ModernWrapSpiderMiddleware, ModernWrapSpider)
await self._test_wrap(ModernWrapSpiderMiddleware, ModernWrapSpider)
async def _test_sleep(self, spider_middlewares):
class TestSpider(Spider):

64
tests/test_spidermiddleware_referer.py
View File

@ -6,6 +6,7 @@ from urllib.parse import urlparse
import pytest
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import Request, Response
from scrapy.settings import Settings
from scrapy.spidermiddlewares.referer import (
@ -842,13 +843,14 @@ class TestRequestMetaSettingFallback:
response = Response(origin, headers=response_headers)
request = Request(target, meta=request_meta)
with warnings.catch_warnings(record=True) as w:
if check_warning:
with pytest.warns(
RuntimeWarning, match="Could not load referrer policy"
):
policy = mw.policy(response, request)
else:
policy = mw.policy(response, request)
assert isinstance(policy, policy_class)
if check_warning:
assert len(w) == 1
assert w[0].category is RuntimeWarning, w[0].message
assert isinstance(policy, policy_class)
class TestSettingsPolicyByName:
@ -973,49 +975,39 @@ class TestPolicyMethodResponseParamRename:
self.response = Response("http://www.example.com")
def test_pos_string(self):
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
ScrapyDeprecationWarning,
match=r"Passing a response URL to RefererMiddleware\.policy\(\)",
):
self.mw.policy("http://old.com", self.request)
found = False
for warning in w:
if "Passing a response URL" in str(warning.message):
found = True
break
assert found
def test_pos_response(self):
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.filterwarnings(
"error",
category=ScrapyDeprecationWarning,
message=r"Passing 'resp_or_url' is deprecated",
)
self.mw.policy(self.response, self.request)
for warning in w:
assert "resp_or_url" not in str(warning.message)
def test_key_resp_or_url(self):
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
ScrapyDeprecationWarning, match=r"Passing 'resp_or_url' is deprecated"
):
self.mw.policy(resp_or_url=self.response, request=self.request)
found = False
for warning in w:
if "Passing 'resp_or_url' is deprecated, use 'response' instead" in str(
warning.message
):
found = True
break
assert found
def test_key_response(self):
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.filterwarnings(
"error",
category=ScrapyDeprecationWarning,
message=r"Passing 'resp_or_url' is deprecated",
)
self.mw.policy(response=self.response, request=self.request)
for warning in w:
assert "resp_or_url" not in str(warning.message)
def test_key_response_string(self):
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
with pytest.warns(ScrapyDeprecationWarning, match="Passing a response URL"):
self.mw.policy(response="http://old.com", request=self.request)
found = False
for warning in w:
if "Passing a response URL" in str(warning.message):
found = True
break
assert found
def test_both_resp_or_url_and_response(self):
with pytest.raises(

9
tests/test_stats.py
View File

@ -94,8 +94,13 @@ class TestStatsCollector:
assert stats.get_value("test2") == 35
stats.min_value("test4", 7)
assert stats.get_value("test4") == 7
stats.set_stats({"replaced": "stats"})
assert stats.get_stats() == {"replaced": "stats"}
stats.clear_stats()
assert stats.get_stats() == {}
def test_dummy_collector(self, crawler: Crawler) -> None:
def test_dummy_collector(self) -> None:
crawler = get_crawler(Spider, {"STATS_DUMP": False})
stats = DummyStatsCollector(crawler)
assert stats.get_stats() == {}
assert stats.get_value("anything") is None
@ -104,9 +109,11 @@ class TestStatsCollector:
stats.inc_value("v1")
stats.max_value("v2", 100)
stats.min_value("v3", 100)
stats.set_stats({"key": "val"})
stats.open_spider()
stats.set_value("test", "value")
assert stats.get_stats() == {}
stats.close_spider()
def test_deprecated_spider_arg(self, crawler: Crawler, spider: Spider) -> None:
stats = StatsCollector(crawler)

8
tests/test_utils_curl.py
View File

@ -213,10 +213,12 @@ class TestCurlToRequestKwargs:
def test_ignore_unknown_options(self):
# case 1: ignore_unknown_options=True:
curl_command = "curl --bar --baz http://www.example.com"
expected_result = {"method": "GET", "url": "http://www.example.com"}
with warnings.catch_warnings(): # avoid warning when executing tests
warnings.simplefilter("ignore")
curl_command = "curl --bar --baz http://www.example.com"
expected_result = {"method": "GET", "url": "http://www.example.com"}
warnings.filterwarnings(
"ignore", category=UserWarning, message="Unrecognized options:"
)
assert curl_to_request_kwargs(curl_command) == expected_result
# case 2: ignore_unknown_options=False (raise exception):

15
tests/test_utils_datatypes.py
View File

@ -1,5 +1,4 @@
import copy
import warnings
from abc import ABC, abstractmethod
from collections.abc import Iterator, Mapping, MutableMapping
from typing import Any
@ -227,18 +226,12 @@ class TestCaselessDict(TestCaseInsensitiveDictBase):
dict_class = CaselessDict
def test_deprecation_message(self):
with warnings.catch_warnings(record=True) as caught:
warnings.filterwarnings("always", category=ScrapyDeprecationWarning)
with pytest.warns(
ScrapyDeprecationWarning,
match=r"scrapy.utils.datatypes.CaselessDict is deprecated",
):
self.dict_class({"foo": "bar"})
assert len(caught) == 1
assert issubclass(caught[0].category, ScrapyDeprecationWarning)
assert (
str(caught[0].message)
== "scrapy.utils.datatypes.CaselessDict is deprecated,"
" please use scrapy.utils.datatypes.CaseInsensitiveDict instead"
)
class TestSequenceExclude:
def test_list(self):

148
tests/test_utils_deprecate.py
View File

@ -1,7 +1,6 @@
import inspect
import warnings
from unittest import mock
from warnings import WarningMessage
import pytest
@ -22,34 +21,26 @@ class NewName(SomeBaseClass):
class TestWarnWhenSubclassed:
def _mywarnings(self, w: list[WarningMessage]) -> list[WarningMessage]:
return [x for x in w if x.category is MyWarning]
def test_no_warning_on_definition(self):
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.simplefilter("error", category=ScrapyDeprecationWarning)
create_deprecated_class("Deprecated", NewName)
w = self._mywarnings(w)
assert w == []
def test_subclassing_warning_message(self):
msg = (
r"tests\.test_utils_deprecate\.UserClass inherits from "
r"deprecated class tests\.test_utils_deprecate\.Deprecated, "
r"please inherit from tests\.test_utils_deprecate\.NewName."
r" \(warning only on first subclass, there may be others\)"
)
Deprecated = create_deprecated_class(
"Deprecated", NewName, warn_category=MyWarning
)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(MyWarning, match=msg) as w:
class UserClass(Deprecated):
pass
w = self._mywarnings(w)
assert len(w) == 1
assert (
str(w[0].message) == "tests.test_utils_deprecate.UserClass inherits from "
"deprecated class tests.test_utils_deprecate.Deprecated, "
"please inherit from tests.test_utils_deprecate.NewName."
" (warning only on first subclass, there may be others)"
)
assert w[0].lineno == inspect.getsourcelines(UserClass)[1]
def test_custom_class_paths(self):
@ -61,62 +52,77 @@ class TestWarnWhenSubclassed:
warn_category=MyWarning,
)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
MyWarning,
match=r"UserClass inherits from deprecated class bar\.OldClass, please inherit from foo\.NewClass",
):
class UserClass(Deprecated):
pass
with pytest.warns(
MyWarning,
match=r"bar\.OldClass is deprecated, instantiate foo\.NewClass instead",
):
_ = Deprecated()
w = self._mywarnings(w)
assert len(w) == 2
assert "foo.NewClass" in str(w[0].message)
assert "bar.OldClass" in str(w[0].message)
assert "foo.NewClass" in str(w[1].message)
assert "bar.OldClass" in str(w[1].message)
def test_subclassing_warns_only_on_direct_children(self):
Deprecated = create_deprecated_class(
"Deprecated", NewName, warn_once=False, warn_category=MyWarning
)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
MyWarning,
match="UserClass inherits from deprecated class",
):
class UserClass(Deprecated):
pass
with warnings.catch_warnings():
warnings.simplefilter("error", MyWarning)
class NoWarnOnMe(UserClass):
pass
w = self._mywarnings(w)
assert len(w) == 1
assert "UserClass" in str(w[0].message)
def test_subclassing_warns_once_by_default(self):
Deprecated = create_deprecated_class(
"Deprecated", NewName, warn_category=MyWarning
)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
MyWarning,
match="UserClass inherits from deprecated class",
):
class UserClass(Deprecated):
pass
with warnings.catch_warnings():
warnings.simplefilter("error", MyWarning)
class FooClass(Deprecated):
pass
class BarClass(Deprecated):
pass
w = self._mywarnings(w)
assert len(w) == 1
assert "UserClass" in str(w[0].message)
def test_warning_on_instance(self):
Deprecated = create_deprecated_class(
"Deprecated", NewName, warn_category=MyWarning
)
with pytest.warns(MyWarning) as w:
_, lineno = Deprecated(), inspect.getlineno(inspect.currentframe())
w = [x for x in w if x.category is MyWarning]
assert len(w) == 1
assert (
str(w[0].message) == "tests.test_utils_deprecate.Deprecated is deprecated, "
"instantiate tests.test_utils_deprecate.NewName instead."
)
assert w[0].lineno == lineno
# ignore subclassing warnings
with warnings.catch_warnings():
warnings.simplefilter("ignore", MyWarning)
@ -124,29 +130,20 @@ class TestWarnWhenSubclassed:
class UserClass(Deprecated):
pass
with warnings.catch_warnings(record=True) as w:
_, lineno = Deprecated(), inspect.getlineno(inspect.currentframe())
_ = UserClass() # subclass instances don't warn
w = self._mywarnings(w)
assert len(w) == 1
assert (
str(w[0].message) == "tests.test_utils_deprecate.Deprecated is deprecated, "
"instantiate tests.test_utils_deprecate.NewName instead."
)
assert w[0].lineno == lineno
with warnings.catch_warnings():
warnings.simplefilter("error", MyWarning)
UserClass() # subclass instances don't warn
def test_warning_auto_message(self):
with warnings.catch_warnings(record=True) as w:
Deprecated = create_deprecated_class("Deprecated", NewName)
Deprecated = create_deprecated_class("Deprecated", NewName)
with pytest.warns(
ScrapyDeprecationWarning,
match=r"UserClass2 inherits from deprecated class tests\.test_utils_deprecate\.Deprecated, please inherit from tests\.test_utils_deprecate\.NewName",
):
class UserClass2(Deprecated):
pass
msg = str(w[0].message)
assert "tests.test_utils_deprecate.NewName" in msg
assert "tests.test_utils_deprecate.Deprecated" in msg
def test_issubclass(self):
with warnings.catch_warnings():
warnings.simplefilter("ignore", ScrapyDeprecationWarning)
@ -222,8 +219,8 @@ class TestWarnWhenSubclassed:
create_deprecated_class("Deprecated", New)
def test_deprecate_subclass_of_deprecated_class(self):
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
with warnings.catch_warnings():
warnings.simplefilter("error", MyWarning)
Deprecated = create_deprecated_class(
"Deprecated", NewName, warn_category=MyWarning
)
@ -234,33 +231,26 @@ class TestWarnWhenSubclassed:
warn_category=MyWarning,
)
w = self._mywarnings(w)
assert len(w) == 0, [str(warning) for warning in w]
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
MyWarning,
match=r"AlsoDeprecated is deprecated, instantiate foo\.Bar instead",
):
AlsoDeprecated()
with pytest.warns(
MyWarning,
match=r"UserClass inherits from deprecated class tests\.test_utils_deprecate\.AlsoDeprecated, please inherit from foo\.Bar",
):
class UserClass(AlsoDeprecated):
pass
w = self._mywarnings(w)
assert len(w) == 2
assert "AlsoDeprecated" in str(w[0].message)
assert "foo.Bar" in str(w[0].message)
assert "AlsoDeprecated" in str(w[1].message)
assert "foo.Bar" in str(w[1].message)
def test_inspect_stack(self):
with (
mock.patch("inspect.stack", side_effect=IndexError),
warnings.catch_warnings(record=True) as w,
pytest.warns(UserWarning, match="Error detecting parent module"),
):
DeprecatedName = create_deprecated_class("DeprecatedName", NewName)
class SubClass(DeprecatedName):
pass
assert "Error detecting parent module" in str(w[0].message)
create_deprecated_class("DeprecatedName", NewName)
@mock.patch(
@ -272,12 +262,12 @@ class TestWarnWhenSubclassed:
)
class TestUpdateClassPath:
def test_old_path_gets_fixed(self):
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
ScrapyDeprecationWarning,
match="`scrapy.contrib.debug.Debug` class is deprecated, use `scrapy.extensions.debug.Debug` instead",
):
output = update_classpath("scrapy.contrib.debug.Debug")
assert output == "scrapy.extensions.debug.Debug"
assert len(w) == 1
assert "scrapy.contrib.debug.Debug" in str(w[0].message)
assert "scrapy.extensions.debug.Debug" in str(w[0].message)
def test_sorted_replacement(self):
with warnings.catch_warnings():
@ -286,10 +276,10 @@ class TestUpdateClassPath:
assert output == "scrapy.pipelines.Pipeline"
def test_unmatched_path_stays_the_same(self):
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.simplefilter("error", ScrapyDeprecationWarning)
output = update_classpath("scrapy.unmatched.Path")
assert output == "scrapy.unmatched.Path"
assert len(w) == 0
def test_returns_nonstring(self):
for notastring in [None, True, [1, 2, 3], object()]:

84
tests/test_utils_misc/test_return_with_argument_inside_generator.py
View File

@ -93,29 +93,27 @@ https://example.org
assert is_generator_with_return_value(h1)
assert is_generator_with_return_value(i1)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
UserWarning,
match='The "MockSpider.top_level_return_something" method is a generator',
):
warn_on_generator_with_return_value(mock_spider, top_level_return_something)
assert len(w) == 1
assert (
'The "MockSpider.top_level_return_something" method is a generator'
in str(w[0].message)
)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
UserWarning, match='The "MockSpider.f1" method is a generator'
):
warn_on_generator_with_return_value(mock_spider, f1)
assert len(w) == 1
assert 'The "MockSpider.f1" method is a generator' in str(w[0].message)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
UserWarning, match='The "MockSpider.g1" method is a generator'
):
warn_on_generator_with_return_value(mock_spider, g1)
assert len(w) == 1
assert 'The "MockSpider.g1" method is a generator' in str(w[0].message)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
UserWarning, match='The "MockSpider.h1" method is a generator'
):
warn_on_generator_with_return_value(mock_spider, h1)
assert len(w) == 1
assert 'The "MockSpider.h1" method is a generator' in str(w[0].message)
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
UserWarning, match='The "MockSpider.i1" method is a generator'
):
warn_on_generator_with_return_value(mock_spider, i1)
assert len(w) == 1
assert 'The "MockSpider.i1" method is a generator' in str(w[0].message)
def test_generators_return_none(self, mock_spider):
def f2():
@ -160,32 +158,18 @@ https://example.org
assert not is_generator_with_return_value(k2) # not recursive
assert not is_generator_with_return_value(l2)
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.simplefilter("error", UserWarning)
warn_on_generator_with_return_value(mock_spider, top_level_return_none)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, f2)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, g2)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, h2)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, i2)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, j2)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, k2)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, l2)
assert len(w) == 0
def test_generators_return_none_with_decorator(self, mock_spider): # noqa: PLR0915
def test_generators_return_none_with_decorator(self, mock_spider):
def decorator(func):
def inner_func():
func()
@ -241,39 +225,23 @@ https://example.org
assert not is_generator_with_return_value(k3) # not recursive
assert not is_generator_with_return_value(l3)
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.simplefilter("error", UserWarning)
warn_on_generator_with_return_value(mock_spider, top_level_return_none)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, f3)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, g3)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, h3)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, i3)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, j3)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, k3)
assert len(w) == 0
with warnings.catch_warnings(record=True) as w:
warn_on_generator_with_return_value(mock_spider, l3)
assert len(w) == 0
@mock.patch(
"scrapy.utils.misc.is_generator_with_return_value", new=_indentation_error
)
def test_indentation_error(self, mock_spider):
with warnings.catch_warnings(record=True) as w:
with pytest.warns(UserWarning, match="Unable to determine"):
warn_on_generator_with_return_value(mock_spider, top_level_return_none)
assert len(w) == 1
assert "Unable to determine" in str(w[0].message)
def test_partial(self):
def cb(arg1, arg2):
@ -300,13 +268,11 @@ https://example.org
yield 1
return "value"
with warnings.catch_warnings(record=True) as w:
with warnings.catch_warnings():
warnings.simplefilter("error", UserWarning)
warn_on_generator_with_return_value(spider, gen_with_return)
assert len(w) == 0
spider.settings.settings_dict["WARN_ON_GENERATOR_RETURN_VALUE"] = True
with warnings.catch_warnings(record=True) as w:
with pytest.warns(UserWarning, match="is a generator"):
warn_on_generator_with_return_value(spider, gen_with_return)
assert len(w) == 1
assert "is a generator" in str(w[0].message)

8
tests/test_utils_project.py
View File

@ -1,5 +1,4 @@
import os
import warnings
from pathlib import Path
import pytest
@ -41,11 +40,8 @@ class TestGetProjectSettings:
envvars = {
"SCRAPY_SETTINGS_MODULE": value,
}
with warnings.catch_warnings():
warnings.simplefilter("error")
with set_environ(**envvars):
settings = get_project_settings()
with set_environ(**envvars):
settings = get_project_settings()
assert settings.get("SETTINGS_MODULE") == value
def test_invalid_envvar(self):

2
tests/test_utils_python.py
View File

@ -175,7 +175,7 @@ def test_get_func_args():
assert get_func_args(partial_f2) == ["a", "c"]
assert get_func_args(partial_f3) == ["c"]
assert get_func_args(cal) == ["a", "b", "c"]
assert get_func_args(object) == [] # pylint: disable=use-implicit-booleaness-not-comparison
assert get_func_args(object) == []
assert get_func_args(str.split, stripself=True) == ["sep", "maxsplit"]
assert get_func_args(" ".join, stripself=True) == ["iterable"]

11
tests/test_utils_sitemap.py
View File

@ -1,5 +1,6 @@
import warnings
import pytest
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.utils.sitemap import Sitemap, sitemap_urls_from_robots
@ -204,7 +205,10 @@ Disallow: /forum/search/
Disallow: /forum/active/
"""
with warnings.catch_warnings(record=True) as w:
with pytest.warns(
ScrapyDeprecationWarning,
match="Passing `str` type as `robots_text` is deprecated",
):
assert list(
sitemap_urls_from_robots(robots, base_url="http://example.com")
) == [
@ -213,9 +217,6 @@ Disallow: /forum/active/
"http://example.com/sitemap-uppercase.xml",
"http://example.com/sitemap-relative-url.xml",
]
assert "Passing `str` type as `robots_text` is deprecated, use `bytes`" in str(
w[0].message
)
def test_sitemap_blanklines():

17
tests/utils/__init__.py
View File

@ -1,5 +1,6 @@
import asyncio
import os
from collections.abc import Callable
from pathlib import Path
from twisted.internet.defer import Deferred
@ -31,3 +32,19 @@ def get_script_run_env() -> dict[str, str]:
env = os.environ.copy()
env["PYTHONPATH"] = pythonpath
return env
class OneShotLoop:
"""Test stub for create_looping_call: run once immediately, no background task."""
def __init__(self, func: Callable[[], None]):
self.func = func
self.running = False
def start(self, _interval: float, now: bool = True) -> None:
self.running = True
if now:
self.func()
def stop(self) -> None:
self.running = False