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 5. (#6712)

This commit is contained in:
Andrey Rakhmatullin 2025-03-09 23:24:45 +04:00 committed by GitHub
parent d442227fa7
commit 7bbe775040
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 446 additions and 521 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
tests/test_logformatter.py
View File

@ -1,11 +1,10 @@
import logging
import unittest
import pytest
from testfixtures import LogCapture
from twisted.internet import defer
from twisted.python.failure import Failure
from twisted.trial.unittest import TestCase as TwistedTestCase
from twisted.trial.unittest import TestCase
from scrapy.exceptions import DropItem
from scrapy.http import Request, Response
@ -24,8 +23,8 @@ class CustomItem(Item):
return f"name: {self['name']}"
class LogFormatterTestCase(unittest.TestCase):
def setUp(self):
class TestLogFormatter:
def setup_method(self):
self.formatter = LogFormatter()
self.spider = Spider("default")
self.spider.crawler = get_crawler()
@ -35,9 +34,7 @@ class LogFormatterTestCase(unittest.TestCase):
res = Response("http://www.example.com")
logkws = self.formatter.crawled(req, res, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline, "Crawled (200) <GET http://www.example.com> (referer: None)"
)
assert logline == "Crawled (200) <GET http://www.example.com> (referer: None)"
def test_crawled_without_referer(self):
req = Request(
@ -46,9 +43,9 @@ class LogFormatterTestCase(unittest.TestCase):
res = Response("http://www.example.com", flags=["cached"])
logkws = self.formatter.crawled(req, res, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline,
"Crawled (200) <GET http://www.example.com> (referer: http://example.com) ['cached']",
assert (
logline
== "Crawled (200) <GET http://www.example.com> (referer: http://example.com) ['cached']"
)
def test_flags_in_request(self):
@ -56,9 +53,9 @@ class LogFormatterTestCase(unittest.TestCase):
res = Response("http://www.example.com")
logkws = self.formatter.crawled(req, res, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline,
"Crawled (200) <GET http://www.example.com> ['test', 'flag'] (referer: None)",
assert (
logline
== "Crawled (200) <GET http://www.example.com> ['test', 'flag'] (referer: None)"
)
def test_dropped(self):
@ -69,7 +66,7 @@ class LogFormatterTestCase(unittest.TestCase):
logline = logkws["msg"] % logkws["args"]
lines = logline.splitlines()
assert all(isinstance(x, str) for x in lines)
self.assertEqual(lines, ["Dropped: \u2018", "{}"])
assert lines == ["Dropped: \u2018", "{}"]
def test_dropitem_default_log_level(self):
item = {}
@ -79,38 +76,38 @@ class LogFormatterTestCase(unittest.TestCase):
spider.crawler = get_crawler(Spider)
logkws = self.formatter.dropped(item, exception, response, spider)
self.assertEqual(logkws["level"], logging.WARNING)
assert logkws["level"] == logging.WARNING
spider.crawler.settings.frozen = False
spider.crawler.settings["DEFAULT_DROPITEM_LOG_LEVEL"] = logging.INFO
spider.crawler.settings.frozen = True
logkws = self.formatter.dropped(item, exception, response, spider)
self.assertEqual(logkws["level"], logging.INFO)
assert logkws["level"] == logging.INFO
spider.crawler.settings.frozen = False
spider.crawler.settings["DEFAULT_DROPITEM_LOG_LEVEL"] = "INFO"
spider.crawler.settings.frozen = True
logkws = self.formatter.dropped(item, exception, response, spider)
self.assertEqual(logkws["level"], logging.INFO)
assert logkws["level"] == logging.INFO
spider.crawler.settings.frozen = False
spider.crawler.settings["DEFAULT_DROPITEM_LOG_LEVEL"] = 10
spider.crawler.settings.frozen = True
logkws = self.formatter.dropped(item, exception, response, spider)
self.assertEqual(logkws["level"], logging.DEBUG)
assert logkws["level"] == logging.DEBUG
spider.crawler.settings.frozen = False
spider.crawler.settings["DEFAULT_DROPITEM_LOG_LEVEL"] = 0
spider.crawler.settings.frozen = True
logkws = self.formatter.dropped(item, exception, response, spider)
self.assertEqual(logkws["level"], logging.NOTSET)
assert logkws["level"] == logging.NOTSET
unsupported_value = object()
spider.crawler.settings.frozen = False
spider.crawler.settings["DEFAULT_DROPITEM_LOG_LEVEL"] = unsupported_value
spider.crawler.settings.frozen = True
logkws = self.formatter.dropped(item, exception, response, spider)
self.assertEqual(logkws["level"], unsupported_value)
assert logkws["level"] == unsupported_value
with pytest.raises(TypeError):
logging.log(logkws["level"], "message")
@ -121,11 +118,11 @@ class LogFormatterTestCase(unittest.TestCase):
exception = DropItem("Test drop", log_level="INFO")
logkws = self.formatter.dropped(item, exception, response, self.spider)
self.assertEqual(logkws["level"], logging.INFO)
assert logkws["level"] == logging.INFO
exception = DropItem("Test drop", log_level="ERROR")
logkws = self.formatter.dropped(item, exception, response, self.spider)
self.assertEqual(logkws["level"], logging.ERROR)
assert logkws["level"] == logging.ERROR
def test_item_error(self):
# In practice, the complete traceback is shown by passing the
@ -135,7 +132,7 @@ class LogFormatterTestCase(unittest.TestCase):
response = Response("http://www.example.com")
logkws = self.formatter.item_error(item, exception, response, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(logline, "Error processing {'key': 'value'}")
assert logline == "Error processing {'key': 'value'}"
def test_spider_error(self):
# In practice, the complete traceback is shown by passing the
@ -147,9 +144,9 @@ class LogFormatterTestCase(unittest.TestCase):
response = Response("http://www.example.com", request=request)
logkws = self.formatter.spider_error(failure, request, response, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline,
"Spider error processing <GET http://www.example.com> (referer: http://example.org)",
assert (
logline
== "Spider error processing <GET http://www.example.com> (referer: http://example.org)"
)
def test_download_error_short(self):
@ -159,7 +156,7 @@ class LogFormatterTestCase(unittest.TestCase):
request = Request("http://www.example.com")
logkws = self.formatter.download_error(failure, request, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(logline, "Error downloading <GET http://www.example.com>")
assert logline == "Error downloading <GET http://www.example.com>"
def test_download_error_long(self):
# In practice, the complete traceback is shown by passing the
@ -170,9 +167,7 @@ class LogFormatterTestCase(unittest.TestCase):
failure, request, self.spider, "Some message"
)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline, "Error downloading <GET http://www.example.com>: Some message"
)
assert logline == "Error downloading <GET http://www.example.com>: Some message"
def test_scraped(self):
item = CustomItem()
@ -182,9 +177,7 @@ class LogFormatterTestCase(unittest.TestCase):
logline = logkws["msg"] % logkws["args"]
lines = logline.splitlines()
assert all(isinstance(x, str) for x in lines)
self.assertEqual(
lines, ["Scraped from <200 http://www.example.com>", "name: \xa3"]
)
assert lines == ["Scraped from <200 http://www.example.com>", "name: \xa3"]
class LogFormatterSubclass(LogFormatter):
@ -200,8 +193,8 @@ class LogFormatterSubclass(LogFormatter):
}
class LogformatterSubclassTest(LogFormatterTestCase):
def setUp(self):
class TestLogformatterSubclass(TestLogFormatter):
def setup_method(self):
self.formatter = LogFormatterSubclass()
self.spider = Spider("default")
self.spider.crawler = get_crawler(Spider)
@ -211,8 +204,8 @@ class LogformatterSubclassTest(LogFormatterTestCase):
res = Response("http://www.example.com")
logkws = self.formatter.crawled(req, res, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline, "Crawled (200) <GET http://www.example.com> (referer: None) []"
assert (
logline == "Crawled (200) <GET http://www.example.com> (referer: None) []"
)
def test_crawled_without_referer(self):
@ -224,9 +217,9 @@ class LogformatterSubclassTest(LogFormatterTestCase):
res = Response("http://www.example.com")
logkws = self.formatter.crawled(req, res, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline,
"Crawled (200) <GET http://www.example.com> (referer: http://example.com) ['cached']",
assert (
logline
== "Crawled (200) <GET http://www.example.com> (referer: http://example.com) ['cached']"
)
def test_flags_in_request(self):
@ -234,9 +227,9 @@ class LogformatterSubclassTest(LogFormatterTestCase):
res = Response("http://www.example.com")
logkws = self.formatter.crawled(req, res, self.spider)
logline = logkws["msg"] % logkws["args"]
self.assertEqual(
logline,
"Crawled (200) <GET http://www.example.com> (referer: None) ['test', 'flag']",
assert (
logline
== "Crawled (200) <GET http://www.example.com> (referer: None) ['test', 'flag']"
)
@ -261,7 +254,7 @@ class DropSomeItemsPipeline:
self.drop = True
class ShowOrSkipMessagesTestCase(TwistedTestCase):
class TestShowOrSkipMessages(TestCase):
@classmethod
def setUpClass(cls):
cls.mockserver = MockServer()
@ -284,9 +277,9 @@ class ShowOrSkipMessagesTestCase(TwistedTestCase):
crawler = get_crawler(ItemSpider, self.base_settings)
with LogCapture() as lc:
yield crawler.crawl(mockserver=self.mockserver)
self.assertIn("Scraped from <200 http://127.0.0.1:", str(lc))
self.assertIn("Crawled (200) <GET http://127.0.0.1:", str(lc))
self.assertIn("Dropped: Ignoring item", str(lc))
assert "Scraped from <200 http://127.0.0.1:" in str(lc)
assert "Crawled (200) <GET http://127.0.0.1:" in str(lc)
assert "Dropped: Ignoring item" in str(lc)
@defer.inlineCallbacks
def test_skip_messages(self):
@ -295,6 +288,6 @@ class ShowOrSkipMessagesTestCase(TwistedTestCase):
crawler = get_crawler(ItemSpider, settings)
with LogCapture() as lc:
yield crawler.crawl(mockserver=self.mockserver)
self.assertNotIn("Scraped from <200 http://127.0.0.1:", str(lc))
self.assertNotIn("Crawled (200) <GET http://127.0.0.1:", str(lc))
self.assertNotIn("Dropped: Ignoring item", str(lc))
assert "Scraped from <200 http://127.0.0.1:" not in str(lc)
assert "Crawled (200) <GET http://127.0.0.1:" not in str(lc)
assert "Dropped: Ignoring item" not in str(lc)

45
tests/test_logstats.py
View File

@ -1,4 +1,3 @@
import unittest
from datetime import datetime
import pytest
@ -8,8 +7,8 @@ from scrapy.utils.test import get_crawler
from tests.spiders import SimpleSpider
class TestLogStats(unittest.TestCase):
def setUp(self):
class TestLogStats:
def setup_method(self):
self.crawler = get_crawler(SimpleSpider)
self.spider = self.crawler._create_spider("spidey")
self.stats = self.crawler.stats
@ -26,34 +25,34 @@ class TestLogStats(unittest.TestCase):
logstats.itemsprev
logstats.spider_opened(self.spider)
self.assertEqual(logstats.pagesprev, 4802)
self.assertEqual(logstats.itemsprev, 3201)
assert logstats.pagesprev == 4802
assert logstats.itemsprev == 3201
logstats.calculate_stats()
self.assertEqual(logstats.items, 3201)
self.assertEqual(logstats.pages, 4802)
self.assertEqual(logstats.irate, 0.0)
self.assertEqual(logstats.prate, 0.0)
self.assertEqual(logstats.pagesprev, 4802)
self.assertEqual(logstats.itemsprev, 3201)
assert logstats.items == 3201
assert logstats.pages == 4802
assert logstats.irate == 0.0
assert logstats.prate == 0.0
assert logstats.pagesprev == 4802
assert logstats.itemsprev == 3201
# Simulate what happens after a minute
self.stats.set_value("response_received_count", 5187)
self.stats.set_value("item_scraped_count", 3492)
logstats.calculate_stats()
self.assertEqual(logstats.items, 3492)
self.assertEqual(logstats.pages, 5187)
self.assertEqual(logstats.irate, 291.0)
self.assertEqual(logstats.prate, 385.0)
self.assertEqual(logstats.pagesprev, 5187)
self.assertEqual(logstats.itemsprev, 3492)
assert logstats.items == 3492
assert logstats.pages == 5187
assert logstats.irate == 291.0
assert logstats.prate == 385.0
assert logstats.pagesprev == 5187
assert logstats.itemsprev == 3492
# Simulate when spider closes after running for 30 mins
self.stats.set_value("start_time", datetime.fromtimestamp(1655100172))
self.stats.set_value("finish_time", datetime.fromtimestamp(1655101972))
logstats.spider_closed(self.spider, "test reason")
self.assertEqual(self.stats.get_value("responses_per_minute"), 172.9)
self.assertEqual(self.stats.get_value("items_per_minute"), 116.4)
assert self.stats.get_value("responses_per_minute") == 172.9
assert self.stats.get_value("items_per_minute") == 116.4
def test_stats_calculations_no_time(self):
"""The stat values should be None since the start and finish time are
@ -61,8 +60,8 @@ class TestLogStats(unittest.TestCase):
"""
logstats = LogStats.from_crawler(self.crawler)
logstats.spider_closed(self.spider, "test reason")
self.assertIsNone(self.stats.get_value("responses_per_minute"))
self.assertIsNone(self.stats.get_value("items_per_minute"))
assert self.stats.get_value("responses_per_minute") is None
assert self.stats.get_value("items_per_minute") is None
def test_stats_calculation_no_elapsed_time(self):
"""The stat values should be None since the elapsed time is 0."""
@ -70,5 +69,5 @@ class TestLogStats(unittest.TestCase):
self.stats.set_value("start_time", datetime.fromtimestamp(1655100172))
self.stats.set_value("finish_time", datetime.fromtimestamp(1655100172))
logstats.spider_closed(self.spider, "test reason")
self.assertIsNone(self.stats.get_value("responses_per_minute"))
self.assertIsNone(self.stats.get_value("items_per_minute"))
assert self.stats.get_value("responses_per_minute") is None
assert self.stats.get_value("items_per_minute") is None

71
tests/test_mail.py
View File

@ -1,4 +1,3 @@
import unittest
from email.charset import Charset
from io import BytesIO
@ -8,7 +7,7 @@ from twisted.internet._sslverify import ClientTLSOptions
from scrapy.mail import MailSender
class MailSenderTest(unittest.TestCase):
class TestMailSender:
def test_send(self):
mailsender = MailSender(debug=True)
mailsender.send(
@ -20,15 +19,15 @@ class MailSenderTest(unittest.TestCase):
assert self.catched_msg
self.assertEqual(self.catched_msg["to"], ["test@scrapy.org"])
self.assertEqual(self.catched_msg["subject"], "subject")
self.assertEqual(self.catched_msg["body"], "body")
assert self.catched_msg["to"] == ["test@scrapy.org"]
assert self.catched_msg["subject"] == "subject"
assert self.catched_msg["body"] == "body"
msg = self.catched_msg["msg"]
self.assertEqual(msg["to"], "test@scrapy.org")
self.assertEqual(msg["subject"], "subject")
self.assertEqual(msg.get_payload(), "body")
self.assertEqual(msg.get("Content-Type"), "text/plain")
assert msg["to"] == "test@scrapy.org"
assert msg["subject"] == "subject"
assert msg.get_payload() == "body"
assert msg.get("Content-Type") == "text/plain"
def test_send_single_values_to_and_cc(self):
mailsender = MailSender(debug=True)
@ -51,8 +50,8 @@ class MailSenderTest(unittest.TestCase):
)
msg = self.catched_msg["msg"]
self.assertEqual(msg.get_payload(), "<p>body</p>")
self.assertEqual(msg.get("Content-Type"), "text/html")
assert msg.get_payload() == "<p>body</p>"
assert msg.get("Content-Type") == "text/html"
def test_send_attach(self):
attach = BytesIO()
@ -70,22 +69,22 @@ class MailSenderTest(unittest.TestCase):
)
assert self.catched_msg
self.assertEqual(self.catched_msg["to"], ["test@scrapy.org"])
self.assertEqual(self.catched_msg["subject"], "subject")
self.assertEqual(self.catched_msg["body"], "body")
assert self.catched_msg["to"] == ["test@scrapy.org"]
assert self.catched_msg["subject"] == "subject"
assert self.catched_msg["body"] == "body"
msg = self.catched_msg["msg"]
self.assertEqual(msg["to"], "test@scrapy.org")
self.assertEqual(msg["subject"], "subject")
assert msg["to"] == "test@scrapy.org"
assert msg["subject"] == "subject"
payload = msg.get_payload()
assert isinstance(payload, list)
self.assertEqual(len(payload), 2)
assert len(payload) == 2
text, attach = payload
self.assertEqual(text.get_payload(decode=True), b"body")
self.assertEqual(text.get_charset(), Charset("us-ascii"))
self.assertEqual(attach.get_payload(decode=True), b"content")
assert text.get_payload(decode=True) == b"body"
assert text.get_charset() == Charset("us-ascii")
assert attach.get_payload(decode=True) == b"content"
def _catch_mail_sent(self, **kwargs):
self.catched_msg = {**kwargs}
@ -103,14 +102,14 @@ class MailSenderTest(unittest.TestCase):
)
assert self.catched_msg
self.assertEqual(self.catched_msg["subject"], subject)
self.assertEqual(self.catched_msg["body"], body)
assert self.catched_msg["subject"] == subject
assert self.catched_msg["body"] == body
msg = self.catched_msg["msg"]
self.assertEqual(msg["subject"], subject)
self.assertEqual(msg.get_payload(decode=True).decode("utf-8"), body)
self.assertEqual(msg.get_charset(), Charset("utf-8"))
self.assertEqual(msg.get("Content-Type"), 'text/plain; charset="utf-8"')
assert msg["subject"] == subject
assert msg.get_payload(decode=True).decode("utf-8") == body
assert msg.get_charset() == Charset("utf-8")
assert msg.get("Content-Type") == 'text/plain; charset="utf-8"'
def test_send_attach_utf8(self):
subject = "sübjèçt"
@ -131,22 +130,22 @@ class MailSenderTest(unittest.TestCase):
)
assert self.catched_msg
self.assertEqual(self.catched_msg["subject"], subject)
self.assertEqual(self.catched_msg["body"], body)
assert self.catched_msg["subject"] == subject
assert self.catched_msg["body"] == body
msg = self.catched_msg["msg"]
self.assertEqual(msg["subject"], subject)
self.assertEqual(msg.get_charset(), Charset("utf-8"))
self.assertEqual(msg.get("Content-Type"), 'multipart/mixed; charset="utf-8"')
assert msg["subject"] == subject
assert msg.get_charset() == Charset("utf-8")
assert msg.get("Content-Type") == 'multipart/mixed; charset="utf-8"'
payload = msg.get_payload()
assert isinstance(payload, list)
self.assertEqual(len(payload), 2)
assert len(payload) == 2
text, attach = payload
self.assertEqual(text.get_payload(decode=True).decode("utf-8"), body)
self.assertEqual(text.get_charset(), Charset("utf-8"))
self.assertEqual(attach.get_payload(decode=True).decode("utf-8"), body)
assert text.get_payload(decode=True).decode("utf-8") == body
assert text.get_charset() == Charset("utf-8")
assert attach.get_payload(decode=True).decode("utf-8") == body
def test_create_sender_factory_with_host(self):
mailsender = MailSender(debug=False, smtphost="smtp.testhost.com")
@ -156,4 +155,4 @@ class MailSenderTest(unittest.TestCase):
)
context = factory.buildProtocol("test@scrapy.org").context
self.assertIsInstance(context, ClientTLSOptions)
assert isinstance(context, ClientTLSOptions)

30
tests/test_middleware.py
View File

@ -1,5 +1,3 @@
from twisted.trial import unittest
from scrapy.exceptions import NotConfigured
from scrapy.middleware import MiddlewareManager
from scrapy.utils.test import get_crawler
@ -51,37 +49,27 @@ class MyMiddlewareManager(MiddlewareManager):
self.methods["process"].append(mw.process)
class MiddlewareManagerTest(unittest.TestCase):
class TestMiddlewareManager:
def test_init(self):
m1, m2, m3 = M1(), M2(), M3()
mwman = MyMiddlewareManager(m1, m2, m3)
self.assertEqual(
list(mwman.methods["open_spider"]), [m1.open_spider, m2.open_spider]
)
self.assertEqual(
list(mwman.methods["close_spider"]), [m2.close_spider, m1.close_spider]
)
self.assertEqual(list(mwman.methods["process"]), [m1.process, m3.process])
assert list(mwman.methods["open_spider"]) == [m1.open_spider, m2.open_spider]
assert list(mwman.methods["close_spider"]) == [m2.close_spider, m1.close_spider]
assert list(mwman.methods["process"]) == [m1.process, m3.process]
def test_methods(self):
mwman = MyMiddlewareManager(M1(), M2(), M3())
self.assertEqual(
[x.__self__.__class__ for x in mwman.methods["open_spider"]], [M1, M2]
)
self.assertEqual(
[x.__self__.__class__ for x in mwman.methods["close_spider"]], [M2, M1]
)
self.assertEqual(
[x.__self__.__class__ for x in mwman.methods["process"]], [M1, M3]
)
assert [x.__self__.__class__ for x in mwman.methods["open_spider"]] == [M1, M2]
assert [x.__self__.__class__ for x in mwman.methods["close_spider"]] == [M2, M1]
assert [x.__self__.__class__ for x in mwman.methods["process"]] == [M1, M3]
def test_enabled(self):
m1, m2, m3 = M1(), M2(), M3()
mwman = MiddlewareManager(m1, m2, m3)
self.assertEqual(mwman.middlewares, (m1, m2, m3))
assert mwman.middlewares == (m1, m2, m3)
def test_enabled_from_settings(self):
crawler = get_crawler()
mwman = MyMiddlewareManager.from_crawler(crawler)
classes = [x.__class__ for x in mwman.middlewares]
self.assertEqual(classes, [M1, M3])
assert classes == [M1, M3]

16
tests/test_pipelines.py
View File

@ -76,7 +76,7 @@ class ItemSpider(Spider):
return {"field": 42}
class PipelineTestCase(unittest.TestCase):
class TestPipeline(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.mockserver = MockServer()
@ -87,8 +87,8 @@ class PipelineTestCase(unittest.TestCase):
cls.mockserver.__exit__(None, None, None)
def _on_item_scraped(self, item):
self.assertIsInstance(item, dict)
self.assertTrue(item.get("pipeline_passed"))
assert isinstance(item, dict)
assert item.get("pipeline_passed")
self.items.append(item)
def _create_crawler(self, pipeline_class):
@ -104,30 +104,30 @@ class PipelineTestCase(unittest.TestCase):
def test_simple_pipeline(self):
crawler = self._create_crawler(SimplePipeline)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 1)
assert len(self.items) == 1
@defer.inlineCallbacks
def test_deferred_pipeline(self):
crawler = self._create_crawler(DeferredPipeline)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 1)
assert len(self.items) == 1
@defer.inlineCallbacks
def test_asyncdef_pipeline(self):
crawler = self._create_crawler(AsyncDefPipeline)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 1)
assert len(self.items) == 1
@pytest.mark.only_asyncio
@defer.inlineCallbacks
def test_asyncdef_asyncio_pipeline(self):
crawler = self._create_crawler(AsyncDefAsyncioPipeline)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 1)
assert len(self.items) == 1
@pytest.mark.only_not_asyncio
@defer.inlineCallbacks
def test_asyncdef_not_asyncio_pipeline(self):
crawler = self._create_crawler(AsyncDefNotAsyncioPipeline)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 1)
assert len(self.items) == 1

