From e48a1bdde393484fe227149e798c41c343d0092d Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Mon, 27 Oct 2025 16:28:16 +0500 Subject: [PATCH] Speed up and refactor command tests (#7118) --- tests/test_command_check.py | 90 ++++---- tests/test_command_crawl.py | 48 +++-- tests/test_command_fetch.py | 45 ++-- tests/test_command_genspider.py | 139 +++++++------ tests/test_command_parse.py | 246 ++++++++++++++-------- tests/test_command_runspider.py | 196 +++++++----------- tests/test_command_shell.py | 144 ++++++------- tests/test_command_startproject.py | 210 +++++++++---------- tests/test_command_version.py | 12 +- tests/test_commands.py | 318 ++++++++++++++--------------- tests/utils/cmdline.py | 38 ++++ 11 files changed, 773 insertions(+), 713 deletions(-) create mode 100644 tests/utils/cmdline.py diff --git a/tests/test_command_check.py b/tests/test_command_check.py index 97bd9d726..dc07ed695 100644 --- a/tests/test_command_check.py +++ b/tests/test_command_check.py @@ -1,19 +1,29 @@ +from __future__ import annotations + import sys from io import StringIO +from typing import TYPE_CHECKING +from unittest import TestCase from unittest.mock import Mock, PropertyMock, call, patch from scrapy.commands.check import Command, TextTestResult -from tests.test_commands import TestCommandBase +from tests.test_commands import TestProjectBase +from tests.utils.cmdline import proc + +if TYPE_CHECKING: + from pathlib import Path -class TestCheckCommand(TestCommandBase): - def setup_method(self): - super().setup_method() - self.spider_name = "check_spider" - self.spider = (self.proj_mod_path / "spiders" / "checkspider.py").resolve() +class DummyTestCase(TestCase): + pass - def _write_contract(self, contracts, parse_def): - self.spider.write_text( + +class TestCheckCommand(TestProjectBase): + spider_name = "check_spider" + + def _write_contract(self, proj_path: Path, contracts: str, parse_def: str) -> None: + spider = proj_path / self.project_name / "spiders" / "checkspider.py" + spider.write_text( f""" import scrapy @@ -21,6 +31,10 @@ class CheckSpider(scrapy.Spider): name = '{self.spider_name}' start_urls = ['data:,'] + custom_settings = {{ + "DOWNLOAD_DELAY": 0, + }} + def parse(self, response, **cb_kwargs): \"\"\" @url data:, @@ -31,32 +45,34 @@ class CheckSpider(scrapy.Spider): encoding="utf-8", ) - def _test_contract(self, contracts="", parse_def="pass"): - self._write_contract(contracts, parse_def) - p, out, err = self.proc("check") + def _test_contract( + self, proj_path: Path, contracts: str = "", parse_def: str = "pass" + ) -> None: + self._write_contract(proj_path, contracts, parse_def) + ret, out, err = proc("check", cwd=proj_path) assert "F" not in out assert "OK" in err - assert p.returncode == 0 + assert ret == 0 - def test_check_returns_requests_contract(self): + def test_check_returns_requests_contract(self, proj_path: Path) -> None: contracts = """ @returns requests 1 """ parse_def = """ yield scrapy.Request(url='http://next-url.com') """ - self._test_contract(contracts, parse_def) + self._test_contract(proj_path, contracts, parse_def) - def test_check_returns_items_contract(self): + def test_check_returns_items_contract(self, proj_path: Path) -> None: contracts = """ @returns items 1 """ parse_def = """ yield {'key1': 'val1', 'key2': 'val2'} """ - self._test_contract(contracts, parse_def) + self._test_contract(proj_path, contracts, parse_def) - def test_check_cb_kwargs_contract(self): + def test_check_cb_kwargs_contract(self, proj_path: Path) -> None: contracts = """ @cb_kwargs {"arg1": "val1", "arg2": "val2"} """ @@ -64,18 +80,18 @@ class CheckSpider(scrapy.Spider): if len(cb_kwargs.items()) == 0: raise Exception("Callback args not set") """ - self._test_contract(contracts, parse_def) + self._test_contract(proj_path, contracts, parse_def) - def test_check_scrapes_contract(self): + def test_check_scrapes_contract(self, proj_path: Path) -> None: contracts = """ @scrapes key1 key2 """ parse_def = """ yield {'key1': 'val1', 'key2': 'val2'} """ - self._test_contract(contracts, parse_def) + self._test_contract(proj_path, contracts, parse_def) - def test_check_all_default_contracts(self): + def test_check_all_default_contracts(self, proj_path: Path) -> None: contracts = """ @returns items 1 @returns requests 1 @@ -88,67 +104,69 @@ class CheckSpider(scrapy.Spider): if len(cb_kwargs.items()) == 0: raise Exception("Callback args not set") """ - self._test_contract(contracts, parse_def) + self._test_contract(proj_path, contracts, parse_def) - def test_SCRAPY_CHECK_set(self): + def test_SCRAPY_CHECK_set(self, proj_path: Path) -> None: parse_def = """ import os if not os.environ.get('SCRAPY_CHECK'): raise Exception('SCRAPY_CHECK not set') """ - self._test_contract(parse_def=parse_def) + self._test_contract(proj_path, parse_def=parse_def) def test_printSummary_with_unsuccessful_test_result_without_errors_and_without_failures( self, - ): + ) -> None: result = TextTestResult(Mock(), descriptions=False, verbosity=1) start_time = 1.0 stop_time = 2.0 result.testsRun = 5 result.failures = [] result.errors = [] - result.unexpectedSuccesses = ["a", "b"] + result.unexpectedSuccesses = [DummyTestCase(), DummyTestCase()] with patch.object(result.stream, "write") as mock_write: result.printSummary(start_time, stop_time) mock_write.assert_has_calls([call("FAILED"), call("\n")]) - def test_printSummary_with_unsuccessful_test_result_with_only_failures(self): + def test_printSummary_with_unsuccessful_test_result_with_only_failures( + self, + ) -> None: result = TextTestResult(Mock(), descriptions=False, verbosity=1) start_time = 1.0 stop_time = 2.0 result.testsRun = 5 - result.failures = [(self, "failure")] + result.failures = [(DummyTestCase(), "failure")] result.errors = [] with patch.object(result.stream, "writeln") as mock_write: result.printSummary(start_time, stop_time) mock_write.assert_called_with(" (failures=1)") - def test_printSummary_with_unsuccessful_test_result_with_only_errors(self): + def test_printSummary_with_unsuccessful_test_result_with_only_errors(self) -> None: result = TextTestResult(Mock(), descriptions=False, verbosity=1) start_time = 1.0 stop_time = 2.0 result.testsRun = 5 result.failures = [] - result.errors = [(self, "error")] + result.errors = [(DummyTestCase(), "error")] with patch.object(result.stream, "writeln") as mock_write: result.printSummary(start_time, stop_time) mock_write.assert_called_with(" (errors=1)") def test_printSummary_with_unsuccessful_test_result_with_both_failures_and_errors( self, - ): + ) -> None: result = TextTestResult(Mock(), descriptions=False, verbosity=1) start_time = 1.0 stop_time = 2.0 result.testsRun = 5 - result.failures = [(self, "failure")] - result.errors = [(self, "error")] + result.failures = [(DummyTestCase(), "failure")] + result.errors = [(DummyTestCase(), "error")] with patch.object(result.stream, "writeln") as mock_write: result.printSummary(start_time, stop_time) mock_write.assert_called_with(" (failures=1, errors=1)") @patch("scrapy.commands.check.ContractsManager") - def test_run_with_opts_list_prints_spider(self, cm_cls_mock): + def test_run_with_opts_list_prints_spider(self, cm_cls_mock) -> None: output = StringIO() sys.stdout = output cmd = Command() @@ -175,7 +193,7 @@ class CheckSpider(scrapy.Spider): @patch("scrapy.commands.check.ContractsManager") def test_run_without_opts_list_does_not_crawl_spider_with_no_tested_methods( self, cm_cls_mock - ): + ) -> None: cmd = Command() cmd.settings = Mock(getwithbase=Mock(return_value={})) cm_cls_mock.return_value = cm_mock = Mock() @@ -186,7 +204,7 @@ class CheckSpider(scrapy.Spider): spider_loader_mock.load.side_effect = lambda x: {spider_name: spider_cls_mock}[ x ] - tested_methods = [] + tested_methods: list[str] = [] cm_mock.tested_methods_from_spidercls.side_effect = lambda x: { spider_cls_mock: tested_methods }[x] diff --git a/tests/test_command_crawl.py b/tests/test_command_crawl.py index 0ab0659b2..5a223d122 100644 --- a/tests/test_command_crawl.py +++ b/tests/test_command_crawl.py @@ -1,22 +1,29 @@ from __future__ import annotations -from pathlib import Path +from typing import TYPE_CHECKING -from tests.test_commands import TestCommandBase +from tests.test_commands import TestProjectBase +from tests.utils.cmdline import proc + +if TYPE_CHECKING: + from collections.abc import Iterable + from pathlib import Path -class TestCrawlCommand(TestCommandBase): - def crawl(self, code, args=()): - Path(self.proj_mod_path, "spiders", "myspider.py").write_text( +class TestCrawlCommand(TestProjectBase): + def crawl( + self, code: str, proj_path: Path, args: Iterable[str] = () + ) -> tuple[int, str, str]: + (proj_path / self.project_name / "spiders" / "myspider.py").write_text( code, encoding="utf-8" ) - return self.proc("crawl", "myspider", *args) + return proc("crawl", "myspider", *args, cwd=proj_path) - def get_log(self, code, args=()): - _, _, stderr = self.crawl(code, args=args) + def get_log(self, code: str, proj_path: Path, args: Iterable[str] = ()) -> str: + _, _, stderr = self.crawl(code, proj_path, args=args) return stderr - def test_no_output(self): + def test_no_output(self, proj_path: Path) -> None: spider_code = """ import scrapy @@ -28,7 +35,7 @@ class MySpider(scrapy.Spider): return yield """ - log = self.get_log(spider_code) + log = self.get_log(spider_code, proj_path) assert "[myspider] DEBUG: It works!" in log assert ( "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" @@ -36,7 +43,7 @@ class MySpider(scrapy.Spider): ) assert "Spider closed (finished)" in log - def test_output(self): + def test_output(self, proj_path: Path) -> None: spider_code = """ import scrapy @@ -49,10 +56,10 @@ class MySpider(scrapy.Spider): yield """ args = ["-o", "example.json"] - log = self.get_log(spider_code, args=args) + log = self.get_log(spider_code, proj_path, args=args) assert "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}" in log - def test_overwrite_output(self): + def test_overwrite_output(self, proj_path: Path) -> None: spider_code = """ import json import scrapy @@ -69,18 +76,19 @@ class MySpider(scrapy.Spider): return yield """ - Path(self.cwd, "example.json").write_text("not empty", encoding="utf-8") + j = proj_path / "example.json" + j.write_text("not empty", encoding="utf-8") args = ["-O", "example.json"] - log = self.get_log(spider_code, args=args) + log = self.get_log(spider_code, proj_path, args=args) assert ( '[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}' in log ) - with Path(self.cwd, "example.json").open(encoding="utf-8") as f2: + with j.open(encoding="utf-8") as f2: first_line = f2.readline() assert first_line != "not empty" - def test_output_and_overwrite_output(self): + def test_output_and_overwrite_output(self, proj_path: Path) -> None: spider_code = """ import scrapy @@ -92,12 +100,12 @@ class MySpider(scrapy.Spider): yield """ args = ["-o", "example1.json", "-O", "example2.json"] - log = self.get_log(spider_code, args=args) + log = self.get_log(spider_code, proj_path, args=args) assert ( "error: Please use only one of -o/--output and -O/--overwrite-output" in log ) - def test_default_reactor(self): + def test_default_reactor(self, proj_path: Path) -> None: spider_code = """ import scrapy @@ -109,7 +117,7 @@ class MySpider(scrapy.Spider): return yield """ - log = self.get_log(spider_code, args=("-s", "TWISTED_REACTOR=")) + log = self.get_log(spider_code, proj_path, args=("-s", "TWISTED_REACTOR=")) assert "[myspider] DEBUG: It works!" in log assert ( "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" diff --git a/tests/test_command_fetch.py b/tests/test_command_fetch.py index b6b443e28..4f4f19c3c 100644 --- a/tests/test_command_fetch.py +++ b/tests/test_command_fetch.py @@ -1,35 +1,32 @@ -from tests.mockserver.http import MockServer -from tests.test_commands import TestProjectBase +from __future__ import annotations + +from typing import TYPE_CHECKING + +from tests.utils.cmdline import proc + +if TYPE_CHECKING: + from tests.mockserver.http import MockServer -class TestFetchCommand(TestProjectBase): - @classmethod - def setup_class(cls): - cls.mockserver = MockServer() - cls.mockserver.__enter__() - - @classmethod - def teardown_class(cls): - cls.mockserver.__exit__(None, None, None) - - def test_output(self): - _, out, _ = self.proc("fetch", self.mockserver.url("/text")) +class TestFetchCommand: + def test_output(self, mockserver: MockServer) -> None: + _, out, _ = proc("fetch", mockserver.url("/text")) assert out.strip() == "Works" - def test_redirect_default(self): - _, out, _ = self.proc("fetch", self.mockserver.url("/redirect")) + def test_redirect_default(self, mockserver: MockServer) -> None: + _, out, _ = proc("fetch", mockserver.url("/redirect")) assert out.strip() == "Redirected here" - def test_redirect_disabled(self): - _, _, err = self.proc( - "fetch", "--no-redirect", self.mockserver.url("/redirect-no-meta-refresh") + def test_redirect_disabled(self, mockserver: MockServer) -> None: + _, _, err = proc( + "fetch", "--no-redirect", mockserver.url("/redirect-no-meta-refresh") ) err = err.strip() - assert "downloader/response_status_count/302" in err, err - assert "downloader/response_status_count/200" not in err, err + assert "downloader/response_status_count/302" in err + assert "downloader/response_status_count/200" not in err - def test_headers(self): - _, out, _ = self.proc("fetch", self.mockserver.url("/text"), "--headers") + def test_headers(self, mockserver: MockServer) -> None: + _, out, _ = proc("fetch", mockserver.url("/text"), "--headers") out = out.replace("\r", "") # required on win32 - assert "Server: TwistedWeb" in out, out + assert "Server: TwistedWeb" in out assert "Content-Type: text/plain" in out diff --git a/tests/test_command_genspider.py b/tests/test_command_genspider.py index c8c73ba15..9e4d90cb0 100644 --- a/tests/test_command_genspider.py +++ b/tests/test_command_genspider.py @@ -1,21 +1,34 @@ from __future__ import annotations -import os +import re from pathlib import Path import pytest -from tests.test_commands import TestCommandBase, TestProjectBase +from tests.test_commands import TestProjectBase +from tests.utils.cmdline import call, proc -class TestGenspiderCommand(TestCommandBase): - def test_arguments(self): +def find_in_file(filename: Path, regex: str) -> re.Match | None: + """Find first pattern occurrence in file""" + pattern = re.compile(regex) + with filename.open("r", encoding="utf-8") as f: + for line in f: + match = pattern.search(line) + if match is not None: + return match + return None + + +class TestGenspiderCommand(TestProjectBase): + def test_arguments(self, proj_path: Path) -> None: + spider = proj_path / self.project_name / "spiders" / "test_name.py" # only pass one argument. spider script shouldn't be created - assert self.call("genspider", "test_name") == 2 - assert not Path(self.proj_mod_path, "spiders", "test_name.py").exists() + assert call("genspider", "test_name", cwd=proj_path) == 2 + assert not spider.exists() # pass two arguments . spider script should be created - assert self.call("genspider", "test_name", "test.com") == 0 - assert Path(self.proj_mod_path, "spiders", "test_name.py").exists() + assert call("genspider", "test_name", "test.com", cwd=proj_path) == 0 + assert spider.exists() @pytest.mark.parametrize( "tplname", @@ -26,44 +39,43 @@ class TestGenspiderCommand(TestCommandBase): "csvfeed", ], ) - def test_template(self, tplname: str) -> None: + def test_template(self, tplname: str, proj_path: Path) -> None: args = [f"--template={tplname}"] if tplname else [] spname = "test_spider" spmodule = f"{self.project_name}.spiders.{spname}" - p, out, err = self.proc("genspider", spname, "test.com", *args) + spfile = proj_path / self.project_name / "spiders" / f"{spname}.py" + _, out, _ = proc("genspider", spname, "test.com", *args, cwd=proj_path) assert ( - f"Created spider {spname!r} using template {tplname!r} in module:{os.linesep} {spmodule}" + f"Created spider {spname!r} using template {tplname!r} in module:\n {spmodule}" in out ) - assert Path(self.proj_mod_path, "spiders", "test_spider.py").exists() - modify_time_before = ( - Path(self.proj_mod_path, "spiders", "test_spider.py").stat().st_mtime - ) - p, out, err = self.proc("genspider", spname, "test.com", *args) + assert spfile.exists() + modify_time_before = spfile.stat().st_mtime + _, out, _ = proc("genspider", spname, "test.com", *args, cwd=proj_path) assert f"Spider {spname!r} already exists in module" in out - modify_time_after = ( - Path(self.proj_mod_path, "spiders", "test_spider.py").stat().st_mtime - ) + modify_time_after = spfile.stat().st_mtime assert modify_time_after == modify_time_before - def test_list(self): - assert self.call("genspider", "--list") == 0 + def test_list(self, proj_path: Path) -> None: + assert call("genspider", "--list", cwd=proj_path) == 0 - def test_dump(self): - assert self.call("genspider", "--dump=basic") == 0 - assert self.call("genspider", "-d", "basic") == 0 + def test_dump(self, proj_path: Path) -> None: + assert call("genspider", "--dump=basic", cwd=proj_path) == 0 + assert call("genspider", "-d", "basic", cwd=proj_path) == 0 - def test_same_name_as_project(self): - assert self.call("genspider", self.project_name) == 2 - assert not Path( - self.proj_mod_path, "spiders", f"{self.project_name}.py" + def test_same_name_as_project(self, proj_path: Path) -> None: + assert call("genspider", self.project_name, cwd=proj_path) == 2 + assert not ( + proj_path / self.project_name / "spiders" / f"{self.project_name}.py" ).exists() @pytest.mark.parametrize("force", [True, False]) - def test_same_filename_as_existing_spider(self, force: bool) -> None: + def test_same_filename_as_existing_spider( + self, force: bool, proj_path: Path + ) -> None: file_name = "example" - file_path = Path(self.proj_mod_path, "spiders", f"{file_name}.py") - assert self.call("genspider", file_name, "example.com") == 0 + file_path = proj_path / self.project_name / "spiders" / f"{file_name}.py" + assert call("genspider", file_name, "example.com", cwd=proj_path) == 0 assert file_path.exists() # change name of spider but not its file name @@ -77,7 +89,9 @@ class TestGenspiderCommand(TestCommandBase): file_contents_before = file_data if force: - p, out, err = self.proc("genspider", "--force", file_name, "example.com") + _, out, _ = proc( + "genspider", "--force", file_name, "example.com", cwd=proj_path + ) assert ( f"Created spider {file_name!r} using template 'basic' in module" in out ) @@ -86,7 +100,7 @@ class TestGenspiderCommand(TestCommandBase): file_contents_after = file_path.read_text(encoding="utf-8") assert file_contents_after != file_contents_before else: - p, out, err = self.proc("genspider", file_name, "example.com") + _, out, _ = proc("genspider", file_name, "example.com", cwd=proj_path) assert f"{file_path.resolve()} already exists" in out modify_time_after = file_path.stat().st_mtime assert modify_time_after == modify_time_before @@ -100,18 +114,13 @@ class TestGenspiderCommand(TestCommandBase): ("https://test.com", "test.com"), ], ) - def test_url(self, url: str, domain: str) -> None: - assert self.call("genspider", "--force", "test_name", url) == 0 - m = self.find_in_file( - self.proj_mod_path / "spiders" / "test_name.py", - r"allowed_domains\s*=\s*\[['\"](.+)['\"]\]", - ) + def test_url(self, url: str, domain: str, proj_path: Path) -> None: + assert call("genspider", "--force", "test_name", url, cwd=proj_path) == 0 + spider = proj_path / self.project_name / "spiders" / "test_name.py" + m = find_in_file(spider, r"allowed_domains\s*=\s*\[['\"](.+)['\"]\]") assert m is not None assert m.group(1) == domain - m = self.find_in_file( - self.proj_mod_path / "spiders" / "test_name.py", - r"start_urls\s*=\s*\[['\"](.+)['\"]\]", - ) + m = find_in_file(spider, r"start_urls\s*=\s*\[['\"](.+)['\"]\]") assert m is not None assert m.group(1) == f"https://{domain}" @@ -139,26 +148,31 @@ class TestGenspiderCommand(TestCommandBase): ("test.com/feed.csv", "https://test.com/feed.csv", "csvfeed"), ], ) - def test_template_start_urls(self, url: str, expected: str, template: str) -> None: - assert self.call("genspider", "-t", template, "--force", "test_name", url) == 0 - m = self.find_in_file( - self.proj_mod_path / "spiders" / "test_name.py", - r"start_urls\s*=\s*\[['\"](.+)['\"]\]", + def test_template_start_urls( + self, url: str, expected: str, template: str, proj_path: Path + ) -> None: + assert ( + call( + "genspider", "-t", template, "--force", "test_name", url, cwd=proj_path + ) + == 0 ) + spider = proj_path / self.project_name / "spiders" / "test_name.py" + m = find_in_file(spider, r"start_urls\s*=\s*\[['\"](.+)['\"]\]") assert m is not None assert m.group(1) == expected -class TestGenspiderStandaloneCommand(TestProjectBase): - def test_generate_standalone_spider(self): - self.call("genspider", "example", "example.com") - assert Path(self.temp_path, "example.py").exists() +class TestGenspiderStandaloneCommand: + def test_generate_standalone_spider(self, tmp_path: Path) -> None: + call("genspider", "example", "example.com", cwd=tmp_path) + assert Path(tmp_path, "example.py").exists() @pytest.mark.parametrize("force", [True, False]) - def test_same_name_as_existing_file(self, force: bool) -> None: + def test_same_name_as_existing_file(self, force: bool, tmp_path: Path) -> None: file_name = "example" - file_path = Path(self.temp_path, file_name + ".py") - p, out, err = self.proc("genspider", file_name, "example.com") + file_path = Path(tmp_path, file_name + ".py") + p, out, err = proc("genspider", file_name, "example.com", cwd=tmp_path) assert f"Created spider {file_name!r} using template 'basic' " in out assert file_path.exists() modify_time_before = file_path.stat().st_mtime @@ -166,8 +180,14 @@ class TestGenspiderStandaloneCommand(TestProjectBase): if force: # use different template to ensure contents were changed - p, out, err = self.proc( - "genspider", "--force", "-t", "crawl", file_name, "example.com" + p, out, err = proc( + "genspider", + "--force", + "-t", + "crawl", + file_name, + "example.com", + cwd=tmp_path, ) assert f"Created spider {file_name!r} using template 'crawl' " in out modify_time_after = file_path.stat().st_mtime @@ -175,10 +195,9 @@ class TestGenspiderStandaloneCommand(TestProjectBase): file_contents_after = file_path.read_text(encoding="utf-8") assert file_contents_after != file_contents_before else: - p, out, err = self.proc("genspider", file_name, "example.com") + _, out, _ = proc("genspider", file_name, "example.com", cwd=tmp_path) assert ( - f"{Path(self.temp_path, file_name + '.py').resolve()} already exists" - in out + f"{Path(tmp_path, file_name + '.py').resolve()} already exists" in out ) modify_time_after = file_path.stat().st_mtime assert modify_time_after == modify_time_before diff --git a/tests/test_command_parse.py b/tests/test_command_parse.py index 7b119d7cf..85529e607 100644 --- a/tests/test_command_parse.py +++ b/tests/test_command_parse.py @@ -1,27 +1,29 @@ +from __future__ import annotations + import argparse import re -from pathlib import Path +from typing import TYPE_CHECKING + +import pytest from scrapy.commands import parse from scrapy.settings import Settings -from tests.mockserver.http import MockServer -from tests.test_commands import TestCommandBase +from tests.test_commands import TestProjectBase +from tests.utils.cmdline import call, proc + +if TYPE_CHECKING: + from pathlib import Path + + from tests.mockserver.http import MockServer -class TestParseCommand(TestCommandBase): - @classmethod - def setup_class(cls): - cls.mockserver = MockServer() - cls.mockserver.__enter__() +class TestParseCommand(TestProjectBase): + spider_name = "parse_spider" - @classmethod - def teardown_class(cls): - cls.mockserver.__exit__(None, None, None) - - def setup_method(self): - super().setup_method() - self.spider_name = "parse_spider" - (self.proj_mod_path / "spiders" / "myspider.py").write_text( + @pytest.fixture(autouse=True) + def create_files(self, proj_path: Path) -> None: + proj_mod_path = proj_path / self.project_name + (proj_mod_path / "spiders" / "myspider.py").write_text( f""" import scrapy from scrapy.linkextractors import LinkExtractor @@ -30,7 +32,13 @@ from scrapy.utils.test import get_from_asyncio_queue import asyncio -class AsyncDefAsyncioReturnSpider(scrapy.Spider): +class BaseSpider(scrapy.Spider): + custom_settings = {{ + "DOWNLOAD_DELAY": 0, + }} + + +class AsyncDefAsyncioReturnSpider(BaseSpider): name = "asyncdef_asyncio_return" async def parse(self, response): @@ -39,7 +47,7 @@ class AsyncDefAsyncioReturnSpider(scrapy.Spider): self.logger.info(f"Got response {{status}}") return [{{'id': 1}}, {{'id': 2}}] -class AsyncDefAsyncioReturnSingleElementSpider(scrapy.Spider): +class AsyncDefAsyncioReturnSingleElementSpider(BaseSpider): name = "asyncdef_asyncio_return_single_element" async def parse(self, response): @@ -48,7 +56,7 @@ class AsyncDefAsyncioReturnSingleElementSpider(scrapy.Spider): self.logger.info(f"Got response {{status}}") return {{'foo': 42}} -class AsyncDefAsyncioGenLoopSpider(scrapy.Spider): +class AsyncDefAsyncioGenLoopSpider(BaseSpider): name = "asyncdef_asyncio_gen_loop" async def parse(self, response): @@ -57,7 +65,7 @@ class AsyncDefAsyncioGenLoopSpider(scrapy.Spider): yield {{'foo': i}} self.logger.info(f"Got response {{response.status}}") -class AsyncDefAsyncioSpider(scrapy.Spider): +class AsyncDefAsyncioSpider(BaseSpider): name = "asyncdef_asyncio" async def parse(self, response): @@ -65,7 +73,7 @@ class AsyncDefAsyncioSpider(scrapy.Spider): status = await get_from_asyncio_queue(response.status) self.logger.debug(f"Got response {{status}}") -class AsyncDefAsyncioGenExcSpider(scrapy.Spider): +class AsyncDefAsyncioGenExcSpider(BaseSpider): name = "asyncdef_asyncio_gen_exc" async def parse(self, response): @@ -87,7 +95,8 @@ class MySpider(scrapy.Spider): custom_settings = {{ "DOWNLOADER_MIDDLEWARES": {{ CallbackSignatureDownloaderMiddleware: 0, - }} + }}, + "DOWNLOAD_DELAY": 0, }} def parse(self, response): @@ -120,6 +129,10 @@ class MySpider(scrapy.Spider): class MyGoodCrawlSpider(CrawlSpider): name = 'goodcrawl{self.spider_name}' + custom_settings = {{ + "DOWNLOAD_DELAY": 0, + }} + rules = ( Rule(LinkExtractor(allow=r'/html'), callback='parse_item', follow=True), Rule(LinkExtractor(allow=r'/text'), follow=True), @@ -136,6 +149,10 @@ class MyBadCrawlSpider(CrawlSpider): '''Spider which doesn't define a parse_item callback while using it in a rule.''' name = 'badcrawl{self.spider_name}' + custom_settings = {{ + "DOWNLOAD_DELAY": 0, + }} + rules = ( Rule(LinkExtractor(allow=r'/html'), callback='parse_item', follow=True), ) @@ -146,7 +163,7 @@ class MyBadCrawlSpider(CrawlSpider): encoding="utf-8", ) - (self.proj_mod_path / "pipelines.py").write_text( + (proj_mod_path / "pipelines.py").write_text( """ import logging @@ -160,15 +177,15 @@ class MyPipeline: encoding="utf-8", ) - with (self.proj_mod_path / "settings.py").open("a", encoding="utf-8") as f: + with (proj_mod_path / "settings.py").open("a", encoding="utf-8") as f: f.write( f""" ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} """ ) - def test_spider_arguments(self): - _, _, stderr = self.proc( + def test_spider_arguments(self, proj_path: Path, mockserver: MockServer) -> None: + _, _, stderr = proc( "parse", "--spider", self.spider_name, @@ -177,13 +194,14 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} "-c", "parse", "--verbose", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "DEBUG: It Works!" in stderr - def test_request_with_meta(self): + def test_request_with_meta(self, proj_path: Path, mockserver: MockServer) -> None: raw_json_string = '{"foo" : "baz"}' - _, _, stderr = self.proc( + _, _, stderr = proc( "parse", "--spider", self.spider_name, @@ -192,11 +210,12 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} "-c", "parse_request_with_meta", "--verbose", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "DEBUG: It Works!" in stderr - _, _, stderr = self.proc( + _, _, stderr = proc( "parse", "--spider", self.spider_name, @@ -205,13 +224,16 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} "-c", "parse_request_with_meta", "--verbose", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "DEBUG: It Works!" in stderr - def test_request_with_cb_kwargs(self): + def test_request_with_cb_kwargs( + self, proj_path: Path, mockserver: MockServer + ) -> None: raw_json_string = '{"foo" : "bar", "key": "value"}' - _, _, stderr = self.proc( + _, _, stderr = proc( "parse", "--spider", self.spider_name, @@ -220,7 +242,8 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} "-c", "parse_request_with_cb_kwargs", "--verbose", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "DEBUG: It Works!" in stderr assert ( @@ -228,20 +251,23 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} in stderr ) - def test_request_without_meta(self): - _, _, stderr = self.proc( + def test_request_without_meta( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, _, stderr = proc( "parse", "--spider", self.spider_name, "-c", "parse_request_without_meta", "--nolinks", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "DEBUG: It Works!" in stderr - def test_pipelines(self): - _, _, stderr = self.proc( + def test_pipelines(self, proj_path: Path, mockserver: MockServer) -> None: + _, _, stderr = proc( "parse", "--spider", self.spider_name, @@ -249,163 +275,210 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} "-c", "parse", "--verbose", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "INFO: It Works!" in stderr - def test_async_def_asyncio_parse_items_list(self): - _, out, stderr = self.proc( + def test_async_def_asyncio_parse_items_list( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, out, stderr = proc( "parse", "--spider", "asyncdef_asyncio_return", "-c", "parse", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "INFO: Got response 200" in stderr assert "{'id': 1}" in out assert "{'id': 2}" in out - def test_async_def_asyncio_parse_items_single_element(self): - _, out, stderr = self.proc( + def test_async_def_asyncio_parse_items_single_element( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, out, stderr = proc( "parse", "--spider", "asyncdef_asyncio_return_single_element", "-c", "parse", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "INFO: Got response 200" in stderr assert "{'foo': 42}" in out - def test_async_def_asyncgen_parse_loop(self): - _, out, stderr = self.proc( + def test_async_def_asyncgen_parse_loop( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, out, stderr = proc( "parse", "--spider", "asyncdef_asyncio_gen_loop", "-c", "parse", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "INFO: Got response 200" in stderr for i in range(10): assert f"{{'foo': {i}}}" in out - def test_async_def_asyncgen_parse_exc(self): - _, out, stderr = self.proc( + def test_async_def_asyncgen_parse_exc( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, out, stderr = proc( "parse", "--spider", "asyncdef_asyncio_gen_exc", "-c", "parse", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "ValueError" in stderr for i in range(7): assert f"{{'foo': {i}}}" in out - def test_async_def_asyncio_parse(self): - _, _, stderr = self.proc( + def test_async_def_asyncio_parse( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, _, stderr = proc( "parse", "--spider", "asyncdef_asyncio", "-c", "parse", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "DEBUG: Got response 200" in stderr - def test_parse_items(self): - _, out, _ = self.proc( + def test_parse_items(self, proj_path: Path, mockserver: MockServer) -> None: + _, out, _ = proc( "parse", "--spider", self.spider_name, "-c", "parse", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "[{}, {'foo': 'bar'}]" in out - def test_parse_items_no_callback_passed(self): - _, out, _ = self.proc( - "parse", "--spider", self.spider_name, self.mockserver.url("/html") + def test_parse_items_no_callback_passed( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, out, _ = proc( + "parse", + "--spider", + self.spider_name, + mockserver.url("/html"), + cwd=proj_path, ) assert "[{}, {'foo': 'bar'}]" in out - def test_wrong_callback_passed(self): - _, out, stderr = self.proc( + def test_wrong_callback_passed( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, out, stderr = proc( "parse", "--spider", self.spider_name, "-c", "dummy", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert re.search(r"# Scraped Items -+\r?\n\[\]", out) assert "Cannot find callback" in stderr - def test_crawlspider_matching_rule_callback_set(self): + def test_crawlspider_matching_rule_callback_set( + self, proj_path: Path, mockserver: MockServer + ) -> None: """If a rule matches the URL, use it's defined callback.""" - _, out, _ = self.proc( + _, out, _ = proc( "parse", "--spider", "goodcrawl" + self.spider_name, "-r", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert "[{}, {'foo': 'bar'}]" in out - def test_crawlspider_matching_rule_default_callback(self): + def test_crawlspider_matching_rule_default_callback( + self, proj_path: Path, mockserver: MockServer + ) -> None: """If a rule match but it has no callback set, use the 'parse' callback.""" - _, out, _ = self.proc( + _, out, _ = proc( "parse", "--spider", "goodcrawl" + self.spider_name, "-r", - self.mockserver.url("/text"), + mockserver.url("/text"), + cwd=proj_path, ) assert "[{}, {'nomatch': 'default'}]" in out - def test_spider_with_no_rules_attribute(self): + def test_spider_with_no_rules_attribute( + self, proj_path: Path, mockserver: MockServer + ) -> None: """Using -r with a spider with no rule should not produce items.""" - _, out, stderr = self.proc( - "parse", "--spider", self.spider_name, "-r", self.mockserver.url("/html") + _, out, stderr = proc( + "parse", + "--spider", + self.spider_name, + "-r", + mockserver.url("/html"), + cwd=proj_path, ) assert re.search(r"# Scraped Items -+\r?\n\[\]", out) assert "No CrawlSpider rules found" in stderr - def test_crawlspider_missing_callback(self): - _, out, _ = self.proc( + def test_crawlspider_missing_callback( + self, proj_path: Path, mockserver: MockServer + ) -> None: + _, out, _ = proc( "parse", "--spider", "badcrawl" + self.spider_name, "-r", - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert re.search(r"# Scraped Items -+\r?\n\[\]", out) - def test_crawlspider_no_matching_rule(self): + def test_crawlspider_no_matching_rule( + self, proj_path: Path, mockserver: MockServer + ) -> None: """The requested URL has no matching rule, so no items should be scraped""" - _, out, stderr = self.proc( + _, out, stderr = proc( "parse", "--spider", "badcrawl" + self.spider_name, "-r", - self.mockserver.url("/enc-gb18030"), + mockserver.url("/enc-gb18030"), + cwd=proj_path, ) assert re.search(r"# Scraped Items -+\r?\n\[\]", out) assert "Cannot find a rule that matches" in stderr - def test_crawlspider_not_exists_with_not_matched_url(self): - assert self.call("parse", self.mockserver.url("/invalid_url")) == 0 + def test_crawlspider_not_exists_with_not_matched_url( + self, proj_path: Path, mockserver: MockServer + ) -> None: + assert call("parse", mockserver.url("/invalid_url"), cwd=proj_path) == 0 - def test_output_flag(self): + def test_output_flag(self, proj_path: Path, mockserver: MockServer) -> None: """Checks if a file was created successfully having correct format containing correct data in it. """ file_name = "data.json" - file_path = Path(self.proj_path, file_name) - self.proc( + file_path = proj_path / file_name + proc( "parse", "--spider", self.spider_name, @@ -413,7 +486,8 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} "parse", "-o", file_name, - self.mockserver.url("/html"), + mockserver.url("/html"), + cwd=proj_path, ) assert file_path.exists() diff --git a/tests/test_command_runspider.py b/tests/test_command_runspider.py index 89670feb3..2fa4ce581 100644 --- a/tests/test_command_runspider.py +++ b/tests/test_command_runspider.py @@ -4,21 +4,19 @@ import asyncio import inspect import platform import sys -from contextlib import contextmanager -from pathlib import Path -from tempfile import TemporaryDirectory, mkdtemp from typing import TYPE_CHECKING import pytest -from tests.test_commands import TestCommandBase from tests.test_crawler import ExceptionSpider, NoRequestsSpider +from tests.utils.cmdline import proc if TYPE_CHECKING: - from collections.abc import Iterator + from collections.abc import Iterable + from pathlib import Path -class TestRunSpiderCommand(TestCommandBase): +class TestRunSpiderCommand: spider_filename = "myspider.py" debug_log_spider = """ @@ -43,26 +41,21 @@ class BadSpider(scrapy.Spider): yield """ - @contextmanager - def _create_file(self, content: str, name: str | None = None) -> Iterator[str]: - with TemporaryDirectory() as tmpdir: - if name: - fname = Path(tmpdir, name).resolve() - else: - fname = Path(tmpdir, self.spider_filename).resolve() - fname.write_text(content, encoding="utf-8") - yield str(fname) + def runspider( + self, cwd: Path, code: str, name: str | None = None, args: Iterable[str] = () + ) -> tuple[int, str, str]: + fname = cwd / (name or self.spider_filename) + fname.write_text(code, encoding="utf-8") + return proc("runspider", str(fname), *args, cwd=cwd) - def runspider(self, code, name=None, args=()): - with self._create_file(code, name) as fname: - return self.proc("runspider", fname, *args) - - def get_log(self, code, name=None, args=()): - _, _, stderr = self.runspider(code, name, args=args) + def get_log( + self, cwd: Path, code: str, name: str | None = None, args: Iterable[str] = () + ) -> str: + _, _, stderr = self.runspider(cwd, code, name, args=args) return stderr - def test_runspider(self): - log = self.get_log(self.debug_log_spider) + def test_runspider(self, tmp_path: Path) -> None: + log = self.get_log(tmp_path, self.debug_log_spider) assert "DEBUG: It Works!" in log assert ( "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" @@ -70,27 +63,29 @@ class BadSpider(scrapy.Spider): ) assert "INFO: Spider closed (finished)" in log - def test_run_fail_spider(self): - proc, _, _ = self.runspider( - "import scrapy\n" + inspect.getsource(ExceptionSpider) + def test_run_fail_spider(self, tmp_path: Path) -> None: + ret, _, _ = self.runspider( + tmp_path, "import scrapy\n" + inspect.getsource(ExceptionSpider) ) - ret = proc.returncode assert ret != 0 - def test_run_good_spider(self): - proc, _, _ = self.runspider( - "import scrapy\n" + inspect.getsource(NoRequestsSpider) + def test_run_good_spider(self, tmp_path: Path) -> None: + ret, _, _ = self.runspider( + tmp_path, "import scrapy\n" + inspect.getsource(NoRequestsSpider) ) - ret = proc.returncode assert ret == 0 - def test_runspider_log_level(self): - log = self.get_log(self.debug_log_spider, args=("-s", "LOG_LEVEL=INFO")) + def test_runspider_log_level(self, tmp_path: Path) -> None: + log = self.get_log( + tmp_path, self.debug_log_spider, args=("-s", "LOG_LEVEL=INFO") + ) assert "DEBUG: It Works!" not in log assert "INFO: Spider opened" in log - def test_runspider_default_reactor(self): - log = self.get_log(self.debug_log_spider, args=("-s", "TWISTED_REACTOR=")) + def test_runspider_default_reactor(self, tmp_path: Path) -> None: + log = self.get_log( + tmp_path, self.debug_log_spider, args=("-s", "TWISTED_REACTOR=") + ) assert "DEBUG: It Works!" in log assert ( "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" @@ -100,7 +95,7 @@ class BadSpider(scrapy.Spider): assert "INFO: Closing spider (finished)" in log assert "INFO: Spider closed (finished)" in log - def test_runspider_dnscache_disabled(self): + def test_runspider_dnscache_disabled(self, tmp_path: Path) -> None: # see https://github.com/scrapy/scrapy/issues/2811 # The spider below should not be able to connect to localhost:12345, # which is intended, @@ -121,40 +116,41 @@ class MySpider(scrapy.Spider): def parse(self, response): return {'test': 'value'} """ - log = self.get_log(dnscache_spider, args=("-s", "DNSCACHE_ENABLED=False")) + log = self.get_log( + tmp_path, dnscache_spider, args=("-s", "DNSCACHE_ENABLED=False") + ) assert "DNSLookupError" not in log assert "INFO: Spider opened" in log - def test_runspider_log_short_names(self): - log1 = self.get_log(self.debug_log_spider, args=("-s", "LOG_SHORT_NAMES=1")) + @pytest.mark.parametrize("value", [False, True]) + def test_runspider_log_short_names(self, tmp_path: Path, value: bool) -> None: + log1 = self.get_log( + tmp_path, self.debug_log_spider, args=("-s", f"LOG_SHORT_NAMES={value}") + ) assert "[myspider] DEBUG: It Works!" in log1 - assert "[scrapy]" in log1 - assert "[scrapy.core.engine]" not in log1 + assert ("[scrapy]" in log1) is value + assert ("[scrapy.core.engine]" in log1) is not value - log2 = self.get_log(self.debug_log_spider, args=("-s", "LOG_SHORT_NAMES=0")) - assert "[myspider] DEBUG: It Works!" in log2 - assert "[scrapy]" not in log2 - assert "[scrapy.core.engine]" in log2 - - def test_runspider_no_spider_found(self): - log = self.get_log("from scrapy.spiders import Spider\n") + def test_runspider_no_spider_found(self, tmp_path: Path) -> None: + log = self.get_log(tmp_path, "from scrapy.spiders import Spider\n") assert "No spider found in file" in log - def test_runspider_file_not_found(self): - _, _, log = self.proc("runspider", "some_non_existent_file") + def test_runspider_file_not_found(self) -> None: + _, _, log = proc("runspider", "some_non_existent_file") assert "File not found: some_non_existent_file" in log - def test_runspider_unable_to_load(self): - log = self.get_log("", name="myspider.txt") + def test_runspider_unable_to_load(self, tmp_path: Path) -> None: + log = self.get_log(tmp_path, "", name="myspider.txt") assert "Unable to load" in log - def test_start_errors(self): - log = self.get_log(self.badspider, name="badspider.py") + def test_start_errors(self, tmp_path: Path) -> None: + log = self.get_log(tmp_path, self.badspider, name="badspider.py") assert "start" in log assert "badspider.py" in log, log - def test_asyncio_enabled_true(self): + def test_asyncio_enabled_true(self, tmp_path: Path) -> None: log = self.get_log( + tmp_path, self.debug_log_spider, args=[ "-s", @@ -166,15 +162,16 @@ class MySpider(scrapy.Spider): in log ) - def test_asyncio_enabled_default(self): - log = self.get_log(self.debug_log_spider, args=[]) + def test_asyncio_enabled_default(self, tmp_path: Path) -> None: + log = self.get_log(tmp_path, self.debug_log_spider) assert ( "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" in log ) - def test_asyncio_enabled_false(self): + def test_asyncio_enabled_false(self, tmp_path: Path) -> None: log = self.get_log( + tmp_path, self.debug_log_spider, args=["-s", "TWISTED_REACTOR=twisted.internet.selectreactor.SelectReactor"], ) @@ -185,8 +182,9 @@ class MySpider(scrapy.Spider): ) @pytest.mark.requires_uvloop - def test_custom_asyncio_loop_enabled_true(self): + def test_custom_asyncio_loop_enabled_true(self, tmp_path: Path) -> None: log = self.get_log( + tmp_path, self.debug_log_spider, args=[ "-s", @@ -197,8 +195,9 @@ class MySpider(scrapy.Spider): ) assert "Using asyncio event loop: uvloop.Loop" in log - def test_custom_asyncio_loop_enabled_false(self): + def test_custom_asyncio_loop_enabled_false(self, tmp_path: Path) -> None: log = self.get_log( + tmp_path, self.debug_log_spider, args=[ "-s", @@ -214,7 +213,7 @@ class MySpider(scrapy.Spider): in log ) - def test_output(self): + def test_output(self, tmp_path: Path) -> None: spider_code = """ import scrapy @@ -227,10 +226,10 @@ class MySpider(scrapy.Spider): yield """ args = ["-o", "example.json"] - log = self.get_log(spider_code, args=args) + log = self.get_log(tmp_path, spider_code, args=args) assert "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}" in log - def test_overwrite_output(self): + def test_overwrite_output(self, tmp_path: Path) -> None: spider_code = """ import json import scrapy @@ -247,18 +246,18 @@ class MySpider(scrapy.Spider): return yield """ - Path(self.cwd, "example.json").write_text("not empty", encoding="utf-8") + (tmp_path / "example.json").write_text("not empty", encoding="utf-8") args = ["-O", "example.json"] - log = self.get_log(spider_code, args=args) + log = self.get_log(tmp_path, spider_code, args=args) assert ( '[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}' in log ) - with Path(self.cwd, "example.json").open(encoding="utf-8") as f2: + with (tmp_path / "example.json").open(encoding="utf-8") as f2: first_line = f2.readline() assert first_line != "not empty" - def test_output_and_overwrite_output(self): + def test_output_and_overwrite_output(self, tmp_path: Path) -> None: spider_code = """ import scrapy @@ -270,12 +269,12 @@ class MySpider(scrapy.Spider): yield """ args = ["-o", "example1.json", "-O", "example2.json"] - log = self.get_log(spider_code, args=args) + log = self.get_log(tmp_path, spider_code, args=args) assert ( "error: Please use only one of -o/--output and -O/--overwrite-output" in log ) - def test_output_stdout(self): + def test_output_stdout(self, tmp_path: Path) -> None: spider_code = """ import scrapy @@ -288,11 +287,11 @@ class MySpider(scrapy.Spider): yield """ args = ["-o", "-:json"] - log = self.get_log(spider_code, args=args) + log = self.get_log(tmp_path, spider_code, args=args) assert "[myspider] DEBUG: FEEDS: {'stdout:': {'format': 'json'}}" in log - @pytest.mark.skipif(platform.system() == "Windows", reason="Linux only") - def test_absolute_path_linux(self): + @pytest.mark.parametrize("arg", ["output.json:json", "output.json"]) + def test_absolute_path(self, tmp_path: Path, arg: str) -> None: spider_code = """ import scrapy @@ -304,52 +303,15 @@ class MySpider(scrapy.Spider): def parse(self, response): yield {"hello": "world"} """ - temp_dir = mkdtemp() - args = ["-o", f"{temp_dir}/output1.json:json"] - log = self.get_log(spider_code, args=args) + args = ["-o", str(tmp_path / arg)] + log = self.get_log(tmp_path, spider_code, args=args) assert ( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}/output1.json" + f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {tmp_path / 'output.json'}" in log ) - args = ["-o", f"{temp_dir}/output2.json"] - log = self.get_log(spider_code, args=args) - assert ( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}/output2.json" - in log - ) - - @pytest.mark.skipif(platform.system() != "Windows", reason="Windows only") - def test_absolute_path_windows(self): - spider_code = """ -import scrapy - -class MySpider(scrapy.Spider): - name = 'myspider' - - start_urls = ["data:,"] - - def parse(self, response): - yield {"hello": "world"} - """ - temp_dir = mkdtemp() - - args = ["-o", f"{temp_dir}\\output1.json:json"] - log = self.get_log(spider_code, args=args) - assert ( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}\\output1.json" - in log - ) - - args = ["-o", f"{temp_dir}\\output2.json"] - log = self.get_log(spider_code, args=args) - assert ( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}\\output2.json" - in log - ) - - def test_args_change_settings(self): + def test_args_change_settings(self, tmp_path: Path) -> None: spider_code = """ import scrapy @@ -368,7 +330,7 @@ class MySpider(scrapy.Spider): yield """ args = ["-a", "foo=42"] - log = self.get_log(spider_code, args=args) + log = self.get_log(tmp_path, spider_code, args=args) assert "Spider closed (finished)" in log assert "The value of FOO is 42" in log @@ -379,10 +341,10 @@ class MySpider(scrapy.Spider): class TestWindowsRunSpiderCommand(TestRunSpiderCommand): spider_filename = "myspider.pyw" - def test_start_errors(self): - log = self.get_log(self.badspider, name="badspider.pyw") + def test_start_errors(self, tmp_path: Path) -> None: + log = self.get_log(tmp_path, self.badspider, name="badspider.pyw") assert "start" in log assert "badspider.pyw" in log - def test_runspider_unable_to_load(self): + def test_runspider_unable_to_load(self, tmp_path: Path) -> None: pytest.skip("Already Tested in 'RunSpiderCommandTest'") diff --git a/tests/test_command_shell.py b/tests/test_command_shell.py index 76635b1aa..d9fb96fb8 100644 --- a/tests/test_command_shell.py +++ b/tests/test_command_shell.py @@ -1,145 +1,133 @@ +from __future__ import annotations + import os import sys from io import BytesIO from pathlib import Path +from typing import TYPE_CHECKING, cast import pytest from pexpect.popen_spawn import PopenSpawn from scrapy.utils.reactor import _asyncio_reactor_path from tests import NON_EXISTING_RESOLVABLE, tests_datadir -from tests.mockserver.http import MockServer -from tests.test_commands import TestProjectBase +from tests.utils.cmdline import proc + +if TYPE_CHECKING: + from tests.mockserver.http import MockServer -class TestShellCommand(TestProjectBase): - @classmethod - def setup_class(cls): - cls.mockserver = MockServer() - cls.mockserver.__enter__() - - @classmethod - def teardown_class(cls): - cls.mockserver.__exit__(None, None, None) - - def test_empty(self): - _, out, _ = self.proc("shell", "-c", "item") +class TestShellCommand: + def test_empty(self) -> None: + _, out, _ = proc("shell", "-c", "item") assert "{}" in out - def test_response_body(self): - _, out, _ = self.proc( - "shell", self.mockserver.url("/text"), "-c", "response.body" - ) + def test_response_body(self, mockserver: MockServer) -> None: + _, out, _ = proc("shell", mockserver.url("/text"), "-c", "response.body") assert "Works" in out - def test_response_type_text(self): - _, out, _ = self.proc( - "shell", self.mockserver.url("/text"), "-c", "type(response)" - ) + def test_response_type_text(self, mockserver: MockServer) -> None: + _, out, _ = proc("shell", mockserver.url("/text"), "-c", "type(response)") assert "TextResponse" in out - def test_response_type_html(self): - _, out, _ = self.proc( - "shell", self.mockserver.url("/html"), "-c", "type(response)" - ) + def test_response_type_html(self, mockserver: MockServer) -> None: + _, out, _ = proc("shell", mockserver.url("/html"), "-c", "type(response)") assert "HtmlResponse" in out - def test_response_selector_html(self): + def test_response_selector_html(self, mockserver: MockServer) -> None: xpath = "response.xpath(\"//p[@class='one']/text()\").get()" - _, out, _ = self.proc("shell", self.mockserver.url("/html"), "-c", xpath) + _, out, _ = proc("shell", mockserver.url("/html"), "-c", xpath) assert out.strip() == "Works" - def test_response_encoding_gb18030(self): - _, out, _ = self.proc( - "shell", self.mockserver.url("/enc-gb18030"), "-c", "response.encoding" + def test_response_encoding_gb18030(self, mockserver: MockServer) -> None: + _, out, _ = proc( + "shell", mockserver.url("/enc-gb18030"), "-c", "response.encoding" ) assert out.strip() == "gb18030" - def test_redirect(self): - _, out, _ = self.proc( - "shell", self.mockserver.url("/redirect"), "-c", "response.url" - ) + def test_redirect(self, mockserver: MockServer) -> None: + _, out, _ = proc("shell", mockserver.url("/redirect"), "-c", "response.url") assert out.strip().endswith("/redirected") - def test_redirect_follow_302(self): - _, out, _ = self.proc( + def test_redirect_follow_302(self, mockserver: MockServer) -> None: + _, out, _ = proc( "shell", - self.mockserver.url("/redirect-no-meta-refresh"), + mockserver.url("/redirect-no-meta-refresh"), "-c", "response.status", ) assert out.strip().endswith("200") - def test_redirect_not_follow_302(self): - _, out, _ = self.proc( + def test_redirect_not_follow_302(self, mockserver: MockServer) -> None: + _, out, _ = proc( "shell", "--no-redirect", - self.mockserver.url("/redirect-no-meta-refresh"), + mockserver.url("/redirect-no-meta-refresh"), "-c", "response.status", ) assert out.strip().endswith("302") - def test_fetch_redirect_follow_302(self): + def test_fetch_redirect_follow_302(self, mockserver: MockServer) -> None: """Test that calling ``fetch(url)`` follows HTTP redirects by default.""" - url = self.mockserver.url("/redirect-no-meta-refresh") + url = mockserver.url("/redirect-no-meta-refresh") code = f"fetch('{url}')" - p, out, errout = self.proc("shell", "-c", code) - assert p.returncode == 0, out - assert "Redirecting (302)" in errout - assert "Crawled (200)" in errout + ret, out, err = proc("shell", "-c", code) + assert ret == 0, out + assert "Redirecting (302)" in err + assert "Crawled (200)" in err - def test_fetch_redirect_not_follow_302(self): + def test_fetch_redirect_not_follow_302(self, mockserver: MockServer) -> None: """Test that calling ``fetch(url, redirect=False)`` disables automatic redirects.""" - url = self.mockserver.url("/redirect-no-meta-refresh") + url = mockserver.url("/redirect-no-meta-refresh") code = f"fetch('{url}', redirect=False)" - p, out, errout = self.proc("shell", "-c", code) - assert p.returncode == 0, out - assert "Crawled (302)" in errout + ret, out, err = proc("shell", "-c", code) + assert ret == 0, out + assert "Crawled (302)" in err - def test_request_replace(self): - url = self.mockserver.url("/text") + def test_request_replace(self, mockserver: MockServer) -> None: + url = mockserver.url("/text") code = f"fetch('{url}') or fetch(response.request.replace(method='POST'))" - p, out, _ = self.proc("shell", "-c", code) - assert p.returncode == 0, out + ret, out, _ = proc("shell", "-c", code) + assert ret == 0, out - def test_scrapy_import(self): - url = self.mockserver.url("/text") + def test_scrapy_import(self, mockserver: MockServer) -> None: + url = mockserver.url("/text") code = f"fetch(scrapy.Request('{url}'))" - p, out, _ = self.proc("shell", "-c", code) - assert p.returncode == 0, out + ret, out, _ = proc("shell", "-c", code) + assert ret == 0, out - def test_local_file(self): + def test_local_file(self) -> None: filepath = Path(tests_datadir, "test_site", "index.html") - _, out, _ = self.proc("shell", str(filepath), "-c", "item") + _, out, _ = proc("shell", str(filepath), "-c", "item") assert "{}" in out - def test_local_nofile(self): + def test_local_nofile(self) -> None: filepath = "file:///tests/sample_data/test_site/nothinghere.html" - p, out, err = self.proc("shell", filepath, "-c", "item") - assert p.returncode == 1, out or err + ret, out, err = proc("shell", filepath, "-c", "item") + assert ret == 1, out or err assert "No such file or directory" in err - def test_dns_failures(self): + def test_dns_failures(self, mockserver: MockServer) -> None: if NON_EXISTING_RESOLVABLE: pytest.skip("Non-existing hosts are resolvable") url = "www.somedomainthatdoesntexi.st" - p, out, err = self.proc("shell", url, "-c", "item") - assert p.returncode == 1, out or err + ret, out, err = proc("shell", url, "-c", "item") + assert ret == 1, out or err assert "DNS lookup failed" in err - def test_shell_fetch_async(self): - url = self.mockserver.url("/html") + def test_shell_fetch_async(self, mockserver: MockServer) -> None: + url = mockserver.url("/html") code = f"fetch('{url}')" - p, _, err = self.proc( + ret, _, err = proc( "shell", "-c", code, "--set", f"TWISTED_REACTOR={_asyncio_reactor_path}" ) - assert p.returncode == 0, err + assert ret == 0, err assert "RuntimeError: There is no current event loop in thread" not in err class TestInteractiveShell: - def test_fetch(self): + def test_fetch(self, mockserver: MockServer) -> None: args = ( sys.executable, "-m", @@ -149,13 +137,13 @@ class TestInteractiveShell: env = os.environ.copy() env["SCRAPY_PYTHON_SHELL"] = "python" logfile = BytesIO() - p = PopenSpawn(args, env=env, timeout=5) + # https://github.com/python/typeshed/issues/14915 + p = PopenSpawn(args, env=cast("os._Environ", env), timeout=5) p.logfile_read = logfile p.expect_exact("Available Scrapy objects") - with MockServer() as mockserver: - p.sendline(f"fetch('{mockserver.url('/')}')") - p.sendline("type(response)") - p.expect_exact("HtmlResponse") + p.sendline(f"fetch('{mockserver.url('/')}')") + p.sendline("type(response)") + p.expect_exact("HtmlResponse") p.sendeof() p.wait() logfile.seek(0) diff --git a/tests/test_command_startproject.py b/tests/test_command_startproject.py index 246066485..fe3c9bcbe 100644 --- a/tests/test_command_startproject.py +++ b/tests/test_command_startproject.py @@ -8,70 +8,74 @@ from itertools import chain from pathlib import Path from shutil import copytree from stat import S_IWRITE as ANYONE_WRITE_PERMISSION -from tempfile import mkdtemp import scrapy from scrapy.commands.startproject import IGNORE -from tests.test_commands import TestProjectBase +from scrapy.utils.test import get_testenv +from tests.utils.cmdline import call, proc -class TestStartprojectCommand(TestProjectBase): - def test_startproject(self): - p, _, _ = self.proc("startproject", self.project_name) - assert p.returncode == 0 +class TestStartprojectCommand: + project_name = "testproject" - assert Path(self.proj_path, "scrapy.cfg").exists() - assert Path(self.proj_path, "testproject").exists() - assert Path(self.proj_mod_path, "__init__.py").exists() - assert Path(self.proj_mod_path, "items.py").exists() - assert Path(self.proj_mod_path, "pipelines.py").exists() - assert Path(self.proj_mod_path, "settings.py").exists() - assert Path(self.proj_mod_path, "spiders", "__init__.py").exists() + @staticmethod + def _assert_files_exist(project_dir: Path, project_name: str) -> None: + assert (project_dir / "scrapy.cfg").exists() + assert (project_dir / project_name).exists() + assert (project_dir / project_name / "__init__.py").exists() + assert (project_dir / project_name / "items.py").exists() + assert (project_dir / project_name / "pipelines.py").exists() + assert (project_dir / project_name / "settings.py").exists() + assert (project_dir / project_name / "spiders" / "__init__.py").exists() - assert self.call("startproject", self.project_name) == 1 - assert self.call("startproject", "wrong---project---name") == 1 - assert self.call("startproject", "sys") == 1 + def test_startproject(self, tmp_path: Path) -> None: + # with no dir argument creates the project in the "self.project_name" subdir of cwd + assert call("startproject", self.project_name, cwd=tmp_path) == 0 + self._assert_files_exist(tmp_path / self.project_name, self.project_name) - def test_startproject_with_project_dir(self): - project_dir = mkdtemp() - assert self.call("startproject", self.project_name, project_dir) == 0 + assert call("startproject", self.project_name, cwd=tmp_path) == 1 + assert call("startproject", "wrong---project---name") == 1 + assert call("startproject", "sys") == 1 - assert Path(project_dir, "scrapy.cfg").exists() - assert Path(project_dir, "testproject").exists() - assert Path(project_dir, self.project_name, "__init__.py").exists() - assert Path(project_dir, self.project_name, "items.py").exists() - assert Path(project_dir, self.project_name, "pipelines.py").exists() - assert Path(project_dir, self.project_name, "settings.py").exists() - assert Path(project_dir, self.project_name, "spiders", "__init__.py").exists() - - assert self.call("startproject", self.project_name, project_dir + "2") == 0 - - assert self.call("startproject", self.project_name, project_dir) == 1 - assert self.call("startproject", self.project_name + "2", project_dir) == 1 - assert self.call("startproject", "wrong---project---name") == 1 - assert self.call("startproject", "sys") == 1 - assert self.call("startproject") == 2 + def test_startproject_with_project_dir(self, tmp_path: Path) -> None: + # with a dir arg creates the project in the specified dir + project_dir = tmp_path / "project" assert ( - self.call("startproject", self.project_name, project_dir, "another_params") + call("startproject", self.project_name, str(project_dir), cwd=tmp_path) == 0 + ) + self._assert_files_exist(project_dir, self.project_name) + + assert ( + call( + "startproject", self.project_name, str(project_dir) + "2", cwd=tmp_path + ) + == 0 + ) + + assert ( + call("startproject", self.project_name, str(project_dir), cwd=tmp_path) == 1 + ) + assert ( + call( + "startproject", self.project_name + "2", str(project_dir), cwd=tmp_path + ) + == 1 + ) + assert call("startproject", "wrong---project---name") == 1 + assert call("startproject", "sys") == 1 + assert call("startproject") == 2 + assert ( + call("startproject", self.project_name, str(project_dir), "another_params") == 2 ) - def test_existing_project_dir(self): - project_dir = mkdtemp() + def test_existing_project_dir(self, tmp_path: Path) -> None: project_name = self.project_name + "_existing" - project_path = Path(project_dir, project_name) + project_path = tmp_path / project_name project_path.mkdir() - p, _, _ = self.proc("startproject", project_name, cwd=project_dir) - assert p.returncode == 0 - - assert Path(project_path, "scrapy.cfg").exists() - assert Path(project_path, project_name).exists() - assert Path(project_path, project_name, "__init__.py").exists() - assert Path(project_path, project_name, "items.py").exists() - assert Path(project_path, project_name, "pipelines.py").exists() - assert Path(project_path, project_name, "settings.py").exists() - assert Path(project_path, project_name, "spiders", "__init__.py").exists() + assert call("startproject", project_name, cwd=tmp_path) == 0 + self._assert_files_exist(project_path, project_name) def get_permissions_dict( @@ -101,26 +105,22 @@ def get_permissions_dict( return permissions_dict -class TestStartprojectTemplates(TestProjectBase): - def setup_method(self): - super().setup_method() - self.tmpl = str(Path(self.temp_path, "templates")) - self.tmpl_proj = str(Path(self.tmpl, "project")) +class TestStartprojectTemplates: + def test_startproject_template_override(self, tmp_path: Path) -> None: + tmpl = tmp_path / "templates" + tmpl_proj = tmpl / "project" + project_name = "testproject" - def test_startproject_template_override(self): - copytree(Path(scrapy.__path__[0], "templates"), self.tmpl) - Path(self.tmpl_proj, "root_template").write_bytes(b"") - assert Path(self.tmpl_proj, "root_template").exists() + copytree(Path(scrapy.__path__[0], "templates"), tmpl) + (tmpl_proj / "root_template").write_bytes(b"") - args = ["--set", f"TEMPLATES_DIR={self.tmpl}"] - p, out, err = self.proc("startproject", self.project_name, *args) - assert ( - f"New Scrapy project '{self.project_name}', using template directory" in out - ) - assert self.tmpl_proj in out - assert Path(self.proj_path, "root_template").exists() + args = ["--set", f"TEMPLATES_DIR={tmpl}"] + _, out, _ = proc("startproject", project_name, *args, cwd=tmp_path) + assert f"New Scrapy project '{project_name}', using template directory" in out + assert str(tmpl_proj) in out + assert (tmp_path / project_name / "root_template").exists() - def test_startproject_permissions_from_writable(self): + def test_startproject_permissions_from_writable(self, tmp_path: Path) -> None: """Check that generated files have the right permissions when the template folder has the same permissions as in the project, i.e. everything is writable.""" @@ -137,7 +137,8 @@ class TestStartprojectTemplates(TestProjectBase): IGNORE, ) - destination = mkdtemp() + destination = tmp_path / "proj" + destination.mkdir() process = subprocess.Popen( ( sys.executable, @@ -149,16 +150,16 @@ class TestStartprojectTemplates(TestProjectBase): cwd=destination, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - env=self.env, + env=get_testenv(), ) process.wait() - project_dir = Path(destination, project_name) + project_dir = destination / project_name actual_permissions = get_permissions_dict(project_dir) assert actual_permissions == expected_permissions - def test_startproject_permissions_from_read_only(self): + def test_startproject_permissions_from_read_only(self, tmp_path: Path) -> None: """Check that generated files have the right permissions when the template folder has been made read-only, which is something that some systems do. @@ -183,37 +184,34 @@ class TestStartprojectTemplates(TestProjectBase): current_permissions = path.stat().st_mode path.chmod(current_permissions & ~ANYONE_WRITE_PERMISSION) - read_only_templates_dir = str(Path(mkdtemp()) / "templates") + read_only_templates_dir = tmp_path / "templates" copytree(templates_dir, read_only_templates_dir) for root, dirs, files in os.walk(read_only_templates_dir): for node in chain(dirs, files): _make_read_only(Path(root, node)) - destination = mkdtemp() - process = subprocess.Popen( - ( - sys.executable, - "-m", - "scrapy.cmdline", + destination = tmp_path / "proj" + destination.mkdir() + assert ( + call( "startproject", project_name, "--set", f"TEMPLATES_DIR={read_only_templates_dir}", - ), - cwd=destination, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - env=self.env, + cwd=destination, + ) + == 0 ) - process.wait() - project_dir = Path(destination, project_name) + project_dir = destination / project_name actual_permissions = get_permissions_dict(project_dir) assert actual_permissions == expected_permissions - def test_startproject_permissions_unchanged_in_destination(self): + def test_startproject_permissions_unchanged_in_destination( + self, tmp_path: Path + ) -> None: """Check that preexisting folders and files in the destination folder do not see their permissions modified.""" scrapy_path = scrapy.__path__[0] @@ -229,8 +227,9 @@ class TestStartprojectTemplates(TestProjectBase): IGNORE, ) - destination = mkdtemp() - project_dir = Path(destination, project_name) + destination = tmp_path / "proj" + project_dir = destination / project_name + project_dir.mkdir(parents=True) existing_nodes = { oct(permissions)[2:] + extension: permissions @@ -244,7 +243,6 @@ class TestStartprojectTemplates(TestProjectBase): 0o777, ) } - project_dir.mkdir() for node, permissions in existing_nodes.items(): path = project_dir / node if node.endswith(".d"): @@ -253,27 +251,13 @@ class TestStartprojectTemplates(TestProjectBase): path.touch(mode=permissions) expected_permissions[node] = oct(path.stat().st_mode) - process = subprocess.Popen( - ( - sys.executable, - "-m", - "scrapy.cmdline", - "startproject", - project_name, - ".", - ), - cwd=project_dir, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - env=self.env, - ) - process.wait() + assert call("startproject", project_name, ".", cwd=project_dir) == 0 actual_permissions = get_permissions_dict(project_dir) assert actual_permissions == expected_permissions - def test_startproject_permissions_umask_022(self): + def test_startproject_permissions_umask_022(self, tmp_path: Path) -> None: """Check that generated files have the right permissions when the system uses a umask value that causes new files to have different permissions than those from the template folder.""" @@ -298,23 +282,11 @@ class TestStartprojectTemplates(TestProjectBase): ) with umask(0o002): - destination = mkdtemp() - process = subprocess.Popen( - ( - sys.executable, - "-m", - "scrapy.cmdline", - "startproject", - project_name, - ), - cwd=destination, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - env=self.env, - ) - process.wait() + destination = tmp_path / "proj" + destination.mkdir() + assert call("startproject", project_name, cwd=destination) == 0 - project_dir = Path(destination, project_name) + project_dir = destination / project_name actual_permissions = get_permissions_dict(project_dir) assert actual_permissions == expected_permissions diff --git a/tests/test_command_version.py b/tests/test_command_version.py index de58203fc..c4ca9d07b 100644 --- a/tests/test_command_version.py +++ b/tests/test_command_version.py @@ -1,14 +1,14 @@ import scrapy -from tests.test_commands import TestProjectBase +from tests.utils.cmdline import proc -class TestVersionCommand(TestProjectBase): - def test_output(self): - _, out, _ = self.proc("version") +class TestVersionCommand: + def test_output(self) -> None: + _, out, _ = proc("version") assert out.strip() == f"Scrapy {scrapy.__version__}" - def test_verbose_output(self): - _, out, _ = self.proc("version", "-v") + def test_verbose_output(self) -> None: + _, out, _ = proc("version", "-v") headers = [line.partition(":")[0].strip() for line in out.strip().splitlines()] assert headers == [ "Scrapy", diff --git a/tests/test_commands.py b/tests/test_commands.py index 9aaeb3e8d..07bd0b45c 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -2,14 +2,9 @@ from __future__ import annotations import argparse import json -import re -import subprocess -import sys from io import StringIO -from pathlib import Path -from shutil import rmtree -from tempfile import TemporaryFile, mkdtemp -from typing import Any +from shutil import copytree +from typing import TYPE_CHECKING from unittest import mock import pytest @@ -18,9 +13,11 @@ import scrapy from scrapy.cmdline import _pop_command_name, _print_unknown_command_msg from scrapy.commands import ScrapyCommand, ScrapyHelpFormatter, view from scrapy.settings import Settings -from scrapy.utils.python import to_unicode from scrapy.utils.reactor import _asyncio_reactor_path -from scrapy.utils.test import get_testenv +from tests.utils.cmdline import call, proc + +if TYPE_CHECKING: + from pathlib import Path class EmptyCommand(ScrapyCommand): @@ -66,68 +63,29 @@ class TestCommandSettings: class TestProjectBase: + """A base class for tests that may need a Scrapy project.""" + project_name = "testproject" - def setup_method(self): - self.temp_path = mkdtemp() - self.cwd = self.temp_path - self.proj_path = Path(self.temp_path, self.project_name) - self.proj_mod_path = self.proj_path / self.project_name - self.env = get_testenv() + @pytest.fixture(scope="session") + def _proj_path_cached(self, tmp_path_factory: pytest.TempPathFactory) -> Path: + """Create a Scrapy project in a temporary directory and return its path. - def teardown_method(self): - rmtree(self.temp_path) + Used as a cache for ``proj_path``. + """ + tmp_path = tmp_path_factory.mktemp("proj") + call("startproject", self.project_name, cwd=tmp_path) + return tmp_path / self.project_name - def call(self, *args: str, **popen_kwargs: Any) -> int: - with TemporaryFile() as out: - args = (sys.executable, "-m", "scrapy.cmdline", *args) - return subprocess.call( - args, stdout=out, stderr=out, cwd=self.cwd, env=self.env, **popen_kwargs - ) - - def proc( - self, *args: str, **popen_kwargs: Any - ) -> tuple[subprocess.Popen[bytes], str, str]: - args = (sys.executable, "-m", "scrapy.cmdline", *args) - p = subprocess.Popen( - args, - cwd=popen_kwargs.pop("cwd", self.cwd), - env=self.env, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - **popen_kwargs, - ) - - try: - stdout, stderr = p.communicate(timeout=15) - except subprocess.TimeoutExpired: - p.kill() - p.communicate() - pytest.fail("Command took too much time to complete") - - return p, to_unicode(stdout), to_unicode(stderr) - - @staticmethod - def find_in_file(filename: Path, regex: str) -> re.Match | None: - """Find first pattern occurrence in file""" - pattern = re.compile(regex) - with filename.open("r", encoding="utf-8") as f: - for line in f: - match = pattern.search(line) - if match is not None: - return match - return None + @pytest.fixture + def proj_path(self, tmp_path: Path, _proj_path_cached: Path) -> Path: + """Copy a pre-generated Scrapy project into a temporary directory and return its path.""" + proj_path = tmp_path / self.project_name + copytree(_proj_path_cached, proj_path) + return proj_path -class TestCommandBase(TestProjectBase): - def setup_method(self): - super().setup_method() - self.call("startproject", self.project_name) - self.cwd = self.proj_path - self.env["SCRAPY_SETTINGS_MODULE"] = f"{self.project_name}.settings" - - -class TestCommandCrawlerProcess(TestCommandBase): +class TestCommandCrawlerProcess(TestProjectBase): """Test that the command uses the expected kind of *CrawlerProcess and produces expected errors when needed.""" @@ -135,9 +93,10 @@ class TestCommandCrawlerProcess(TestCommandBase): NORMAL_MSG = "Using CrawlerProcess" ASYNC_MSG = "Using AsyncCrawlerProcess" - def setup_method(self): - super().setup_method() - (self.cwd / self.project_name / "spiders" / "sp.py").write_text(""" + @pytest.fixture(autouse=True) + def create_files(self, proj_path: Path) -> None: + proj_mod_path = proj_path / self.project_name + (proj_mod_path / "spiders" / "sp.py").write_text(""" import scrapy class MySpider(scrapy.Spider): @@ -151,7 +110,7 @@ class MySpider(scrapy.Spider): yield """) - (self.cwd / self.project_name / "spiders" / "aiosp.py").write_text(""" + (proj_mod_path / "spiders" / "aiosp.py").write_text(""" import asyncio import scrapy @@ -168,18 +127,20 @@ class MySpider(scrapy.Spider): yield """) - self._append_settings("LOG_LEVEL = 'DEBUG'\n") + self._append_settings(proj_mod_path, "LOG_LEVEL = 'DEBUG'\n") - def _append_settings(self, text: str) -> None: + @staticmethod + def _append_settings(proj_mod_path: Path, text: str) -> None: """Add text to the end of the project settings.py.""" - with (self.cwd / self.project_name / "settings.py").open( - "a", encoding="utf-8" - ) as f: + with (proj_mod_path / "settings.py").open("a", encoding="utf-8") as f: f.write(text) - def _replace_custom_settings(self, spider_name: str, text: str) -> None: + @staticmethod + def _replace_custom_settings( + proj_mod_path: Path, spider_name: str, text: str + ) -> None: """Replace custom_settings in the given spider file with the given text.""" - spider_path = self.cwd / self.project_name / "spiders" / f"{spider_name}.py" + spider_path = proj_mod_path / "spiders" / f"{spider_name}.py" with spider_path.open("r+", encoding="utf-8") as f: content = f.read() content = content.replace( @@ -189,74 +150,87 @@ class MySpider(scrapy.Spider): f.write(content) f.truncate() - def _assert_spider_works(self, msg: str, *args: str) -> None: + def _assert_spider_works(self, msg: str, proj_path: Path, *args: str) -> None: """The command uses the expected *CrawlerProcess, the spider works.""" - _, _, err = self.proc(self.name, *args) + _, _, err = proc(self.name, *args, cwd=proj_path) assert msg in err assert "It works!" in err assert "Spider closed (finished)" in err - def _assert_spider_asyncio_fail(self, msg: str, *args: str) -> None: + def _assert_spider_asyncio_fail( + self, msg: str, proj_path: Path, *args: str + ) -> None: """The command uses the expected *CrawlerProcess, the spider fails to use asyncio.""" - _, _, err = self.proc(self.name, *args) + _, _, err = proc(self.name, *args, cwd=proj_path) assert msg in err assert "no running event loop" in err - def test_project_settings(self): + def test_project_settings(self, proj_path: Path) -> None: """The reactor is set via the project default settings (to the asyncio value). AsyncCrawlerProcess, the asyncio reactor, both spiders work.""" for spider in ["sp", "aiosp"]: - self._assert_spider_works(self.ASYNC_MSG, spider) + self._assert_spider_works(self.ASYNC_MSG, proj_path, spider) - def test_cmdline_asyncio(self): + def test_cmdline_asyncio(self, proj_path: Path) -> None: """The reactor is set via the command line to the asyncio value. AsyncCrawlerProcess, the asyncio reactor, both spiders work.""" for spider in ["sp", "aiosp"]: self._assert_spider_works( - self.ASYNC_MSG, spider, "-s", f"TWISTED_REACTOR={_asyncio_reactor_path}" + self.ASYNC_MSG, + proj_path, + spider, + "-s", + f"TWISTED_REACTOR={_asyncio_reactor_path}", ) - def test_project_settings_explicit_asyncio(self): + def test_project_settings_explicit_asyncio(self, proj_path: Path) -> None: """The reactor explicitly is set via the project settings to the asyncio value. AsyncCrawlerProcess, the asyncio reactor, both spiders work.""" - self._append_settings(f"TWISTED_REACTOR = '{_asyncio_reactor_path}'\n") + self._append_settings( + proj_path / self.project_name, + f"TWISTED_REACTOR = '{_asyncio_reactor_path}'\n", + ) for spider in ["sp", "aiosp"]: - self._assert_spider_works(self.ASYNC_MSG, spider) + self._assert_spider_works(self.ASYNC_MSG, proj_path, spider) - def test_cmdline_empty(self): + def test_cmdline_empty(self, proj_path: Path) -> None: """The reactor is set via the command line to the empty value. CrawlerProcess, the default reactor, only the normal spider works.""" - self._assert_spider_works(self.NORMAL_MSG, "sp", "-s", "TWISTED_REACTOR=") + self._assert_spider_works( + self.NORMAL_MSG, proj_path, "sp", "-s", "TWISTED_REACTOR=" + ) self._assert_spider_asyncio_fail( - self.NORMAL_MSG, "aiosp", "-s", "TWISTED_REACTOR=" + self.NORMAL_MSG, proj_path, "aiosp", "-s", "TWISTED_REACTOR=" ) - def test_project_settings_empty(self): + def test_project_settings_empty(self, proj_path: Path) -> None: """The reactor is set via the project settings to the empty value. CrawlerProcess, the default reactor, only the normal spider works.""" - self._append_settings("TWISTED_REACTOR = None\n") + self._append_settings(proj_path / self.project_name, "TWISTED_REACTOR = None\n") - self._assert_spider_works(self.NORMAL_MSG, "sp") + self._assert_spider_works(self.NORMAL_MSG, proj_path, "sp") self._assert_spider_asyncio_fail( - self.NORMAL_MSG, "aiosp", "-s", "TWISTED_REACTOR=" + self.NORMAL_MSG, proj_path, "aiosp", "-s", "TWISTED_REACTOR=" ) - def test_spider_settings_asyncio(self): + def test_spider_settings_asyncio(self, proj_path: Path) -> None: """The reactor is set via the spider settings to the asyncio value. AsyncCrawlerProcess, the asyncio reactor, both spiders work.""" for spider in ["sp", "aiosp"]: self._replace_custom_settings( - spider, f"{{'TWISTED_REACTOR': '{_asyncio_reactor_path}'}}" + proj_path / self.project_name, + spider, + f"{{'TWISTED_REACTOR': '{_asyncio_reactor_path}'}}", ) - self._assert_spider_works(self.ASYNC_MSG, spider) + self._assert_spider_works(self.ASYNC_MSG, proj_path, spider) - def test_spider_settings_asyncio_cmdline_empty(self): + def test_spider_settings_asyncio_cmdline_empty(self, proj_path: Path) -> None: """The reactor is set via the spider settings to the asyncio value and via command line to the empty value. The command line value takes precedence so the spider settings don't matter. @@ -264,29 +238,35 @@ class MySpider(scrapy.Spider): CrawlerProcess, the default reactor, only the normal spider works.""" for spider in ["sp", "aiosp"]: self._replace_custom_settings( - spider, f"{{'TWISTED_REACTOR': '{_asyncio_reactor_path}'}}" + proj_path / self.project_name, + spider, + f"{{'TWISTED_REACTOR': '{_asyncio_reactor_path}'}}", ) - self._assert_spider_works(self.NORMAL_MSG, "sp", "-s", "TWISTED_REACTOR=") + self._assert_spider_works( + self.NORMAL_MSG, proj_path, "sp", "-s", "TWISTED_REACTOR=" + ) self._assert_spider_asyncio_fail( - self.NORMAL_MSG, "aiosp", "-s", "TWISTED_REACTOR=" + self.NORMAL_MSG, proj_path, "aiosp", "-s", "TWISTED_REACTOR=" ) - def test_project_empty_spider_settings_asyncio(self): + def test_project_empty_spider_settings_asyncio(self, proj_path: Path) -> None: """The reactor is set via the project settings to the empty value and via the spider settings to the asyncio value. CrawlerProcess is chosen based on the project settings, but the asyncio reactor is chosen based on the spider settings. CrawlerProcess, the asyncio reactor, both spiders work.""" - self._append_settings("TWISTED_REACTOR = None\n") + self._append_settings(proj_path / self.project_name, "TWISTED_REACTOR = None\n") for spider in ["sp", "aiosp"]: self._replace_custom_settings( - spider, f"{{'TWISTED_REACTOR': '{_asyncio_reactor_path}'}}" + proj_path / self.project_name, + spider, + f"{{'TWISTED_REACTOR': '{_asyncio_reactor_path}'}}", ) - self._assert_spider_works(self.NORMAL_MSG, spider) + self._assert_spider_works(self.NORMAL_MSG, proj_path, spider) - def test_project_asyncio_spider_settings_select(self): + def test_project_asyncio_spider_settings_select(self, proj_path: Path) -> None: """The reactor is set via the project settings to the asyncio value and via the spider settings to the select value. AsyncCrawlerProcess is chosen based on the project settings, and the conflicting reactor @@ -294,13 +274,17 @@ class MySpider(scrapy.Spider): AsyncCrawlerProcess, the asyncio reactor, both spiders produce a mismatched reactor exception.""" - self._append_settings(f"TWISTED_REACTOR = '{_asyncio_reactor_path}'\n") + self._append_settings( + proj_path / self.project_name, + f"TWISTED_REACTOR = '{_asyncio_reactor_path}'\n", + ) for spider in ["sp", "aiosp"]: self._replace_custom_settings( + proj_path / self.project_name, spider, "{'TWISTED_REACTOR': 'twisted.internet.selectreactor.SelectReactor'}", ) - _, _, err = self.proc(self.name, spider) + _, _, err = proc(self.name, spider, cwd=proj_path) assert self.ASYNC_MSG in err assert ( "The installed reactor (twisted.internet.asyncioreactor.AsyncioSelectorReactor)" @@ -308,29 +292,40 @@ class MySpider(scrapy.Spider): " (twisted.internet.selectreactor.SelectReactor)" ) in err - def test_project_asyncio_spider_settings_select_forced(self): + def test_project_asyncio_spider_settings_select_forced( + self, proj_path: Path + ) -> None: """The reactor is set via the project settings to the asyncio value and via the spider settings to the select value, CrawlerProcess is forced via the project settings. The reactor is chosen based on the spider settings. CrawlerProcess, the select reactor, only the normal spider works.""" - self._append_settings("FORCE_CRAWLER_PROCESS = True\n") + self._append_settings( + proj_path / self.project_name, "FORCE_CRAWLER_PROCESS = True\n" + ) for spider in ["sp", "aiosp"]: self._replace_custom_settings( + proj_path / self.project_name, spider, "{'TWISTED_REACTOR': 'twisted.internet.selectreactor.SelectReactor'}", ) - self._assert_spider_works(self.NORMAL_MSG, "sp") - self._assert_spider_asyncio_fail(self.NORMAL_MSG, "aiosp") + self._assert_spider_works(self.NORMAL_MSG, proj_path, "sp") + self._assert_spider_asyncio_fail(self.NORMAL_MSG, proj_path, "aiosp") -class TestMiscCommands(TestCommandBase): - def test_list(self): - assert self.call("list") == 0 +class TestMiscCommands(TestProjectBase): + def test_list(self, proj_path: Path) -> None: + assert call("list", cwd=proj_path) == 0 - def test_command_not_found(self): + def test_list_subdir(self, proj_path: Path) -> None: + """Test that commands work in a subdirectory of the project.""" + subdir = proj_path / "subdir" + subdir.mkdir(exist_ok=True) + assert call("list", cwd=subdir) == 0 + + def test_command_not_found(self) -> None: na_msg = """ The list command is not available from this location. These commands are only available from within a project: check, crawl, edit, list, parse. @@ -339,9 +334,9 @@ These commands are only available from within a project: check, crawl, edit, lis Unknown command: abc """ params = [ - ("list", 0, na_msg), - ("abc", 0, not_found_msg), - ("abc", 1, not_found_msg), + ("list", False, na_msg), + ("abc", False, not_found_msg), + ("abc", True, not_found_msg), ] for cmdname, inproject, message in params: with mock.patch("sys.stdout", new=StringIO()) as out: @@ -349,31 +344,22 @@ Unknown command: abc assert out.getvalue().strip() == message.strip() -class TestProjectSubdir(TestProjectBase): - """Test that commands work in a subdirectory of the project.""" - - def setup_method(self): - super().setup_method() - self.call("startproject", self.project_name) - self.cwd = self.proj_path / "subdir" - self.cwd.mkdir(exist_ok=True) - - def test_list(self): - assert self.call("list") == 0 - - -class TestBenchCommand(TestCommandBase): - def test_run(self): - _, _, log = self.proc( - "bench", "-s", "LOGSTATS_INTERVAL=0.001", "-s", "CLOSESPIDER_TIMEOUT=0.01" +class TestBenchCommand: + def test_run(self) -> None: + _, _, err = proc( + "bench", + "-s", + "LOGSTATS_INTERVAL=0.001", + "-s", + "CLOSESPIDER_TIMEOUT=0.01", ) - assert "INFO: Crawled" in log - assert "Unhandled Error" not in log - assert "log_count/ERROR" not in log + assert "INFO: Crawled" in err + assert "Unhandled Error" not in err + assert "log_count/ERROR" not in err -class TestViewCommand(TestCommandBase): - def test_methods(self): +class TestViewCommand: + def test_methods(self) -> None: command = view.Command() command.settings = Settings() parser = argparse.ArgumentParser( @@ -387,52 +373,50 @@ class TestViewCommand(TestCommandBase): assert "URL using the Scrapy downloader and show its" in command.long_desc() -class TestHelpMessage(TestCommandBase): - def setup_method(self): - super().setup_method() - self.commands = [ - "parse", - "startproject", - "view", - "crawl", - "edit", - "list", - "fetch", - "settings", - "shell", - "runspider", - "version", - "genspider", - "check", - "bench", - ] +class TestHelpMessage(TestProjectBase): + COMMANDS = [ + "parse", + "startproject", + "view", + "crawl", + "edit", + "list", + "fetch", + "settings", + "shell", + "runspider", + "version", + "genspider", + "check", + "bench", + ] - def test_help_messages(self): - for command in self.commands: - _, out, _ = self.proc(command, "-h") + def test_help_messages(self, proj_path: Path) -> None: + for command in self.COMMANDS: + _, out, _ = proc(command, "-h", cwd=proj_path) assert "Usage" in out class TestPopCommandName: - def test_valid_command(self): + def test_valid_command(self) -> None: argv = ["scrapy", "crawl", "my_spider"] command = _pop_command_name(argv) assert command == "crawl" assert argv == ["scrapy", "my_spider"] - def test_no_command(self): + def test_no_command(self) -> None: argv = ["scrapy"] command = _pop_command_name(argv) assert command is None assert argv == ["scrapy"] - def test_option_before_command(self): + def test_option_before_command(self) -> None: argv = ["scrapy", "-h", "crawl"] command = _pop_command_name(argv) assert command == "crawl" assert argv == ["scrapy", "-h"] - def test_option_after_command(self): + def test_option_after_command(self) -> None: argv = ["scrapy", "crawl", "-h"] command = _pop_command_name(argv) assert command == "crawl" diff --git a/tests/utils/cmdline.py b/tests/utils/cmdline.py new file mode 100644 index 000000000..122e0236d --- /dev/null +++ b/tests/utils/cmdline.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import subprocess +import sys +from typing import Any + +import pytest + +from scrapy.utils.test import get_testenv + + +def call(*args: str, **popen_kwargs: Any) -> int: + args = (sys.executable, "-m", "scrapy.cmdline", *args) + return subprocess.call( + args, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + env=get_testenv(), + **popen_kwargs, + ) + + +def proc(*args: str, **popen_kwargs: Any) -> tuple[int, str, str]: + args = (sys.executable, "-m", "scrapy.cmdline", *args) + try: + p = subprocess.run( + args, + check=False, + capture_output=True, + encoding="utf-8", + timeout=15, + env=get_testenv(), + **popen_kwargs, + ) + except subprocess.TimeoutExpired: + pytest.fail("Command took too much time to complete") + + return p.returncode, p.stdout, p.stderr