From 5a605969bdc102e0193ad15ccc571dc6164e5d26 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Thu, 6 Mar 2025 23:52:41 +0400 Subject: [PATCH] Converting tests to plain asserts, part 2. (#6699) --- tests/test_addons.py | 48 +-- tests/test_closespider.py | 28 +- tests/test_cmdline/__init__.py | 30 +- .../__init__.py | 7 +- tests/test_command_check.py | 14 +- tests/test_command_fetch.py | 10 +- tests/test_command_parse.py | 81 ++-- tests/test_command_shell.py | 28 +- tests/test_command_version.py | 36 +- tests/test_commands.py | 371 +++++++++--------- tests/test_contracts.py | 112 +++--- tests/test_core_downloader.py | 25 +- tests/test_crawl.py | 288 +++++++------- tests/test_crawler.py | 351 +++++++++-------- tests/test_dependencies.py | 7 +- tests/test_downloaderslotssettings.py | 2 +- tests/test_dupefilters.py | 21 +- tests/test_engine.py | 137 ++++--- tests/test_engine_stop_download_bytes.py | 13 +- tests/test_engine_stop_download_headers.py | 12 +- tests/test_webclient.py | 40 +- 21 files changed, 796 insertions(+), 865 deletions(-) diff --git a/tests/test_addons.py b/tests/test_addons.py index a0caa3511..686bf9952 100644 --- a/tests/test_addons.py +++ b/tests/test_addons.py @@ -39,7 +39,7 @@ class CreateInstanceAddon: settings.update(self.config, "addon") -class AddonTest(unittest.TestCase): +class TestAddon: def test_update_settings(self): settings = BaseSettings() settings.set("KEY1", "default", priority="default") @@ -47,19 +47,19 @@ class AddonTest(unittest.TestCase): addon_config = {"KEY1": "addon", "KEY2": "addon", "KEY3": "addon"} testaddon = get_addon_cls(addon_config)() testaddon.update_settings(settings) - self.assertEqual(settings["KEY1"], "addon") - self.assertEqual(settings["KEY2"], "project") - self.assertEqual(settings["KEY3"], "addon") + assert settings["KEY1"] == "addon" + assert settings["KEY2"] == "project" + assert settings["KEY3"] == "addon" -class AddonManagerTest(unittest.TestCase): +class TestAddonManager(unittest.TestCase): def test_load_settings(self): settings_dict = { "ADDONS": {"tests.test_addons.SimpleAddon": 0}, } crawler = get_crawler(settings_dict=settings_dict) manager = crawler.addons - self.assertIsInstance(manager.addons[0], SimpleAddon) + assert isinstance(manager.addons[0], SimpleAddon) def test_notconfigured(self): class NotConfiguredAddon: @@ -71,7 +71,7 @@ class AddonManagerTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings_dict) manager = crawler.addons - self.assertFalse(manager.addons) + assert not manager.addons def test_load_settings_order(self): # Get three addons with different settings @@ -86,8 +86,8 @@ class AddonManagerTest(unittest.TestCase): settings = {"ADDONS": {a: i for i, a in enumerate(ordered_addons)}} crawler = get_crawler(settings_dict=settings) manager = crawler.addons - self.assertEqual([a.number for a in manager.addons], expected_order) - self.assertEqual(crawler.settings.getint("KEY1"), expected_order[-1]) + assert [a.number for a in manager.addons] == expected_order + assert crawler.settings.getint("KEY1") == expected_order[-1] def test_build_from_crawler(self): settings_dict = { @@ -96,8 +96,8 @@ class AddonManagerTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings_dict) manager = crawler.addons - self.assertIsInstance(manager.addons[0], CreateInstanceAddon) - self.assertEqual(crawler.settings.get("MYADDON_KEY"), "val") + assert isinstance(manager.addons[0], CreateInstanceAddon) + assert crawler.settings.get("MYADDON_KEY") == "val" def test_settings_priority(self): config = { @@ -107,14 +107,14 @@ class AddonManagerTest(unittest.TestCase): "ADDONS": {get_addon_cls(config): 1}, } crawler = get_crawler(settings_dict=settings_dict) - self.assertEqual(crawler.settings.getint("KEY"), 15) + assert crawler.settings.getint("KEY") == 15 settings = Settings(settings_dict) settings.set("KEY", 0, priority="default") runner = CrawlerRunner(settings) crawler = runner.create_crawler(Spider) crawler._apply_settings() - self.assertEqual(crawler.settings.getint("KEY"), 15) + assert crawler.settings.getint("KEY") == 15 settings_dict = { "KEY": 20, # priority=project @@ -124,7 +124,7 @@ class AddonManagerTest(unittest.TestCase): settings.set("KEY", 0, priority="default") runner = CrawlerRunner(settings) crawler = runner.create_crawler(Spider) - self.assertEqual(crawler.settings.getint("KEY"), 20) + assert crawler.settings.getint("KEY") == 20 def test_fallback_workflow(self): FALLBACK_SETTING = "MY_FALLBACK_DOWNLOAD_HANDLER" @@ -143,12 +143,12 @@ class AddonManagerTest(unittest.TestCase): "ADDONS": {AddonWithFallback: 1}, } crawler = get_crawler(settings_dict=settings_dict) - self.assertEqual( - crawler.settings.getwithbase("DOWNLOAD_HANDLERS")["https"], "AddonHandler" + assert ( + crawler.settings.getwithbase("DOWNLOAD_HANDLERS")["https"] == "AddonHandler" ) - self.assertEqual( - crawler.settings.get(FALLBACK_SETTING), - "scrapy.core.downloader.handlers.http.HTTPDownloadHandler", + assert ( + crawler.settings.get(FALLBACK_SETTING) + == "scrapy.core.downloader.handlers.http.HTTPDownloadHandler" ) settings_dict = { @@ -156,10 +156,10 @@ class AddonManagerTest(unittest.TestCase): "DOWNLOAD_HANDLERS": {"https": "UserHandler"}, } crawler = get_crawler(settings_dict=settings_dict) - self.assertEqual( - crawler.settings.getwithbase("DOWNLOAD_HANDLERS")["https"], "AddonHandler" + assert ( + crawler.settings.getwithbase("DOWNLOAD_HANDLERS")["https"] == "AddonHandler" ) - self.assertEqual(crawler.settings.get(FALLBACK_SETTING), "UserHandler") + assert crawler.settings.get(FALLBACK_SETTING) == "UserHandler" def test_logging_message(self): class LoggedAddon: @@ -199,6 +199,6 @@ class AddonManagerTest(unittest.TestCase): settings.set("KEY", "default", priority="default") runner = CrawlerRunner(settings) crawler = runner.create_crawler(MySpider) - self.assertEqual(crawler.settings.get("KEY"), "default") + assert crawler.settings.get("KEY") == "default" yield crawler.crawl() - self.assertEqual(crawler.settings.get("KEY"), "addon") + assert crawler.settings.get("KEY") == "addon" diff --git a/tests/test_closespider.py b/tests/test_closespider.py index ecde301d1..476662789 100644 --- a/tests/test_closespider.py +++ b/tests/test_closespider.py @@ -28,9 +28,9 @@ class TestCloseSpider(TestCase): crawler = get_crawler(ItemSpider, {"CLOSESPIDER_ITEMCOUNT": close_on}) yield crawler.crawl(mockserver=self.mockserver) reason = crawler.spider.meta["close_reason"] - self.assertEqual(reason, "closespider_itemcount") + assert reason == "closespider_itemcount" itemcount = crawler.stats.get_value("item_scraped_count") - self.assertTrue(itemcount >= close_on) + assert itemcount >= close_on @defer.inlineCallbacks def test_closespider_pagecount(self): @@ -38,9 +38,9 @@ class TestCloseSpider(TestCase): crawler = get_crawler(FollowAllSpider, {"CLOSESPIDER_PAGECOUNT": close_on}) yield crawler.crawl(mockserver=self.mockserver) reason = crawler.spider.meta["close_reason"] - self.assertEqual(reason, "closespider_pagecount") + assert reason == "closespider_pagecount" pagecount = crawler.stats.get_value("response_received_count") - self.assertTrue(pagecount >= close_on) + assert pagecount >= close_on @defer.inlineCallbacks def test_closespider_pagecount_no_item(self): @@ -57,10 +57,10 @@ class TestCloseSpider(TestCase): max_items=max_items, max_requests=max_requests, mockserver=self.mockserver ) reason = crawler.spider.meta["close_reason"] - self.assertEqual(reason, "closespider_pagecount_no_item") + assert reason == "closespider_pagecount_no_item" pagecount = crawler.stats.get_value("response_received_count") itemcount = crawler.stats.get_value("item_scraped_count") - self.assertLessEqual(pagecount, close_on + itemcount) + assert pagecount <= close_on + itemcount @defer.inlineCallbacks def test_closespider_pagecount_no_item_with_pagecount(self): @@ -75,9 +75,9 @@ class TestCloseSpider(TestCase): ) yield crawler.crawl(mockserver=self.mockserver) reason = crawler.spider.meta["close_reason"] - self.assertEqual(reason, "closespider_pagecount_no_item") + assert reason == "closespider_pagecount_no_item" pagecount = crawler.stats.get_value("response_received_count") - self.assertLess(pagecount, close_on_pagecount) + assert pagecount < close_on_pagecount @defer.inlineCallbacks def test_closespider_errorcount(self): @@ -85,10 +85,10 @@ class TestCloseSpider(TestCase): crawler = get_crawler(ErrorSpider, {"CLOSESPIDER_ERRORCOUNT": close_on}) yield crawler.crawl(total=1000000, mockserver=self.mockserver) reason = crawler.spider.meta["close_reason"] - self.assertEqual(reason, "closespider_errorcount") + assert reason == "closespider_errorcount" key = f"spider_exceptions/{crawler.spider.exception_cls.__name__}" errorcount = crawler.stats.get_value(key) - self.assertTrue(errorcount >= close_on) + assert errorcount >= close_on @defer.inlineCallbacks def test_closespider_timeout(self): @@ -96,9 +96,9 @@ class TestCloseSpider(TestCase): crawler = get_crawler(FollowAllSpider, {"CLOSESPIDER_TIMEOUT": close_on}) yield crawler.crawl(total=1000000, mockserver=self.mockserver) reason = crawler.spider.meta["close_reason"] - self.assertEqual(reason, "closespider_timeout") + assert reason == "closespider_timeout" total_seconds = crawler.stats.get_value("elapsed_time_seconds") - self.assertTrue(total_seconds >= close_on) + assert total_seconds >= close_on @defer.inlineCallbacks def test_closespider_timeout_no_item(self): @@ -106,6 +106,6 @@ class TestCloseSpider(TestCase): crawler = get_crawler(SlowSpider, {"CLOSESPIDER_TIMEOUT_NO_ITEM": timeout}) yield crawler.crawl(n=3, mockserver=self.mockserver) reason = crawler.spider.meta["close_reason"] - self.assertEqual(reason, "closespider_timeout_no_item") + assert reason == "closespider_timeout_no_item" total_seconds = crawler.stats.get_value("elapsed_time_seconds") - self.assertTrue(total_seconds >= timeout) + assert total_seconds >= timeout diff --git a/tests/test_cmdline/__init__.py b/tests/test_cmdline/__init__.py index acd524ea4..98a85bc17 100644 --- a/tests/test_cmdline/__init__.py +++ b/tests/test_cmdline/__init__.py @@ -4,7 +4,6 @@ import pstats import shutil import sys import tempfile -import unittest from io import StringIO from pathlib import Path from subprocess import PIPE, Popen @@ -12,8 +11,8 @@ from subprocess import PIPE, Popen from scrapy.utils.test import get_testenv -class CmdlineTest(unittest.TestCase): - def setUp(self): +class TestCmdline: + def setup_method(self): self.env = get_testenv() tests_path = Path(__file__).parent.parent self.env["PYTHONPATH"] += os.pathsep + str(tests_path.parent) @@ -27,12 +26,12 @@ class CmdlineTest(unittest.TestCase): return comm.decode(encoding) def test_default_settings(self): - self.assertEqual(self._execute("settings", "--get", "TEST1"), "default") + assert self._execute("settings", "--get", "TEST1") == "default" def test_override_settings_using_set_arg(self): - self.assertEqual( - self._execute("settings", "--get", "TEST1", "-s", "TEST1=override"), - "override", + assert ( + self._execute("settings", "--get", "TEST1", "-s", "TEST1=override") + == "override" ) def test_profiling(self): @@ -40,14 +39,14 @@ class CmdlineTest(unittest.TestCase): filename = path / "res.prof" try: self._execute("version", "--profile", str(filename)) - self.assertTrue(filename.exists()) + assert filename.exists() out = StringIO() stats = pstats.Stats(str(filename), stream=out) stats.print_stats() out.seek(0) stats = out.read() - self.assertIn(str(Path("scrapy", "commands", "version.py")), stats) - self.assertIn("tottime", stats) + assert str(Path("scrapy", "commands", "version.py")) in stats + assert "tottime" in stats finally: shutil.rmtree(path) @@ -62,15 +61,14 @@ class CmdlineTest(unittest.TestCase): "EXTENSIONS=" + json.dumps(EXTENSIONS), ) # XXX: There's gotta be a smarter way to do this... - self.assertNotIn("...", settingsstr) + assert "..." not in settingsstr for char in ("'", "<", ">"): settingsstr = settingsstr.replace(char, '"') settingsdict = json.loads(settingsstr) - self.assertCountEqual(settingsdict.keys(), EXTENSIONS.keys()) - self.assertEqual(200, settingsdict[EXT_PATH]) + assert set(settingsdict.keys()) == set(EXTENSIONS.keys()) + assert settingsdict[EXT_PATH] == 200 def test_pathlib_path_as_feeds_key(self): - self.assertEqual( - self._execute("settings", "--get", "FEEDS"), - json.dumps({"items.csv": {"format": "csv", "fields": ["price", "name"]}}), + assert self._execute("settings", "--get", "FEEDS") == json.dumps( + {"items.csv": {"format": "csv", "fields": ["price", "name"]}} ) diff --git a/tests/test_cmdline_crawl_with_pipeline/__init__.py b/tests/test_cmdline_crawl_with_pipeline/__init__.py index 5cb09b5c0..5228f6abd 100644 --- a/tests/test_cmdline_crawl_with_pipeline/__init__.py +++ b/tests/test_cmdline_crawl_with_pipeline/__init__.py @@ -1,10 +1,9 @@ import sys -import unittest from pathlib import Path from subprocess import PIPE, Popen -class CmdlineCrawlPipelineTest(unittest.TestCase): +class TestCmdlineCrawlPipeline: def _execute(self, spname): args = (sys.executable, "-m", "scrapy.cmdline", "crawl", spname) cwd = Path(__file__).resolve().parent @@ -13,7 +12,7 @@ class CmdlineCrawlPipelineTest(unittest.TestCase): return proc.returncode def test_open_spider_normally_in_pipeline(self): - self.assertEqual(self._execute("normal"), 0) + assert self._execute("normal") == 0 def test_exception_at_open_spider_in_pipeline(self): - self.assertEqual(self._execute("exception"), 1) + assert self._execute("exception") == 1 diff --git a/tests/test_command_check.py b/tests/test_command_check.py index b0f1cd38a..975f31dfe 100644 --- a/tests/test_command_check.py +++ b/tests/test_command_check.py @@ -3,10 +3,10 @@ from io import StringIO from unittest.mock import Mock, PropertyMock, call, patch from scrapy.commands.check import Command, TextTestResult -from tests.test_commands import CommandTest +from tests.test_commands import TestCommandBase -class CheckCommandTest(CommandTest): +class TestCheckCommand(TestCommandBase): command = "check" def setUp(self): @@ -36,9 +36,9 @@ class CheckSpider(scrapy.Spider): def _test_contract(self, contracts="", parse_def="pass"): self._write_contract(contracts, parse_def) p, out, err = self.proc("check") - self.assertNotIn("F", out) - self.assertIn("OK", err) - self.assertEqual(p.returncode, 0) + assert "F" not in out + assert "OK" in err + assert p.returncode == 0 def test_check_returns_requests_contract(self): contracts = """ @@ -171,9 +171,7 @@ class CheckSpider(scrapy.Spider): cmd.run([spider_name], Mock(list=True)) - self.assertEqual( - "FakeSpider\n * fakeMethod1\n * fakeMethod2\n", output.getvalue() - ) + assert output.getvalue() == "FakeSpider\n * fakeMethod1\n * fakeMethod2\n" sys.stdout = sys.__stdout__ @patch("scrapy.commands.check.ContractsManager") diff --git a/tests/test_command_fetch.py b/tests/test_command_fetch.py index a4d7fdd30..a31cada85 100644 --- a/tests/test_command_fetch.py +++ b/tests/test_command_fetch.py @@ -5,18 +5,18 @@ from tests.utils.testproc import ProcessTest from tests.utils.testsite import SiteTest -class FetchTest(ProcessTest, SiteTest, unittest.TestCase): +class TestFetchCommand(ProcessTest, SiteTest, unittest.TestCase): command = "fetch" @defer.inlineCallbacks def test_output(self): _, out, _ = yield self.execute([self.url("/text")]) - self.assertEqual(out.strip(), b"Works") + assert out.strip() == b"Works" @defer.inlineCallbacks def test_redirect_default(self): _, out, _ = yield self.execute([self.url("/redirect")]) - self.assertEqual(out.strip(), b"Redirected here") + assert out.strip() == b"Redirected here" @defer.inlineCallbacks def test_redirect_disabled(self): @@ -24,8 +24,8 @@ class FetchTest(ProcessTest, SiteTest, unittest.TestCase): ["--no-redirect", self.url("/redirect-no-meta-refresh")] ) err = err.strip() - self.assertIn(b"downloader/response_status_count/302", err, err) - self.assertNotIn(b"downloader/response_status_count/200", err, err) + assert b"downloader/response_status_count/302" in err, err + assert b"downloader/response_status_count/200" not in err, err @defer.inlineCallbacks def test_headers(self): diff --git a/tests/test_command_parse.py b/tests/test_command_parse.py index 9f2c7fa13..9e66d319c 100644 --- a/tests/test_command_parse.py +++ b/tests/test_command_parse.py @@ -1,5 +1,6 @@ import argparse import os +import re from pathlib import Path from twisted.internet import defer @@ -7,18 +8,18 @@ from twisted.internet import defer from scrapy.commands import parse from scrapy.settings import Settings from scrapy.utils.python import to_unicode -from tests.test_commands import CommandTest +from tests.test_commands import TestCommandBase from tests.utils.testproc import ProcessTest from tests.utils.testsite import SiteTest -def _textmode(bstr): +def _textmode(bstr: bytes) -> str: """Normalize input the same as writing to a file and reading from it in text mode""" return to_unicode(bstr).replace(os.linesep, "\n") -class ParseCommandTest(ProcessTest, SiteTest, CommandTest): +class TestParseCommand(ProcessTest, SiteTest, TestCommandBase): command = "parse" def setUp(self): @@ -184,7 +185,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("DEBUG: It Works!", _textmode(stderr)) + assert "DEBUG: It Works!" in _textmode(stderr) @defer.inlineCallbacks def test_request_with_meta(self): @@ -201,7 +202,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("DEBUG: It Works!", _textmode(stderr)) + assert "DEBUG: It Works!" in _textmode(stderr) _, _, stderr = yield self.execute( [ @@ -215,7 +216,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("DEBUG: It Works!", _textmode(stderr)) + assert "DEBUG: It Works!" in _textmode(stderr) @defer.inlineCallbacks def test_request_with_cb_kwargs(self): @@ -233,9 +234,9 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} ] ) log = _textmode(stderr) - self.assertIn("DEBUG: It Works!", log) - self.assertIn( - "DEBUG: request.callback signature: (response, foo=None, key=None)", log + assert "DEBUG: It Works!" in log + assert ( + "DEBUG: request.callback signature: (response, foo=None, key=None)" in log ) @defer.inlineCallbacks @@ -250,7 +251,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("DEBUG: It Works!", _textmode(stderr)) + assert "DEBUG: It Works!" in _textmode(stderr) @defer.inlineCallbacks def test_pipelines(self): @@ -265,7 +266,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("INFO: It Works!", _textmode(stderr)) + assert "INFO: It Works!" in _textmode(stderr) @defer.inlineCallbacks def test_async_def_asyncio_parse_items_list(self): @@ -278,9 +279,9 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("INFO: Got response 200", _textmode(stderr)) - self.assertIn("{'id': 1}", _textmode(out)) - self.assertIn("{'id': 2}", _textmode(out)) + assert "INFO: Got response 200" in _textmode(stderr) + assert "{'id': 1}" in _textmode(out) + assert "{'id': 2}" in _textmode(out) @defer.inlineCallbacks def test_async_def_asyncio_parse_items_single_element(self): @@ -293,8 +294,8 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("INFO: Got response 200", _textmode(stderr)) - self.assertIn("{'foo': 42}", _textmode(out)) + assert "INFO: Got response 200" in _textmode(stderr) + assert "{'foo': 42}" in _textmode(out) @defer.inlineCallbacks def test_async_def_asyncgen_parse_loop(self): @@ -307,9 +308,9 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("INFO: Got response 200", _textmode(stderr)) + assert "INFO: Got response 200" in _textmode(stderr) for i in range(10): - self.assertIn(f"{{'foo': {i}}}", _textmode(out)) + assert f"{{'foo': {i}}}" in _textmode(out) @defer.inlineCallbacks def test_async_def_asyncgen_parse_exc(self): @@ -322,9 +323,9 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("ValueError", _textmode(stderr)) + assert "ValueError" in _textmode(stderr) for i in range(7): - self.assertIn(f"{{'foo': {i}}}", _textmode(out)) + assert f"{{'foo': {i}}}" in _textmode(out) @defer.inlineCallbacks def test_async_def_asyncio_parse(self): @@ -337,29 +338,29 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} self.url("/html"), ] ) - self.assertIn("DEBUG: Got response 200", _textmode(stderr)) + assert "DEBUG: Got response 200" in _textmode(stderr) @defer.inlineCallbacks def test_parse_items(self): status, out, stderr = yield self.execute( ["--spider", self.spider_name, "-c", "parse", self.url("/html")] ) - self.assertIn("""[{}, {'foo': 'bar'}]""", _textmode(out)) + assert "[{}, {'foo': 'bar'}]" in _textmode(out) @defer.inlineCallbacks def test_parse_items_no_callback_passed(self): status, out, stderr = yield self.execute( ["--spider", self.spider_name, self.url("/html")] ) - self.assertIn("""[{}, {'foo': 'bar'}]""", _textmode(out)) + assert "[{}, {'foo': 'bar'}]" in _textmode(out) @defer.inlineCallbacks def test_wrong_callback_passed(self): status, out, stderr = yield self.execute( ["--spider", self.spider_name, "-c", "dummy", self.url("/html")] ) - self.assertRegex(_textmode(out), r"""# Scraped Items -+\n\[\]""") - self.assertIn("""Cannot find callback""", _textmode(stderr)) + assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out)) + assert "Cannot find callback" in _textmode(stderr) @defer.inlineCallbacks def test_crawlspider_matching_rule_callback_set(self): @@ -367,7 +368,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} status, out, stderr = yield self.execute( ["--spider", "goodcrawl" + self.spider_name, "-r", self.url("/html")] ) - self.assertIn("""[{}, {'foo': 'bar'}]""", _textmode(out)) + assert "[{}, {'foo': 'bar'}]" in _textmode(out) @defer.inlineCallbacks def test_crawlspider_matching_rule_default_callback(self): @@ -375,7 +376,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} status, out, stderr = yield self.execute( ["--spider", "goodcrawl" + self.spider_name, "-r", self.url("/text")] ) - self.assertIn("""[{}, {'nomatch': 'default'}]""", _textmode(out)) + assert "[{}, {'nomatch': 'default'}]" in _textmode(out) @defer.inlineCallbacks def test_spider_with_no_rules_attribute(self): @@ -383,15 +384,15 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} status, out, stderr = yield self.execute( ["--spider", self.spider_name, "-r", self.url("/html")] ) - self.assertRegex(_textmode(out), r"""# Scraped Items -+\n\[\]""") - self.assertIn("""No CrawlSpider rules found""", _textmode(stderr)) + assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out)) + assert "No CrawlSpider rules found" in _textmode(stderr) @defer.inlineCallbacks def test_crawlspider_missing_callback(self): status, out, stderr = yield self.execute( ["--spider", "badcrawl" + self.spider_name, "-r", self.url("/html")] ) - self.assertRegex(_textmode(out), r"""# Scraped Items -+\n\[\]""") + assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out)) @defer.inlineCallbacks def test_crawlspider_no_matching_rule(self): @@ -399,13 +400,13 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} status, out, stderr = yield self.execute( ["--spider", "badcrawl" + self.spider_name, "-r", self.url("/enc-gb18030")] ) - self.assertRegex(_textmode(out), r"""# Scraped Items -+\n\[\]""") - self.assertIn("""Cannot find a rule that matches""", _textmode(stderr)) + assert re.search(r"# Scraped Items -+\n\[\]", _textmode(out)) + assert "Cannot find a rule that matches" in _textmode(stderr) @defer.inlineCallbacks def test_crawlspider_not_exists_with_not_matched_url(self): status, out, stderr = yield self.execute([self.url("/invalid_url")]) - self.assertEqual(status, 0) + assert status == 0 @defer.inlineCallbacks def test_output_flag(self): @@ -426,11 +427,11 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} ] ) - self.assertTrue(file_path.exists()) - self.assertTrue(file_path.is_file()) + assert file_path.exists() + assert file_path.is_file() content = '[\n{},\n{"foo": "bar"}\n]' - self.assertEqual(file_path.read_text(encoding="utf-8"), content) + assert file_path.read_text(encoding="utf-8") == content def test_parse_add_options(self): command = parse.Command() @@ -445,7 +446,7 @@ ITEM_PIPELINES = {{'{self.project_name}.pipelines.MyPipeline': 1}} namespace = parser.parse_args( ["--verbose", "--nolinks", "-d", "2", "--spider", self.spider_name] ) - self.assertTrue(namespace.nolinks) - self.assertEqual(namespace.depth, 2) - self.assertEqual(namespace.spider, self.spider_name) - self.assertTrue(namespace.verbose) + assert namespace.nolinks + assert namespace.depth == 2 + assert namespace.spider == self.spider_name + assert namespace.verbose diff --git a/tests/test_command_shell.py b/tests/test_command_shell.py index 9ca5e05dc..0f45a7ee8 100644 --- a/tests/test_command_shell.py +++ b/tests/test_command_shell.py @@ -13,7 +13,7 @@ from tests.utils.testproc import ProcessTest from tests.utils.testsite import SiteTest -class ShellTest(ProcessTest, SiteTest, unittest.TestCase): +class TestShellCommand(ProcessTest, SiteTest, unittest.TestCase): command = "shell" @defer.inlineCallbacks @@ -40,14 +40,14 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase): def test_response_selector_html(self): xpath = "response.xpath(\"//p[@class='one']/text()\").get()" _, out, _ = yield self.execute([self.url("/html"), "-c", xpath]) - self.assertEqual(out.strip(), b"Works") + assert out.strip() == b"Works" @defer.inlineCallbacks def test_response_encoding_gb18030(self): _, out, _ = yield self.execute( [self.url("/enc-gb18030"), "-c", "response.encoding"] ) - self.assertEqual(out.strip(), b"gb18030") + assert out.strip() == b"gb18030" @defer.inlineCallbacks def test_redirect(self): @@ -79,7 +79,7 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase): url = self.url("/redirect-no-meta-refresh") code = f"fetch('{url}')" errcode, out, errout = yield self.execute(["-c", code]) - self.assertEqual(errcode, 0, out) + assert errcode == 0, out assert b"Redirecting (302)" in errout assert b"Crawled (200)" in errout @@ -89,7 +89,7 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase): url = self.url("/redirect-no-meta-refresh") code = f"fetch('{url}', redirect=False)" errcode, out, errout = yield self.execute(["-c", code]) - self.assertEqual(errcode, 0, out) + assert errcode == 0, out assert b"Crawled (302)" in errout @defer.inlineCallbacks @@ -97,14 +97,14 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase): url = self.url("/text") code = f"fetch('{url}') or fetch(response.request.replace(method='POST'))" errcode, out, _ = yield self.execute(["-c", code]) - self.assertEqual(errcode, 0, out) + assert errcode == 0, out @defer.inlineCallbacks def test_scrapy_import(self): url = self.url("/text") code = f"fetch(scrapy.Request('{url}'))" errcode, out, _ = yield self.execute(["-c", code]) - self.assertEqual(errcode, 0, out) + assert errcode == 0, out @defer.inlineCallbacks def test_local_file(self): @@ -118,8 +118,8 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase): errcode, out, err = yield self.execute( [filepath, "-c", "item"], check_code=False ) - self.assertEqual(errcode, 1, out or err) - self.assertIn(b"No such file or directory", err) + assert errcode == 1, out or err + assert b"No such file or directory" in err @defer.inlineCallbacks def test_dns_failures(self): @@ -127,8 +127,8 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase): raise unittest.SkipTest("Non-existing hosts are resolvable") url = "www.somedomainthatdoesntexi.st" errcode, out, err = yield self.execute([url, "-c", "item"], check_code=False) - self.assertEqual(errcode, 1, out or err) - self.assertIn(b"DNS lookup failed", err) + assert errcode == 1, out or err + assert b"DNS lookup failed" in err @defer.inlineCallbacks def test_shell_fetch_async(self): @@ -137,10 +137,10 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase): code = f"fetch('{url}')" args = ["-c", code, "--set", f"TWISTED_REACTOR={reactor_path}"] _, _, err = yield self.execute(args, check_code=True) - self.assertNotIn(b"RuntimeError: There is no current event loop in thread", err) + assert b"RuntimeError: There is no current event loop in thread" not in err -class InteractiveShellTest(unittest.TestCase): +class TestInteractiveShell: def test_fetch(self): args = ( sys.executable, @@ -161,4 +161,4 @@ class InteractiveShellTest(unittest.TestCase): p.sendeof() p.wait() logfile.seek(0) - self.assertNotIn("Traceback", logfile.read().decode()) + assert "Traceback" not in logfile.read().decode() diff --git a/tests/test_command_version.py b/tests/test_command_version.py index 917f457cb..a61a6a32b 100644 --- a/tests/test_command_version.py +++ b/tests/test_command_version.py @@ -7,17 +7,14 @@ import scrapy from tests.utils.testproc import ProcessTest -class VersionTest(ProcessTest, unittest.TestCase): +class TestVersionCommand(ProcessTest, unittest.TestCase): command = "version" @defer.inlineCallbacks def test_output(self): encoding = sys.stdout.encoding or "utf-8" _, out, _ = yield self.execute([]) - self.assertEqual( - out.strip().decode(encoding), - f"Scrapy {scrapy.__version__}", - ) + assert out.strip().decode(encoding) == f"Scrapy {scrapy.__version__}" @defer.inlineCallbacks def test_verbose_output(self): @@ -27,19 +24,16 @@ class VersionTest(ProcessTest, unittest.TestCase): line.partition(":")[0].strip() for line in out.strip().decode(encoding).splitlines() ] - self.assertEqual( - headers, - [ - "Scrapy", - "lxml", - "libxml2", - "cssselect", - "parsel", - "w3lib", - "Twisted", - "Python", - "pyOpenSSL", - "cryptography", - "Platform", - ], - ) + assert headers == [ + "Scrapy", + "lxml", + "libxml2", + "cssselect", + "parsel", + "w3lib", + "Twisted", + "Python", + "pyOpenSSL", + "cryptography", + "Platform", + ] diff --git a/tests/test_commands.py b/tests/test_commands.py index 1a0db1e03..f63e05628 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -14,7 +14,7 @@ from itertools import chain from pathlib import Path from shutil import copytree, rmtree from stat import S_IWRITE as ANYONE_WRITE_PERMISSION -from tempfile import TemporaryFile, mkdtemp +from tempfile import TemporaryDirectory, TemporaryFile, mkdtemp from threading import Timer from typing import TYPE_CHECKING from unittest import mock, skipIf @@ -35,8 +35,8 @@ if TYPE_CHECKING: from collections.abc import Iterator -class CommandSettings(unittest.TestCase): - def setUp(self): +class TestCommandSettings: + def setup_method(self): self.command = ScrapyCommand() self.command.settings = Settings() self.parser = argparse.ArgumentParser( @@ -50,10 +50,8 @@ class CommandSettings(unittest.TestCase): args=["-s", f"FEEDS={feeds_json}", "spider.py"] ) self.command.process_options(args, opts) - self.assertIsInstance( - self.command.settings["FEEDS"], scrapy.settings.BaseSettings - ) - self.assertEqual(dict(self.command.settings["FEEDS"]), json.loads(feeds_json)) + assert isinstance(self.command.settings["FEEDS"], scrapy.settings.BaseSettings) + assert dict(self.command.settings["FEEDS"]) == json.loads(feeds_json) def test_help_formatter(self): formatter = ScrapyHelpFormatter(prog="scrapy") @@ -64,17 +62,14 @@ class CommandSettings(unittest.TestCase): "\n", "Global Options:\n", ] - self.assertEqual( - formatter._join_parts(part_strings), - ( - "Usage\n=====\n scrapy genspider [options] \n\n\n" - "Optional Arguments\n==================\n\n" - "Global Options\n--------------\n" - ), + assert formatter._join_parts(part_strings) == ( + "Usage\n=====\n scrapy genspider [options] \n\n\n" + "Optional Arguments\n==================\n\n" + "Global Options\n--------------\n" ) -class ProjectTest(unittest.TestCase): +class TestProjectBase(unittest.TestCase): project_name = "testproject" def setUp(self): @@ -130,12 +125,12 @@ class ProjectTest(unittest.TestCase): return None -class StartprojectTest(ProjectTest): +class TestStartprojectCommand(TestProjectBase): def test_startproject(self): p, out, err = self.proc("startproject", self.project_name) print(out) print(err, file=sys.stderr) - self.assertEqual(p.returncode, 0) + assert p.returncode == 0 assert Path(self.proj_path, "scrapy.cfg").exists() assert Path(self.proj_path, "testproject").exists() @@ -145,13 +140,13 @@ class StartprojectTest(ProjectTest): assert Path(self.proj_mod_path, "settings.py").exists() assert Path(self.proj_mod_path, "spiders", "__init__.py").exists() - self.assertEqual(1, self.call("startproject", self.project_name)) - self.assertEqual(1, self.call("startproject", "wrong---project---name")) - self.assertEqual(1, self.call("startproject", "sys")) + 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_with_project_dir(self): project_dir = mkdtemp() - self.assertEqual(0, self.call("startproject", self.project_name, project_dir)) + assert self.call("startproject", self.project_name, project_dir) == 0 assert Path(project_dir, "scrapy.cfg").exists() assert Path(project_dir, "testproject").exists() @@ -161,20 +156,16 @@ class StartprojectTest(ProjectTest): assert Path(project_dir, self.project_name, "settings.py").exists() assert Path(project_dir, self.project_name, "spiders", "__init__.py").exists() - self.assertEqual( - 0, self.call("startproject", self.project_name, project_dir + "2") - ) + assert self.call("startproject", self.project_name, project_dir + "2") == 0 - self.assertEqual(1, self.call("startproject", self.project_name, project_dir)) - self.assertEqual( - 1, self.call("startproject", self.project_name + "2", project_dir) - ) - self.assertEqual(1, self.call("startproject", "wrong---project---name")) - self.assertEqual(1, self.call("startproject", "sys")) - self.assertEqual(2, self.call("startproject")) - self.assertEqual( - 2, - self.call("startproject", self.project_name, project_dir, "another_params"), + 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 + assert ( + self.call("startproject", self.project_name, project_dir, "another_params") + == 2 ) def test_existing_project_dir(self): @@ -186,7 +177,7 @@ class StartprojectTest(ProjectTest): p, out, err = self.proc("startproject", project_name, cwd=project_dir) print(out) print(err, file=sys.stderr) - self.assertEqual(p.returncode, 0) + assert p.returncode == 0 assert Path(project_path, "scrapy.cfg").exists() assert Path(project_path, project_name).exists() @@ -224,7 +215,7 @@ def get_permissions_dict( return permissions_dict -class StartprojectTemplatesTest(ProjectTest): +class TestStartprojectTemplates(TestProjectBase): maxDiff = None def setUp(self): @@ -239,11 +230,10 @@ class StartprojectTemplatesTest(ProjectTest): args = ["--set", f"TEMPLATES_DIR={self.tmpl}"] p, out, err = self.proc("startproject", self.project_name, *args) - self.assertIn( - f"New Scrapy project '{self.project_name}', using template directory", - out, + assert ( + f"New Scrapy project '{self.project_name}', using template directory" in out ) - self.assertIn(self.tmpl_proj, out) + assert self.tmpl_proj in out assert Path(self.proj_path, "root_template").exists() def test_startproject_permissions_from_writable(self): @@ -280,7 +270,7 @@ class StartprojectTemplatesTest(ProjectTest): project_dir = Path(destination, project_name) actual_permissions = get_permissions_dict(project_dir) - self.assertEqual(actual_permissions, expected_permissions) + assert actual_permissions == expected_permissions def test_startproject_permissions_from_read_only(self): """Check that generated files have the right permissions when the @@ -333,7 +323,7 @@ class StartprojectTemplatesTest(ProjectTest): project_dir = Path(destination, project_name) actual_permissions = get_permissions_dict(project_dir) - self.assertEqual(actual_permissions, expected_permissions) + assert actual_permissions == expected_permissions def test_startproject_permissions_unchanged_in_destination(self): """Check that preexisting folders and files in the destination folder @@ -391,7 +381,7 @@ class StartprojectTemplatesTest(ProjectTest): actual_permissions = get_permissions_dict(project_dir) - self.assertEqual(actual_permissions, expected_permissions) + assert actual_permissions == expected_permissions def test_startproject_permissions_umask_022(self): """Check that generated files have the right permissions when the @@ -435,10 +425,10 @@ class StartprojectTemplatesTest(ProjectTest): project_dir = Path(destination, project_name) actual_permissions = get_permissions_dict(project_dir) - self.assertEqual(actual_permissions, expected_permissions) + assert actual_permissions == expected_permissions -class CommandTest(ProjectTest): +class TestCommandBase(TestProjectBase): def setUp(self): super().setUp() self.call("startproject", self.project_name) @@ -446,13 +436,13 @@ class CommandTest(ProjectTest): self.env["SCRAPY_SETTINGS_MODULE"] = f"{self.project_name}.settings" -class GenspiderCommandTest(CommandTest): +class TestGenspiderCommand(TestCommandBase): def test_arguments(self): # only pass one argument. spider script shouldn't be created - self.assertEqual(2, self.call("genspider", "test_name")) + assert self.call("genspider", "test_name") == 2 assert not Path(self.proj_mod_path, "spiders", "test_name.py").exists() # pass two arguments . spider script should be created - self.assertEqual(0, self.call("genspider", "test_name", "test.com")) + assert self.call("genspider", "test_name", "test.com") == 0 assert Path(self.proj_mod_path, "spiders", "test_name.py").exists() def test_template(self, tplname="crawl"): @@ -460,20 +450,20 @@ class GenspiderCommandTest(CommandTest): spname = "test_spider" spmodule = f"{self.project_name}.spiders.{spname}" p, out, err = self.proc("genspider", spname, "test.com", *args) - self.assertIn( - f"Created spider {spname!r} using template {tplname!r} in module:{os.linesep} {spmodule}", - out, + assert ( + f"Created spider {spname!r} using template {tplname!r} in module:{os.linesep} {spmodule}" + in out ) - self.assertTrue(Path(self.proj_mod_path, "spiders", "test_spider.py").exists()) + 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) - self.assertIn(f"Spider {spname!r} already exists in module", out) + 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 ) - self.assertEqual(modify_time_after, modify_time_before) + assert modify_time_after == modify_time_before def test_template_basic(self): self.test_template("basic") @@ -485,14 +475,14 @@ class GenspiderCommandTest(CommandTest): self.test_template("xmlfeed") def test_list(self): - self.assertEqual(0, self.call("genspider", "--list")) + assert self.call("genspider", "--list") == 0 def test_dump(self): - self.assertEqual(0, self.call("genspider", "--dump=basic")) - self.assertEqual(0, self.call("genspider", "-d", "basic")) + assert self.call("genspider", "--dump=basic") == 0 + assert self.call("genspider", "-d", "basic") == 0 def test_same_name_as_project(self): - self.assertEqual(2, self.call("genspider", self.project_name)) + assert self.call("genspider", self.project_name) == 2 assert not Path( self.proj_mod_path, "spiders", f"{self.project_name}.py" ).exists() @@ -500,7 +490,7 @@ class GenspiderCommandTest(CommandTest): def test_same_filename_as_existing_spider(self, force=False): file_name = "example" file_path = Path(self.proj_mod_path, "spiders", f"{file_name}.py") - self.assertEqual(0, self.call("genspider", file_name, "example.com")) + assert self.call("genspider", file_name, "example.com") == 0 assert file_path.exists() # change name of spider but not its file name @@ -515,39 +505,39 @@ class GenspiderCommandTest(CommandTest): if force: p, out, err = self.proc("genspider", "--force", file_name, "example.com") - self.assertIn( - f"Created spider {file_name!r} using template 'basic' in module", out + assert ( + f"Created spider {file_name!r} using template 'basic' in module" in out ) modify_time_after = file_path.stat().st_mtime - self.assertNotEqual(modify_time_after, modify_time_before) + assert modify_time_after != modify_time_before file_contents_after = file_path.read_text(encoding="utf-8") - self.assertNotEqual(file_contents_after, file_contents_before) + assert file_contents_after != file_contents_before else: p, out, err = self.proc("genspider", file_name, "example.com") - self.assertIn(f"{file_path.resolve()} already exists", out) + assert f"{file_path.resolve()} already exists" in out modify_time_after = file_path.stat().st_mtime - self.assertEqual(modify_time_after, modify_time_before) + assert modify_time_after == modify_time_before file_contents_after = file_path.read_text(encoding="utf-8") - self.assertEqual(file_contents_after, file_contents_before) + assert file_contents_after == file_contents_before def test_same_filename_as_existing_spider_force(self): self.test_same_filename_as_existing_spider(force=True) def test_url(self, url="test.com", domain="test.com"): - self.assertEqual(0, self.call("genspider", "--force", "test_name", url)) - self.assertEqual( - domain, + assert self.call("genspider", "--force", "test_name", url) == 0 + assert ( self.find_in_file( Path(self.proj_mod_path, "spiders", "test_name.py"), r"allowed_domains\s*=\s*\[['\"](.+)['\"]\]", - ).group(1), + ).group(1) + == domain ) - self.assertEqual( - f"https://{domain}", + assert ( self.find_in_file( Path(self.proj_mod_path, "spiders", "test_name.py"), r"start_urls\s*=\s*\[['\"](.+)['\"]\]", - ).group(1), + ).group(1) + == f"https://{domain}" ) def test_url_schema(self): @@ -556,15 +546,13 @@ class GenspiderCommandTest(CommandTest): def test_template_start_urls( self, url="test.com", expected="https://test.com", template="basic" ): - self.assertEqual( - 0, self.call("genspider", "-t", template, "--force", "test_name", url) - ) - self.assertEqual( - expected, + assert self.call("genspider", "-t", template, "--force", "test_name", url) == 0 + assert ( self.find_in_file( Path(self.proj_mod_path, "spiders", "test_name.py"), r"start_urls\s*=\s*\[['\"](.+)['\"]\]", - ).group(1), + ).group(1) + == expected ) def test_genspider_basic_start_urls(self): @@ -611,7 +599,7 @@ class GenspiderCommandTest(CommandTest): ) -class GenspiderStandaloneCommandTest(ProjectTest): +class TestGenspiderStandaloneCommand(TestProjectBase): def test_generate_standalone_spider(self): self.call("genspider", "example", "example.com") assert Path(self.temp_path, "example.py").exists() @@ -620,7 +608,7 @@ class GenspiderStandaloneCommandTest(ProjectTest): file_name = "example" file_path = Path(self.temp_path, file_name + ".py") p, out, err = self.proc("genspider", file_name, "example.com") - self.assertIn(f"Created spider {file_name!r} using template 'basic' ", out) + assert f"Created spider {file_name!r} using template 'basic' " in out assert file_path.exists() modify_time_before = file_path.stat().st_mtime file_contents_before = file_path.read_text(encoding="utf-8") @@ -630,29 +618,29 @@ class GenspiderStandaloneCommandTest(ProjectTest): p, out, err = self.proc( "genspider", "--force", "-t", "crawl", file_name, "example.com" ) - self.assertIn(f"Created spider {file_name!r} using template 'crawl' ", out) + assert f"Created spider {file_name!r} using template 'crawl' " in out modify_time_after = file_path.stat().st_mtime - self.assertNotEqual(modify_time_after, modify_time_before) + assert modify_time_after != modify_time_before file_contents_after = file_path.read_text(encoding="utf-8") - self.assertNotEqual(file_contents_after, file_contents_before) + assert file_contents_after != file_contents_before else: p, out, err = self.proc("genspider", file_name, "example.com") - self.assertIn( - f"{Path(self.temp_path, file_name + '.py').resolve()} already exists", - out, + assert ( + f"{Path(self.temp_path, file_name + '.py').resolve()} already exists" + in out ) modify_time_after = file_path.stat().st_mtime - self.assertEqual(modify_time_after, modify_time_before) + assert modify_time_after == modify_time_before file_contents_after = file_path.read_text(encoding="utf-8") - self.assertEqual(file_contents_after, file_contents_before) + assert file_contents_after == file_contents_before def test_same_name_as_existing_file_force(self): self.test_same_name_as_existing_file(force=True) -class MiscCommandsTest(CommandTest): +class TestMiscCommands(TestCommandBase): def test_list(self): - self.assertEqual(0, self.call("list")) + assert self.call("list") == 0 def test_command_not_found(self): na_msg = """ @@ -670,10 +658,10 @@ Unknown command: abc for cmdname, inproject, message in params: with mock.patch("sys.stdout", new=StringIO()) as out: _print_unknown_command_msg(Settings(), cmdname, inproject) - self.assertEqual(out.getvalue().strip(), message.strip()) + assert out.getvalue().strip() == message.strip() -class RunSpiderCommandTest(CommandTest): +class TestRunSpiderCommand(TestCommandBase): spider_filename = "myspider.py" debug_log_spider = """ @@ -697,18 +685,14 @@ class BadSpider(scrapy.Spider): """ @contextmanager - def _create_file(self, content, name=None) -> Iterator[str]: - tmpdir = Path(self.mktemp()) - tmpdir.mkdir() - if name: - fname = (tmpdir / name).resolve() - else: - fname = (tmpdir / self.spider_filename).resolve() - fname.write_text(content, encoding="utf-8") - try: + 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) - finally: - rmtree(tmpdir) def runspider(self, code, name=None, args=()): with self._create_file(code, name) as fname: @@ -720,29 +704,29 @@ class BadSpider(scrapy.Spider): def test_runspider(self): log = self.get_log(self.debug_log_spider) - self.assertIn("DEBUG: It Works!", log) - self.assertIn("INFO: Spider opened", log) - self.assertIn("INFO: Closing spider (finished)", log) - self.assertIn("INFO: Spider closed (finished)", log) + assert "DEBUG: It Works!" in log + assert "INFO: Spider opened" in log + assert "INFO: Closing spider (finished)" in log + assert "INFO: Spider closed (finished)" in log def test_run_fail_spider(self): proc, _, _ = self.runspider( "import scrapy\n" + inspect.getsource(ExceptionSpider) ) ret = proc.returncode - self.assertNotEqual(ret, 0) + assert ret != 0 def test_run_good_spider(self): proc, _, _ = self.runspider( "import scrapy\n" + inspect.getsource(NoRequestsSpider) ) ret = proc.returncode - self.assertEqual(ret, 0) + assert ret == 0 def test_runspider_log_level(self): log = self.get_log(self.debug_log_spider, args=("-s", "LOG_LEVEL=INFO")) - self.assertNotIn("DEBUG: It Works!", log) - self.assertIn("INFO: Spider opened", log) + assert "DEBUG: It Works!" not in log + assert "INFO: Spider opened" in log def test_runspider_dnscache_disabled(self): # see https://github.com/scrapy/scrapy/issues/2811 @@ -761,36 +745,36 @@ class MySpider(scrapy.Spider): return {'test': 'value'} """ log = self.get_log(dnscache_spider, args=("-s", "DNSCACHE_ENABLED=False")) - self.assertNotIn("DNSLookupError", log) - self.assertIn("INFO: Spider opened", log) + 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")) - self.assertIn("[myspider] DEBUG: It Works!", log1) - self.assertIn("[scrapy]", log1) - self.assertNotIn("[scrapy.core.engine]", log1) + assert "[myspider] DEBUG: It Works!" in log1 + assert "[scrapy]" in log1 + assert "[scrapy.core.engine]" not in log1 log2 = self.get_log(self.debug_log_spider, args=("-s", "LOG_SHORT_NAMES=0")) - self.assertIn("[myspider] DEBUG: It Works!", log2) - self.assertNotIn("[scrapy]", log2) - self.assertIn("[scrapy.core.engine]", log2) + 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") - self.assertIn("No spider found in file", log) + assert "No spider found in file" in log def test_runspider_file_not_found(self): _, _, log = self.proc("runspider", "some_non_existent_file") - self.assertIn("File not found: some_non_existent_file", log) + assert "File not found: some_non_existent_file" in log def test_runspider_unable_to_load(self): log = self.get_log("", name="myspider.txt") - self.assertIn("Unable to load", log) + assert "Unable to load" in log def test_start_requests_errors(self): log = self.get_log(self.badspider, name="badspider.py") - self.assertIn("start_requests", log) - self.assertIn("badspider.py", log) + assert "start_requests" in log + assert "badspider.py" in log def test_asyncio_enabled_true(self): log = self.get_log( @@ -800,14 +784,16 @@ class MySpider(scrapy.Spider): "TWISTED_REACTOR=twisted.internet.asyncioreactor.AsyncioSelectorReactor", ], ) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) def test_asyncio_enabled_default(self): log = self.get_log(self.debug_log_spider, args=[]) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) def test_asyncio_enabled_false(self): @@ -815,11 +801,10 @@ class MySpider(scrapy.Spider): self.debug_log_spider, args=["-s", "TWISTED_REACTOR=twisted.internet.selectreactor.SelectReactor"], ) - self.assertIn( - "Using reactor: twisted.internet.selectreactor.SelectReactor", log - ) - self.assertNotIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Using reactor: twisted.internet.selectreactor.SelectReactor" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + not in log ) @pytest.mark.requires_uvloop @@ -833,7 +818,7 @@ class MySpider(scrapy.Spider): "ASYNCIO_EVENT_LOOP=uvloop.Loop", ], ) - self.assertIn("Using asyncio event loop: uvloop.Loop", log) + assert "Using asyncio event loop: uvloop.Loop" in log def test_custom_asyncio_loop_enabled_false(self): log = self.get_log( @@ -849,9 +834,9 @@ class MySpider(scrapy.Spider): loop = asyncio.new_event_loop() else: loop = asyncio.SelectorEventLoop() - self.assertIn( - f"Using asyncio event loop: {loop.__module__}.{loop.__class__.__name__}", - log, + assert ( + f"Using asyncio event loop: {loop.__module__}.{loop.__class__.__name__}" + in log ) def test_output(self): @@ -867,9 +852,7 @@ class MySpider(scrapy.Spider): """ args = ["-o", "example.json"] log = self.get_log(spider_code, args=args) - self.assertIn( - "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}", log - ) + assert "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}" in log def test_overwrite_output(self): spider_code = """ @@ -890,13 +873,13 @@ class MySpider(scrapy.Spider): Path(self.cwd, "example.json").write_text("not empty", encoding="utf-8") args = ["-O", "example.json"] log = self.get_log(spider_code, args=args) - self.assertIn( - '[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}', - log, + assert ( + '[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}' + in log ) with Path(self.cwd, "example.json").open(encoding="utf-8") as f2: first_line = f2.readline() - self.assertNotEqual(first_line, "not empty") + assert first_line != "not empty" def test_output_and_overwrite_output(self): spider_code = """ @@ -910,8 +893,8 @@ class MySpider(scrapy.Spider): """ args = ["-o", "example1.json", "-O", "example2.json"] log = self.get_log(spider_code, args=args) - self.assertIn( - "error: Please use only one of -o/--output and -O/--overwrite-output", log + assert ( + "error: Please use only one of -o/--output and -O/--overwrite-output" in log ) def test_output_stdout(self): @@ -927,7 +910,7 @@ class MySpider(scrapy.Spider): """ args = ["-o", "-:json"] log = self.get_log(spider_code, args=args) - self.assertIn("[myspider] DEBUG: FEEDS: {'stdout:': {'format': 'json'}}", log) + assert "[myspider] DEBUG: FEEDS: {'stdout:': {'format': 'json'}}" in log @skipIf(platform.system() == "Windows", reason="Linux only") def test_absolute_path_linux(self): @@ -946,16 +929,16 @@ class MySpider(scrapy.Spider): args = ["-o", f"{temp_dir}/output1.json:json"] log = self.get_log(spider_code, args=args) - self.assertIn( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}/output1.json", - log, + 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) - self.assertIn( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}/output2.json", - log, + assert ( + f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}/output2.json" + in log ) @skipIf(platform.system() != "Windows", reason="Windows only") @@ -975,16 +958,16 @@ class MySpider(scrapy.Spider): args = ["-o", f"{temp_dir}\\output1.json:json"] log = self.get_log(spider_code, args=args) - self.assertIn( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}\\output1.json", - log, + 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) - self.assertIn( - f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}\\output2.json", - log, + assert ( + f"[scrapy.extensions.feedexport] INFO: Stored json feed (1 items) in: {temp_dir}\\output2.json" + in log ) def test_args_change_settings(self): @@ -1006,11 +989,11 @@ class MySpider(scrapy.Spider): """ args = ["-a", "foo=42"] log = self.get_log(spider_code, args=args) - self.assertIn("Spider closed (finished)", log) - self.assertIn("The value of FOO is 42", log) + assert "Spider closed (finished)" in log + assert "The value of FOO is 42" in log -class WindowsRunSpiderCommandTest(RunSpiderCommandTest): +class TestWindowsRunSpiderCommand(TestRunSpiderCommand): spider_filename = "myspider.pyw" def setUp(self): @@ -1020,24 +1003,24 @@ class WindowsRunSpiderCommandTest(RunSpiderCommandTest): def test_start_requests_errors(self): log = self.get_log(self.badspider, name="badspider.pyw") - self.assertIn("start_requests", log) - self.assertIn("badspider.pyw", log) + assert "start_requests" in log + assert "badspider.pyw" in log def test_runspider_unable_to_load(self): raise unittest.SkipTest("Already Tested in 'RunSpiderCommandTest' ") -class BenchCommandTest(CommandTest): +class TestBenchCommand(TestCommandBase): def test_run(self): _, _, log = self.proc( "bench", "-s", "LOGSTATS_INTERVAL=0.001", "-s", "CLOSESPIDER_TIMEOUT=0.01" ) - self.assertIn("INFO: Crawled", log) - self.assertNotIn("Unhandled Error", log) - self.assertNotIn("log_count/ERROR", log) + assert "INFO: Crawled" in log + assert "Unhandled Error" not in log + assert "log_count/ERROR" not in log -class ViewCommandTest(CommandTest): +class TestViewCommand(TestCommandBase): def test_methods(self): command = view.Command() command.settings = Settings() @@ -1048,13 +1031,11 @@ class ViewCommandTest(CommandTest): conflict_handler="resolve", ) command.add_options(parser) - self.assertEqual(command.short_desc(), "Open URL in browser, as seen by Scrapy") - self.assertIn( - "URL using the Scrapy downloader and show its", command.long_desc() - ) + assert command.short_desc() == "Open URL in browser, as seen by Scrapy" + assert "URL using the Scrapy downloader and show its" in command.long_desc() -class CrawlCommandTest(CommandTest): +class TestCrawlCommand(TestCommandBase): def crawl(self, code, args=()): Path(self.proj_mod_path, "spiders", "myspider.py").write_text( code, encoding="utf-8" @@ -1077,7 +1058,7 @@ class MySpider(scrapy.Spider): return [] """ log = self.get_log(spider_code) - self.assertIn("[myspider] DEBUG: It works!", log) + assert "[myspider] DEBUG: It works!" in log def test_output(self): spider_code = """ @@ -1092,9 +1073,7 @@ class MySpider(scrapy.Spider): """ args = ["-o", "example.json"] log = self.get_log(spider_code, args=args) - self.assertIn( - "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}", log - ) + assert "[myspider] DEBUG: FEEDS: {'example.json': {'format': 'json'}}" in log def test_overwrite_output(self): spider_code = """ @@ -1115,13 +1094,13 @@ class MySpider(scrapy.Spider): Path(self.cwd, "example.json").write_text("not empty", encoding="utf-8") args = ["-O", "example.json"] log = self.get_log(spider_code, args=args) - self.assertIn( - '[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}', - log, + assert ( + '[myspider] DEBUG: FEEDS: {"example.json": {"format": "json", "overwrite": true}}' + in log ) with Path(self.cwd, "example.json").open(encoding="utf-8") as f2: first_line = f2.readline() - self.assertNotEqual(first_line, "not empty") + assert first_line != "not empty" def test_output_and_overwrite_output(self): spider_code = """ @@ -1135,12 +1114,12 @@ class MySpider(scrapy.Spider): """ args = ["-o", "example1.json", "-O", "example2.json"] log = self.get_log(spider_code, args=args) - self.assertIn( - "error: Please use only one of -o/--output and -O/--overwrite-output", log + assert ( + "error: Please use only one of -o/--output and -O/--overwrite-output" in log ) -class HelpMessageTest(CommandTest): +class TestHelpMessage(TestCommandBase): def setUp(self): super().setUp() self.commands = [ @@ -1163,30 +1142,30 @@ class HelpMessageTest(CommandTest): def test_help_messages(self): for command in self.commands: _, out, _ = self.proc(command, "-h") - self.assertIn("Usage", out) + assert "Usage" in out -class PopCommandNameTest(unittest.TestCase): +class TestPopCommandName: def test_valid_command(self): argv = ["scrapy", "crawl", "my_spider"] command = _pop_command_name(argv) - self.assertEqual(command, "crawl") - self.assertEqual(argv, ["scrapy", "my_spider"]) + assert command == "crawl" + assert argv == ["scrapy", "my_spider"] def test_no_command(self): argv = ["scrapy"] command = _pop_command_name(argv) - self.assertIsNone(command) - self.assertEqual(argv, ["scrapy"]) + assert command is None + assert argv == ["scrapy"] def test_option_before_command(self): argv = ["scrapy", "-h", "crawl"] command = _pop_command_name(argv) - self.assertEqual(command, "crawl") - self.assertEqual(argv, ["scrapy", "-h"]) + assert command == "crawl" + assert argv == ["scrapy", "-h"] def test_option_after_command(self): argv = ["scrapy", "crawl", "-h"] command = _pop_command_name(argv) - self.assertEqual(command, "crawl") - self.assertEqual(argv, ["scrapy", "-h"]) + assert command == "crawl" + assert argv == ["scrapy", "-h"] diff --git a/tests/test_contracts.py b/tests/test_contracts.py index 0f7d7b54c..fb961ace2 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -1,5 +1,6 @@ from unittest import TextTestResult +import pytest from twisted.internet import defer from twisted.python import failure from twisted.trial import unittest @@ -246,7 +247,7 @@ class InheritsDemoSpider(DemoSpider): name = "inherits_demo_spider" -class ContractsManagerTest(unittest.TestCase): +class TestContractsManager(unittest.TestCase): contracts = [ UrlContract, CallbackKeywordArgumentsContract, @@ -263,34 +264,33 @@ class ContractsManagerTest(unittest.TestCase): self.results = TextTestResult(stream=None, descriptions=False, verbosity=0) def should_succeed(self): - self.assertFalse(self.results.failures) - self.assertFalse(self.results.errors) + assert not self.results.failures + assert not self.results.errors def should_fail(self): - self.assertTrue(self.results.failures) - self.assertFalse(self.results.errors) + assert self.results.failures + assert not self.results.errors def should_error(self): - self.assertTrue(self.results.errors) + assert self.results.errors def test_contracts(self): spider = DemoSpider() # extract contracts correctly contracts = self.conman.extract_contracts(spider.returns_request) - self.assertEqual(len(contracts), 2) - self.assertEqual( - frozenset(type(x) for x in contracts), - frozenset([UrlContract, ReturnsContract]), + assert len(contracts) == 2 + assert frozenset(type(x) for x in contracts) == frozenset( + [UrlContract, ReturnsContract] ) # returns request for valid method request = self.conman.from_method(spider.returns_request, self.results) - self.assertNotEqual(request, None) + assert request is not None # no request for missing url request = self.conman.from_method(spider.parse_no_url, self.results) - self.assertEqual(request, None) + assert request is None def test_cb_kwargs(self): spider = DemoSpider() @@ -298,35 +298,31 @@ class ContractsManagerTest(unittest.TestCase): # extract contracts correctly contracts = self.conman.extract_contracts(spider.returns_request_cb_kwargs) - self.assertEqual(len(contracts), 3) - self.assertEqual( - frozenset(type(x) for x in contracts), - frozenset([UrlContract, CallbackKeywordArgumentsContract, ReturnsContract]), + assert len(contracts) == 3 + assert frozenset(type(x) for x in contracts) == frozenset( + [UrlContract, CallbackKeywordArgumentsContract, ReturnsContract] ) contracts = self.conman.extract_contracts(spider.returns_item_cb_kwargs) - self.assertEqual(len(contracts), 3) - self.assertEqual( - frozenset(type(x) for x in contracts), - frozenset([UrlContract, CallbackKeywordArgumentsContract, ReturnsContract]), + assert len(contracts) == 3 + assert frozenset(type(x) for x in contracts) == frozenset( + [UrlContract, CallbackKeywordArgumentsContract, ReturnsContract] ) contracts = self.conman.extract_contracts( spider.returns_item_cb_kwargs_error_unexpected_keyword ) - self.assertEqual(len(contracts), 3) - self.assertEqual( - frozenset(type(x) for x in contracts), - frozenset([UrlContract, CallbackKeywordArgumentsContract, ReturnsContract]), + assert len(contracts) == 3 + assert frozenset(type(x) for x in contracts) == frozenset( + [UrlContract, CallbackKeywordArgumentsContract, ReturnsContract] ) contracts = self.conman.extract_contracts( spider.returns_item_cb_kwargs_error_missing_argument ) - self.assertEqual(len(contracts), 2) - self.assertEqual( - frozenset(type(x) for x in contracts), - frozenset([UrlContract, ReturnsContract]), + assert len(contracts) == 2 + assert frozenset(type(x) for x in contracts) == frozenset( + [UrlContract, ReturnsContract] ) # returns_request @@ -360,17 +356,15 @@ class ContractsManagerTest(unittest.TestCase): # extract contracts correctly contracts = self.conman.extract_contracts(spider.returns_request_meta) - self.assertEqual(len(contracts), 3) - self.assertEqual( - frozenset(type(x) for x in contracts), - frozenset([UrlContract, MetadataContract, ReturnsContract]), + assert len(contracts) == 3 + assert frozenset(type(x) for x in contracts) == frozenset( + [UrlContract, MetadataContract, ReturnsContract] ) contracts = self.conman.extract_contracts(spider.returns_item_meta) - self.assertEqual(len(contracts), 3) - self.assertEqual( - frozenset(type(x) for x in contracts), - frozenset([UrlContract, MetadataContract, ReturnsContract]), + assert len(contracts) == 3 + assert frozenset(type(x) for x in contracts) == frozenset( + [UrlContract, MetadataContract, ReturnsContract] ) response = ResponseMetaMock() @@ -505,8 +499,8 @@ class ContractsManagerTest(unittest.TestCase): request = self.conman.from_method(spider.returns_request, self.results) request.errback(failure_mock) - self.assertFalse(self.results.failures) - self.assertTrue(self.results.errors) + assert not self.results.failures + assert self.results.errors @defer.inlineCallbacks def test_same_url(self): @@ -537,19 +531,19 @@ class ContractsManagerTest(unittest.TestCase): crawler = get_crawler(TestSameUrlSpider) yield crawler.crawl() - self.assertEqual(crawler.spider.visited, 2) + assert crawler.spider.visited == 2 def test_form_contract(self): spider = DemoSpider() request = self.conman.from_method(spider.custom_form, self.results) - self.assertEqual(request.method, "POST") - self.assertIsInstance(request, FormRequest) + assert request.method == "POST" + assert isinstance(request, FormRequest) def test_inherited_contracts(self): spider = InheritsDemoSpider() requests = self.conman.from_spider(spider, self.results) - self.assertTrue(requests) + assert requests class CustomFailContractPreProcess(Contract): @@ -566,8 +560,8 @@ class CustomFailContractPostProcess(Contract): raise KeyboardInterrupt("Post-process exception") -class CustomContractPrePostProcess(unittest.TestCase): - def setUp(self): +class TestCustomContractPrePostProcess: + def setup_method(self): self.results = TextTestResult(stream=None, descriptions=False, verbosity=0) def test_pre_hook_keyboard_interrupt(self): @@ -576,18 +570,13 @@ class CustomContractPrePostProcess(unittest.TestCase): contract = CustomFailContractPreProcess(spider.returns_request) conman = ContractsManager([contract]) - try: - request = conman.from_method(spider.returns_request, self.results) - contract.add_pre_hook(request, self.results) - # Expect this to raise a KeyboardInterrupt + request = conman.from_method(spider.returns_request, self.results) + contract.add_pre_hook(request, self.results) + with pytest.raises(KeyboardInterrupt, match="Pre-process exception"): request.callback(response, **request.cb_kwargs) - except KeyboardInterrupt as e: - self.assertEqual(str(e), "Pre-process exception") - else: - self.fail("KeyboardInterrupt not raised") - self.assertFalse(self.results.failures) - self.assertFalse(self.results.errors) + assert not self.results.failures + assert not self.results.errors def test_post_hook_keyboard_interrupt(self): spider = DemoSpider() @@ -595,15 +584,10 @@ class CustomContractPrePostProcess(unittest.TestCase): contract = CustomFailContractPostProcess(spider.returns_request) conman = ContractsManager([contract]) - try: - request = conman.from_method(spider.returns_request, self.results) - contract.add_post_hook(request, self.results) - # Expect this to raise a KeyboardInterrupt + request = conman.from_method(spider.returns_request, self.results) + contract.add_post_hook(request, self.results) + with pytest.raises(KeyboardInterrupt, match="Post-process exception"): request.callback(response, **request.cb_kwargs) - except KeyboardInterrupt as e: - self.assertEqual(str(e), "Post-process exception") - else: - self.fail("KeyboardInterrupt not raised") - self.assertFalse(self.results.failures) - self.assertFalse(self.results.errors) + assert not self.results.failures + assert not self.results.errors diff --git a/tests/test_core_downloader.py b/tests/test_core_downloader.py index dffba303f..1bffd69ed 100644 --- a/tests/test_core_downloader.py +++ b/tests/test_core_downloader.py @@ -30,16 +30,13 @@ from scrapy.utils.test import get_crawler from tests.mockserver import PayloadResource, ssl_context_factory -class SlotTest(unittest.TestCase): +class TestSlot: def test_repr(self): slot = Slot(concurrency=8, delay=0.1, randomize_delay=True) - self.assertEqual( - repr(slot), - "Slot(concurrency=8, delay=0.10, randomize_delay=True)", - ) + assert repr(slot) == "Slot(concurrency=8, delay=0.10, randomize_delay=True)" -class ContextFactoryBaseTestCase(unittest.TestCase): +class TestContextFactoryBase(unittest.TestCase): context_factory = None def _listen(self, site): @@ -90,7 +87,7 @@ class ContextFactoryBaseTestCase(unittest.TestCase): return await maybe_deferred_to_future(d) -class ContextFactoryTestCase(ContextFactoryBaseTestCase): +class TestContextFactory(TestContextFactoryBase): @deferred_f_from_coro_f async def testPayload(self): s = "0123456789" * 10 @@ -100,7 +97,7 @@ class ContextFactoryTestCase(ContextFactoryBaseTestCase): body = await self.get_page( self.getURL("payload"), client_context_factory, body=s ) - self.assertEqual(body, to_bytes(s)) + assert body == to_bytes(s) def test_override_getContext(self): class MyFactory(ScrapyClientContextFactory): @@ -112,14 +109,14 @@ class ContextFactoryTestCase(ContextFactoryBaseTestCase): with warnings.catch_warnings(record=True) as w: MyFactory() - self.assertEqual(len(w), 1) - self.assertIn( - "Overriding ScrapyClientContextFactory.getContext() is deprecated", - str(w[0].message), + assert len(w) == 1 + assert ( + "Overriding ScrapyClientContextFactory.getContext() is deprecated" + in str(w[0].message) ) -class ContextFactoryTLSMethodTestCase(ContextFactoryBaseTestCase): +class TestContextFactoryTLSMethod(TestContextFactoryBase): async def _assert_factory_works( self, client_context_factory: ScrapyClientContextFactory ) -> None: @@ -127,7 +124,7 @@ class ContextFactoryTLSMethodTestCase(ContextFactoryBaseTestCase): body = await self.get_page( self.getURL("payload"), client_context_factory, body=s ) - self.assertEqual(body, to_bytes(s)) + assert body == to_bytes(s) @deferred_f_from_coro_f async def test_setting_default(self): diff --git a/tests/test_crawl.py b/tests/test_crawl.py index 3aca2bbce..5766f9313 100644 --- a/tests/test_crawl.py +++ b/tests/test_crawl.py @@ -55,7 +55,7 @@ from tests.spiders import ( ) -class CrawlTestCase(TestCase): +class TestCrawl(TestCase): @classmethod def setUpClass(cls): cls.mockserver = MockServer() @@ -69,7 +69,7 @@ class CrawlTestCase(TestCase): def test_follow_all(self): crawler = get_crawler(FollowAllSpider) yield crawler.crawl(mockserver=self.mockserver) - self.assertEqual(len(crawler.spider.urls_visited), 11) # 10 + start_url + assert len(crawler.spider.urls_visited) == 11 # 10 + start_url @defer.inlineCallbacks def test_fixed_delay(self): @@ -94,9 +94,7 @@ class CrawlTestCase(TestCase): times = crawler.spider.times total_time = times[-1] - times[0] average = total_time / (len(times) - 1) - self.assertTrue( - average > delay * tolerance, f"download delay too small: {average}" - ) + assert average > delay * tolerance, f"download delay too small: {average}" # Ensure that the same test parameters would cause a failure if no # download delay is set. Otherwise, it means we are using a combination @@ -108,34 +106,32 @@ class CrawlTestCase(TestCase): times = crawler.spider.times total_time = times[-1] - times[0] average = total_time / (len(times) - 1) - self.assertFalse( - average > delay / tolerance, "test total or delay values are too small" - ) + assert average <= delay / tolerance, "test total or delay values are too small" @defer.inlineCallbacks def test_timeout_success(self): crawler = get_crawler(DelaySpider) yield crawler.crawl(n=0.5, mockserver=self.mockserver) - self.assertTrue(crawler.spider.t1 > 0) - self.assertTrue(crawler.spider.t2 > 0) - self.assertTrue(crawler.spider.t2 > crawler.spider.t1) + assert crawler.spider.t1 > 0 + assert crawler.spider.t2 > 0 + assert crawler.spider.t2 > crawler.spider.t1 @defer.inlineCallbacks def test_timeout_failure(self): crawler = get_crawler(DelaySpider, {"DOWNLOAD_TIMEOUT": 0.35}) yield crawler.crawl(n=0.5, mockserver=self.mockserver) - self.assertTrue(crawler.spider.t1 > 0) - self.assertTrue(crawler.spider.t2 == 0) - self.assertTrue(crawler.spider.t2_err > 0) - self.assertTrue(crawler.spider.t2_err > crawler.spider.t1) + assert crawler.spider.t1 > 0 + assert crawler.spider.t2 == 0 + assert crawler.spider.t2_err > 0 + assert crawler.spider.t2_err > crawler.spider.t1 # server hangs after receiving response headers crawler = get_crawler(DelaySpider, {"DOWNLOAD_TIMEOUT": 0.35}) yield crawler.crawl(n=0.5, b=1, mockserver=self.mockserver) - self.assertTrue(crawler.spider.t1 > 0) - self.assertTrue(crawler.spider.t2 == 0) - self.assertTrue(crawler.spider.t2_err > 0) - self.assertTrue(crawler.spider.t2_err > crawler.spider.t1) + assert crawler.spider.t1 > 0 + assert crawler.spider.t2 == 0 + assert crawler.spider.t2_err > 0 + assert crawler.spider.t2_err > crawler.spider.t1 @defer.inlineCallbacks def test_retry_503(self): @@ -173,10 +169,10 @@ class CrawlTestCase(TestCase): crawler = get_crawler(BrokenStartRequestsSpider) yield crawler.crawl(fail_before_yield=1, mockserver=self.mockserver) - self.assertEqual(len(log.records), 1) + assert len(log.records) == 1 record = log.records[0] - self.assertIsNotNone(record.exc_info) - self.assertIs(record.exc_info[0], ZeroDivisionError) + assert record.exc_info is not None + assert record.exc_info[0] is ZeroDivisionError @defer.inlineCallbacks def test_start_requests_bug_yielding(self): @@ -184,10 +180,10 @@ class CrawlTestCase(TestCase): crawler = get_crawler(BrokenStartRequestsSpider) yield crawler.crawl(fail_yielding=1, mockserver=self.mockserver) - self.assertEqual(len(log.records), 1) + assert len(log.records) == 1 record = log.records[0] - self.assertIsNotNone(record.exc_info) - self.assertIs(record.exc_info[0], ZeroDivisionError) + assert record.exc_info is not None + assert record.exc_info[0] is ZeroDivisionError @defer.inlineCallbacks def test_start_requests_items(self): @@ -195,7 +191,7 @@ class CrawlTestCase(TestCase): crawler = get_crawler(StartRequestsItemSpider) yield crawler.crawl(mockserver=self.mockserver) - self.assertEqual(len(log.records), 0) + assert len(log.records) == 0 @defer.inlineCallbacks def test_start_requests_unsupported_output(self): @@ -203,23 +199,18 @@ class CrawlTestCase(TestCase): crawler = get_crawler(StartRequestsGoodAndBadOutput) yield crawler.crawl(mockserver=self.mockserver) - self.assertEqual(len(log.records), 2) - self.assertEqual( - log.records[0].msg, - ( - "Got 'data:,b' among start requests. Only requests and items " - "are supported. It will be ignored." - ), + assert len(log.records) == 2 + assert log.records[0].msg == ( + "Got 'data:,b' among start requests. Only requests and items " + "are supported. It will be ignored." ) - self.assertTrue( - re.match( - ( - r"^Got among start " - r"requests\. Only requests and items are supported\. It " - r"will be ignored\.$" - ), - log.records[1].msg, - ) + assert re.match( + ( + r"^Got among start " + r"requests\. Only requests and items are supported\. It " + r"will be ignored\.$" + ), + log.records[1].msg, ) @defer.inlineCallbacks @@ -227,10 +218,9 @@ class CrawlTestCase(TestCase): settings = {"CONCURRENT_REQUESTS": 1} crawler = get_crawler(BrokenStartRequestsSpider, settings) yield crawler.crawl(mockserver=self.mockserver) - self.assertTrue( - crawler.spider.seedsseen.index(None) < crawler.spider.seedsseen.index(99), - crawler.spider.seedsseen, - ) + assert crawler.spider.seedsseen.index(None) < crawler.spider.seedsseen.index( + 99 + ), crawler.spider.seedsseen @defer.inlineCallbacks def test_start_requests_dupes(self): @@ -239,7 +229,7 @@ class CrawlTestCase(TestCase): yield crawler.crawl( dont_filter=True, distinct_urls=2, dupe_factor=3, mockserver=self.mockserver ) - self.assertEqual(crawler.spider.visited, 6) + assert crawler.spider.visited == 6 crawler = get_crawler(DuplicateStartRequestsSpider, settings) yield crawler.crawl( @@ -248,7 +238,7 @@ class CrawlTestCase(TestCase): dupe_factor=4, mockserver=self.mockserver, ) - self.assertEqual(crawler.spider.visited, 3) + assert crawler.spider.visited == 3 @defer.inlineCallbacks def test_unbounded_response(self): @@ -282,7 +272,7 @@ with multiples lines yield crawler.crawl( self.mockserver.url(f"/raw?{query}"), mockserver=self.mockserver ) - self.assertEqual(str(log).count("Got response 200"), 1) + assert str(log).count("Got response 200") == 1 @defer.inlineCallbacks def test_retry_conn_lost(self): @@ -305,8 +295,8 @@ with multiples lines self._assert_retried(log) def _assert_retried(self, log): - self.assertEqual(str(log).count("Retrying"), 2) - self.assertEqual(str(log).count("Gave up retrying"), 1) + assert str(log).count("Retrying") == 2 + assert str(log).count("Gave up retrying") == 1 @defer.inlineCallbacks def test_referer_header(self): @@ -321,20 +311,20 @@ with multiples lines crawler = get_crawler(SingleRequestSpider) yield crawler.crawl(seed=req0, mockserver=self.mockserver) # basic asserts in case of weird communication errors - self.assertIn("responses", crawler.spider.meta) - self.assertNotIn("failures", crawler.spider.meta) + assert "responses" in crawler.spider.meta + assert "failures" not in crawler.spider.meta # start requests doesn't set Referer header echo0 = json.loads(to_unicode(crawler.spider.meta["responses"][2].body)) - self.assertNotIn("Referer", echo0["headers"]) + assert "Referer" not in echo0["headers"] # following request sets Referer to start request url echo1 = json.loads(to_unicode(crawler.spider.meta["responses"][1].body)) - self.assertEqual(echo1["headers"].get("Referer"), [req0.url]) + assert echo1["headers"].get("Referer") == [req0.url] # next request avoids Referer header echo2 = json.loads(to_unicode(crawler.spider.meta["responses"][2].body)) - self.assertNotIn("Referer", echo2["headers"]) + assert "Referer" not in echo2["headers"] # last request explicitly sets a Referer header echo3 = json.loads(to_unicode(crawler.spider.meta["responses"][3].body)) - self.assertEqual(echo3["headers"].get("Referer"), ["http://example.com"]) + assert echo3["headers"].get("Referer") == ["http://example.com"] @defer.inlineCallbacks def test_engine_status(self): @@ -349,10 +339,10 @@ with multiples lines yield crawler.crawl( seed=self.mockserver.url("/"), callback_func=cb, mockserver=self.mockserver ) - self.assertEqual(len(est), 1, est) + assert len(est) == 1, est s = dict(est[0]) - self.assertEqual(s["engine.spider.name"], crawler.spider.name) - self.assertEqual(s["len(engine.scraper.slot.active)"], 1) + assert s["engine.spider.name"] == crawler.spider.name + assert s["len(engine.scraper.slot.active)"] == 1 @defer.inlineCallbacks def test_format_engine_status(self): @@ -367,7 +357,7 @@ with multiples lines yield crawler.crawl( seed=self.mockserver.url("/"), callback_func=cb, mockserver=self.mockserver ) - self.assertEqual(len(est), 1, est) + assert len(est) == 1, est est = est[0].split("\n")[2:-2] # remove header & footer # convert to dict est = [x.split(":") for x in est] @@ -376,8 +366,8 @@ with multiples lines it = iter(est) s = dict(zip(it, it)) - self.assertEqual(s["engine.spider.name"], crawler.spider.name) - self.assertEqual(s["len(engine.scraper.slot.active)"], "1") + assert s["engine.spider.name"] == crawler.spider.name + assert s["len(engine.scraper.slot.active)"] == "1" @defer.inlineCallbacks def test_graceful_crawl_error_handling(self): @@ -398,7 +388,7 @@ with multiples lines crawler = get_crawler(FaultySpider) yield self.assertFailure(crawler.crawl(mockserver=self.mockserver), TestError) - self.assertFalse(crawler.crawling) + assert not crawler.crawling @defer.inlineCallbacks def test_open_spider_error_on_faulty_pipeline(self): @@ -414,7 +404,7 @@ with multiples lines ), ZeroDivisionError, ) - self.assertFalse(crawler.crawling) + assert not crawler.crawling @defer.inlineCallbacks def test_crawlerrunner_accepts_crawler(self): @@ -426,7 +416,7 @@ with multiples lines self.mockserver.url("/status?n=200"), mockserver=self.mockserver, ) - self.assertIn("Got response 200", str(log)) + assert "Got response 200" in str(log) @defer.inlineCallbacks def test_crawl_multiple(self): @@ -446,10 +436,10 @@ with multiples lines yield runner.join() self._assert_retried(log) - self.assertIn("Got response 200", str(log)) + assert "Got response 200" in str(log) -class CrawlSpiderTestCase(TestCase): +class TestCrawlSpider(TestCase): @classmethod def setUpClass(cls): cls.mockserver = MockServer() @@ -480,9 +470,9 @@ class CrawlSpiderTestCase(TestCase): with LogCapture() as log: yield crawler.crawl(mockserver=self.mockserver) - self.assertIn("[parse] status 200 (foo: None)", str(log)) - self.assertIn("[parse] status 201 (foo: None)", str(log)) - self.assertIn("[parse] status 202 (foo: bar)", str(log)) + assert "[parse] status 200 (foo: None)" in str(log) + assert "[parse] status 201 (foo: None)" in str(log) + assert "[parse] status 202 (foo: bar)" in str(log) @defer.inlineCallbacks def test_crawlspider_with_async_callback(self): @@ -490,9 +480,9 @@ class CrawlSpiderTestCase(TestCase): with LogCapture() as log: yield crawler.crawl(mockserver=self.mockserver) - self.assertIn("[parse_async] status 200 (foo: None)", str(log)) - self.assertIn("[parse_async] status 201 (foo: None)", str(log)) - self.assertIn("[parse_async] status 202 (foo: bar)", str(log)) + assert "[parse_async] status 200 (foo: None)" in str(log) + assert "[parse_async] status 201 (foo: None)" in str(log) + assert "[parse_async] status 202 (foo: bar)" in str(log) @defer.inlineCallbacks def test_crawlspider_with_async_generator_callback(self): @@ -500,9 +490,9 @@ class CrawlSpiderTestCase(TestCase): with LogCapture() as log: yield crawler.crawl(mockserver=self.mockserver) - self.assertIn("[parse_async_gen] status 200 (foo: None)", str(log)) - self.assertIn("[parse_async_gen] status 201 (foo: None)", str(log)) - self.assertIn("[parse_async_gen] status 202 (foo: bar)", str(log)) + assert "[parse_async_gen] status 200 (foo: None)" in str(log) + assert "[parse_async_gen] status 201 (foo: None)" in str(log) + assert "[parse_async_gen] status 202 (foo: bar)" in str(log) @defer.inlineCallbacks def test_crawlspider_with_errback(self): @@ -510,12 +500,12 @@ class CrawlSpiderTestCase(TestCase): with LogCapture() as log: yield crawler.crawl(mockserver=self.mockserver) - self.assertIn("[parse] status 200 (foo: None)", str(log)) - self.assertIn("[parse] status 201 (foo: None)", str(log)) - self.assertIn("[parse] status 202 (foo: bar)", str(log)) - self.assertIn("[errback] status 404", str(log)) - self.assertIn("[errback] status 500", str(log)) - self.assertIn("[errback] status 501", str(log)) + assert "[parse] status 200 (foo: None)" in str(log) + assert "[parse] status 201 (foo: None)" in str(log) + assert "[parse] status 202 (foo: bar)" in str(log) + assert "[errback] status 404" in str(log) + assert "[errback] status 500" in str(log) + assert "[errback] status 501" in str(log) @defer.inlineCallbacks def test_crawlspider_process_request_cb_kwargs(self): @@ -523,9 +513,9 @@ class CrawlSpiderTestCase(TestCase): with LogCapture() as log: yield crawler.crawl(mockserver=self.mockserver) - self.assertIn("[parse] status 200 (foo: process_request)", str(log)) - self.assertIn("[parse] status 201 (foo: process_request)", str(log)) - self.assertIn("[parse] status 202 (foo: bar)", str(log)) + assert "[parse] status 200 (foo: process_request)" in str(log) + assert "[parse] status 201 (foo: process_request)" in str(log) + assert "[parse] status 202 (foo: bar)" in str(log) @defer.inlineCallbacks def test_async_def_parse(self): @@ -534,7 +524,7 @@ class CrawlSpiderTestCase(TestCase): yield crawler.crawl( self.mockserver.url("/status?n=200"), mockserver=self.mockserver ) - self.assertIn("Got response 200", str(log)) + assert "Got response 200" in str(log) @pytest.mark.only_asyncio @defer.inlineCallbacks @@ -549,15 +539,15 @@ class CrawlSpiderTestCase(TestCase): yield crawler.crawl( self.mockserver.url("/status?n=200"), mockserver=self.mockserver ) - self.assertIn("Got response 200", str(log)) + assert "Got response 200" in str(log) @pytest.mark.only_asyncio @defer.inlineCallbacks def test_async_def_asyncio_parse_items_list(self): log, items, _ = yield self._run_spider(AsyncDefAsyncioReturnSpider) - self.assertIn("Got response 200", str(log)) - self.assertIn({"id": 1}, items) - self.assertIn({"id": 2}, items) + assert "Got response 200" in str(log) + assert {"id": 1} in items + assert {"id": 2} in items @pytest.mark.only_asyncio @defer.inlineCallbacks @@ -573,81 +563,81 @@ class CrawlSpiderTestCase(TestCase): yield crawler.crawl( self.mockserver.url("/status?n=200"), mockserver=self.mockserver ) - self.assertIn("Got response 200", str(log)) - self.assertIn({"foo": 42}, items) + assert "Got response 200" in str(log) + assert {"foo": 42} in items @pytest.mark.only_asyncio @defer.inlineCallbacks def test_async_def_asyncgen_parse(self): log, _, stats = yield self._run_spider(AsyncDefAsyncioGenSpider) - self.assertIn("Got response 200", str(log)) + assert "Got response 200" in str(log) itemcount = stats.get_value("item_scraped_count") - self.assertEqual(itemcount, 1) + assert itemcount == 1 @pytest.mark.only_asyncio @defer.inlineCallbacks def test_async_def_asyncgen_parse_loop(self): log, items, stats = yield self._run_spider(AsyncDefAsyncioGenLoopSpider) - self.assertIn("Got response 200", str(log)) + assert "Got response 200" in str(log) itemcount = stats.get_value("item_scraped_count") - self.assertEqual(itemcount, 10) + assert itemcount == 10 for i in range(10): - self.assertIn({"foo": i}, items) + assert {"foo": i} in items @pytest.mark.only_asyncio @defer.inlineCallbacks def test_async_def_asyncgen_parse_exc(self): log, items, stats = yield self._run_spider(AsyncDefAsyncioGenExcSpider) log = str(log) - self.assertIn("Spider error processing", log) - self.assertIn("ValueError", log) + assert "Spider error processing" in log + assert "ValueError" in log itemcount = stats.get_value("item_scraped_count") - self.assertEqual(itemcount, 7) + assert itemcount == 7 for i in range(7): - self.assertIn({"foo": i}, items) + assert {"foo": i} in items @pytest.mark.only_asyncio @defer.inlineCallbacks def test_async_def_asyncgen_parse_complex(self): _, items, stats = yield self._run_spider(AsyncDefAsyncioGenComplexSpider) itemcount = stats.get_value("item_scraped_count") - self.assertEqual(itemcount, 156) + assert itemcount == 156 # some random items for i in [1, 4, 21, 22, 207, 311]: - self.assertIn({"index": i}, items) + assert {"index": i} in items for i in [10, 30, 122]: - self.assertIn({"index2": i}, items) + assert {"index2": i} in items @pytest.mark.only_asyncio @defer.inlineCallbacks def test_async_def_asyncio_parse_reqs_list(self): log, *_ = yield self._run_spider(AsyncDefAsyncioReqsReturnSpider) for req_id in range(3): - self.assertIn(f"Got response 200, req_id {req_id}", str(log)) + assert f"Got response 200, req_id {req_id}" in str(log) @pytest.mark.only_not_asyncio @defer.inlineCallbacks def test_async_def_deferred_direct(self): _, items, _ = yield self._run_spider(AsyncDefDeferredDirectSpider) - self.assertEqual(items, [{"code": 200}]) + assert items == [{"code": 200}] @pytest.mark.only_asyncio @defer.inlineCallbacks def test_async_def_deferred_wrapped(self): log, items, _ = yield self._run_spider(AsyncDefDeferredWrappedSpider) - self.assertEqual(items, [{"code": 200}]) + assert items == [{"code": 200}] @defer.inlineCallbacks def test_async_def_deferred_maybe_wrapped(self): _, items, _ = yield self._run_spider(AsyncDefDeferredMaybeWrappedSpider) - self.assertEqual(items, [{"code": 200}]) + assert items == [{"code": 200}] @defer.inlineCallbacks def test_response_ssl_certificate_none(self): crawler = get_crawler(SingleRequestSpider) url = self.mockserver.url("/echo?body=test", is_secure=False) yield crawler.crawl(seed=url, mockserver=self.mockserver) - self.assertIsNone(crawler.spider.meta["responses"][0].certificate) + assert crawler.spider.meta["responses"][0].certificate is None @defer.inlineCallbacks def test_response_ssl_certificate(self): @@ -655,9 +645,9 @@ class CrawlSpiderTestCase(TestCase): url = self.mockserver.url("/echo?body=test", is_secure=True) yield crawler.crawl(seed=url, mockserver=self.mockserver) cert = crawler.spider.meta["responses"][0].certificate - self.assertIsInstance(cert, Certificate) - self.assertEqual(cert.getSubject().commonName, b"localhost") - self.assertEqual(cert.getIssuer().commonName, b"localhost") + assert isinstance(cert, Certificate) + assert cert.getSubject().commonName == b"localhost" + assert cert.getIssuer().commonName == b"localhost" @pytest.mark.xfail( reason="Responses with no body return early and contain no certificate" @@ -668,9 +658,9 @@ class CrawlSpiderTestCase(TestCase): url = self.mockserver.url("/status?n=200", is_secure=True) yield crawler.crawl(seed=url, mockserver=self.mockserver) cert = crawler.spider.meta["responses"][0].certificate - self.assertIsInstance(cert, Certificate) - self.assertEqual(cert.getSubject().commonName, b"localhost") - self.assertEqual(cert.getIssuer().commonName, b"localhost") + assert isinstance(cert, Certificate) + assert cert.getSubject().commonName == b"localhost" + assert cert.getIssuer().commonName == b"localhost" @defer.inlineCallbacks def test_dns_server_ip_address_none(self): @@ -678,7 +668,7 @@ class CrawlSpiderTestCase(TestCase): url = self.mockserver.url("/status?n=200") yield crawler.crawl(seed=url, mockserver=self.mockserver) ip_address = crawler.spider.meta["responses"][0].ip_address - self.assertIsNone(ip_address) + assert ip_address is None @defer.inlineCallbacks def test_dns_server_ip_address(self): @@ -687,61 +677,57 @@ class CrawlSpiderTestCase(TestCase): expected_netloc, _ = urlparse(url).netloc.split(":") yield crawler.crawl(seed=url, mockserver=self.mockserver) ip_address = crawler.spider.meta["responses"][0].ip_address - self.assertIsInstance(ip_address, IPv4Address) - self.assertEqual(str(ip_address), gethostbyname(expected_netloc)) + assert isinstance(ip_address, IPv4Address) + assert str(ip_address) == gethostbyname(expected_netloc) @defer.inlineCallbacks def test_bytes_received_stop_download_callback(self): crawler = get_crawler(BytesReceivedCallbackSpider) yield crawler.crawl(mockserver=self.mockserver) - self.assertIsNone(crawler.spider.meta.get("failure")) - self.assertIsInstance(crawler.spider.meta["response"], Response) - self.assertEqual( - crawler.spider.meta["response"].body, - crawler.spider.meta.get("bytes_received"), + assert crawler.spider.meta.get("failure") is None + assert isinstance(crawler.spider.meta["response"], Response) + assert crawler.spider.meta["response"].body == crawler.spider.meta.get( + "bytes_received" ) - self.assertLess( - len(crawler.spider.meta["response"].body), - crawler.spider.full_response_length, + assert ( + len(crawler.spider.meta["response"].body) + < crawler.spider.full_response_length ) @defer.inlineCallbacks def test_bytes_received_stop_download_errback(self): crawler = get_crawler(BytesReceivedErrbackSpider) yield crawler.crawl(mockserver=self.mockserver) - self.assertIsNone(crawler.spider.meta.get("response")) - self.assertIsInstance(crawler.spider.meta["failure"], Failure) - self.assertIsInstance(crawler.spider.meta["failure"].value, StopDownload) - self.assertIsInstance(crawler.spider.meta["failure"].value.response, Response) - self.assertEqual( - crawler.spider.meta["failure"].value.response.body, - crawler.spider.meta.get("bytes_received"), - ) - self.assertLess( - len(crawler.spider.meta["failure"].value.response.body), - crawler.spider.full_response_length, + assert crawler.spider.meta.get("response") is None + assert isinstance(crawler.spider.meta["failure"], Failure) + assert isinstance(crawler.spider.meta["failure"].value, StopDownload) + assert isinstance(crawler.spider.meta["failure"].value.response, Response) + assert crawler.spider.meta[ + "failure" + ].value.response.body == crawler.spider.meta.get("bytes_received") + assert ( + len(crawler.spider.meta["failure"].value.response.body) + < crawler.spider.full_response_length ) @defer.inlineCallbacks def test_headers_received_stop_download_callback(self): crawler = get_crawler(HeadersReceivedCallbackSpider) yield crawler.crawl(mockserver=self.mockserver) - self.assertIsNone(crawler.spider.meta.get("failure")) - self.assertIsInstance(crawler.spider.meta["response"], Response) - self.assertEqual( - crawler.spider.meta["response"].headers, - crawler.spider.meta.get("headers_received"), + assert crawler.spider.meta.get("failure") is None + assert isinstance(crawler.spider.meta["response"], Response) + assert crawler.spider.meta["response"].headers == crawler.spider.meta.get( + "headers_received" ) @defer.inlineCallbacks def test_headers_received_stop_download_errback(self): crawler = get_crawler(HeadersReceivedErrbackSpider) yield crawler.crawl(mockserver=self.mockserver) - self.assertIsNone(crawler.spider.meta.get("response")) - self.assertIsInstance(crawler.spider.meta["failure"], Failure) - self.assertIsInstance(crawler.spider.meta["failure"].value, StopDownload) - self.assertIsInstance(crawler.spider.meta["failure"].value.response, Response) - self.assertEqual( - crawler.spider.meta["failure"].value.response.headers, - crawler.spider.meta.get("headers_received"), - ) + assert crawler.spider.meta.get("response") is None + assert isinstance(crawler.spider.meta["failure"], Failure) + assert isinstance(crawler.spider.meta["failure"].value, StopDownload) + assert isinstance(crawler.spider.meta["failure"].value.response, Response) + assert crawler.spider.meta[ + "failure" + ].value.response.headers == crawler.spider.meta.get("headers_received") diff --git a/tests/test_crawler.py b/tests/test_crawler.py index df5ebfa7b..0bbcc0843 100644 --- a/tests/test_crawler.py +++ b/tests/test_crawler.py @@ -39,13 +39,13 @@ def get_raw_crawler(spidercls=None, settings_dict=None): return Crawler(spidercls or DefaultSpider, settings) -class BaseCrawlerTest(unittest.TestCase): +class TestBaseCrawler(unittest.TestCase): def assertOptionIsDefault(self, settings, key): - self.assertIsInstance(settings, Settings) - self.assertEqual(settings[key], getattr(default_settings, key)) + assert isinstance(settings, Settings) + assert settings[key] == getattr(default_settings, key) -class CrawlerTestCase(BaseCrawlerTest): +class TestCrawler(TestBaseCrawler): def test_populate_spidercls_settings(self): spider_settings = {"TEST1": "spider", "TEST2": "spider"} project_settings = {**BASE_SETTINGS, "TEST1": "project", "TEST3": "project"} @@ -58,16 +58,16 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = Crawler(CustomSettingsSpider, settings) crawler._apply_settings() - self.assertEqual(crawler.settings.get("TEST1"), "spider") - self.assertEqual(crawler.settings.get("TEST2"), "spider") - self.assertEqual(crawler.settings.get("TEST3"), "project") + assert crawler.settings.get("TEST1") == "spider" + assert crawler.settings.get("TEST2") == "spider" + assert crawler.settings.get("TEST3") == "project" - self.assertFalse(settings.frozen) - self.assertTrue(crawler.settings.frozen) + assert not settings.frozen + assert crawler.settings.frozen def test_crawler_accepts_dict(self): crawler = get_crawler(DefaultSpider, {"foo": "bar"}) - self.assertEqual(crawler.settings["foo"], "bar") + assert crawler.settings["foo"] == "bar" self.assertOptionIsDefault(crawler.settings, "RETRY_ENABLED") def test_crawler_accepts_None(self): @@ -107,23 +107,23 @@ class CrawlerTestCase(BaseCrawlerTest): }, } crawler = get_crawler(settings_dict=settings) - self.assertEqual(len(TrackingAddon.instances), 1) + assert len(TrackingAddon.instances) == 1 expected = TrackingAddon.instances[-1] addon = crawler.get_addon(TrackingAddon) - self.assertEqual(addon, expected) + assert addon == expected addon = crawler.get_addon(DefaultSpider) - self.assertIsNone(addon) + assert addon is None addon = crawler.get_addon(ParentAddon) - self.assertEqual(addon, expected) + assert addon == expected class ChildAddon(TrackingAddon): pass addon = crawler.get_addon(ChildAddon) - self.assertIsNone(addon) + assert addon is None @inlineCallbacks def test_get_downloader_middleware(self): @@ -162,18 +162,18 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = TrackingDownloaderMiddleware yield crawler.crawl() - self.assertEqual(len(TrackingDownloaderMiddleware.instances), 1) - self.assertEqual(MySpider.result, TrackingDownloaderMiddleware.instances[-1]) + assert len(TrackingDownloaderMiddleware.instances) == 1 + assert MySpider.result == TrackingDownloaderMiddleware.instances[-1] crawler = get_raw_crawler(MySpider, settings) MySpider.cls = DefaultSpider yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ParentDownloaderMiddleware yield crawler.crawl() - self.assertEqual(MySpider.result, TrackingDownloaderMiddleware.instances[-1]) + assert MySpider.result == TrackingDownloaderMiddleware.instances[-1] class ChildDownloaderMiddleware(TrackingDownloaderMiddleware): pass @@ -181,7 +181,7 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ChildDownloaderMiddleware yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None def test_get_downloader_middleware_not_crawling(self): crawler = get_raw_crawler(settings_dict=BASE_SETTINGS) @@ -242,18 +242,18 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = TrackingExtension yield crawler.crawl() - self.assertEqual(len(TrackingExtension.instances), 1) - self.assertEqual(MySpider.result, TrackingExtension.instances[-1]) + assert len(TrackingExtension.instances) == 1 + assert MySpider.result == TrackingExtension.instances[-1] crawler = get_raw_crawler(MySpider, settings) MySpider.cls = DefaultSpider yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ParentExtension yield crawler.crawl() - self.assertEqual(MySpider.result, TrackingExtension.instances[-1]) + assert MySpider.result == TrackingExtension.instances[-1] class ChildExtension(TrackingExtension): pass @@ -261,7 +261,7 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ChildExtension yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None def test_get_extension_not_crawling(self): crawler = get_raw_crawler(settings_dict=BASE_SETTINGS) @@ -322,18 +322,18 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = TrackingItemPipeline yield crawler.crawl() - self.assertEqual(len(TrackingItemPipeline.instances), 1) - self.assertEqual(MySpider.result, TrackingItemPipeline.instances[-1]) + assert len(TrackingItemPipeline.instances) == 1 + assert MySpider.result == TrackingItemPipeline.instances[-1] crawler = get_raw_crawler(MySpider, settings) MySpider.cls = DefaultSpider yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ParentItemPipeline yield crawler.crawl() - self.assertEqual(MySpider.result, TrackingItemPipeline.instances[-1]) + assert MySpider.result == TrackingItemPipeline.instances[-1] class ChildItemPipeline(TrackingItemPipeline): pass @@ -341,7 +341,7 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ChildItemPipeline yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None def test_get_item_pipeline_not_crawling(self): crawler = get_raw_crawler(settings_dict=BASE_SETTINGS) @@ -402,18 +402,18 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = TrackingSpiderMiddleware yield crawler.crawl() - self.assertEqual(len(TrackingSpiderMiddleware.instances), 1) - self.assertEqual(MySpider.result, TrackingSpiderMiddleware.instances[-1]) + assert len(TrackingSpiderMiddleware.instances) == 1 + assert MySpider.result == TrackingSpiderMiddleware.instances[-1] crawler = get_raw_crawler(MySpider, settings) MySpider.cls = DefaultSpider yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ParentSpiderMiddleware yield crawler.crawl() - self.assertEqual(MySpider.result, TrackingSpiderMiddleware.instances[-1]) + assert MySpider.result == TrackingSpiderMiddleware.instances[-1] class ChildSpiderMiddleware(TrackingSpiderMiddleware): pass @@ -421,7 +421,7 @@ class CrawlerTestCase(BaseCrawlerTest): crawler = get_raw_crawler(MySpider, settings) MySpider.cls = ChildSpiderMiddleware yield crawler.crawl() - self.assertIsNone(MySpider.result) + assert MySpider.result is None def test_get_spider_middleware_not_crawling(self): crawler = get_raw_crawler(settings_dict=BASE_SETTINGS) @@ -446,7 +446,7 @@ class CrawlerTestCase(BaseCrawlerTest): yield crawler.crawl() -class SpiderSettingsTestCase(unittest.TestCase): +class TestSpiderSettings: def test_spider_custom_settings(self): class MySpider(scrapy.Spider): name = "spider" @@ -454,10 +454,10 @@ class SpiderSettingsTestCase(unittest.TestCase): crawler = get_crawler(MySpider) enabled_exts = [e.__class__ for e in crawler.extensions.middlewares] - self.assertIn(AutoThrottle, enabled_exts) + assert AutoThrottle in enabled_exts -class CrawlerLoggingTestCase(unittest.TestCase): +class TestCrawlerLogging: def test_no_root_handler_installed(self): handler = get_scrapy_root_handler() if handler is not None: @@ -469,8 +469,8 @@ class CrawlerLoggingTestCase(unittest.TestCase): get_crawler(MySpider) assert get_scrapy_root_handler() is None - def test_spider_custom_settings_log_level(self): - log_file = Path(self.mktemp()) + def test_spider_custom_settings_log_level(self, tmp_path): + log_file = Path(tmp_path, "log.txt") log_file.write_text("previous message\n", encoding="utf-8") class MySpider(scrapy.Spider): @@ -481,9 +481,9 @@ class CrawlerLoggingTestCase(unittest.TestCase): } configure_logging() - self.assertEqual(get_scrapy_root_handler().level, logging.DEBUG) + assert get_scrapy_root_handler().level == logging.DEBUG crawler = get_crawler(MySpider) - self.assertEqual(get_scrapy_root_handler().level, logging.INFO) + assert get_scrapy_root_handler().level == logging.INFO info_count = crawler.stats.get_value("log_count/INFO") logging.debug("debug message") logging.info("info message") @@ -492,18 +492,18 @@ class CrawlerLoggingTestCase(unittest.TestCase): logged = log_file.read_text(encoding="utf-8") - self.assertIn("previous message", logged) - self.assertNotIn("debug message", logged) - self.assertIn("info message", logged) - self.assertIn("warning message", logged) - self.assertIn("error message", logged) - self.assertEqual(crawler.stats.get_value("log_count/ERROR"), 1) - self.assertEqual(crawler.stats.get_value("log_count/WARNING"), 1) - self.assertEqual(crawler.stats.get_value("log_count/INFO") - info_count, 1) - self.assertEqual(crawler.stats.get_value("log_count/DEBUG", 0), 0) + assert "previous message" in logged + assert "debug message" not in logged + assert "info message" in logged + assert "warning message" in logged + assert "error message" in logged + assert crawler.stats.get_value("log_count/ERROR") == 1 + assert crawler.stats.get_value("log_count/WARNING") == 1 + assert crawler.stats.get_value("log_count/INFO") - info_count == 1 + assert crawler.stats.get_value("log_count/DEBUG", 0) == 0 - def test_spider_custom_settings_log_append(self): - log_file = Path(self.mktemp()) + def test_spider_custom_settings_log_append(self, tmp_path): + log_file = Path(tmp_path, "log.txt") log_file.write_text("previous message\n", encoding="utf-8") class MySpider(scrapy.Spider): @@ -519,8 +519,8 @@ class CrawlerLoggingTestCase(unittest.TestCase): logged = log_file.read_text(encoding="utf-8") - self.assertNotIn("previous message", logged) - self.assertIn("debug message", logged) + assert "previous message" not in logged + assert "debug message" in logged class SpiderLoaderWithWrongInterface: @@ -532,7 +532,7 @@ class CustomSpiderLoader(SpiderLoader): pass -class CrawlerRunnerTestCase(BaseCrawlerTest): +class TestCrawlerRunner(TestBaseCrawler): def test_spider_manager_verify_interface(self): settings = Settings( { @@ -544,7 +544,7 @@ class CrawlerRunnerTestCase(BaseCrawlerTest): def test_crawler_runner_accepts_dict(self): runner = CrawlerRunner({"foo": "bar"}) - self.assertEqual(runner.settings["foo"], "bar") + assert runner.settings["foo"] == "bar" self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED") def test_crawler_runner_accepts_None(self): @@ -552,10 +552,10 @@ class CrawlerRunnerTestCase(BaseCrawlerTest): self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED") -class CrawlerProcessTest(BaseCrawlerTest): +class TestCrawlerProcess(TestBaseCrawler): def test_crawler_process_accepts_dict(self): runner = CrawlerProcess({"foo": "bar"}) - self.assertEqual(runner.settings["foo"], "bar") + assert runner.settings["foo"] == "bar" self.assertOptionIsDefault(runner.settings, "RETRY_ENABLED") def test_crawler_process_accepts_None(self): @@ -579,7 +579,7 @@ class NoRequestsSpider(scrapy.Spider): @pytest.mark.usefixtures("reactor_pytest") -class CrawlerRunnerHasSpider(unittest.TestCase): +class TestCrawlerRunnerHasSpider(unittest.TestCase): def _runner(self): return CrawlerRunner() @@ -587,14 +587,14 @@ class CrawlerRunnerHasSpider(unittest.TestCase): def test_crawler_runner_bootstrap_successful(self): runner = self._runner() yield runner.crawl(NoRequestsSpider) - self.assertFalse(runner.bootstrap_failed) + assert not runner.bootstrap_failed @inlineCallbacks def test_crawler_runner_bootstrap_successful_for_several(self): runner = self._runner() yield runner.crawl(NoRequestsSpider) yield runner.crawl(NoRequestsSpider) - self.assertFalse(runner.bootstrap_failed) + assert not runner.bootstrap_failed @inlineCallbacks def test_crawler_runner_bootstrap_failed(self): @@ -605,9 +605,9 @@ class CrawlerRunnerHasSpider(unittest.TestCase): except ValueError: pass else: - self.fail("Exception should be raised from spider") + pytest.fail("Exception should be raised from spider") - self.assertTrue(runner.bootstrap_failed) + assert runner.bootstrap_failed @inlineCallbacks def test_crawler_runner_bootstrap_failed_for_several(self): @@ -618,11 +618,11 @@ class CrawlerRunnerHasSpider(unittest.TestCase): except ValueError: pass else: - self.fail("Exception should be raised from spider") + pytest.fail("Exception should be raised from spider") yield runner.crawl(NoRequestsSpider) - self.assertTrue(runner.bootstrap_failed) + assert runner.bootstrap_failed @inlineCallbacks def test_crawler_runner_asyncio_enabled_true(self): @@ -664,31 +664,34 @@ class ScriptRunnerMixin: return stderr.decode("utf-8") -class CrawlerProcessSubprocess(ScriptRunnerMixin, unittest.TestCase): +class TestCrawlerProcessSubprocess(ScriptRunnerMixin, unittest.TestCase): script_dir = Path(__file__).parent.resolve() / "CrawlerProcess" def test_simple(self): log = self.run_script("simple.py") - self.assertIn("Spider closed (finished)", log) - self.assertNotIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + not in log ) def test_multi(self): log = self.run_script("multi.py") - self.assertIn("Spider closed (finished)", log) - self.assertNotIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + not in log ) - self.assertNotIn("ReactorAlreadyInstalledError", log) + assert "ReactorAlreadyInstalledError" not in log def test_reactor_default(self): log = self.run_script("reactor_default.py") - self.assertIn("Spider closed (finished)", log) - self.assertNotIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + not in log ) - self.assertNotIn("ReactorAlreadyInstalledError", log) + assert "ReactorAlreadyInstalledError" not in log def test_reactor_default_twisted_reactor_select(self): log = self.run_script("reactor_default_twisted_reactor_select.py") @@ -703,50 +706,46 @@ class CrawlerProcessSubprocess(ScriptRunnerMixin, unittest.TestCase): # If that ever becomes the case on more platforms (i.e. if Linux # also starts using the select reactor by default in a future # version of Twisted), then we will need to rethink this test. - self.assertIn("Spider closed (finished)", log) + assert "Spider closed (finished)" in log else: - self.assertNotIn("Spider closed (finished)", log) - self.assertIn( - ( - "does not match the requested one " - "(twisted.internet.selectreactor.SelectReactor)" - ), - log, - ) + assert "Spider closed (finished)" not in log + assert ( + "does not match the requested one " + "(twisted.internet.selectreactor.SelectReactor)" + ) in log def test_reactor_select(self): log = self.run_script("reactor_select.py") - self.assertIn("Spider closed (finished)", log) - self.assertNotIn("ReactorAlreadyInstalledError", log) + assert "Spider closed (finished)" in log + assert "ReactorAlreadyInstalledError" not in log def test_reactor_select_twisted_reactor_select(self): log = self.run_script("reactor_select_twisted_reactor_select.py") - self.assertIn("Spider closed (finished)", log) - self.assertNotIn("ReactorAlreadyInstalledError", log) + assert "Spider closed (finished)" in log + assert "ReactorAlreadyInstalledError" not in log def test_reactor_select_subclass_twisted_reactor_select(self): log = self.run_script("reactor_select_subclass_twisted_reactor_select.py") - self.assertNotIn("Spider closed (finished)", log) - self.assertIn( - ( - "does not match the requested one " - "(twisted.internet.selectreactor.SelectReactor)" - ), - log, - ) + assert "Spider closed (finished)" not in log + assert ( + "does not match the requested one " + "(twisted.internet.selectreactor.SelectReactor)" + ) in log def test_asyncio_enabled_no_reactor(self): log = self.run_script("asyncio_enabled_no_reactor.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) def test_asyncio_enabled_reactor(self): log = self.run_script("asyncio_enabled_reactor.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) @pytest.mark.skipif( @@ -755,129 +754,129 @@ class CrawlerProcessSubprocess(ScriptRunnerMixin, unittest.TestCase): ) def test_ipv6_default_name_resolver(self): log = self.run_script("default_name_resolver.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "'downloader/exception_type_count/twisted.internet.error.DNSLookupError': 1,", - log, + assert "Spider closed (finished)" in log + assert ( + "'downloader/exception_type_count/twisted.internet.error.DNSLookupError': 1," + in log ) - self.assertIn( - "twisted.internet.error.DNSLookupError: DNS lookup failed: no results for hostname lookup: ::1.", - log, + assert ( + "twisted.internet.error.DNSLookupError: DNS lookup failed: no results for hostname lookup: ::1." + in log ) def test_caching_hostname_resolver_ipv6(self): log = self.run_script("caching_hostname_resolver_ipv6.py") - self.assertIn("Spider closed (finished)", log) - self.assertNotIn("twisted.internet.error.DNSLookupError", log) + assert "Spider closed (finished)" in log + assert "twisted.internet.error.DNSLookupError" not in log def test_caching_hostname_resolver_finite_execution(self): with MockServer() as mock_server: http_address = mock_server.http_address.replace("0.0.0.0", "127.0.0.1") log = self.run_script("caching_hostname_resolver.py", http_address) - self.assertIn("Spider closed (finished)", log) - self.assertNotIn("ERROR: Error downloading", log) - self.assertNotIn("TimeoutError", log) - self.assertNotIn("twisted.internet.error.DNSLookupError", log) + assert "Spider closed (finished)" in log + assert "ERROR: Error downloading" not in log + assert "TimeoutError" not in log + assert "twisted.internet.error.DNSLookupError" not in log def test_twisted_reactor_select(self): log = self.run_script("twisted_reactor_select.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.selectreactor.SelectReactor", log - ) + assert "Spider closed (finished)" in log + assert "Using reactor: twisted.internet.selectreactor.SelectReactor" in log @pytest.mark.skipif( platform.system() == "Windows", reason="PollReactor is not supported on Windows" ) def test_twisted_reactor_poll(self): log = self.run_script("twisted_reactor_poll.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn("Using reactor: twisted.internet.pollreactor.PollReactor", log) + assert "Spider closed (finished)" in log + assert "Using reactor: twisted.internet.pollreactor.PollReactor" in log def test_twisted_reactor_asyncio(self): log = self.run_script("twisted_reactor_asyncio.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) def test_twisted_reactor_asyncio_custom_settings(self): log = self.run_script("twisted_reactor_custom_settings.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) def test_twisted_reactor_asyncio_custom_settings_same(self): log = self.run_script("twisted_reactor_custom_settings_same.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) def test_twisted_reactor_asyncio_custom_settings_conflict(self): log = self.run_script("twisted_reactor_custom_settings_conflict.py") - self.assertIn( - "Using reactor: twisted.internet.selectreactor.SelectReactor", log - ) - self.assertIn( - "(twisted.internet.selectreactor.SelectReactor) does not match the requested one", - log, + assert "Using reactor: twisted.internet.selectreactor.SelectReactor" in log + assert ( + "(twisted.internet.selectreactor.SelectReactor) does not match the requested one" + in log ) @pytest.mark.requires_uvloop def test_custom_loop_asyncio(self): log = self.run_script("asyncio_custom_loop.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) - self.assertIn("Using asyncio event loop: uvloop.Loop", log) + assert "Using asyncio event loop: uvloop.Loop" in log @pytest.mark.requires_uvloop def test_custom_loop_asyncio_deferred_signal(self): log = self.run_script("asyncio_deferred_signal.py", "uvloop.Loop") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) - self.assertIn("Using asyncio event loop: uvloop.Loop", log) - self.assertIn("async pipeline opened!", log) + assert "Using asyncio event loop: uvloop.Loop" in log + assert "async pipeline opened!" in log @pytest.mark.requires_uvloop def test_asyncio_enabled_reactor_same_loop(self): log = self.run_script("asyncio_enabled_reactor_same_loop.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) - self.assertIn("Using asyncio event loop: uvloop.Loop", log) + assert "Using asyncio event loop: uvloop.Loop" in log @pytest.mark.requires_uvloop def test_asyncio_enabled_reactor_different_loop(self): log = self.run_script("asyncio_enabled_reactor_different_loop.py") - self.assertNotIn("Spider closed (finished)", log) - self.assertIn( - ( - "does not match the one specified in the ASYNCIO_EVENT_LOOP " - "setting (uvloop.Loop)" - ), - log, - ) + assert "Spider closed (finished)" not in log + assert ( + "does not match the one specified in the ASYNCIO_EVENT_LOOP " + "setting (uvloop.Loop)" + ) in log def test_default_loop_asyncio_deferred_signal(self): log = self.run_script("asyncio_deferred_signal.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn( - "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", log + assert "Spider closed (finished)" in log + assert ( + "Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) - self.assertNotIn("Using asyncio event loop: uvloop.Loop", log) - self.assertIn("async pipeline opened!", log) + assert "Using asyncio event loop: uvloop.Loop" not in log + assert "async pipeline opened!" in log def test_args_change_settings(self): log = self.run_script("args_settings.py") - self.assertIn("Spider closed (finished)", log) - self.assertIn("The value of FOO is 42", log) + assert "Spider closed (finished)" in log + assert "The value of FOO is 42" in log def test_shutdown_graceful(self): sig = signal.SIGINT if sys.platform != "win32" else signal.SIGBREAK @@ -910,23 +909,23 @@ class CrawlerProcessSubprocess(ScriptRunnerMixin, unittest.TestCase): p.wait() -class CrawlerRunnerSubprocess(ScriptRunnerMixin, unittest.TestCase): +class TestCrawlerRunnerSubprocess(ScriptRunnerMixin): script_dir = Path(__file__).parent.resolve() / "CrawlerRunner" def test_response_ip_address(self): log = self.run_script("ip_address.py") - self.assertIn("INFO: Spider closed (finished)", log) - self.assertIn("INFO: Host: not.a.real.domain", log) - self.assertIn("INFO: Type: ", log) - self.assertIn("INFO: IP address: 127.0.0.1", log) + assert "INFO: Spider closed (finished)" in log + assert "INFO: Host: not.a.real.domain" in log + assert "INFO: Type: " in log + assert "INFO: IP address: 127.0.0.1" in log def test_change_default_reactor(self): log = self.run_script("change_reactor.py") - self.assertIn( - "DEBUG: Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor", - log, + assert ( + "DEBUG: Using reactor: twisted.internet.asyncioreactor.AsyncioSelectorReactor" + in log ) - self.assertIn("DEBUG: Using asyncio event loop", log) + assert "DEBUG: Using asyncio event loop" in log @pytest.mark.parametrize( diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py index a39ed0694..162747581 100644 --- a/tests/test_dependencies.py +++ b/tests/test_dependencies.py @@ -4,11 +4,12 @@ from configparser import ConfigParser from importlib import import_module from pathlib import Path +import pytest from twisted import version as twisted_version from twisted.trial import unittest -class ScrapyUtilsTest(unittest.TestCase): +class TestScrapyUtils: def test_required_openssl_version(self): try: module = import_module("OpenSSL") @@ -27,7 +28,7 @@ class ScrapyUtilsTest(unittest.TestCase): See https://github.com/scrapy/scrapy/pull/4814#issuecomment-706230011 """ if not os.environ.get("_SCRAPY_PINNED", None): - self.skipTest("Not in a pinned environment") + pytest.skip("Not in a pinned environment") tox_config_file_path = Path(__file__).parent / ".." / "tox.ini" config_parser = ConfigParser() @@ -36,4 +37,4 @@ class ScrapyUtilsTest(unittest.TestCase): match = re.search(pattern, config_parser["pinned"]["deps"]) pinned_twisted_version_string = match[1] - self.assertEqual(twisted_version.short(), pinned_twisted_version_string) + assert twisted_version.short() == pinned_twisted_version_string diff --git a/tests/test_downloaderslotssettings.py b/tests/test_downloaderslotssettings.py index 4f8b005d7..15b3ad5af 100644 --- a/tests/test_downloaderslotssettings.py +++ b/tests/test_downloaderslotssettings.py @@ -76,7 +76,7 @@ class CrawlTestCase(TestCase): for k, v in slots.items() } - self.assertTrue(max(list(error_delta.values())) < tolerance) + assert max(list(error_delta.values())) < tolerance def test_params(): diff --git a/tests/test_dupefilters.py b/tests/test_dupefilters.py index 703c23529..d5e1b37f7 100644 --- a/tests/test_dupefilters.py +++ b/tests/test_dupefilters.py @@ -2,7 +2,6 @@ import hashlib import shutil import sys import tempfile -import unittest from pathlib import Path from warnings import catch_warnings @@ -39,7 +38,7 @@ class DirectDupeFilter: method = "n/a" -class RFPDupeFilterTest(unittest.TestCase): +class TestRFPDupeFilter: def test_df_from_crawler_scheduler(self): settings = { "DUPEFILTER_DEBUG": True, @@ -47,8 +46,8 @@ class RFPDupeFilterTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings) scheduler = Scheduler.from_crawler(crawler) - self.assertTrue(scheduler.df.debug) - self.assertEqual(scheduler.df.method, "from_crawler") + assert scheduler.df.debug + assert scheduler.df.method == "from_crawler" def test_df_direct_scheduler(self): settings = { @@ -56,7 +55,7 @@ class RFPDupeFilterTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings) scheduler = Scheduler.from_crawler(crawler) - self.assertEqual(scheduler.df.method, "n/a") + assert scheduler.df.method == "n/a" def test_filter(self): dupefilter = _get_dupefilter() @@ -256,16 +255,16 @@ class RFPDupeFilterTest(unittest.TestCase): dupefilter.close("finished") -class BaseDupeFilterTestCase(unittest.TestCase): +class TestBaseDupeFilter: def test_log_deprecation(self): dupefilter = _get_dupefilter( settings={"DUPEFILTER_CLASS": BaseDupeFilter}, ) with catch_warnings(record=True) as warning_list: dupefilter.log(None, None) - self.assertEqual(len(warning_list), 1) - self.assertEqual( - str(warning_list[0].message), - "Calling BaseDupeFilter.log() is deprecated.", + assert len(warning_list) == 1 + assert ( + str(warning_list[0].message) + == "Calling BaseDupeFilter.log() is deprecated." ) - self.assertEqual(warning_list[0].category, ScrapyDeprecationWarning) + assert warning_list[0].category == ScrapyDeprecationWarning diff --git a/tests/test_engine.py b/tests/test_engine.py index e9470493f..4bac8d273 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -243,8 +243,9 @@ class CrawlerRun: self.signals_caught[sig] = signalargs -class EngineTestBase(unittest.TestCase): - def _assert_visited_urls(self, run: CrawlerRun): +class TestEngineBase(unittest.TestCase): + @staticmethod + def _assert_visited_urls(run: CrawlerRun) -> None: must_be_visited = [ "/", "/redirect", @@ -259,8 +260,9 @@ class EngineTestBase(unittest.TestCase): f"URLs not visited: {list(urls_expected - urls_visited)}" ) - def _assert_scheduled_requests(self, run: CrawlerRun, count=None): - self.assertEqual(count, len(run.reqplug)) + @staticmethod + def _assert_scheduled_requests(run: CrawlerRun, count: int) -> None: + assert len(run.reqplug) == count paths_expected = ["/item999.html", "/item2.html", "/item1.html"] @@ -270,101 +272,104 @@ class EngineTestBase(unittest.TestCase): scheduled_requests_count = len(run.reqplug) dropped_requests_count = len(run.reqdropped) responses_count = len(run.respplug) - self.assertEqual( - scheduled_requests_count, dropped_requests_count + responses_count - ) - self.assertEqual(len(run.reqreached), responses_count) + assert scheduled_requests_count == dropped_requests_count + responses_count + assert len(run.reqreached) == responses_count - def _assert_dropped_requests(self, run: CrawlerRun): - self.assertEqual(len(run.reqdropped), 1) + @staticmethod + def _assert_dropped_requests(run: CrawlerRun) -> None: + assert len(run.reqdropped) == 1 - def _assert_downloaded_responses(self, run: CrawlerRun, count): + @staticmethod + def _assert_downloaded_responses(run: CrawlerRun, count: int) -> None: # response tests - self.assertEqual(count, len(run.respplug)) - self.assertEqual(count, len(run.reqreached)) + assert len(run.respplug) == count + assert len(run.reqreached) == count for response, _ in run.respplug: if run.getpath(response.url) == "/item999.html": - self.assertEqual(404, response.status) + assert response.status == 404 if run.getpath(response.url) == "/redirect": - self.assertEqual(302, response.status) + assert response.status == 302 - def _assert_items_error(self, run: CrawlerRun): - self.assertEqual(2, len(run.itemerror)) + @staticmethod + def _assert_items_error(run: CrawlerRun) -> None: + assert len(run.itemerror) == 2 for item, response, spider, failure in run.itemerror: - self.assertEqual(failure.value.__class__, ZeroDivisionError) - self.assertEqual(spider, run.spider) + assert failure.value.__class__ is ZeroDivisionError + assert spider == run.spider - self.assertEqual(item["url"], response.url) + assert item["url"] == response.url if "item1.html" in item["url"]: - self.assertEqual("Item 1 name", item["name"]) - self.assertEqual("100", item["price"]) + assert item["name"] == "Item 1 name" + assert item["price"] == "100" if "item2.html" in item["url"]: - self.assertEqual("Item 2 name", item["name"]) - self.assertEqual("200", item["price"]) + assert item["name"] == "Item 2 name" + assert item["price"] == "200" - def _assert_scraped_items(self, run: CrawlerRun): - self.assertEqual(2, len(run.itemresp)) + @staticmethod + def _assert_scraped_items(run: CrawlerRun) -> None: + assert len(run.itemresp) == 2 for item, response in run.itemresp: item = ItemAdapter(item) - self.assertEqual(item["url"], response.url) + assert item["url"] == response.url if "item1.html" in item["url"]: - self.assertEqual("Item 1 name", item["name"]) - self.assertEqual("100", item["price"]) + assert item["name"] == "Item 1 name" + assert item["price"] == "100" if "item2.html" in item["url"]: - self.assertEqual("Item 2 name", item["name"]) - self.assertEqual("200", item["price"]) + assert item["name"] == "Item 2 name" + assert item["price"] == "200" - def _assert_headers_received(self, run: CrawlerRun): + @staticmethod + def _assert_headers_received(run: CrawlerRun) -> None: for headers in run.headers.values(): - self.assertIn(b"Server", headers) - self.assertIn(b"TwistedWeb", headers[b"Server"]) - self.assertIn(b"Date", headers) - self.assertIn(b"Content-Type", headers) + assert b"Server" in headers + assert b"TwistedWeb" in headers[b"Server"] + assert b"Date" in headers + assert b"Content-Type" in headers - def _assert_bytes_received(self, run: CrawlerRun): - self.assertEqual(9, len(run.bytes)) + @staticmethod + def _assert_bytes_received(run: CrawlerRun) -> None: + assert len(run.bytes) == 9 for request, data in run.bytes.items(): joined_data = b"".join(data) if run.getpath(request.url) == "/": - self.assertEqual(joined_data, get_testdata("test_site", "index.html")) + assert joined_data == get_testdata("test_site", "index.html") elif run.getpath(request.url) == "/item1.html": - self.assertEqual(joined_data, get_testdata("test_site", "item1.html")) + assert joined_data == get_testdata("test_site", "item1.html") elif run.getpath(request.url) == "/item2.html": - self.assertEqual(joined_data, get_testdata("test_site", "item2.html")) + assert joined_data == get_testdata("test_site", "item2.html") elif run.getpath(request.url) == "/redirected": - self.assertEqual(joined_data, b"Redirected here") + assert joined_data == b"Redirected here" elif run.getpath(request.url) == "/redirect": - self.assertEqual( - joined_data, - b"\n\n" + assert ( + joined_data == b"\n\n" b" \n" b' \n' b" \n" b' \n' b' click here\n' b" \n" - b"\n", + b"\n" ) elif run.getpath(request.url) == "/tem999.html": - self.assertEqual( - joined_data, - b"\n\n" + assert ( + joined_data == b"\n\n" b" 404 - No Such Resource\n" b" \n" b"

No Such Resource

\n" b"

File not found.

\n" b" \n" - b"\n", + b"\n" ) elif run.getpath(request.url) == "/numbers": # signal was fired multiple times - self.assertTrue(len(data) > 1) + assert len(data) > 1 # bytes were received in order numbers = [str(x).encode("utf8") for x in range(2**18)] - self.assertEqual(joined_data, b"".join(numbers)) + assert joined_data == b"".join(numbers) - def _assert_signals_caught(self, run: CrawlerRun): + @staticmethod + def _assert_signals_caught(run: CrawlerRun) -> None: assert signals.engine_started in run.signals_caught assert signals.engine_stopped in run.signals_caught assert signals.spider_opened in run.signals_caught @@ -372,19 +377,14 @@ class EngineTestBase(unittest.TestCase): assert signals.spider_closed in run.signals_caught assert signals.headers_received in run.signals_caught - self.assertEqual( - {"spider": run.spider}, run.signals_caught[signals.spider_opened] - ) - self.assertEqual( - {"spider": run.spider}, run.signals_caught[signals.spider_idle] - ) - self.assertEqual( - {"spider": run.spider, "reason": "finished"}, - run.signals_caught[signals.spider_closed], - ) + assert {"spider": run.spider} == run.signals_caught[signals.spider_opened] + assert {"spider": run.spider} == run.signals_caught[signals.spider_idle] + assert {"spider": run.spider, "reason": "finished"} == run.signals_caught[ + signals.spider_closed + ] -class EngineTest(EngineTestBase): +class TestEngine(TestEngineBase): @defer.inlineCallbacks def test_crawler(self): for spider in ( @@ -419,10 +419,9 @@ class EngineTest(EngineTestBase): def test_crawler_change_close_reason_on_idle(self): run = CrawlerRun(ChangeCloseReasonSpider) yield run.run() - self.assertEqual( - {"spider": run.spider, "reason": "custom_reason"}, - run.signals_caught[signals.spider_closed], - ) + assert {"spider": run.spider, "reason": "custom_reason"} == run.signals_caught[ + signals.spider_closed + ] @defer.inlineCallbacks def test_close_downloader(self): @@ -470,7 +469,7 @@ class EngineTest(EngineTestBase): finally: timer.cancel() - self.assertNotIn(b"Traceback", stderr) + assert b"Traceback" not in stderr def test_request_scheduled_signal(caplog): diff --git a/tests/test_engine_stop_download_bytes.py b/tests/test_engine_stop_download_bytes.py index 5dd04c310..f09b0e091 100644 --- a/tests/test_engine_stop_download_bytes.py +++ b/tests/test_engine_stop_download_bytes.py @@ -7,8 +7,8 @@ from tests.test_engine import ( CrawlerRun, DataClassItemsSpider, DictItemsSpider, - EngineTestBase, MySpider, + TestEngineBase, ) @@ -18,7 +18,7 @@ class BytesReceivedCrawlerRun(CrawlerRun): raise StopDownload(fail=False) -class BytesReceivedEngineTest(EngineTestBase): +class TestBytesReceivedEngine(TestEngineBase): @defer.inlineCallbacks def test_crawler(self): for spider in ( @@ -61,14 +61,15 @@ class BytesReceivedEngineTest(EngineTestBase): self._assert_headers_received(run) self._assert_bytes_received(run) - def _assert_bytes_received(self, run: CrawlerRun): - self.assertEqual(9, len(run.bytes)) + @staticmethod + def _assert_bytes_received(run: CrawlerRun) -> None: + assert len(run.bytes) == 9 for request, data in run.bytes.items(): joined_data = b"".join(data) - self.assertTrue(len(data) == 1) # signal was fired only once + assert len(data) == 1 # signal was fired only once if run.getpath(request.url) == "/numbers": # Received bytes are not the complete response. The exact amount depends # on the buffer size, which can vary, so we only check that the amount # of received bytes is strictly less than the full response. numbers = [str(x).encode("utf8") for x in range(2**18)] - self.assertTrue(len(joined_data) < len(b"".join(numbers))) + assert len(joined_data) < len(b"".join(numbers)) diff --git a/tests/test_engine_stop_download_headers.py b/tests/test_engine_stop_download_headers.py index 06929d1e4..dbb0ea0d2 100644 --- a/tests/test_engine_stop_download_headers.py +++ b/tests/test_engine_stop_download_headers.py @@ -7,8 +7,8 @@ from tests.test_engine import ( CrawlerRun, DataClassItemsSpider, DictItemsSpider, - EngineTestBase, MySpider, + TestEngineBase, ) @@ -18,7 +18,7 @@ class HeadersReceivedCrawlerRun(CrawlerRun): raise StopDownload(fail=False) -class HeadersReceivedEngineTest(EngineTestBase): +class TestHeadersReceivedEngine(TestEngineBase): @defer.inlineCallbacks def test_crawler(self): for spider in ( @@ -60,10 +60,12 @@ class HeadersReceivedEngineTest(EngineTestBase): self._assert_bytes_received(run) self._assert_headers_received(run) - def _assert_bytes_received(self, run: CrawlerRun): - self.assertEqual(0, len(run.bytes)) + @staticmethod + def _assert_bytes_received(run: CrawlerRun) -> None: + assert len(run.bytes) == 0 - def _assert_visited_urls(self, run: CrawlerRun): + @staticmethod + def _assert_visited_urls(run: CrawlerRun) -> None: must_be_visited = ["/", "/redirect", "/redirected"] urls_visited = {rp[0].url for rp in run.respplug} urls_expected = {run.geturl(p) for p in must_be_visited} diff --git a/tests/test_webclient.py b/tests/test_webclient.py index 1b4ad2f2f..c3c03d6c3 100644 --- a/tests/test_webclient.py +++ b/tests/test_webclient.py @@ -36,7 +36,7 @@ from tests.mockserver import ( PayloadResource, ssl_context_factory, ) -from tests.test_core_downloader import ContextFactoryBaseTestCase +from tests.test_core_downloader import TestContextFactoryBase def getPage(url, contextFactory=None, response_transform=None, *args, **kwargs): @@ -63,7 +63,7 @@ def getPage(url, contextFactory=None, response_transform=None, *args, **kwargs): @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") -class ScrapyHTTPPageGetterTests(unittest.TestCase): +class TestScrapyHTTPPageGetter: def test_earlyHeaders(self): # basic test stolen from twisted HTTPageGetter factory = client.ScrapyHTTPClientFactory( @@ -177,9 +177,7 @@ class ScrapyHTTPPageGetterTests(unittest.TestCase): protocol = client.ScrapyHTTPPageGetter() protocol.factory = factory protocol.makeConnection(transport) - self.assertEqual( - set(transport.value().splitlines()), set(testvalue.splitlines()) - ) + assert set(transport.value().splitlines()) == set(testvalue.splitlines()) return testvalue def test_non_standard_line_endings(self): @@ -192,9 +190,7 @@ class ScrapyHTTPPageGetterTests(unittest.TestCase): protocol.dataReceived(b"Hello: World\n") protocol.dataReceived(b"Foo: Bar\n") protocol.dataReceived(b"\n") - self.assertEqual( - protocol.headers, Headers({"Hello": ["World"], "Foo": ["Bar"]}) - ) + assert protocol.headers == Headers({"Hello": ["World"], "Foo": ["Bar"]}) class EncodingResource(resource.Resource): @@ -207,7 +203,7 @@ class EncodingResource(resource.Resource): @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") -class WebClientTestCase(unittest.TestCase): +class TestWebClient(unittest.TestCase): def _listen(self, site): return reactor.listenTCP(0, site, interface="127.0.0.1") @@ -319,7 +315,7 @@ class WebClientTestCase(unittest.TestCase): return getPage(self.getURL("notsuchfile")).addCallback(self._cbNoSuchFile) def _cbNoSuchFile(self, pageData): - self.assertIn(b"404 - No Such Resource", pageData) + assert b"404 - No Such Resource" in pageData def testFactoryInfo(self): url = self.getURL("file") @@ -329,20 +325,20 @@ class WebClientTestCase(unittest.TestCase): return factory.deferred.addCallback(self._cbFactoryInfo, factory) def _cbFactoryInfo(self, ignoredResult, factory): - self.assertEqual(factory.status, b"200") - self.assertTrue(factory.version.startswith(b"HTTP/")) - self.assertEqual(factory.message, b"OK") - self.assertEqual(factory.response_headers[b"content-length"], b"10") + assert factory.status == b"200" + assert factory.version.startswith(b"HTTP/") + assert factory.message == b"OK" + assert factory.response_headers[b"content-length"] == b"10" def testRedirect(self): return getPage(self.getURL("redirect")).addCallback(self._cbRedirect) def _cbRedirect(self, pageData): - self.assertEqual( - pageData, - b'\n\n \n \n' + assert ( + pageData + == b'\n\n \n \n' b' \n \n ' - b'click here\n \n\n', + b'click here\n \n\n' ) def test_encoding(self): @@ -356,14 +352,12 @@ class WebClientTestCase(unittest.TestCase): def _check_Encoding(self, response, original_body): content_encoding = to_unicode(response.headers[b"Content-Encoding"]) - self.assertEqual(content_encoding, EncodingResource.out_encoding) - self.assertEqual( - response.body.decode(content_encoding), to_unicode(original_body) - ) + assert content_encoding == EncodingResource.out_encoding + assert response.body.decode(content_encoding) == to_unicode(original_body) @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") -class WebClientSSLTestCase(ContextFactoryBaseTestCase): +class WebClientSSLTestCase(TestContextFactoryBase): def testPayload(self): s = "0123456789" * 10 return getPage(self.getURL("payload"), body=s).addCallback(