113
tests/test_pqueues.py
View File

@ -1,5 +1,4 @@
import tempfile
import unittest
import pytest
import queuelib
@ -12,8 +11,8 @@ from scrapy.utils.test import get_crawler
from tests.test_scheduler import MockDownloader, MockEngine
class PriorityQueueTest(unittest.TestCase):
def setUp(self):
class TestPriorityQueue:
def setup_method(self):
self.crawler = get_crawler(Spider)
self.spider = self.crawler._create_spider("foo")
@ -22,20 +21,20 @@ class PriorityQueueTest(unittest.TestCase):
queue = ScrapyPriorityQueue.from_crawler(
self.crawler, FifoMemoryQueue, temp_dir
)
self.assertIsNone(queue.pop())
self.assertEqual(len(queue), 0)
assert queue.pop() is None
assert len(queue) == 0
req1 = Request("https://example.org/1", priority=1)
queue.push(req1)
self.assertEqual(len(queue), 1)
assert len(queue) == 1
dequeued = queue.pop()
self.assertEqual(len(queue), 0)
self.assertEqual(dequeued.url, req1.url)
self.assertEqual(dequeued.priority, req1.priority)
self.assertEqual(queue.close(), [])
assert len(queue) == 0
assert dequeued.url == req1.url
assert dequeued.priority == req1.priority
assert not queue.close()
def test_no_peek_raises(self):
if hasattr(queuelib.queue.FifoMemoryQueue, "peek"):
raise unittest.SkipTest("queuelib.queue.FifoMemoryQueue.peek is defined")
pytest.skip("queuelib.queue.FifoMemoryQueue.peek is defined")
temp_dir = tempfile.mkdtemp()
queue = ScrapyPriorityQueue.from_crawler(
self.crawler, FifoMemoryQueue, temp_dir
@ -50,53 +49,53 @@ class PriorityQueueTest(unittest.TestCase):
def test_peek(self):
if not hasattr(queuelib.queue.FifoMemoryQueue, "peek"):
raise unittest.SkipTest("queuelib.queue.FifoMemoryQueue.peek is undefined")
pytest.skip("queuelib.queue.FifoMemoryQueue.peek is undefined")
temp_dir = tempfile.mkdtemp()
queue = ScrapyPriorityQueue.from_crawler(
self.crawler, FifoMemoryQueue, temp_dir
)
self.assertEqual(len(queue), 0)
self.assertIsNone(queue.peek())
assert len(queue) == 0
assert queue.peek() is None
req1 = Request("https://example.org/1")
req2 = Request("https://example.org/2")
req3 = Request("https://example.org/3")
queue.push(req1)
queue.push(req2)
queue.push(req3)
self.assertEqual(len(queue), 3)
self.assertEqual(queue.peek().url, req1.url)
self.assertEqual(queue.pop().url, req1.url)
self.assertEqual(len(queue), 2)
self.assertEqual(queue.peek().url, req2.url)
self.assertEqual(queue.pop().url, req2.url)
self.assertEqual(len(queue), 1)
self.assertEqual(queue.peek().url, req3.url)
self.assertEqual(queue.pop().url, req3.url)
self.assertEqual(queue.close(), [])
assert len(queue) == 3
assert queue.peek().url == req1.url
assert queue.pop().url == req1.url
assert len(queue) == 2
assert queue.peek().url == req2.url
assert queue.pop().url == req2.url
assert len(queue) == 1
assert queue.peek().url == req3.url
assert queue.pop().url == req3.url
assert not queue.close()
def test_queue_push_pop_priorities(self):
temp_dir = tempfile.mkdtemp()
queue = ScrapyPriorityQueue.from_crawler(
self.crawler, FifoMemoryQueue, temp_dir, [-1, -2, -3]
)
self.assertIsNone(queue.pop())
self.assertEqual(len(queue), 0)
assert queue.pop() is None
assert len(queue) == 0
req1 = Request("https://example.org/1", priority=1)
req2 = Request("https://example.org/2", priority=2)
req3 = Request("https://example.org/3", priority=3)
queue.push(req1)
queue.push(req2)
queue.push(req3)
self.assertEqual(len(queue), 3)
assert len(queue) == 3
dequeued = queue.pop()
self.assertEqual(len(queue), 2)
self.assertEqual(dequeued.url, req3.url)
self.assertEqual(dequeued.priority, req3.priority)
self.assertEqual(queue.close(), [-1, -2])
assert len(queue) == 2
assert dequeued.url == req3.url
assert dequeued.priority == req3.priority
assert queue.close() == [-1, -2]
class DownloaderAwarePriorityQueueTest(unittest.TestCase):
def setUp(self):
class TestDownloaderAwarePriorityQueue:
def setup_method(self):
crawler = get_crawler(Spider)
crawler.engine = MockEngine(downloader=MockDownloader())
self.queue = DownloaderAwarePriorityQueue.from_crawler(
@ -105,30 +104,30 @@ class DownloaderAwarePriorityQueueTest(unittest.TestCase):
key="foo/bar",
)
def tearDown(self):
def teardown_method(self):
self.queue.close()
def test_push_pop(self):
self.assertEqual(len(self.queue), 0)
self.assertIsNone(self.queue.pop())
assert len(self.queue) == 0
assert self.queue.pop() is None
req1 = Request("http://www.example.com/1")
req2 = Request("http://www.example.com/2")
req3 = Request("http://www.example.com/3")
self.queue.push(req1)
self.queue.push(req2)
self.queue.push(req3)
self.assertEqual(len(self.queue), 3)
self.assertEqual(self.queue.pop().url, req1.url)
self.assertEqual(len(self.queue), 2)
self.assertEqual(self.queue.pop().url, req2.url)
self.assertEqual(len(self.queue), 1)
self.assertEqual(self.queue.pop().url, req3.url)
self.assertEqual(len(self.queue), 0)
self.assertIsNone(self.queue.pop())
assert len(self.queue) == 3
assert self.queue.pop().url == req1.url
assert len(self.queue) == 2
assert self.queue.pop().url == req2.url
assert len(self.queue) == 1
assert self.queue.pop().url == req3.url
assert len(self.queue) == 0
assert self.queue.pop() is None
def test_no_peek_raises(self):
if hasattr(queuelib.queue.FifoMemoryQueue, "peek"):
raise unittest.SkipTest("queuelib.queue.FifoMemoryQueue.peek is defined")
pytest.skip("queuelib.queue.FifoMemoryQueue.peek is defined")
self.queue.push(Request("https://example.org"))
with pytest.raises(
NotImplementedError,
@ -138,21 +137,21 @@ class DownloaderAwarePriorityQueueTest(unittest.TestCase):
def test_peek(self):
if not hasattr(queuelib.queue.FifoMemoryQueue, "peek"):
raise unittest.SkipTest("queuelib.queue.FifoMemoryQueue.peek is undefined")
self.assertEqual(len(self.queue), 0)
pytest.skip("queuelib.queue.FifoMemoryQueue.peek is undefined")
assert len(self.queue) == 0
req1 = Request("https://example.org/1")
req2 = Request("https://example.org/2")
req3 = Request("https://example.org/3")
self.queue.push(req1)
self.queue.push(req2)
self.queue.push(req3)
self.assertEqual(len(self.queue), 3)
self.assertEqual(self.queue.peek().url, req1.url)
self.assertEqual(self.queue.pop().url, req1.url)
self.assertEqual(len(self.queue), 2)
self.assertEqual(self.queue.peek().url, req2.url)
self.assertEqual(self.queue.pop().url, req2.url)
self.assertEqual(len(self.queue), 1)
self.assertEqual(self.queue.peek().url, req3.url)
self.assertEqual(self.queue.pop().url, req3.url)
self.assertIsNone(self.queue.peek())
assert len(self.queue) == 3
assert self.queue.peek().url == req1.url
assert self.queue.pop().url == req1.url
assert len(self.queue) == 2
assert self.queue.peek().url == req2.url
assert self.queue.pop().url == req2.url
assert len(self.queue) == 1
assert self.queue.peek().url == req3.url
assert self.queue.pop().url == req3.url
assert self.queue.peek() is None

11
tests/test_proxy_connect.py
View File

@ -6,6 +6,7 @@ from pathlib import Path
from subprocess import PIPE, Popen
from urllib.parse import urlsplit, urlunsplit
import pytest
from testfixtures import LogCapture
from twisted.internet import defer
from twisted.trial.unittest import TestCase
@ -61,7 +62,7 @@ def _wrong_credentials(proxy_url):
return urlunsplit(bad_auth_proxy)
class ProxyConnectTestCase(TestCase):
class TestProxyConnect(TestCase):
@classmethod
def setUpClass(cls):
cls.mockserver = MockServer()
@ -75,7 +76,7 @@ class ProxyConnectTestCase(TestCase):
try:
import mitmproxy # noqa: F401
except ImportError:
self.skipTest("mitmproxy is not installed")
pytest.skip("mitmproxy is not installed")
self._oldenv = os.environ.copy()
@ -113,12 +114,12 @@ class ProxyConnectTestCase(TestCase):
yield crawler.crawl(seed=request)
self._assert_got_response_code(200, log)
echo = json.loads(crawler.spider.meta["responses"][0].text)
self.assertTrue("Proxy-Authorization" not in echo["headers"])
assert "Proxy-Authorization" not in echo["headers"]
def _assert_got_response_code(self, code, log):
print(log)
self.assertEqual(str(log).count(f"Crawled ({code})"), 1)
assert str(log).count(f"Crawled ({code})") == 1
def _assert_got_tunnel_error(self, log):
print(log)
self.assertIn("TunnelError", str(log))
assert "TunnelError" in str(log)

26
tests/test_request_attribute_binding.py
View File

@ -56,7 +56,7 @@ class AlternativeCallbacksMiddleware:
return response.replace(request=new_request)
class CrawlTestCase(TestCase):
class TestCrawl(TestCase):
@classmethod
def setUpClass(cls):
cls.mockserver = MockServer()
@ -72,7 +72,7 @@ class CrawlTestCase(TestCase):
crawler = get_crawler(SingleRequestSpider)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.request.url, url)
assert response.request.url == url
@defer.inlineCallbacks
def test_response_error(self):
@ -82,8 +82,8 @@ class CrawlTestCase(TestCase):
yield crawler.crawl(seed=url, mockserver=self.mockserver)
failure = crawler.spider.meta["failure"]
response = failure.value.response
self.assertEqual(failure.request.url, url)
self.assertEqual(response.request.url, url)
assert failure.request.url == url
assert response.request.url == url
@defer.inlineCallbacks
def test_downloader_middleware_raise_exception(self):
@ -98,8 +98,8 @@ class CrawlTestCase(TestCase):
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
failure = crawler.spider.meta["failure"]
self.assertEqual(failure.request.url, url)
self.assertIsInstance(failure.value, ZeroDivisionError)
assert failure.request.url == url
assert isinstance(failure.value, ZeroDivisionError)
@defer.inlineCallbacks
def test_downloader_middleware_override_request_in_process_response(self):
@ -131,10 +131,10 @@ class CrawlTestCase(TestCase):
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.request.url, OVERRIDDEN_URL)
assert response.request.url == OVERRIDDEN_URL
self.assertEqual(signal_params["response"].url, url)
self.assertEqual(signal_params["request"].url, OVERRIDDEN_URL)
assert signal_params["response"].url == url
assert signal_params["request"].url == OVERRIDDEN_URL
log.check_present(
(
@ -164,8 +164,8 @@ class CrawlTestCase(TestCase):
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.body, b"Caught ZeroDivisionError")
self.assertEqual(response.request.url, OVERRIDDEN_URL)
assert response.body == b"Caught ZeroDivisionError"
assert response.request.url == OVERRIDDEN_URL
@defer.inlineCallbacks
def test_downloader_middleware_do_not_override_in_process_exception(self):
@ -187,8 +187,8 @@ class CrawlTestCase(TestCase):
)
yield crawler.crawl(seed=url, mockserver=self.mockserver)
response = crawler.spider.meta["responses"][0]
self.assertEqual(response.body, b"Caught ZeroDivisionError")
self.assertEqual(response.request.url, url)
assert response.body == b"Caught ZeroDivisionError"
assert response.request.url == url
@defer.inlineCallbacks
def test_downloader_middleware_alternative_callback(self):

