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

Converting tests to plain asserts, part 2. (#6699)

This commit is contained in:
Andrey Rakhmatullin 2025-03-06 23:52:41 +04:00 committed by GitHub
parent 1843a4f753
commit 5a605969bd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 796 additions and 865 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
tests/test_addons.py
View File

@ -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"

28
tests/test_closespider.py
View File

@ -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

30
tests/test_cmdline/__init__.py
View File

@ -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"]}}
)

7
tests/test_cmdline_crawl_with_pipeline/__init__.py
View File

@ -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

14
tests/test_command_check.py
View File

@ -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")

10
tests/test_command_fetch.py
View File

@ -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):

81
tests/test_command_parse.py
View File

@ -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

28
tests/test_command_shell.py
View File

@ -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()

36
tests/test_command_version.py
View File

@ -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",
]

371
tests/test_commands.py
View File

@ -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] <name> <domain>\n\n\n"
"Optional Arguments\n==================\n\n"
"Global Options\n--------------\n"
),
assert formatter._join_parts(part_strings) == (
"Usage\n=====\n scrapy genspider [options] <name> <domain>\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 <name> <domain>. 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"]

112
tests/test_contracts.py
View File

@ -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

25
tests/test_core_downloader.py
View File

@ -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):

288
tests/test_crawl.py
View File

@ -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 <object object at 0x[0-9a-fA-F]+> among start "
r"requests\. Only requests and items are supported\. It "
r"will be ignored\.$"
),
log.records[1].msg,
)
assert re.match(
(
r"^Got <object object at 0x[0-9a-fA-F]+> 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")

351
tests/test_crawler.py
View File

@ -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: <class 'ipaddress.IPv4Address'>", 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: <class 'ipaddress.IPv4Address'>" 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(

7
tests/test_dependencies.py
View File

@ -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

2
tests/test_downloaderslotssettings.py
View File

@ -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():

21
tests/test_dupefilters.py
View File

@ -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

137
tests/test_engine.py
View File

@ -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<html>\n"
assert (
joined_data == b"\n<html>\n"
b" <head>\n"
b' <meta http-equiv="refresh" content="0;URL=/redirected">\n'
b" </head>\n"
b' <body bgcolor="#FFFFFF" text="#000000">\n'
b' <a href="/redirected">click here</a>\n'
b" </body>\n"
b"</html>\n",
b"</html>\n"
)
elif run.getpath(request.url) == "/tem999.html":
self.assertEqual(
joined_data,
b"\n<html>\n"
assert (
joined_data == b"\n<html>\n"
b" <head><title>404 - No Such Resource</title></head>\n"
b" <body>\n"
b" <h1>No Such Resource</h1>\n"
b" <p>File not found.</p>\n"
b" </body>\n"
b"</html>\n",
b"</html>\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):

13
tests/test_engine_stop_download_bytes.py
View File

@ -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))

12
tests/test_engine_stop_download_headers.py
View File

@ -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}

40
tests/test_webclient.py
View File

@ -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<html>\n <head>\n <meta http-equiv="refresh" content="0;URL=/file">\n'
assert (
pageData
== b'\n<html>\n <head>\n <meta http-equiv="refresh" content="0;URL=/file">\n'
b' </head>\n <body bgcolor="#FFFFFF" text="#000000">\n '
b'<a href="/file">click here</a>\n </body>\n</html>\n',
b'<a href="/file">click here</a>\n </body>\n</html>\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(