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

Merge pull request #6601 from wRAR/ruff-rules-5 

Add more Ruff rules, do some pylint cleanups
This commit is contained in:
Andrey Rakhmatullin 2025-01-02 17:38:15 +04:00 committed by GitHub
parent 6ae5b92671 176ae348c5
commit c330a399dc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
77 changed files with 302 additions and 297 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/_ext/scrapydocs.py
View File

@ -1,3 +1,4 @@
# pylint: disable=import-error
from operator import itemgetter
from docutils import nodes

2
docs/conf.py
View File

@ -8,6 +8,8 @@
#
# All configuration values have a default; values that are commented out
# serve to show the default.
# pylint: disable=import-error
import os
import sys
from pathlib import Path

102
pyproject.toml
View File

@ -124,44 +124,34 @@ extension-pkg-allow-list=[
]
[tool.pylint."MESSAGES CONTROL"]
enable = [
"useless-suppression",
]
disable = [
"abstract-method",
"arguments-differ",
"arguments-renamed",
# Ones we want to ignore
"attribute-defined-outside-init",
"broad-exception-caught",
"consider-using-with",
"cyclic-import",
"dangerous-default-value",
"disallowed-name",
"duplicate-code", # https://github.com/PyCQA/pylint/issues/214
"eval-used",
"duplicate-code", # https://github.com/pylint-dev/pylint/issues/214
"fixme",
"import-error",
"import-outside-toplevel",
"inherit-non-class",
"inherit-non-class", # false positives with create_deprecated_class()
"invalid-name",
"invalid-overridden-method",
"isinstance-second-argument-not-valid-type",
"keyword-arg-before-vararg",
"isinstance-second-argument-not-valid-type", # false positives with create_deprecated_class()
"line-too-long",
"logging-format-interpolation",
"logging-fstring-interpolation",
"logging-not-lazy",
"missing-docstring",
"no-member",
"no-method-argument",
"no-name-in-module",
"no-self-argument",
"no-value-for-parameter", # https://github.com/pylint-dev/pylint/issues/3268
"no-value-for-parameter", # https://github.com/pylint-dev/pylint/issues/3268
"not-callable",
"pointless-statement",
"pointless-string-statement",
"protected-access",
"raise-missing-from",
"redefined-builtin",
"redefined-outer-name",
"signature-differs",
"too-few-public-methods",
"too-many-ancestors",
"too-many-arguments",
@ -173,14 +163,23 @@ disable = [
"too-many-positional-arguments",
"too-many-public-methods",
"too-many-return-statements",
"unbalanced-tuple-unpacking",
"unnecessary-dunder-call",
"unused-argument",
"unused-import",
"unused-variable",
"used-before-assignment",
"useless-return",
"useless-return", # https://github.com/pylint-dev/pylint/issues/6530
"wrong-import-position",
# Ones that we may want to address (fix, ignore per-line or move to "don't want to fix")
"abstract-method",
"arguments-differ",
"arguments-renamed",
"dangerous-default-value",
"keyword-arg-before-vararg",
"pointless-statement",
"raise-missing-from",
"unbalanced-tuple-unpacking",
"unnecessary-dunder-call",
"used-before-assignment",
]
[tool.pytest.ini_options]
@ -222,6 +221,8 @@ extend-select = [
"D",
# flake8-future-annotations
"FA",
# flynt
"FLY",
# refurb
"FURB",
# isort
@ -238,6 +239,8 @@ extend-select = [
"PIE",
# pylint
"PL",
# flake8-use-pathlib
"PTH",
# flake8-pyi
"PYI",
# flake8-quotes
@ -246,8 +249,12 @@ extend-select = [
"RET",
# flake8-raise
"RSE",
# Ruff-specific rules
"RUF",
# flake8-bandit
"S",
# flake8-simplify
"SIM",
# flake8-slots
"SLOT",
# flake8-debugger
@ -262,22 +269,8 @@ extend-select = [
"YTT",
]
ignore = [
# Assigning to `os.environ` doesn't clear the environment.
"B003",
# Do not use mutable data structures for argument defaults.
"B006",
# Loop control variable not used within the loop body.
"B007",
# Do not perform function calls in argument defaults.
"B008",
# Star-arg unpacking after a keyword argument is strongly discouraged.
"B026",
# Found useless expression.
"B018",
# No explicit stacklevel argument found.
"B028",
# Within an `except` clause, raise exceptions with `raise ... from`
"B904",
# Ones we want to ignore
# Missing docstring in public module
"D100",
# Missing docstring in public class
@ -324,12 +317,45 @@ ignore = [
"PLR2004",
# `for` loop variable overwritten by assignment target
"PLW2901",
# String contains ambiguous {}.
"RUF001",
# Docstring contains ambiguous {}.
"RUF002",
# Comment contains ambiguous {}.
"RUF003",
# Mutable class attributes should be annotated with `typing.ClassVar`
"RUF012",
# Use of `assert` detected; needed for mypy
"S101",
# FTP-related functions are being called; https://github.com/scrapy/scrapy/issues/4180
"S321",
# Argument default set to insecure SSL protocol
"S503",
# Use a context manager for opening files
"SIM115",
# Yoda condition detected
"SIM300",
# Ones that we may want to address (fix, ignore per-line or move to "don't want to fix")
# Assigning to `os.environ` doesn't clear the environment.
"B003",
# Do not use mutable data structures for argument defaults.
"B006",
# Loop control variable not used within the loop body.
"B007",
# Do not perform function calls in argument defaults.
"B008",
# Found useless expression.
"B018",
# Star-arg unpacking after a keyword argument is strongly discouraged.
"B026",
# No explicit stacklevel argument found.
"B028",
# Within an `except` clause, raise exceptions with `raise ... from`
"B904",
# Use capitalized environment variable
"SIM112",
]
[tool.ruff.lint.per-file-ignores]

12
scrapy/__init__.py
View File

@ -13,14 +13,14 @@ from scrapy.selector import Selector
from scrapy.spiders import Spider
__all__ = [
"Field",
"FormRequest",
"Item",
"Request",
"Selector",
"Spider",
"__version__",
"version_info",
"Spider",
"Request",
"FormRequest",
"Selector",
"Item",
"Field",
]

4
scrapy/cmdline.py
View File

@ -90,12 +90,10 @@ def _get_commands_dict(
def _pop_command_name(argv: list[str]) -> str | None:
i = 0
for arg in argv[1:]:
for i, arg in enumerate(argv[1:]):
if not arg.startswith("-"):
del argv[i]
return arg
i += 1
return None

5
scrapy/commands/crawl.py
View File

@ -39,9 +39,8 @@ class Command(BaseRunSpiderCommand):
else:
self.crawler_process.start()
if (
self.crawler_process.bootstrap_failed
or hasattr(self.crawler_process, "has_exception")
if self.crawler_process.bootstrap_failed or (
hasattr(self.crawler_process, "has_exception")
and self.crawler_process.has_exception
):
self.exitcode = 1

2
scrapy/commands/genspider.py
View File

@ -154,7 +154,7 @@ class Command(ScrapyCommand):
spiders_dir = Path(spiders_module.__file__).parent.resolve()
else:
spiders_module = None
spiders_dir = Path(".")
spiders_dir = Path()
spider_file = f"{spiders_dir / module}.py"
shutil.copyfile(template_file, spider_file)
render_templatefile(spider_file, **tvars)

13
scrapy/commands/parse.py
View File

@ -174,13 +174,12 @@ class Command(BaseRunSpiderCommand):
display.pprint([ItemAdapter(x).asdict() for x in items], colorize=colour)
def print_requests(self, lvl: int | None = None, colour: bool = True) -> None:
if lvl is None:
if self.requests:
requests = self.requests[max(self.requests)]
else:
requests = []
else:
if lvl is not None:
requests = self.requests.get(lvl, [])
elif self.requests:
requests = self.requests[max(self.requests)]
else:
requests = []
print("# Requests ", "-" * 65)
display.pprint(requests, colorize=colour)
@ -269,7 +268,7 @@ class Command(BaseRunSpiderCommand):
assert self.crawler_process
assert self.spidercls
self.crawler_process.crawl(self.spidercls, **opts.spargs)
self.pcrawler = list(self.crawler_process.crawlers)[0]
self.pcrawler = next(iter(self.crawler_process.crawlers))
self.crawler_process.start()
if not self.first_response:

2
scrapy/commands/runspider.py
View File

@ -20,7 +20,7 @@ def _import_file(filepath: str | PathLike[str]) -> ModuleType:
if abspath.suffix not in (".py", ".pyw"):
raise ValueError(f"Not a Python source file: {abspath}")
dirname = str(abspath.parent)
sys.path = [dirname] + sys.path
sys.path = [dirname, *sys.path]
try:
module = import_module(abspath.stem)
finally:

12
scrapy/commands/startproject.py
View File

@ -1,6 +1,5 @@
from __future__ import annotations
import os
import re
import string
from importlib.util import find_spec
@ -28,9 +27,9 @@ TEMPLATES_TO_RENDER: tuple[tuple[str, ...], ...] = (
IGNORE = ignore_patterns("*.pyc", "__pycache__", ".svn")
def _make_writable(path: str | os.PathLike) -> None:
current_permissions = os.stat(path).st_mode
os.chmod(path, current_permissions | OWNER_WRITE_PERMISSION)
def _make_writable(path: Path) -> None:
current_permissions = path.stat().st_mode
path.chmod(current_permissions | OWNER_WRITE_PERMISSION)
class Command(ScrapyCommand):
@ -96,10 +95,7 @@ class Command(ScrapyCommand):
project_name = args[0]
if len(args) == 2:
project_dir = Path(args[1])
else:
project_dir = Path(args[0])
project_dir = Path(args[-1])
if (project_dir / "scrapy.cfg").exists():
self.exitcode = 1

2
scrapy/contracts/__init__.py
View File

@ -199,7 +199,7 @@ def _create_testcase(method: Callable, desc: str) -> TestCase:
spider = method.__self__.name # type: ignore[attr-defined]
class ContractTestCase(TestCase):
def __str__(_self) -> str:
def __str__(_self) -> str: # pylint: disable=no-self-argument
return f"[{spider}] {method.__name__} ({desc})"
name = f"{spider}_{method.__name__}"

15
scrapy/core/downloader/handlers/ftp.py
View File

@ -32,6 +32,7 @@ from __future__ import annotations
import re
from io import BytesIO
from pathlib import Path
from typing import TYPE_CHECKING, Any, BinaryIO
from urllib.parse import unquote
@ -56,9 +57,11 @@ if TYPE_CHECKING:
class ReceivedDataProtocol(Protocol):
def __init__(self, filename: str | None = None):
self.__filename: str | None = filename
self.body: BinaryIO = open(filename, "wb") if filename else BytesIO()
def __init__(self, filename: bytes | None = None):
self.__filename: bytes | None = filename
self.body: BinaryIO = (
Path(filename.decode()).open("wb") if filename else BytesIO()
)
self.size: int = 0
def dataReceived(self, data: bytes) -> None:
@ -66,7 +69,7 @@ class ReceivedDataProtocol(Protocol):
self.size += len(data)
@property
def filename(self) -> str | None:
def filename(self) -> bytes | None:
return self.__filename
def close(self) -> None:
@ -128,8 +131,8 @@ class FTPDownloadHandler:
) -> Response:
self.result = result
protocol.close()
headers = {"local filename": protocol.filename or "", "size": protocol.size}
body = to_bytes(protocol.filename or protocol.body.read())
headers = {"local filename": protocol.filename or b"", "size": protocol.size}
body = protocol.filename or protocol.body.read()
respcls = responsetypes.from_args(url=request.url, body=body)
# hints for Headers-related types may need to be fixed to not use AnyStr
return respcls(url=request.url, status=200, body=body, headers=headers) # type: ignore[arg-type]

5
scrapy/core/downloader/handlers/http11.py
View File

@ -424,10 +424,7 @@ class ScrapyAgent:
headers = TxHeaders(request.headers)
if isinstance(agent, self._TunnelingAgent):
headers.removeHeader(b"Proxy-Authorization")
if request.body:
bodyproducer = _RequestBodyProducer(request.body)
else:
bodyproducer = None
bodyproducer = _RequestBodyProducer(request.body) if request.body else None
start_time = time()
d: Deferred[TxResponse] = agent.request(
method, to_bytes(url, encoding="ascii"), headers, bodyproducer

9
scrapy/core/engine.py
View File

@ -291,9 +291,7 @@ class ExecutionEngine:
return False
if self.slot.start_requests is not None: # not all start requests are handled
return False
if self.slot.scheduler.has_pending_requests():
return False
return True
return not self.slot.scheduler.has_pending_requests()
def crawl(self, request: Request) -> None:
"""Inject the request into the spider <-> downloader pipeline"""
@ -388,9 +386,8 @@ class ExecutionEngine:
)
self.slot = Slot(start_requests, close_if_idle, nextcall, scheduler)
self.spider = spider
if hasattr(scheduler, "open"):
if d := scheduler.open(spider):
yield d
if hasattr(scheduler, "open") and (d := scheduler.open(spider)):
yield d
yield self.scraper.open_spider(spider)
assert self.crawler.stats
self.crawler.stats.open_spider(spider)

5
scrapy/core/spidermw.py
View File

@ -198,10 +198,7 @@ class SpiderMiddlewareManager(MiddlewareManager):
# chain, they went through it already from the process_spider_exception method
recovered: MutableChain[_T] | MutableAsyncChain[_T]
last_result_is_async = isinstance(result, AsyncIterable)
if last_result_is_async:
recovered = MutableAsyncChain()
else:
recovered = MutableChain()
recovered = MutableAsyncChain() if last_result_is_async else MutableChain()
# There are three cases for the middleware: def foo, async def foo, def foo + async def foo_async.
# 1. def foo. Sync iterables are passed as is, async ones are downgraded.

6
scrapy/crawler.py
View File

@ -1,5 +1,6 @@
from __future__ import annotations
import contextlib
import logging
import pprint
import signal
@ -503,7 +504,6 @@ class CrawlerProcess(CrawlerRunner):
def _stop_reactor(self, _: Any = None) -> None:
from twisted.internet import reactor
try:
# raised if already stopped or in shutdown stage
with contextlib.suppress(RuntimeError):
reactor.stop()
except RuntimeError: # raised if already stopped or in shutdown stage
pass

9
scrapy/downloadermiddlewares/httpauth.py
View File

@ -42,7 +42,10 @@ class HttpAuthMiddleware:
self, request: Request, spider: Spider
) -> Request | Response | None:
auth = getattr(self, "auth", None)
if auth and b"Authorization" not in request.headers:
if not self.domain or url_is_from_any_domain(request.url, [self.domain]):
request.headers[b"Authorization"] = auth
if (
auth
and b"Authorization" not in request.headers
and (not self.domain or url_is_from_any_domain(request.url, [self.domain]))
):
request.headers[b"Authorization"] = auth
return None

5
scrapy/downloadermiddlewares/httpproxy.py
View File

@ -51,10 +51,7 @@ class HttpProxyMiddleware:
proxy_type, user, password, hostport = _parse_proxy(url)
proxy_url = urlunparse((proxy_type or orig_type, hostport, "", "", "", ""))
if user:
creds = self._basic_auth_header(user, password)
else:
creds = None
creds = self._basic_auth_header(user, password) if user else None
return creds, proxy_url

12
scrapy/downloadermiddlewares/redirect.py
View File

@ -101,12 +101,14 @@ class BaseRedirectMiddleware:
if ttl and redirects <= self.max_redirect_times:
redirected.meta["redirect_times"] = redirects
redirected.meta["redirect_ttl"] = ttl - 1
redirected.meta["redirect_urls"] = request.meta.get("redirect_urls", []) + [
request.url
redirected.meta["redirect_urls"] = [
*request.meta.get("redirect_urls", []),
request.url,
]
redirected.meta["redirect_reasons"] = [
*request.meta.get("redirect_reasons", []),
reason,
]
redirected.meta["redirect_reasons"] = request.meta.get(
"redirect_reasons", []
) + [reason]
redirected.dont_filter = request.dont_filter
redirected.priority = request.priority + self.priority_adjust
logger.debug(

13
scrapy/exporters.py
View File

@ -25,13 +25,13 @@ if TYPE_CHECKING:
__all__ = [
"BaseItemExporter",
"PprintItemExporter",
"PickleItemExporter",
"CsvItemExporter",
"XmlItemExporter",
"JsonLinesItemExporter",
"JsonItemExporter",
"JsonLinesItemExporter",
"MarshalItemExporter",
"PickleItemExporter",
"PprintItemExporter",
"XmlItemExporter",
]
@ -81,10 +81,7 @@ class BaseItemExporter:
include_empty = self.export_empty_fields
if self.fields_to_export is None:
if include_empty:
field_iter = item.field_names()
else:
field_iter = item.keys()
field_iter = item.field_names() if include_empty else item.keys()
elif isinstance(self.fields_to_export, Mapping):
if include_empty:
field_iter = self.fields_to_export.items()

9
scrapy/extensions/debug.py
View File

@ -6,6 +6,7 @@ See documentation in docs/topics/extensions.rst
from __future__ import annotations
import contextlib
import logging
import signal
import sys
@ -69,12 +70,10 @@ class StackTraceDump:
class Debugger:
def __init__(self) -> None:
try:
# win32 platforms don't support SIGUSR signals
with contextlib.suppress(AttributeError):
signal.signal(signal.SIGUSR2, self._enter_debugger) # type: ignore[attr-defined]
except AttributeError:
# win32 platforms don't support SIGUSR signals
pass
def _enter_debugger(self, signum: int, frame: FrameType | None) -> None:
assert frame
Pdb().set_trace(frame.f_back) # noqa: T100
Pdb().set_trace(frame.f_back)

7
scrapy/extensions/feedexport.py
View File

@ -6,6 +6,7 @@ See documentation in docs/topics/feed-exports.rst
from __future__ import annotations
import contextlib
import logging
import re
import sys
@ -109,6 +110,8 @@ class ItemFilter:
class IFeedStorage(Interface):
"""Interface that all Feed Storages must implement"""
# pylint: disable=no-self-argument
def __init__(uri, *, feed_options=None): # pylint: disable=super-init-not-called
"""Initialize the storage with the parameters given in the URI and the
feed-specific options (see :setting:`FEEDS`)"""
@ -642,10 +645,8 @@ class FeedExporter:
)
d = {}
for k, v in conf.items():
try:
with contextlib.suppress(NotConfigured):
d[k] = load_object(v)
except NotConfigured:
pass
return d
def _exporter_supported(self, format: str) -> bool:

5
scrapy/extensions/httpcache.py
View File

@ -89,10 +89,7 @@ class RFC2616Policy:
return False
cc = self._parse_cachecontrol(request)
# obey user-agent directive "Cache-Control: no-store"
if b"no-store" in cc:
return False
# Any other is eligible for caching
return True
return b"no-store" not in cc
def should_cache_response(self, response: Response, request: Request) -> bool:
# What is cacheable - https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.9.1

5
scrapy/extensions/periodic_log.py
View File

@ -151,10 +151,7 @@ class PeriodicLog:
return False
if exclude and not include:
return True
for p in include:
if p in stat_name:
return True
return False
return any(p in stat_name for p in include)
def spider_closed(self, spider: Spider, reason: str) -> None:
self.log()

4
scrapy/extensions/telnet.py
View File

@ -84,7 +84,9 @@ class TelnetConsole(protocol.ServerFactory):
"""An implementation of IPortal"""
@defers
def login(self_, credentials, mind, *interfaces):
def login(
self_, credentials, mind, *interfaces
): # pylint: disable=no-self-argument
if not (
credentials.username == self.username.encode("utf8")
and credentials.checkPassword(self.password.encode("utf8"))

5
scrapy/http/cookies.py
View File

@ -64,9 +64,8 @@ class CookieJar:
cookies += self.jar._cookies_for_domain(host, wreq) # type: ignore[attr-defined]
attrs = self.jar._cookie_attrs(cookies) # type: ignore[attr-defined]
if attrs:
if not wreq.has_header("Cookie"):
wreq.add_unredirected_header("Cookie", "; ".join(attrs))
if attrs and not wreq.has_header("Cookie"):
wreq.add_unredirected_header("Cookie", "; ".join(attrs))
self.processed += 1
if self.processed % self.check_expired_frequency == 0:

6
scrapy/http/request/json_request.py
View File

@ -20,7 +20,7 @@ if TYPE_CHECKING:
class JsonRequest(Request):
attributes: tuple[str, ...] = Request.attributes + ("dumps_kwargs",)
attributes: tuple[str, ...] = (*Request.attributes, "dumps_kwargs")
def __init__(
self, *args: Any, dumps_kwargs: dict[str, Any] | None = None, **kwargs: Any
@ -29,7 +29,7 @@ class JsonRequest(Request):
dumps_kwargs.setdefault("sort_keys", True)
self._dumps_kwargs: dict[str, Any] = dumps_kwargs
body_passed = kwargs.get("body", None) is not None
body_passed = kwargs.get("body") is not None
data: Any = kwargs.pop("data", None)
data_passed: bool = data is not None
@ -61,7 +61,7 @@ class JsonRequest(Request):
def replace(
self, *args: Any, cls: type[Request] | None = None, **kwargs: Any
) -> Request:
body_passed = kwargs.get("body", None) is not None
body_passed = kwargs.get("body") is not None
data: Any = kwargs.pop("data", None)
data_passed: bool = data is not None

2
scrapy/http/response/text.py
View File

@ -43,7 +43,7 @@ class TextResponse(Response):
_DEFAULT_ENCODING = "ascii"
_cached_decoded_json = _NONE
attributes: tuple[str, ...] = Response.attributes + ("encoding",)
attributes: tuple[str, ...] = (*Response.attributes, "encoding")
def __init__(self, *args: Any, **kwargs: Any):
self._encoding: str | None = kwargs.pop("encoding", None)

2
scrapy/interfaces.py
View File

@ -1,3 +1,5 @@
# pylint: disable=no-method-argument,no-self-argument
from zope.interface import Interface

2
scrapy/link.py
View File

@ -24,7 +24,7 @@ class Link:
of the anchor tag.
"""
__slots__ = ["url", "text", "fragment", "nofollow"]
__slots__ = ["fragment", "nofollow", "text", "url"]
def __init__(
self, url: str, text: str = "", fragment: str = "", nofollow: bool = False

13
scrapy/linkextractors/lxmlhtml.py
View File

@ -41,9 +41,12 @@ _collect_string_content = etree.XPath("string()")
def _nons(tag: Any) -> Any:
if isinstance(tag, str):
if tag[0] == "{" and tag[1 : len(XHTML_NAMESPACE) + 1] == XHTML_NAMESPACE:
return tag.split("}")[-1]
if (
isinstance(tag, str)
and tag[0] == "{"
and tag[1 : len(XHTML_NAMESPACE) + 1] == XHTML_NAMESPACE
):
return tag.split("}")[-1]
return tag
@ -230,9 +233,7 @@ class LxmlLinkExtractor:
parsed_url, self.deny_extensions
):
return False
if self.restrict_text and not _matches(link.text, self.restrict_text):
return False
return True
return not self.restrict_text or _matches(link.text, self.restrict_text)
def matches(self, url: str) -> bool:
if self.allow_domains and not url_is_from_any_domain(url, self.allow_domains):

4
scrapy/logformatter.py
View File

@ -76,8 +76,8 @@ class LogFormatter:
self, request: Request, response: Response, spider: Spider
) -> LogFormatterResult:
"""Logs a message when the crawler finds a webpage."""
request_flags = f" {str(request.flags)}" if request.flags else ""
response_flags = f" {str(response.flags)}" if response.flags else ""
request_flags = f" {request.flags!s}" if request.flags else ""
response_flags = f" {response.flags!s}" if response.flags else ""
return {
"level": logging.DEBUG,
"msg": CRAWLEDMSG,

8
scrapy/mail.py
View File

@ -111,11 +111,9 @@ class MailSender:
) -> Deferred[None] | None:
from twisted.internet import reactor
msg: MIMEBase
if attachs:
msg = MIMEMultipart()
else:
msg = MIMENonMultipart(*mimetype.split("/", 1))
msg: MIMEBase = (
MIMEMultipart() if attachs else MIMENonMultipart(*mimetype.split("/", 1))
)
to = list(arg_to_iter(to))
cc = list(arg_to_iter(cc))

6
scrapy/pipelines/files.py
View File

@ -553,10 +553,8 @@ class FilesPipeline(MediaPipeline):
ftp_store.USE_ACTIVE_MODE = settings.getbool("FEED_STORAGE_FTP_ACTIVE")
def _get_store(self, uri: str) -> FilesStoreProtocol:
if Path(uri).is_absolute(): # to support win32 paths like: C:\\some\dir
scheme = "file"
else:
scheme = urlparse(uri).scheme
# to support win32 paths like: C:\\some\dir
scheme = "file" if Path(uri).is_absolute() else urlparse(uri).scheme
store_cls = self.STORE_SCHEMES[scheme]
return store_cls(uri)

3
scrapy/pipelines/media.py
View File

@ -127,8 +127,7 @@ class MediaPipeline(ABC):
if (
not base_class_name
or class_name == base_class_name
or settings
and not settings.get(formatted_key)
or (settings and not settings.get(formatted_key))
):
return key
return formatted_key

2
scrapy/pqueues.py
View File

@ -34,7 +34,7 @@ def _path_safe(text: str) -> str:
# as we replace some letters we can get collision for different slots
# add we add unique part
unique_slot = hashlib.md5(text.encode("utf8")).hexdigest() # noqa: S324
return "-".join([pathable_slot, unique_slot])
return f"{pathable_slot}-{unique_slot}"
class QueueProtocol(Protocol):

22
scrapy/shell.py
View File

@ -6,6 +6,7 @@ See documentation in docs/topics/shell.rst
from __future__ import annotations
import contextlib
import os
import signal
from typing import TYPE_CHECKING, Any
@ -70,17 +71,16 @@ class Shell:
else:
self.populate_vars()
if self.code:
# pylint: disable-next=eval-used
print(eval(self.code, globals(), self.vars)) # noqa: S307
else:
"""
Detect interactive shell setting in scrapy.cfg
e.g.: ~/.config/scrapy.cfg or ~/.scrapy.cfg
[settings]
# shell can be one of ipython, bpython or python;
# to be used as the interactive python console, if available.
# (default is ipython, fallbacks in the order listed above)
shell = python
"""
# Detect interactive shell setting in scrapy.cfg
# e.g.: ~/.config/scrapy.cfg or ~/.scrapy.cfg
# [settings]
# # shell can be one of ipython, bpython or python;
# # to be used as the interactive python console, if available.
# # (default is ipython, fallbacks in the order listed above)
# shell = python
cfg = get_config()
section, option = "settings", "shell"
env = os.environ.get("SCRAPY_PYTHON_SHELL")
@ -143,12 +143,10 @@ class Shell:
else:
request.meta["handle_httpstatus_all"] = True
response = None
try:
with contextlib.suppress(IgnoreRequest):
response, spider = threads.blockingCallFromThread(
reactor, self._schedule, request, spider
)
except IgnoreRequest:
pass
self.populate_vars(response, request, spider)
def populate_vars(

17
scrapy/spidermiddlewares/referer.py
View File

@ -195,8 +195,7 @@ class StrictOriginPolicy(ReferrerPolicy):
if (
self.tls_protected(response_url)
and self.potentially_trustworthy(request_url)
or not self.tls_protected(response_url)
):
) or not self.tls_protected(response_url):
return self.origin_referrer(response_url)
return None
@ -249,8 +248,7 @@ class StrictOriginWhenCrossOriginPolicy(ReferrerPolicy):
if (
self.tls_protected(response_url)
and self.potentially_trustworthy(request_url)
or not self.tls_protected(response_url)
):
) or not self.tls_protected(response_url):
return self.origin_referrer(response_url)
return None
@ -282,7 +280,7 @@ class DefaultReferrerPolicy(NoReferrerWhenDowngradePolicy):
using ``file://`` or ``s3://`` scheme.
"""
NOREFERRER_SCHEMES: tuple[str, ...] = LOCAL_SCHEMES + ("file", "s3")
NOREFERRER_SCHEMES: tuple[str, ...] = (*LOCAL_SCHEMES, "file", "s3")
name: str = POLICY_SCRAPY_DEFAULT
@ -362,11 +360,10 @@ class RefererMiddleware:
- otherwise, the policy from settings is used.
"""
policy_name = request.meta.get("referrer_policy")
if policy_name is None:
if isinstance(resp_or_url, Response):
policy_header = resp_or_url.headers.get("Referrer-Policy")
if policy_header is not None:
policy_name = to_unicode(policy_header.decode("latin1"))
if policy_name is None and isinstance(resp_or_url, Response):
policy_header = resp_or_url.headers.get("Referrer-Policy")
if policy_header is not None:
policy_name = to_unicode(policy_header.decode("latin1"))
if policy_name is None:
return self.default_policy()

5
scrapy/utils/_compression.py
View File

@ -1,3 +1,4 @@
import contextlib
import zlib
from io import BytesIO
from warnings import warn
@ -37,10 +38,8 @@ else:
return decompressor.process(data)
try:
with contextlib.suppress(ImportError):
import zstandard
except ImportError:
pass
_CHUNK_SIZE = 65536 # 64 KiB

2
scrapy/utils/console.py
View File

@ -59,7 +59,7 @@ def _embed_ptpython_shell(
namespace: dict[str, Any] = {}, banner: str = ""
) -> EmbedFuncT:
"""Start a ptpython shell"""
import ptpython.repl
import ptpython.repl # pylint: disable=import-error
@wraps(_embed_ptpython_shell)
def wrapper(namespace: dict[str, Any] = namespace, banner: str = "") -> None:

6
scrapy/utils/datatypes.py
View File

@ -8,6 +8,7 @@ This module must not depend on any module outside the Standard Library.
from __future__ import annotations
import collections
import contextlib
import warnings
import weakref
from collections import OrderedDict
@ -173,10 +174,9 @@ class LocalWeakReferencedCache(weakref.WeakKeyDictionary):
self.data: LocalCache = LocalCache(limit=limit)
def __setitem__(self, key: _KT, value: _VT) -> None:
try:
# if raised, key is not weak-referenceable, skip caching
with contextlib.suppress(TypeError):
super().__setitem__(key, value)
except TypeError:
pass # key is not weak-referenceable, skip caching
def __getitem__(self, key: _KT) -> _VT | None: # type: ignore[override]
try:

1
scrapy/utils/deprecate.py
View File

@ -57,6 +57,7 @@ def create_deprecated_class(
# https://github.com/python/mypy/issues/4177
class DeprecatedClass(new_class.__class__): # type: ignore[misc, name-defined]
# pylint: disable=no-self-argument
deprecated_class: type | None = None
warned_on_subclass: bool = False

1
scrapy/utils/display.py
View File

@ -30,6 +30,7 @@ def _tty_supports_color() -> bool:
def _colorize(text: str, colorize: bool = True) -> str:
# pylint: disable=no-name-in-module
if not colorize or not sys.stdout.isatty() or not _tty_supports_color():
return text
try:

2
scrapy/utils/engine.py
View File

@ -32,7 +32,7 @@ def get_engine_status(engine: ExecutionEngine) -> list[tuple[str, Any]]:
checks: list[tuple[str, Any]] = []
for test in tests:
try:
checks += [(test, eval(test))] # noqa: S307
checks += [(test, eval(test))] # noqa: S307 # pylint: disable=eval-used
except Exception as e:
checks += [(test, f"{type(e).__name__} (exception)")]

4
scrapy/utils/misc.py
View File

@ -252,7 +252,9 @@ def is_generator_with_return_value(callable: Callable[..., Any]) -> bool:
def returns_none(return_node: ast.Return) -> bool:
value = return_node.value
return value is None or isinstance(value, ast.Constant) and value.value is None
return value is None or (
isinstance(value, ast.Constant) and value.value is None
)
if inspect.isgeneratorfunction(callable):
func = callable

2
scrapy/utils/reactor.py
View File

@ -100,7 +100,7 @@ def install_reactor(reactor_path: str, event_loop_path: str | None = None) -> No
asyncioreactor.install(eventloop=event_loop)
else:
*module, _ = reactor_path.split(".")
installer_path = module + ["install"]
installer_path = [*module, "install"]
installer = load_object(".".join(installer_path))
with suppress(error.ReactorAlreadyInstalledError):
installer()

3
scrapy/utils/request.py
View File

@ -229,7 +229,8 @@ def request_to_curl(request: Request) -> str:
cookies = f"--cookie '{cookie}'"
elif isinstance(request.cookies, list):
cookie = "; ".join(
f"{list(c.keys())[0]}={list(c.values())[0]}" for c in request.cookies
f"{next(iter(c.keys()))}={next(iter(c.values()))}"
for c in request.cookies
)
cookies = f"--cookie '{cookie}'"

4
scrapy/utils/signal.py
View File

@ -36,7 +36,7 @@ def send_catch_log(
dont_log = named.pop("dont_log", ())
dont_log = tuple(dont_log) if isinstance(dont_log, Sequence) else (dont_log,)
dont_log += (StopDownload,)
spider = named.get("spider", None)
spider = named.get("spider")
responses: list[tuple[TypingAny, TypingAny]] = []
for receiver in liveReceivers(getAllReceivers(sender, signal)):
result: TypingAny
@ -88,7 +88,7 @@ def send_catch_log_deferred(
return failure
dont_log = named.pop("dont_log", None)
spider = named.get("spider", None)
spider = named.get("spider")
dfds: list[Deferred[tuple[TypingAny, TypingAny]]] = []
for receiver in liveReceivers(getAllReceivers(sender, signal)):
d: Deferred[TypingAny] = maybeDeferred_coro(

4
scrapy/utils/testproc.py
View File

@ -17,7 +17,7 @@ if TYPE_CHECKING:
class ProcessTest:
command: str | None = None
prefix = [sys.executable, "-m", "scrapy.cmdline"]
cwd = os.getcwd() # trial chdirs to temp dir
cwd = os.getcwd() # trial chdirs to temp dir # noqa: PTH109
def execute(
self,
@ -31,7 +31,7 @@ class ProcessTest:
if settings is not None:
env["SCRAPY_SETTINGS_MODULE"] = settings
assert self.command
cmd = self.prefix + [self.command] + list(args)
cmd = [*self.prefix, self.command, *args]
pp = TestProcessProtocol()
pp.deferred.addCallback(self._process_finished, cmd, check_code)
reactor.spawnProcess(pp, cmd[0], cmd, env=env, path=self.cwd)

16
scrapy/utils/url.py
View File

@ -51,7 +51,7 @@ def url_is_from_any_domain(url: UrlT, domains: Iterable[str]) -> bool:
def url_is_from_spider(url: UrlT, spider: type[Spider]) -> bool:
"""Return True if the url belongs to the given spider"""
return url_is_from_any_domain(
url, [spider.name] + list(getattr(spider, "allowed_domains", []))
url, [spider.name, *getattr(spider, "allowed_domains", [])]
)
@ -173,13 +173,19 @@ def strip_url(
parsed_url.username or parsed_url.password
):
netloc = netloc.split("@")[-1]
if strip_default_port and parsed_url.port:
if (parsed_url.scheme, parsed_url.port) in (
if (
strip_default_port
and parsed_url.port
and (parsed_url.scheme, parsed_url.port)
in (
("http", 80),
("https", 443),
("ftp", 21),
):
netloc = netloc.replace(f":{parsed_url.port}", "")
)
):
netloc = netloc.replace(f":{parsed_url.port}", "")
return urlunparse(
(
parsed_url.scheme,

2
scrapy/utils/versions.py
View File

@ -11,7 +11,7 @@ from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.settings.default_settings import LOG_VERSIONS
from scrapy.utils.ssl import get_openssl_version
_DEFAULT_SOFTWARE = ["Scrapy"] + LOG_VERSIONS
_DEFAULT_SOFTWARE = ["Scrapy", *LOG_VERSIONS]
def _version(item):

28
tests/test_addons.py
View File

@ -166,19 +166,21 @@ class AddonManagerTest(unittest.TestCase):
def update_settings(self, settings):
pass
with patch("scrapy.addons.logger") as logger_mock:
with patch("scrapy.addons.build_from_crawler") as build_from_crawler_mock:
settings_dict = {
"ADDONS": {LoggedAddon: 1},
}
addon = LoggedAddon()
build_from_crawler_mock.return_value = addon
crawler = get_crawler(settings_dict=settings_dict)
logger_mock.info.assert_called_once_with(
"Enabled addons:\n%(addons)s",
{"addons": [addon]},
extra={"crawler": crawler},
)
with (
patch("scrapy.addons.logger") as logger_mock,
patch("scrapy.addons.build_from_crawler") as build_from_crawler_mock,
):
settings_dict = {
"ADDONS": {LoggedAddon: 1},
}
addon = LoggedAddon()
build_from_crawler_mock.return_value = addon
crawler = get_crawler(settings_dict=settings_dict)
logger_mock.info.assert_called_once_with(
"Enabled addons:\n%(addons)s",
{"addons": [addon]},
extra={"crawler": crawler},
)
@inlineCallbacks
def test_enable_addon_in_spider(self):

2
tests/test_cmdline/__init__.py
View File

@ -21,7 +21,7 @@ class CmdlineTest(unittest.TestCase):
def _execute(self, *new_args, **kwargs):
encoding = sys.stdout.encoding or "utf-8"
args = (sys.executable, "-m", "scrapy.cmdline") + new_args
args = (sys.executable, "-m", "scrapy.cmdline", *new_args)
proc = Popen(args, stdout=PIPE, stderr=PIPE, env=self.env, **kwargs)
comm = proc.communicate()[0].strip()
return comm.decode(encoding)

4
tests/test_commands.py
View File

@ -87,13 +87,13 @@ class ProjectTest(unittest.TestCase):
def call(self, *new_args, **kwargs):
with TemporaryFile() as out:
args = (sys.executable, "-m", "scrapy.cmdline") + new_args
args = (sys.executable, "-m", "scrapy.cmdline", *new_args)
return subprocess.call(
args, stdout=out, stderr=out, cwd=self.cwd, env=self.env, **kwargs
)
def proc(self, *new_args, **popen_kwargs):
args = (sys.executable, "-m", "scrapy.cmdline") + new_args
args = (sys.executable, "-m", "scrapy.cmdline", *new_args)
p = subprocess.Popen(
args,
cwd=popen_kwargs.pop("cwd", self.cwd),

4
tests/test_contracts.py
View File

@ -517,8 +517,8 @@ class ContractsManagerTest(unittest.TestCase):
super().__init__(*args, **kwargs)
self.visited = 0
def start_requests(s):
return self.conman.from_spider(s, self.results)
def start_requests(self_): # pylint: disable=no-self-argument
return self.conman.from_spider(self_, self.results)
def parse_first(self, response):
self.visited += 1

4
tests/test_crawler.py
View File

@ -1,5 +1,4 @@
import logging
import os
import platform
import re
import signal
@ -643,11 +642,10 @@ class CrawlerRunnerHasSpider(unittest.TestCase):
class ScriptRunnerMixin:
script_dir: Path
cwd = os.getcwd()
def get_script_args(self, script_name: str, *script_args: str) -> list[str]:
script_path = self.script_dir / script_name
return [sys.executable, str(script_path)] + list(script_args)
return [sys.executable, str(script_path), *script_args]
def run_script(self, script_name: str, *script_args: str) -> str:
args = self.get_script_args(script_name, *script_args)

9
tests/test_downloader_handlers.py
View File

@ -116,7 +116,7 @@ class FileTestCase(unittest.TestCase):
def tearDown(self):
os.close(self.fd)
os.remove(self.tmpname)
Path(self.tmpname).unlink()
def test_download(self):
def _test(response):
@ -530,9 +530,10 @@ class Http11TestCase(HttpTestCase):
d = self.download_request(request, Spider("foo"))
def checkDataLoss(failure):
if failure.check(ResponseFailed):
if any(r.check(_DataLoss) for r in failure.value.reasons):
return None
if failure.check(ResponseFailed) and any(
r.check(_DataLoss) for r in failure.value.reasons
):
return None
return failure
d.addCallback(lambda _: self.fail("No DataLoss exception"))

2
tests/test_downloadermiddleware_retry.py
View File

@ -114,7 +114,7 @@ class RetryTest(unittest.TestCase):
def test_exception_to_retry_added(self):
exc = ValueError
settings_dict = {
"RETRY_EXCEPTIONS": list(RETRY_EXCEPTIONS) + [exc],
"RETRY_EXCEPTIONS": [*RETRY_EXCEPTIONS, exc],
}
crawler = get_crawler(Spider, settings_dict=settings_dict)
mw = RetryMiddleware.from_crawler(crawler)

2
tests/test_downloaderslotssettings.py
View File

@ -31,7 +31,7 @@ class DownloaderSlotsSettingsTestSpider(MetaSpider):
def start_requests(self):
self.times = {None: []}
slots = list(self.custom_settings.get("DOWNLOAD_SLOTS", {}).keys()) + [None]
slots = [*self.custom_settings.get("DOWNLOAD_SLOTS", {}), None]
for slot in slots:
url = self.mockserver.url(f"/?downloader_slot={slot}")

6
tests/test_exporters.py
View File

@ -116,7 +116,7 @@ class BaseItemExporterTest(unittest.TestCase):
)
ie = self._get_exporter(fields_to_export=["name"], encoding="latin-1")
_, name = list(ie._get_serialized_fields(self.i))[0]
_, name = next(iter(ie._get_serialized_fields(self.i)))
assert isinstance(name, str)
self.assertEqual(name, "John\xa3")
@ -216,7 +216,9 @@ class PprintItemExporterTest(BaseItemExporterTest):
return PprintItemExporter(self.output, **kwargs)
def _check_output(self):
self._assert_expected_item(eval(self.output.getvalue()))
self._assert_expected_item(
eval(self.output.getvalue()) # pylint: disable=eval-used
)
class PprintItemExporterDataclassTest(PprintItemExporterTest):

12
tests/test_feedexport.py
View File

@ -756,7 +756,7 @@ class FeedExportTest(FeedExportTestBase):
)
finally:
for file_path in FEEDS.keys():
for file_path in FEEDS:
if not Path(file_path).exists():
continue
@ -1229,15 +1229,13 @@ class FeedExportTest(FeedExportTestBase):
class CustomFilter2(scrapy.extensions.feedexport.ItemFilter):
def accepts(self, item):
if "foo" not in item.fields:
return False
return True
return "foo" in item.fields
class CustomFilter3(scrapy.extensions.feedexport.ItemFilter):
def accepts(self, item):
if isinstance(item, tuple(self.item_classes)) and item["foo"] == "bar1":
return True
return False
return (
isinstance(item, tuple(self.item_classes)) and item["foo"] == "bar1"
)
formats = {
"json": b'[\n{"foo": "bar1", "egg": "spam1"}\n]',

5
tests/test_http_request.py
View File

@ -1488,10 +1488,7 @@ def _buildresponse(body, **kwargs):
def _qs(req, encoding="utf-8", to_unicode=False):
if req.method == "POST":
qs = req.body
else:
qs = req.url.partition("?")[2]
qs = req.body if req.method == "POST" else req.url.partition("?")[2]
uqs = unquote_to_bytes(qs)
if to_unicode:
uqs = uqs.decode(encoding)

2
tests/test_http_response.py
View File

@ -960,7 +960,7 @@ class XmlResponseTest(TextResponseTest):
class CustomResponse(TextResponse):
attributes = TextResponse.attributes + ("foo", "bar")
attributes = (*TextResponse.attributes, "foo", "bar")
def __init__(self, *args, **kwargs) -> None:
self.foo = kwargs.pop("foo", None)

2
tests/test_item.py
View File

@ -54,7 +54,7 @@ class ItemTest(unittest.TestCase):
self.assertEqual(itemrepr, "{'name': 'John Doe', 'number': 123}")
i2 = eval(itemrepr)
i2 = eval(itemrepr) # pylint: disable=eval-used
self.assertEqual(i2["name"], "John Doe")
self.assertEqual(i2["number"], 123)

2
tests/test_link.py
View File

@ -49,7 +49,7 @@ class LinkTest(unittest.TestCase):
l1 = Link(
"http://www.example.com", text="test", fragment="something", nofollow=True
)
l2 = eval(repr(l1))
l2 = eval(repr(l1)) # pylint: disable=eval-used
self._assert_same_links(l1, l2)
def test_bytes_url(self):

28
tests/test_pipeline_files.py
View File

@ -634,19 +634,21 @@ class TestGCSFilesStore(unittest.TestCase):
import google.cloud.storage # noqa: F401
except ModuleNotFoundError:
raise unittest.SkipTest("google-cloud-storage is not installed")
with mock.patch("google.cloud.storage") as _:
with mock.patch("scrapy.pipelines.files.time") as _:
uri = "gs://my_bucket/my_prefix/"
store = GCSFilesStore(uri)
store.bucket = mock.Mock()
path = "full/my_data.txt"
yield store.persist_file(
path, mock.Mock(), info=None, meta=None, headers=None
)
yield store.stat_file(path, info=None)
expected_blob_path = store.prefix + path
store.bucket.blob.assert_called_with(expected_blob_path)
store.bucket.get_blob.assert_called_with(expected_blob_path)
with (
mock.patch("google.cloud.storage"),
mock.patch("scrapy.pipelines.files.time"),
):
uri = "gs://my_bucket/my_prefix/"
store = GCSFilesStore(uri)
store.bucket = mock.Mock()
path = "full/my_data.txt"
yield store.persist_file(
path, mock.Mock(), info=None, meta=None, headers=None
)
yield store.stat_file(path, info=None)
expected_blob_path = store.prefix + path
store.bucket.blob.assert_called_with(expected_blob_path)
store.bucket.get_blob.assert_called_with(expected_blob_path)
class TestFTPFileStore(unittest.TestCase):

2
tests/test_scheduler.py
View File

@ -277,14 +277,12 @@ class DownloaderAwareSchedulerTestMixin:
downloader = self.mock_crawler.engine.downloader
while self.scheduler.has_pending_requests():
request = self.scheduler.next_request()
# pylint: disable=protected-access
slot = downloader.get_slot_key(request)
dequeued_slots.append(slot)
downloader.increment(slot)
requests.append(request)
for request in requests:
# pylint: disable=protected-access
slot = downloader.get_slot_key(request)
downloader.decrement(slot)

2
tests/test_scrapy__getattr__.py
View File

@ -3,7 +3,7 @@ import warnings
def test_deprecated_twisted_version():
with warnings.catch_warnings(record=True) as warns:
from scrapy import twisted_version
from scrapy import twisted_version # pylint: disable=no-name-in-module
assert twisted_version is not None
assert isinstance(twisted_version, tuple)

2
tests/test_selector.py
View File

@ -264,7 +264,7 @@ class JMESPathTestCase(unittest.TestCase):
)
@pytest.mark.skipif(PARSEL_18_PLUS, reason="parsel >= 1.8 supports jmespath")
def test_jmespath_not_available(my_json_page) -> None:
def test_jmespath_not_available(self) -> None:
body = """
{
"website": {"name": "Example"}

2
tests/test_settings/__init__.py
View File

@ -170,7 +170,7 @@ class BaseSettingsTest(unittest.TestCase):
self.assertCountEqual(self.settings.attributes.keys(), ctrl_attributes.keys())
for key in ctrl_attributes.keys():
for key in ctrl_attributes:
attr = self.settings.attributes[key]
ctrl_attr = ctrl_attributes[key]
self.assertEqual(attr.value, ctrl_attr.value)

5
tests/test_spiderloader/__init__.py
View File

@ -1,3 +1,4 @@
import contextlib
import shutil
import sys
import tempfile
@ -22,10 +23,8 @@ module_dir = Path(__file__).resolve().parent
def _copytree(source: Path, target: Path):
try:
with contextlib.suppress(shutil.Error):
shutil.copytree(source, target)
except shutil.Error:
pass
class SpiderLoaderTest(unittest.TestCase):

32
tests/test_spidermiddleware_referer.py
View File

@ -891,13 +891,11 @@ class TestSettingsPolicyByName(TestCase):
# test parsing without space(s) after the comma
settings1 = Settings(
{
"REFERRER_POLICY": ",".join(
[
"some-custom-unknown-policy",
POLICY_SAME_ORIGIN,
POLICY_STRICT_ORIGIN_WHEN_CROSS_ORIGIN,
"another-custom-unknown-policy",
]
"REFERRER_POLICY": (
f"some-custom-unknown-policy,"
f"{POLICY_SAME_ORIGIN},"
f"{POLICY_STRICT_ORIGIN_WHEN_CROSS_ORIGIN},"
f"another-custom-unknown-policy"
)
}
)
@ -907,12 +905,10 @@ class TestSettingsPolicyByName(TestCase):
# test parsing with space(s) after the comma
settings2 = Settings(
{
"REFERRER_POLICY": ", ".join(
[
POLICY_STRICT_ORIGIN_WHEN_CROSS_ORIGIN,
"another-custom-unknown-policy",
POLICY_UNSAFE_URL,
]
"REFERRER_POLICY": (
f"{POLICY_STRICT_ORIGIN_WHEN_CROSS_ORIGIN},"
f" another-custom-unknown-policy,"
f" {POLICY_UNSAFE_URL}"
)
}
)
@ -922,12 +918,10 @@ class TestSettingsPolicyByName(TestCase):
def test_multiple_policy_tokens_all_invalid(self):
settings = Settings(
{
"REFERRER_POLICY": ",".join(
[
"some-custom-unknown-policy",
"another-custom-unknown-policy",
"yet-another-custom-unknown-policy",
]
"REFERRER_POLICY": (
"some-custom-unknown-policy,"
"another-custom-unknown-policy,"
"yet-another-custom-unknown-policy"
)
}
)

8
tests/test_squeues_request.py
View File

@ -1,3 +1,7 @@
"""
Queues that handle requests
"""
import shutil
import tempfile
import unittest
@ -16,10 +20,6 @@ from scrapy.squeues import (
)
from scrapy.utils.test import get_crawler
"""
Queues that handle requests
"""
class BaseQueueTestCase(unittest.TestCase):
def setUp(self):

12
tests/test_utils_deprecate.py
View File

@ -259,12 +259,14 @@ class WarnWhenSubclassedTest(unittest.TestCase):
self.assertIn("foo.Bar", str(w[1].message))
def test_inspect_stack(self):
with mock.patch("inspect.stack", side_effect=IndexError):
with warnings.catch_warnings(record=True) as w:
DeprecatedName = create_deprecated_class("DeprecatedName", NewName)
with (
mock.patch("inspect.stack", side_effect=IndexError),
warnings.catch_warnings(record=True) as w,
):
DeprecatedName = create_deprecated_class("DeprecatedName", NewName)
class SubClass(DeprecatedName):
pass
class SubClass(DeprecatedName):
pass
self.assertIn("Error detecting parent module", str(w[0].message))

2
tests/test_utils_iterators.py
View File

@ -366,7 +366,7 @@ class UtilsCsvTestCase(unittest.TestCase):
# explicit type check cuz' we no like stinkin' autocasting! yarrr
for result_row in result:
self.assertTrue(all(isinstance(k, str) for k in result_row.keys()))
self.assertTrue(all(isinstance(k, str) for k in result_row))
self.assertTrue(all(isinstance(v, str) for v in result_row.values()))
def test_csviter_delimiter(self):

2
tests/test_utils_project.py
View File

@ -12,7 +12,7 @@ from scrapy.utils.project import data_path, get_project_settings
@contextlib.contextmanager
def inside_a_project():
prev_dir = os.getcwd()
prev_dir = Path.cwd()
project_dir = tempfile.mkdtemp()
try:

8
tests/test_utils_trackref.py
View File

@ -61,11 +61,11 @@ Foo 1 oldest: 0s ago\n\n""",
)
def test_get_oldest(self):
o1 = Foo() # noqa: F841
o1 = Foo()
o1_time = time()
o2 = Bar() # noqa: F841
o2 = Bar()
o3_time = time()
if o3_time <= o1_time:
@ -80,9 +80,9 @@ Foo 1 oldest: 0s ago\n\n""",
self.assertIsNone(trackref.get_oldest("XXX"))
def test_iter_all(self):
o1 = Foo() # noqa: F841
o1 = Foo()
o2 = Bar() # noqa: F841
o3 = Foo() # noqa: F841
o3 = Foo()
self.assertEqual(
set(trackref.iter_all("Foo")),
{o1, o3},