30
tests/test_request_cb_kwargs.py
View File

@ -151,7 +151,7 @@ class KeywordArgumentsSpider(MockServerSpider):
self.crawler.stats.inc_value("boolean_checks", 1)
class CallbackKeywordArgumentsTestCase(TestCase):
class TestCallbackKeywordArguments(TestCase):
maxDiff = None
@classmethod
@ -168,27 +168,19 @@ class CallbackKeywordArgumentsTestCase(TestCase):
crawler = get_crawler(KeywordArgumentsSpider)
with LogCapture() as log:
yield crawler.crawl(mockserver=self.mockserver)
self.assertTrue(all(crawler.spider.checks))
self.assertEqual(
len(crawler.spider.checks), crawler.stats.get_value("boolean_checks")
)
assert all(crawler.spider.checks)
assert len(crawler.spider.checks) == crawler.stats.get_value("boolean_checks")
# check exceptions for argument mismatch
exceptions = {}
for line in log.records:
for key in ("takes_less", "takes_more"):
if key in line.getMessage():
exceptions[key] = line
self.assertEqual(exceptions["takes_less"].exc_info[0], TypeError)
self.assertTrue(
str(exceptions["takes_less"].exc_info[1]).endswith(
"parse_takes_less() got an unexpected keyword argument 'number'"
),
msg="Exception message: " + str(exceptions["takes_less"].exc_info[1]),
)
self.assertEqual(exceptions["takes_more"].exc_info[0], TypeError)
self.assertTrue(
str(exceptions["takes_more"].exc_info[1]).endswith(
"parse_takes_more() missing 1 required positional argument: 'other'"
),
msg="Exception message: " + str(exceptions["takes_more"].exc_info[1]),
)
assert exceptions["takes_less"].exc_info[0] is TypeError
assert str(exceptions["takes_less"].exc_info[1]).endswith(
"parse_takes_less() got an unexpected keyword argument 'number'"
), "Exception message: " + str(exceptions["takes_less"].exc_info[1])
assert exceptions["takes_more"].exc_info[0] is TypeError
assert str(exceptions["takes_more"].exc_info[1]).endswith(
"parse_takes_more() missing 1 required positional argument: 'other'"
), "Exception message: " + str(exceptions["takes_more"].exc_info[1])

