Do not run item pipelines or feed exports on scrapy check

This commit is contained in:
Adrian Chaves 2026-08-09 02:00:54 +02:00
parent 1bd839b57d
commit 806e938106
3 changed files with 75 additions and 3 deletions

View File

@ -40,7 +40,13 @@ You can use the following contracts:
.. autoclass:: ScrapesContract
Use the :command:`check` command to run the contract checks.
Use the :command:`check` command to run the contract checks. It ignores
:setting:`ITEM_PIPELINES` and :setting:`FEEDS`, since contracts check the
output of callbacks instead of sending it to item processing; use the ``-s``
command-line option to set them back for a check run.
.. versionchanged:: VERSION
:setting:`ITEM_PIPELINES` and :setting:`FEEDS` are now ignored.
Custom Contracts
================

View File

@ -9,6 +9,7 @@ from unittest import TextTestRunner
from scrapy import Spider
from scrapy.commands import ScrapyCommand
from scrapy.contracts import ContractsManager
from scrapy.settings import SETTINGS_PRIORITIES
from scrapy.utils.conf import build_component_list
from scrapy.utils.misc import load_object, set_environ
@ -70,6 +71,18 @@ class Command(ScrapyCommand):
help="print contract tests for all spiders",
)
def process_options(self, args: list[str], opts: argparse.Namespace) -> None:
super().process_options(args, opts)
assert self.settings is not None
# Contracts discard the output of callbacks, so item pipelines and
# feed exports get no items, and opening them only causes side
# effects, such as an empty output file. The priority is above spider
# custom settings, which also define them, and below the command line,
# which can hence set them back.
priority = (SETTINGS_PRIORITIES["spider"] + SETTINGS_PRIORITIES["cmdline"]) // 2
self.settings.set("ITEM_PIPELINES", {}, priority=priority)
self.settings.set("FEEDS", {}, priority=priority)
def run(self, args: list[str], opts: argparse.Namespace) -> None:
# load contracts
assert self.settings is not None

View File

@ -21,7 +21,13 @@ class DummyTestCase(TestCase):
class TestCheckCommand(TestProjectBase):
spider_name = "check_spider"
def _write_contract(self, proj_path: Path, contracts: str, parse_def: str) -> None:
def _write_contract(
self,
proj_path: Path,
contracts: str,
parse_def: str,
custom_settings: str = "",
) -> None:
spider = proj_path / self.project_name / "spiders" / "checkspider.py"
spider.write_text(
f"""
@ -33,6 +39,7 @@ class CheckSpider(scrapy.Spider):
custom_settings = {{
"DOWNLOAD_DELAY": 0,
{custom_settings}
}}
def parse(self, response, **cb_kwargs):
@ -51,8 +58,9 @@ class CheckSpider(scrapy.Spider):
contracts: str = "",
parse_def: str = "pass",
use_reactor: bool = True,
custom_settings: str = "",
) -> None:
self._write_contract(proj_path, contracts, parse_def)
self._write_contract(proj_path, contracts, parse_def, custom_settings)
args = ["check"]
if not use_reactor:
args += ["-s", "TWISTED_REACTOR_ENABLED=False"]
@ -122,6 +130,51 @@ class CheckSpider(scrapy.Spider):
"""
self._test_contract(proj_path, contracts, parse_def)
def test_check_no_item_processing(self, proj_path: Path) -> None:
(proj_path / self.project_name / "pipelines.py").write_text(
"""
from pathlib import Path
class MarkerPipeline:
def open_spider(self, spider):
Path("pipeline.txt").touch()
def process_item(self, item, spider):
return item
""",
encoding="utf-8",
)
self._append_settings(
proj_path / self.project_name,
"""
FEEDS = {"items.jsonl": {"format": "jsonlines"}}
""",
)
contracts = """
@returns items 1
"""
parse_def = """
yield {'key1': 'val1'}
"""
custom_settings = f'"ITEM_PIPELINES": {{"{self.project_name}.pipelines.MarkerPipeline": 100}},'
self._test_contract(
proj_path, contracts, parse_def, custom_settings=custom_settings
)
assert not (proj_path / "items.jsonl").exists()
assert not (proj_path / "pipeline.txt").exists()
ret, _, err = proc(
"check",
"-s",
f'ITEM_PIPELINES={{"{self.project_name}.pipelines.MarkerPipeline": 100}}',
"-s",
'FEEDS={"items.jsonl": {"format": "jsonlines"}}',
cwd=proj_path,
)
assert ret == 0, err
assert (proj_path / "items.jsonl").exists()
assert (proj_path / "pipeline.txt").exists()
def test_SCRAPY_CHECK_set(self, proj_path: Path) -> None:
parse_def = """
import os