50
tests/test_request_dict.py
View File

@ -1,5 +1,3 @@
import unittest
import pytest
from scrapy import Request, Spider
@ -11,8 +9,8 @@ class CustomRequest(Request):
pass
class RequestSerializationTest(unittest.TestCase):
def setUp(self):
class TestRequestSerialization:
def setup_method(self):
self.spider = MethodsSpider()
def test_basic(self):
@ -50,23 +48,23 @@ class RequestSerializationTest(unittest.TestCase):
self._assert_same_request(request, request2)
def _assert_same_request(self, r1, r2):
self.assertEqual(r1.__class__, r2.__class__)
self.assertEqual(r1.url, r2.url)
self.assertEqual(r1.callback, r2.callback)
self.assertEqual(r1.errback, r2.errback)
self.assertEqual(r1.method, r2.method)
self.assertEqual(r1.body, r2.body)
self.assertEqual(r1.headers, r2.headers)
self.assertEqual(r1.cookies, r2.cookies)
self.assertEqual(r1.meta, r2.meta)
self.assertEqual(r1.cb_kwargs, r2.cb_kwargs)
self.assertEqual(r1.encoding, r2.encoding)
self.assertEqual(r1._encoding, r2._encoding)
self.assertEqual(r1.priority, r2.priority)
self.assertEqual(r1.dont_filter, r2.dont_filter)
self.assertEqual(r1.flags, r2.flags)
assert r1.__class__ == r2.__class__
assert r1.url == r2.url
assert r1.callback == r2.callback
assert r1.errback == r2.errback
assert r1.method == r2.method
assert r1.body == r2.body
assert r1.headers == r2.headers
assert r1.cookies == r2.cookies
assert r1.meta == r2.meta
assert r1.cb_kwargs == r2.cb_kwargs
assert r1.encoding == r2.encoding
assert r1._encoding == r2._encoding
assert r1.priority == r2.priority
assert r1.dont_filter == r2.dont_filter
assert r1.flags == r2.flags
if isinstance(r1, JsonRequest):
self.assertEqual(r1.dumps_kwargs, r2.dumps_kwargs)
assert r1.dumps_kwargs == r2.dumps_kwargs
def test_request_class(self):
r1 = FormRequest("http://www.example.com")
@ -92,8 +90,8 @@ class RequestSerializationTest(unittest.TestCase):
)
self._assert_serializes_ok(r, spider=self.spider)
request_dict = r.to_dict(spider=self.spider)
self.assertEqual(request_dict["callback"], "parse_item_reference")
self.assertEqual(request_dict["errback"], "handle_error_reference")
assert request_dict["callback"] == "parse_item_reference"
assert request_dict["errback"] == "handle_error_reference"
def test_private_reference_callback_serialization(self):
r = Request(
@ -103,12 +101,8 @@ class RequestSerializationTest(unittest.TestCase):
)
self._assert_serializes_ok(r, spider=self.spider)
request_dict = r.to_dict(spider=self.spider)
self.assertEqual(
request_dict["callback"], "_MethodsSpider__parse_item_reference"
)
self.assertEqual(
request_dict["errback"], "_MethodsSpider__handle_error_reference"
)
assert request_dict["callback"] == "_MethodsSpider__parse_item_reference"
assert request_dict["errback"] == "_MethodsSpider__handle_error_reference"
def test_private_callback_serialization(self):
r = Request(

8
tests/test_request_left.py
View File

@ -38,22 +38,22 @@ class TestCatching(TestCase):
def test_success(self):
crawler = get_crawler(SignalCatcherSpider)
yield crawler.crawl(self.mockserver.url("/status?n=200"))
self.assertEqual(crawler.spider.caught_times, 1)
assert crawler.spider.caught_times == 1
@defer.inlineCallbacks
def test_timeout(self):
crawler = get_crawler(SignalCatcherSpider, {"DOWNLOAD_TIMEOUT": 0.1})
yield crawler.crawl(self.mockserver.url("/delay?n=0.2"))
self.assertEqual(crawler.spider.caught_times, 1)
assert crawler.spider.caught_times == 1
@defer.inlineCallbacks
def test_disconnect(self):
crawler = get_crawler(SignalCatcherSpider)
yield crawler.crawl(self.mockserver.url("/drop"))
self.assertEqual(crawler.spider.caught_times, 1)
assert crawler.spider.caught_times == 1
@defer.inlineCallbacks
def test_noconnect(self):
crawler = get_crawler(SignalCatcherSpider)
yield crawler.crawl("http://thereisdefinetelynosuchdomain.com")
self.assertEqual(crawler.spider.caught_times, 1)
assert crawler.spider.caught_times == 1

8
tests/test_responsetypes.py
View File

@ -1,5 +1,3 @@
import unittest
from scrapy.http import (
Headers,
HtmlResponse,
@ -11,7 +9,7 @@ from scrapy.http import (
from scrapy.responsetypes import responsetypes
class ResponseTypesTest(unittest.TestCase):
class TestResponseTypes:
def test_from_filename(self):
mappings = [
("data.bin", Response),
@ -123,6 +121,4 @@ class ResponseTypesTest(unittest.TestCase):
def test_custom_mime_types_loaded(self):
# check that mime.types files shipped with scrapy are loaded
self.assertEqual(
responsetypes.mimetypes.guess_type("x.scrapytest")[0], "x-scrapy/test"
)
assert responsetypes.mimetypes.guess_type("x.scrapytest")[0] == "x-scrapy/test"

106
tests/test_robotstxt_interface.py
View File

@ -1,4 +1,4 @@
from twisted.trial import unittest
import pytest
from scrapy.robotstxt import decode_robotstxt
@ -32,8 +32,8 @@ class BaseRobotParserTest:
rp = self.parser_cls.from_crawler(
crawler=None, robotstxt_body=robotstxt_robotstxt_body
)
self.assertTrue(rp.allowed("https://www.site.local/allowed", "*"))
self.assertFalse(rp.allowed("https://www.site.local/disallowed", "*"))
assert rp.allowed("https://www.site.local/allowed", "*")
assert not rp.allowed("https://www.site.local/disallowed", "*")
def test_allowed_wildcards(self):
robotstxt_robotstxt_body = b"""User-agent: first
@ -47,42 +47,36 @@ class BaseRobotParserTest:
crawler=None, robotstxt_body=robotstxt_robotstxt_body
)
self.assertTrue(rp.allowed("https://www.site.local/disallowed", "first"))
self.assertFalse(
rp.allowed("https://www.site.local/disallowed/xyz/end", "first")
)
self.assertFalse(
rp.allowed("https://www.site.local/disallowed/abc/end", "first")
)
self.assertTrue(
rp.allowed("https://www.site.local/disallowed/xyz/endinglater", "first")
)
assert rp.allowed("https://www.site.local/disallowed", "first")
assert not rp.allowed("https://www.site.local/disallowed/xyz/end", "first")
assert not rp.allowed("https://www.site.local/disallowed/abc/end", "first")
assert rp.allowed("https://www.site.local/disallowed/xyz/endinglater", "first")
self.assertTrue(rp.allowed("https://www.site.local/allowed", "second"))
self.assertTrue(rp.allowed("https://www.site.local/is_still_allowed", "second"))
self.assertTrue(rp.allowed("https://www.site.local/is_allowed_too", "second"))
assert rp.allowed("https://www.site.local/allowed", "second")
assert rp.allowed("https://www.site.local/is_still_allowed", "second")
assert rp.allowed("https://www.site.local/is_allowed_too", "second")
def test_length_based_precedence(self):
robotstxt_robotstxt_body = b"User-agent: * \nDisallow: / \nAllow: /page"
rp = self.parser_cls.from_crawler(
crawler=None, robotstxt_body=robotstxt_robotstxt_body
)
self.assertTrue(rp.allowed("https://www.site.local/page", "*"))
assert rp.allowed("https://www.site.local/page", "*")
def test_order_based_precedence(self):
robotstxt_robotstxt_body = b"User-agent: * \nDisallow: / \nAllow: /page"
rp = self.parser_cls.from_crawler(
crawler=None, robotstxt_body=robotstxt_robotstxt_body
)
self.assertFalse(rp.allowed("https://www.site.local/page", "*"))
assert not rp.allowed("https://www.site.local/page", "*")
def test_empty_response(self):
"""empty response should equal 'allow all'"""
rp = self.parser_cls.from_crawler(crawler=None, robotstxt_body=b"")
self.assertTrue(rp.allowed("https://site.local/", "*"))
self.assertTrue(rp.allowed("https://site.local/", "chrome"))
self.assertTrue(rp.allowed("https://site.local/index.html", "*"))
self.assertTrue(rp.allowed("https://site.local/disallowed", "*"))
assert rp.allowed("https://site.local/", "*")
assert rp.allowed("https://site.local/", "chrome")
assert rp.allowed("https://site.local/index.html", "*")
assert rp.allowed("https://site.local/disallowed", "*")
def test_garbage_response(self):
"""garbage response should be discarded, equal 'allow all'"""
@ -90,10 +84,10 @@ class BaseRobotParserTest:
rp = self.parser_cls.from_crawler(
crawler=None, robotstxt_body=robotstxt_robotstxt_body
)
self.assertTrue(rp.allowed("https://site.local/", "*"))
self.assertTrue(rp.allowed("https://site.local/", "chrome"))
self.assertTrue(rp.allowed("https://site.local/index.html", "*"))
self.assertTrue(rp.allowed("https://site.local/disallowed", "*"))
assert rp.allowed("https://site.local/", "*")
assert rp.allowed("https://site.local/", "chrome")
assert rp.allowed("https://site.local/index.html", "*")
assert rp.allowed("https://site.local/disallowed", "*")
def test_unicode_url_and_useragent(self):
robotstxt_robotstxt_body = """
@ -109,79 +103,67 @@ class BaseRobotParserTest:
rp = self.parser_cls.from_crawler(
crawler=None, robotstxt_body=robotstxt_robotstxt_body
)
self.assertTrue(rp.allowed("https://site.local/", "*"))
self.assertFalse(rp.allowed("https://site.local/admin/", "*"))
self.assertFalse(rp.allowed("https://site.local/static/", "*"))
self.assertTrue(rp.allowed("https://site.local/admin/", "UnicödeBöt"))
self.assertFalse(
rp.allowed("https://site.local/wiki/K%C3%A4ytt%C3%A4j%C3%A4:", "*")
)
self.assertFalse(rp.allowed("https://site.local/wiki/Käyttäjä:", "*"))
self.assertTrue(rp.allowed("https://site.local/some/randome/page.html", "*"))
self.assertFalse(
rp.allowed("https://site.local/some/randome/page.html", "UnicödeBöt")
)
assert rp.allowed("https://site.local/", "*")
assert not rp.allowed("https://site.local/admin/", "*")
assert not rp.allowed("https://site.local/static/", "*")
assert rp.allowed("https://site.local/admin/", "UnicödeBöt")
assert not rp.allowed("https://site.local/wiki/K%C3%A4ytt%C3%A4j%C3%A4:", "*")
assert not rp.allowed("https://site.local/wiki/Käyttäjä:", "*")
assert rp.allowed("https://site.local/some/randome/page.html", "*")
assert not rp.allowed("https://site.local/some/randome/page.html", "UnicödeBöt")
class DecodeRobotsTxtTest(unittest.TestCase):
class TestDecodeRobotsTxt:
def test_native_string_conversion(self):
robotstxt_body = b"User-agent: *\nDisallow: /\n"
decoded_content = decode_robotstxt(
robotstxt_body, spider=None, to_native_str_type=True
)
self.assertEqual(decoded_content, "User-agent: *\nDisallow: /\n")
assert decoded_content == "User-agent: *\nDisallow: /\n"
def test_decode_utf8(self):
robotstxt_body = b"User-agent: *\nDisallow: /\n"
decoded_content = decode_robotstxt(robotstxt_body, spider=None)
self.assertEqual(decoded_content, "User-agent: *\nDisallow: /\n")
assert decoded_content == "User-agent: *\nDisallow: /\n"
def test_decode_non_utf8(self):
robotstxt_body = b"User-agent: *\n\xffDisallow: /\n"
decoded_content = decode_robotstxt(robotstxt_body, spider=None)
self.assertEqual(decoded_content, "User-agent: *\nDisallow: /\n")
assert decoded_content == "User-agent: *\nDisallow: /\n"
class PythonRobotParserTest(BaseRobotParserTest, unittest.TestCase):
def setUp(self):
class TestPythonRobotParser(BaseRobotParserTest):
def setup_method(self):
from scrapy.robotstxt import PythonRobotParser
super()._setUp(PythonRobotParser)
def test_length_based_precedence(self):
raise unittest.SkipTest(
pytest.skip(
"RobotFileParser does not support length based directives precedence."
)
def test_allowed_wildcards(self):
raise unittest.SkipTest("RobotFileParser does not support wildcards.")
pytest.skip("RobotFileParser does not support wildcards.")
class RerpRobotParserTest(BaseRobotParserTest, unittest.TestCase):
if not rerp_available():
skip = "Rerp parser is not installed"
def setUp(self):
@pytest.mark.skipif(not rerp_available(), reason="Rerp parser is not installed")
class TestRerpRobotParser(BaseRobotParserTest):
def setup_method(self):
from scrapy.robotstxt import RerpRobotParser
super()._setUp(RerpRobotParser)
def test_length_based_precedence(self):
raise unittest.SkipTest(
"Rerp does not support length based directives precedence."
)
pytest.skip("Rerp does not support length based directives precedence.")
class ProtegoRobotParserTest(BaseRobotParserTest, unittest.TestCase):
if not protego_available():
skip = "Protego parser is not installed"
def setUp(self):
@pytest.mark.skipif(not protego_available(), reason="Protego parser is not installed")
class TestProtegoRobotParser(BaseRobotParserTest):
def setup_method(self):
from scrapy.robotstxt import ProtegoRobotParser
super()._setUp(ProtegoRobotParser)
def test_order_based_precedence(self):
raise unittest.SkipTest(
"Protego does not support order based directives precedence."
)
pytest.skip("Protego does not support order based directives precedence.")

119
tests/test_scheduler.py
View File

@ -2,7 +2,7 @@ from __future__ import annotations
import shutil
import tempfile
import unittest
from abc import ABC, abstractmethod
from typing import Any, NamedTuple
import pytest
@ -65,10 +65,14 @@ class MockCrawler(Crawler):
self.stats = load_object(self.settings["STATS_CLASS"])(self)
class SchedulerHandler:
priority_queue_cls: str | None = None
class SchedulerHandler(ABC):
jobdir = None
@property
@abstractmethod
def priority_queue_cls(self) -> str:
raise NotImplementedError
def create_scheduler(self):
self.mock_crawler = MockCrawler(self.priority_queue_cls, self.jobdir)
self.scheduler = Scheduler.from_crawler(self.mock_crawler)
@ -80,10 +84,10 @@ class SchedulerHandler:
self.mock_crawler.stop()
self.mock_crawler.engine.downloader.close()
def setUp(self):
def setup_method(self):
self.create_scheduler()
def tearDown(self):
def teardown_method(self):
self.close_scheduler()
@ -99,16 +103,16 @@ _PRIORITIES = [
_URLS = {"http://foo.com/a", "http://foo.com/b", "http://foo.com/c"}
class BaseSchedulerInMemoryTester(SchedulerHandler):
class TestSchedulerInMemoryBase(SchedulerHandler):
def test_length(self):
self.assertFalse(self.scheduler.has_pending_requests())
self.assertEqual(len(self.scheduler), 0)
assert not self.scheduler.has_pending_requests()
assert len(self.scheduler) == 0
for url in _URLS:
self.scheduler.enqueue_request(Request(url))
self.assertTrue(self.scheduler.has_pending_requests())
self.assertEqual(len(self.scheduler), len(_URLS))
assert self.scheduler.has_pending_requests()
assert len(self.scheduler) == len(_URLS)
def test_dequeue(self):
for url in _URLS:
@ -118,7 +122,7 @@ class BaseSchedulerInMemoryTester(SchedulerHandler):
while self.scheduler.has_pending_requests():
urls.add(self.scheduler.next_request().url)
self.assertEqual(urls, _URLS)
assert urls == _URLS
def test_dequeue_priorities(self):
for url, priority in _PRIORITIES:
@ -128,25 +132,23 @@ class BaseSchedulerInMemoryTester(SchedulerHandler):
while self.scheduler.has_pending_requests():
priorities.append(self.scheduler.next_request().priority)
self.assertEqual(
priorities, sorted([x[1] for x in _PRIORITIES], key=lambda x: -x)
)
assert priorities == sorted([x[1] for x in _PRIORITIES], key=lambda x: -x)
class BaseSchedulerOnDiskTester(SchedulerHandler):
def setUp(self):
class TestSchedulerOnDiskBase(SchedulerHandler):
def setup_method(self):
self.jobdir = tempfile.mkdtemp()
self.create_scheduler()
def tearDown(self):
def teardown_method(self):
self.close_scheduler()
shutil.rmtree(self.jobdir)
self.jobdir = None
def test_length(self):
self.assertFalse(self.scheduler.has_pending_requests())
self.assertEqual(len(self.scheduler), 0)
assert not self.scheduler.has_pending_requests()
assert len(self.scheduler) == 0
for url in _URLS:
self.scheduler.enqueue_request(Request(url))
@ -154,8 +156,8 @@ class BaseSchedulerOnDiskTester(SchedulerHandler):
self.close_scheduler()
self.create_scheduler()
self.assertTrue(self.scheduler.has_pending_requests())
self.assertEqual(len(self.scheduler), len(_URLS))
assert self.scheduler.has_pending_requests()
assert len(self.scheduler) == len(_URLS)
def test_dequeue(self):
for url in _URLS:
@ -168,7 +170,7 @@ class BaseSchedulerOnDiskTester(SchedulerHandler):
while self.scheduler.has_pending_requests():
urls.add(self.scheduler.next_request().url)
self.assertEqual(urls, _URLS)
assert urls == _URLS
def test_dequeue_priorities(self):
for url, priority in _PRIORITIES:
@ -181,17 +183,19 @@ class BaseSchedulerOnDiskTester(SchedulerHandler):
while self.scheduler.has_pending_requests():
priorities.append(self.scheduler.next_request().priority)
self.assertEqual(
priorities, sorted([x[1] for x in _PRIORITIES], key=lambda x: -x)
)
assert priorities == sorted([x[1] for x in _PRIORITIES], key=lambda x: -x)
class TestSchedulerInMemory(BaseSchedulerInMemoryTester, unittest.TestCase):
priority_queue_cls = "scrapy.pqueues.ScrapyPriorityQueue"
class TestSchedulerInMemory(TestSchedulerInMemoryBase):
@property
def priority_queue_cls(self) -> str:
return "scrapy.pqueues.ScrapyPriorityQueue"
class TestSchedulerOnDisk(BaseSchedulerOnDiskTester, unittest.TestCase):
priority_queue_cls = "scrapy.pqueues.ScrapyPriorityQueue"
class TestSchedulerOnDisk(TestSchedulerOnDiskBase):
@property
def priority_queue_cls(self) -> str:
return "scrapy.pqueues.ScrapyPriorityQueue"
_URLS_WITH_SLOTS = [
@ -204,37 +208,34 @@ _URLS_WITH_SLOTS = [
]
class TestMigration(unittest.TestCase):
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
class TestMigration:
def test_migration(self, tmpdir):
class PrevSchedulerHandler(SchedulerHandler):
jobdir = tmpdir
def tearDown(self):
shutil.rmtree(self.tmpdir)
@property
def priority_queue_cls(self) -> str:
return "scrapy.pqueues.ScrapyPriorityQueue"
def _migration(self, tmp_dir):
prev_scheduler_handler = SchedulerHandler()
prev_scheduler_handler.priority_queue_cls = "scrapy.pqueues.ScrapyPriorityQueue"
prev_scheduler_handler.jobdir = tmp_dir
class NextSchedulerHandler(SchedulerHandler):
jobdir = tmpdir
@property
def priority_queue_cls(self) -> str:
return "scrapy.pqueues.DownloaderAwarePriorityQueue"
prev_scheduler_handler = PrevSchedulerHandler()
prev_scheduler_handler.create_scheduler()
for url in _URLS:
prev_scheduler_handler.scheduler.enqueue_request(Request(url))
prev_scheduler_handler.close_scheduler()
next_scheduler_handler = SchedulerHandler()
next_scheduler_handler.priority_queue_cls = (
"scrapy.pqueues.DownloaderAwarePriorityQueue"
)
next_scheduler_handler.jobdir = tmp_dir
next_scheduler_handler.create_scheduler()
def test_migration(self):
next_scheduler_handler = NextSchedulerHandler()
with pytest.raises(
ValueError,
match="DownloaderAwarePriorityQueue accepts ``slot_startprios`` as a dict",
):
self._migration(self.tmpdir)
next_scheduler_handler.create_scheduler()
def _is_scheduling_fair(enqueued_slots, dequeued_slots):
@ -263,9 +264,12 @@ def _is_scheduling_fair(enqueued_slots, dequeued_slots):
class DownloaderAwareSchedulerTestMixin:
priority_queue_cls: str | None = "scrapy.pqueues.DownloaderAwarePriorityQueue"
reopen = False
@property
def priority_queue_cls(self) -> str:
return "scrapy.pqueues.DownloaderAwarePriorityQueue"
def test_logic(self):
for url, slot in _URLS_WITH_SLOTS:
request = Request(url)
@ -290,20 +294,18 @@ class DownloaderAwareSchedulerTestMixin:
slot = downloader.get_slot_key(request)
downloader.decrement(slot)
self.assertTrue(
_is_scheduling_fair([s for u, s in _URLS_WITH_SLOTS], dequeued_slots)
)
self.assertEqual(sum(len(s.active) for s in downloader.slots.values()), 0)
assert _is_scheduling_fair([s for u, s in _URLS_WITH_SLOTS], dequeued_slots)
assert sum(len(s.active) for s in downloader.slots.values()) == 0
class TestSchedulerWithDownloaderAwareInMemory(
DownloaderAwareSchedulerTestMixin, BaseSchedulerInMemoryTester, unittest.TestCase
DownloaderAwareSchedulerTestMixin, TestSchedulerInMemoryBase
):
pass
class TestSchedulerWithDownloaderAwareOnDisk(
DownloaderAwareSchedulerTestMixin, BaseSchedulerOnDiskTester, unittest.TestCase
DownloaderAwareSchedulerTestMixin, TestSchedulerOnDiskBase
):
reopen = True
@ -337,13 +339,12 @@ class TestIntegrationWithDownloaderAwareInMemory(TestCase):
url = mockserver.url("/status?n=200", is_secure=False)
start_urls = [url] * 6
yield self.crawler.crawl(start_urls)
self.assertEqual(
self.crawler.stats.get_value("downloader/response_count"),
len(start_urls),
assert self.crawler.stats.get_value("downloader/response_count") == len(
start_urls
)
class TestIncompatibility(unittest.TestCase):
class TestIncompatibility:
def _incompatible(self):
settings = {
"SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.DownloaderAwarePriorityQueue",

59
tests/test_scheduler_base.py
View File

@ -1,12 +1,11 @@
from __future__ import annotations
from unittest import TestCase
from urllib.parse import urljoin
import pytest
from testfixtures import LogCapture
from twisted.internet import defer
from twisted.trial.unittest import TestCase as TwistedTestCase
from twisted.trial.unittest import TestCase
from scrapy.core.scheduler import BaseScheduler
from scrapy.http import Request
@ -65,17 +64,17 @@ class PathsSpider(Spider):
class InterfaceCheckMixin:
def test_scheduler_class(self):
self.assertTrue(isinstance(self.scheduler, BaseScheduler))
self.assertTrue(issubclass(self.scheduler.__class__, BaseScheduler))
assert isinstance(self.scheduler, BaseScheduler)
assert issubclass(self.scheduler.__class__, BaseScheduler)
class BaseSchedulerTest(TestCase, InterfaceCheckMixin):
def setUp(self):
class TestBaseScheduler(InterfaceCheckMixin):
def setup_method(self):
self.scheduler = BaseScheduler()
def test_methods(self):
self.assertIsNone(self.scheduler.open(Spider("foo")))
self.assertIsNone(self.scheduler.close("finished"))
assert self.scheduler.open(Spider("foo")) is None
assert self.scheduler.close("finished") is None
with pytest.raises(NotImplementedError):
self.scheduler.has_pending_requests()
with pytest.raises(NotImplementedError):
@ -84,8 +83,8 @@ class BaseSchedulerTest(TestCase, InterfaceCheckMixin):
self.scheduler.next_request()
class MinimalSchedulerTest(TestCase, InterfaceCheckMixin):
def setUp(self):
class TestMinimalScheduler(InterfaceCheckMixin):
def setup_method(self):
self.scheduler = MinimalScheduler()
def test_open_close(self):
@ -101,51 +100,51 @@ class MinimalSchedulerTest(TestCase, InterfaceCheckMixin):
len(self.scheduler)
def test_enqueue_dequeue(self):
self.assertFalse(self.scheduler.has_pending_requests())
assert not self.scheduler.has_pending_requests()
for url in URLS:
self.assertTrue(self.scheduler.enqueue_request(Request(url)))
self.assertFalse(self.scheduler.enqueue_request(Request(url)))
self.assertTrue(self.scheduler.has_pending_requests)
assert self.scheduler.enqueue_request(Request(url))
assert not self.scheduler.enqueue_request(Request(url))
assert self.scheduler.has_pending_requests
dequeued = []
while self.scheduler.has_pending_requests():
request = self.scheduler.next_request()
dequeued.append(request.url)
self.assertEqual(set(dequeued), set(URLS))
self.assertFalse(self.scheduler.has_pending_requests())
assert set(dequeued) == set(URLS)
assert not self.scheduler.has_pending_requests()
class SimpleSchedulerTest(TwistedTestCase, InterfaceCheckMixin):
class SimpleSchedulerTest(TestCase, InterfaceCheckMixin):
def setUp(self):
self.scheduler = SimpleScheduler()
@defer.inlineCallbacks
def test_enqueue_dequeue(self):
open_result = yield self.scheduler.open(Spider("foo"))
self.assertEqual(open_result, "open")
self.assertFalse(self.scheduler.has_pending_requests())
assert open_result == "open"
assert not self.scheduler.has_pending_requests()
for url in URLS:
self.assertTrue(self.scheduler.enqueue_request(Request(url)))
self.assertFalse(self.scheduler.enqueue_request(Request(url)))
assert self.scheduler.enqueue_request(Request(url))
assert not self.scheduler.enqueue_request(Request(url))
self.assertTrue(self.scheduler.has_pending_requests())
self.assertEqual(len(self.scheduler), len(URLS))
assert self.scheduler.has_pending_requests()
assert len(self.scheduler) == len(URLS)
dequeued = []
while self.scheduler.has_pending_requests():
request = self.scheduler.next_request()
dequeued.append(request.url)
self.assertEqual(set(dequeued), set(URLS))
assert set(dequeued) == set(URLS)
self.assertFalse(self.scheduler.has_pending_requests())
self.assertEqual(len(self.scheduler), 0)
assert not self.scheduler.has_pending_requests()
assert len(self.scheduler) == 0
close_result = yield self.scheduler.close("")
self.assertEqual(close_result, "close")
assert close_result == "close"
class MinimalSchedulerCrawlTest(TwistedTestCase):
class MinimalSchedulerCrawlTest(TestCase):
scheduler_cls = MinimalScheduler
@defer.inlineCallbacks
@ -158,8 +157,8 @@ class MinimalSchedulerCrawlTest(TwistedTestCase):
crawler = get_crawler(PathsSpider, settings)
yield crawler.crawl(mockserver)
for path in PATHS:
self.assertIn(f"{{'path': '{path}'}}", str(log))
self.assertIn(f"'item_scraped_count': {len(PATHS)}", str(log))
assert f"{{'path': '{path}'}}" in str(log)
assert f"'item_scraped_count': {len(PATHS)}" in str(log)
class SimpleSchedulerCrawlTest(MinimalSchedulerCrawlTest):

145
tests/test_selector.py
View File

@ -3,7 +3,6 @@ import weakref
import parsel
import pytest
from packaging import version
from twisted.trial import unittest
from scrapy.http import HtmlResponse, TextResponse, XmlResponse
from scrapy.selector import Selector
@ -12,7 +11,7 @@ PARSEL_VERSION = version.parse(getattr(parsel, "__version__", "0.0"))
PARSEL_18_PLUS = PARSEL_VERSION >= version.parse("1.8.0")
class SelectorTestCase(unittest.TestCase):
class TestSelector:
def test_simple_selection(self):
"""Simple selector tests"""
body = b"<p><input name='a'value='1'/><input name='b'value='2'/></p>"
@ -20,57 +19,46 @@ class SelectorTestCase(unittest.TestCase):
sel = Selector(response)
xl = sel.xpath("//input")
self.assertEqual(2, len(xl))
assert len(xl) == 2
for x in xl:
assert isinstance(x, Selector)
self.assertEqual(
sel.xpath("//input").getall(), [x.get() for x in sel.xpath("//input")]
)
self.assertEqual(
[x.get() for x in sel.xpath("//input[@name='a']/@name")], ["a"]
)
self.assertEqual(
[
x.get()
for x in sel.xpath(
"number(concat(//input[@name='a']/@value, //input[@name='b']/@value))"
)
],
["12.0"],
)
self.assertEqual(sel.xpath("concat('xpath', 'rules')").getall(), ["xpathrules"])
self.assertEqual(
[
x.get()
for x in sel.xpath(
"concat(//input[@name='a']/@value, //input[@name='b']/@value)"
)
],
["12"],
)
assert sel.xpath("//input").getall() == [x.get() for x in sel.xpath("//input")]
assert [x.get() for x in sel.xpath("//input[@name='a']/@name")] == ["a"]
assert [
x.get()
for x in sel.xpath(
"number(concat(//input[@name='a']/@value, //input[@name='b']/@value))"
)
] == ["12.0"]
assert sel.xpath("concat('xpath', 'rules')").getall() == ["xpathrules"]
assert [
x.get()
for x in sel.xpath(
"concat(//input[@name='a']/@value, //input[@name='b']/@value)"
)
] == ["12"]
def test_root_base_url(self):
body = b'<html><form action="/path"><input name="a" /></form></html>'
url = "http://example.com"
response = TextResponse(url=url, body=body, encoding="utf-8")
sel = Selector(response)
self.assertEqual(url, sel.root.base)
assert url == sel.root.base
def test_flavor_detection(self):
text = b'<div><img src="a.jpg"><p>Hello</div>'
sel = Selector(XmlResponse("http://example.com", body=text, encoding="utf-8"))
self.assertEqual(sel.type, "xml")
self.assertEqual(
sel.xpath("//div").getall(),
['<div><img src="a.jpg"><p>Hello</p></img></div>'],
)
assert sel.type == "xml"
assert sel.xpath("//div").getall() == [
'<div><img src="a.jpg"><p>Hello</p></img></div>'
]
sel = Selector(HtmlResponse("http://example.com", body=text, encoding="utf-8"))
self.assertEqual(sel.type, "html")
self.assertEqual(
sel.xpath("//div").getall(), ['<div><img src="a.jpg"><p>Hello</p></div>']
)
assert sel.type == "html"
assert sel.xpath("//div").getall() == [
'<div><img src="a.jpg"><p>Hello</p></div>'
]
def test_http_header_encoding_precedence(self):
# '\xa3' = pound symbol in unicode
@ -92,7 +80,7 @@ class SelectorTestCase(unittest.TestCase):
url="http://example.com", headers=headers, body=html_utf8
)
x = Selector(response)
self.assertEqual(x.xpath("//span[@id='blank']/text()").getall(), ["\xa3"])
assert x.xpath("//span[@id='blank']/text()").getall() == ["\xa3"]
def test_badly_encoded_body(self):
# \xe9 alone isn't valid utf8 sequence
@ -116,7 +104,7 @@ class SelectorTestCase(unittest.TestCase):
Selector(TextResponse(url="http://example.com", body=b""), text="")
class JMESPathTestCase(unittest.TestCase):
class TestJMESPath:
@pytest.mark.skipif(
not PARSEL_18_PLUS, reason="parsel < 1.8 doesn't support jmespath"
)
@ -149,16 +137,13 @@ class JMESPathTestCase(unittest.TestCase):
}
"""
resp = TextResponse(url="http://example.com", body=body, encoding="utf-8")
self.assertEqual(
resp.jmespath("html").get(),
"<div><a>a<br>b</a>c</div><div><a>d</a>e<b>f</b></div>",
assert (
resp.jmespath("html").get()
== "<div><a>a<br>b</a>c</div><div><a>d</a>e<b>f</b></div>"
)
self.assertEqual(
resp.jmespath("html").xpath("//div/a/text()").getall(),
["a", "b", "d"],
)
self.assertEqual(resp.jmespath("html").css("div > b").getall(), ["<b>f</b>"])
self.assertEqual(resp.jmespath("content").jmespath("name.age").get(), "18")
assert resp.jmespath("html").xpath("//div/a/text()").getall() == ["a", "b", "d"]
assert resp.jmespath("html").css("div > b").getall() == ["<b>f</b>"]
assert resp.jmespath("content").jmespath("name.age").get() == "18"
@pytest.mark.skipif(
not PARSEL_18_PLUS, reason="parsel < 1.8 doesn't support jmespath"
@ -194,15 +179,19 @@ class JMESPathTestCase(unittest.TestCase):
</div>
"""
resp = TextResponse(url="http://example.com", body=body, encoding="utf-8")
self.assertEqual(
resp.xpath("//div/content/text()").jmespath("user[*].name").getall(),
["A", "B", "C", "D"],
)
self.assertEqual(
resp.xpath("//div/content").jmespath("user[*].name").getall(),
["A", "B", "C", "D"],
)
self.assertEqual(resp.xpath("//div/content").jmespath("total").get(), "4")
assert resp.xpath("//div/content/text()").jmespath("user[*].name").getall() == [
"A",
"B",
"C",
"D",
]
assert resp.xpath("//div/content").jmespath("user[*].name").getall() == [
"A",
"B",
"C",
"D",
]
assert resp.xpath("//div/content").jmespath("total").get() == "4"
@pytest.mark.skipif(
not PARSEL_18_PLUS, reason="parsel < 1.8 doesn't support jmespath"
@ -238,30 +227,26 @@ class JMESPathTestCase(unittest.TestCase):
</div>
"""
resp = TextResponse(url="http://example.com", body=body, encoding="utf-8")
self.assertEqual(
resp.xpath("//div/content/text()").jmespath("user[*].name").re(r"(\w+)"),
["A", "B", "C", "D"],
)
self.assertEqual(
resp.xpath("//div/content").jmespath("user[*].name").re(r"(\w+)"),
["A", "B", "C", "D"],
assert resp.xpath("//div/content/text()").jmespath("user[*].name").re(
r"(\w+)"
) == ["A", "B", "C", "D"]
assert resp.xpath("//div/content").jmespath("user[*].name").re(r"(\w+)") == [
"A",
"B",
"C",
"D",
]
assert resp.xpath("//div/content").jmespath("unavailable").re(r"(\d+)") == []
assert (
resp.xpath("//div/content").jmespath("unavailable").re_first(r"(\d+)")
is None
)
self.assertEqual(
resp.xpath("//div/content").jmespath("unavailable").re(r"(\d+)"), []
)
self.assertEqual(
resp.xpath("//div/content").jmespath("unavailable").re_first(r"(\d+)"),
None,
)
self.assertEqual(
resp.xpath("//div/content")
.jmespath("user[*].age.to_string(@)")
.re(r"(\d+)"),
["18", "32", "22", "25"],
)
assert resp.xpath("//div/content").jmespath("user[*].age.to_string(@)").re(
r"(\d+)"
) == ["18", "32", "22", "25"]
@pytest.mark.skipif(PARSEL_18_PLUS, reason="parsel >= 1.8 supports jmespath")
def test_jmespath_not_available(self) -> None:

6
tests/test_signals.py
View File

@ -20,7 +20,7 @@ class ItemSpider(Spider):
return {"index": response.meta["index"]}
class AsyncSignalTestCase(unittest.TestCase):
class TestAsyncSignal(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.mockserver = MockServer()
@ -43,6 +43,6 @@ class AsyncSignalTestCase(unittest.TestCase):
crawler = get_crawler(ItemSpider)
crawler.signals.connect(self._on_item_scraped, signals.item_scraped)
yield crawler.crawl(mockserver=self.mockserver)
self.assertEqual(len(self.items), 10)
assert len(self.items) == 10
for index in range(10):
self.assertIn({"index": index}, self.items)
assert {"index": index} in self.items

20
tests/test_toplevel.py
View File

@ -1,33 +1,31 @@
from unittest import TestCase
import scrapy
class ToplevelTestCase(TestCase):
class TestToplevel:
def test_version(self):
self.assertIs(type(scrapy.__version__), str)
assert isinstance(scrapy.__version__, str)
def test_version_info(self):
self.assertIs(type(scrapy.version_info), tuple)
assert isinstance(scrapy.version_info, tuple)
def test_request_shortcut(self):
from scrapy.http import FormRequest, Request
self.assertIs(scrapy.Request, Request)
self.assertIs(scrapy.FormRequest, FormRequest)
assert scrapy.Request is Request
assert scrapy.FormRequest is FormRequest
def test_spider_shortcut(self):
from scrapy.spiders import Spider
self.assertIs(scrapy.Spider, Spider)
assert scrapy.Spider is Spider
def test_selector_shortcut(self):
from scrapy.selector import Selector
self.assertIs(scrapy.Selector, Selector)
assert scrapy.Selector is Selector
def test_item_shortcut(self):
from scrapy.item import Field, Item
self.assertIs(scrapy.Item, Item)
self.assertIs(scrapy.Field, Field)
assert scrapy.Item is Item
assert scrapy.Field is Field

11
tests/test_urlparse_monkeypatches.py
View File

@ -1,11 +1,10 @@
import unittest
from urllib.parse import urlparse
class UrlparseTestCase(unittest.TestCase):
class TestUrlparse:
def test_s3_url(self):
p = urlparse("s3://bucket/key/name?param=value")
self.assertEqual(p.scheme, "s3")
self.assertEqual(p.hostname, "bucket")
self.assertEqual(p.path, "/key/name")
self.assertEqual(p.query, "param=value")
assert p.scheme == "s3"
assert p.hostname == "bucket"
assert p.path == "/key/name"
assert p.query == "param=value"