diff --git a/scrapy/spiders/init.py b/scrapy/spiders/init.py index 770ec495c..d88b109b2 100644 --- a/scrapy/spiders/init.py +++ b/scrapy/spiders/init.py @@ -1,9 +1,11 @@ from __future__ import annotations +import warnings from collections.abc import AsyncIterator, Iterable from typing import TYPE_CHECKING, Any, cast from scrapy import Request +from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.spiders import Spider from scrapy.utils.spider import iterate_spider_output @@ -12,7 +14,20 @@ if TYPE_CHECKING: class InitSpider(Spider): - """Base Spider with initialization facilities""" + """Base Spider with initialization facilities + + .. warning:: This class is deprecated. Copy its code into your project if needed. + It will be removed in a future Scrapy version. + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + warnings.warn( + "InitSpider is deprecated. Copy its code from Scrapy's source if needed. " + "Will be removed in a future version.", + ScrapyDeprecationWarning, + stacklevel=2, + ) async def yield_seeds(self) -> AsyncIterator[Any]: async for seed in super().yield_seeds(): diff --git a/tests/test_downloader_handlers.py b/tests/test_downloader_handlers.py index 323a51002..19bd02498 100644 --- a/tests/test_downloader_handlers.py +++ b/tests/test_downloader_handlers.py @@ -69,46 +69,46 @@ class OffDH: return cls(crawler) -class LoadTestCase(unittest.TestCase): +class TestLoad: def test_enabled_handler(self): handlers = {"scheme": DummyDH} crawler = get_crawler(settings_dict={"DOWNLOAD_HANDLERS": handlers}) dh = DownloadHandlers(crawler) - self.assertIn("scheme", dh._schemes) - self.assertIn("scheme", dh._handlers) - self.assertNotIn("scheme", dh._notconfigured) + assert "scheme" in dh._schemes + assert "scheme" in dh._handlers + assert "scheme" not in dh._notconfigured def test_not_configured_handler(self): handlers = {"scheme": OffDH} crawler = get_crawler(settings_dict={"DOWNLOAD_HANDLERS": handlers}) dh = DownloadHandlers(crawler) - self.assertIn("scheme", dh._schemes) - self.assertNotIn("scheme", dh._handlers) - self.assertIn("scheme", dh._notconfigured) + assert "scheme" in dh._schemes + assert "scheme" not in dh._handlers + assert "scheme" in dh._notconfigured def test_disabled_handler(self): handlers = {"scheme": None} crawler = get_crawler(settings_dict={"DOWNLOAD_HANDLERS": handlers}) dh = DownloadHandlers(crawler) - self.assertNotIn("scheme", dh._schemes) + assert "scheme" not in dh._schemes for scheme in handlers: # force load handlers dh._get_handler(scheme) - self.assertNotIn("scheme", dh._handlers) - self.assertIn("scheme", dh._notconfigured) + assert "scheme" not in dh._handlers + assert "scheme" in dh._notconfigured def test_lazy_handlers(self): handlers = {"scheme": DummyLazyDH} crawler = get_crawler(settings_dict={"DOWNLOAD_HANDLERS": handlers}) dh = DownloadHandlers(crawler) - self.assertIn("scheme", dh._schemes) - self.assertNotIn("scheme", dh._handlers) + assert "scheme" in dh._schemes + assert "scheme" not in dh._handlers for scheme in handlers: # force load lazy handler dh._get_handler(scheme) - self.assertIn("scheme", dh._handlers) - self.assertNotIn("scheme", dh._notconfigured) + assert "scheme" in dh._handlers + assert "scheme" not in dh._notconfigured -class FileTestCase(unittest.TestCase): +class TestFile(unittest.TestCase): def setUp(self): # add a special char to check that they are handled correctly self.fd, self.tmpname = mkstemp(suffix="^") @@ -122,10 +122,10 @@ class FileTestCase(unittest.TestCase): def test_download(self): def _test(response): - self.assertEqual(response.url, request.url) - self.assertEqual(response.status, 200) - self.assertEqual(response.body, b"0123456789") - self.assertEqual(response.protocol, None) + assert response.url == request.url + assert response.status == 200 + assert response.body == b"0123456789" + assert response.protocol is None request = Request(path_to_file_uri(self.tmpname)) assert request.url.upper().endswith("%5E") @@ -217,7 +217,7 @@ class DuplicateHeaderResource(resource.Resource): return b"" -class HttpTestCase(unittest.TestCase, ABC): +class TestHttp(unittest.TestCase, ABC): scheme = "http" # only used for HTTPS tests @@ -336,8 +336,8 @@ class HttpTestCase(unittest.TestCase, ABC): def test_host_header_not_in_request_headers(self): def _test(response): - self.assertEqual(response.body, to_bytes(f"{self.host}:{self.portno}")) - self.assertEqual(request.headers, {}) + assert response.body == to_bytes(f"{self.host}:{self.portno}") + assert not request.headers request = Request(self.getURL("host")) return self.download_request(request, Spider("foo")).addCallback(_test) @@ -346,8 +346,8 @@ class HttpTestCase(unittest.TestCase, ABC): host = self.host + ":" + str(self.portno) def _test(response): - self.assertEqual(response.body, host.encode()) - self.assertEqual(request.headers.get("Host"), host.encode()) + assert response.body == host.encode() + assert request.headers.get("Host") == host.encode() request = Request(self.getURL("host"), headers={"Host": host}) return self.download_request(request, Spider("foo")).addCallback(_test) @@ -365,7 +365,7 @@ class HttpTestCase(unittest.TestCase, ABC): """ def _test(response): - self.assertEqual(response.body, b"0") + assert response.body == b"0" request = Request(self.getURL("contentlength"), method="POST") return self.download_request(request, Spider("foo")).addCallback(_test) @@ -376,8 +376,8 @@ class HttpTestCase(unittest.TestCase, ABC): headers = Headers(json.loads(response.text)["headers"]) contentlengths = headers.getlist("Content-Length") - self.assertEqual(len(contentlengths), 1) - self.assertEqual(contentlengths, [b"0"]) + assert len(contentlengths) == 1 + assert contentlengths == [b"0"] request = Request(self.getURL("echo"), method="POST") return self.download_request(request, Spider("foo")).addCallback(_test) @@ -399,7 +399,7 @@ class HttpTestCase(unittest.TestCase, ABC): def _test_response_class(self, filename, body, response_class): def _test(response): - self.assertEqual(type(response), response_class) + assert type(response) is response_class # pylint: disable=unidiomatic-typecheck request = Request(self.getURL(filename), body=body) return self.download_request(request, Spider("foo")).addCallback(_test) @@ -416,17 +416,14 @@ class HttpTestCase(unittest.TestCase, ABC): def test_get_duplicate_header(self): def _test(response): - self.assertEqual( - response.headers.getlist(b"Set-Cookie"), - [b"a=b", b"c=d"], - ) + assert response.headers.getlist(b"Set-Cookie") == [b"a=b", b"c=d"] request = Request(self.getURL("duplicate-header")) return self.download_request(request, Spider("foo")).addCallback(_test) @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") -class Http10TestCase(HttpTestCase): +class TestHttp10(TestHttp): """HTTP 1.0 test case""" @property @@ -441,11 +438,11 @@ class Http10TestCase(HttpTestCase): return d -class Https10TestCase(Http10TestCase): +class TestHttps10(TestHttp10): scheme = "https" -class Http11TestCase(HttpTestCase): +class TestHttp11(TestHttp): """HTTP 1.1 test case""" @property @@ -466,7 +463,7 @@ class Http11TestCase(HttpTestCase): body = b"Some plain text\ndata with tabs\t and null bytes\0" def _test_type(response): - self.assertEqual(type(response), TextResponse) + assert type(response) is TextResponse # pylint: disable=unidiomatic-typecheck request = Request(self.getURL("nocontenttype"), body=body) d = self.download_request(request, Spider("foo")) @@ -583,7 +580,7 @@ class Http11TestCase(HttpTestCase): return d -class Https11TestCase(Http11TestCase): +class TestHttps11(TestHttp11): scheme = "https" tls_log_message = ( @@ -611,7 +608,7 @@ class Https11TestCase(Http11TestCase): yield download_handler.close() -class SimpleHttpsTest(unittest.TestCase): +class TestSimpleHttps(unittest.TestCase): """Base class for special cases tested with just one simple request""" keyfile = "keys/localhost.key" @@ -663,7 +660,7 @@ class SimpleHttpsTest(unittest.TestCase): return d -class Https11WrongHostnameTestCase(SimpleHttpsTest): +class TestHttps11WrongHostname(TestSimpleHttps): # above tests use a server certificate for "localhost", # client connection to "localhost" too. # here we test that even if the server certificate is for another domain, @@ -673,7 +670,7 @@ class Https11WrongHostnameTestCase(SimpleHttpsTest): certfile = "keys/example-com.cert.pem" -class Https11InvalidDNSId(SimpleHttpsTest): +class TestHttps11InvalidDNSId(TestSimpleHttps): """Connect to HTTPS hosts with IP while certificate uses domain names IDs.""" def setUp(self): @@ -681,18 +678,18 @@ class Https11InvalidDNSId(SimpleHttpsTest): self.host = "127.0.0.1" -class Https11InvalidDNSPattern(SimpleHttpsTest): +class TestHttps11InvalidDNSPattern(TestSimpleHttps): """Connect to HTTPS hosts where the certificate are issued to an ip instead of a domain.""" keyfile = "keys/localhost.ip.key" certfile = "keys/localhost.ip.crt" -class Https11CustomCiphers(SimpleHttpsTest): +class TestHttps11CustomCiphers(TestSimpleHttps): cipher_string = "CAMELLIA256-SHA" -class Http11MockServerTestCase(unittest.TestCase): +class TestHttp11MockServer(unittest.TestCase): """HTTP 1.1 test case with MockServer""" settings_dict: dict | None = None @@ -719,7 +716,7 @@ class Http11MockServerTestCase(unittest.TestCase): ) ) failure = crawler.spider.meta["failure"] - self.assertIsInstance(failure.value, defer.CancelledError) + assert isinstance(failure.value, defer.CancelledError) @defer.inlineCallbacks def test_download(self): @@ -728,9 +725,9 @@ class Http11MockServerTestCase(unittest.TestCase): seed=Request(url=self.mockserver.url("", is_secure=self.is_secure)) ) failure = crawler.spider.meta.get("failure") - self.assertTrue(failure is None) + assert failure is None reason = crawler.spider.meta["close_reason"] - self.assertTrue(reason, "finished") + assert reason == "finished" class UriResource(resource.Resource): @@ -748,7 +745,7 @@ class UriResource(resource.Resource): return b"" -class HttpProxyTestCase(unittest.TestCase, ABC): +class TestHttpProxy(unittest.TestCase, ABC): expected_http_proxy_request_body = b"http://example.com" @property @@ -777,9 +774,9 @@ class HttpProxyTestCase(unittest.TestCase, ABC): def test_download_with_proxy(self): def _test(response): - self.assertEqual(response.status, 200) - self.assertEqual(response.url, request.url) - self.assertEqual(response.body, self.expected_http_proxy_request_body) + assert response.status == 200 + assert response.url == request.url + assert response.body == self.expected_http_proxy_request_body http_proxy = self.getURL("") request = Request("http://example.com", meta={"proxy": http_proxy}) @@ -787,22 +784,22 @@ class HttpProxyTestCase(unittest.TestCase, ABC): def test_download_without_proxy(self): def _test(response): - self.assertEqual(response.status, 200) - self.assertEqual(response.url, request.url) - self.assertEqual(response.body, b"/path/to/resource") + assert response.status == 200 + assert response.url == request.url + assert response.body == b"/path/to/resource" request = Request(self.getURL("path/to/resource")) return self.download_request(request, Spider("foo")).addCallback(_test) @pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning") -class Http10ProxyTestCase(HttpProxyTestCase): +class TestHttp10Proxy(TestHttpProxy): @property def download_handler_cls(self) -> type[DownloadHandlerProtocol]: return HTTP10DownloadHandler -class Http11ProxyTestCase(HttpProxyTestCase): +class TestHttp11Proxy(TestHttpProxy): @property def download_handler_cls(self) -> type[DownloadHandlerProtocol]: return HTTP11DownloadHandler @@ -817,13 +814,13 @@ class Http11ProxyTestCase(HttpProxyTestCase): request = Request(domain, meta={"proxy": http_proxy, "download_timeout": 0.2}) d = self.download_request(request, Spider("foo")) timeout = yield self.assertFailure(d, error.TimeoutError) - self.assertIn(domain, timeout.osError) + assert domain in timeout.osError def test_download_with_proxy_without_http_scheme(self): def _test(response): - self.assertEqual(response.status, 200) - self.assertEqual(response.url, request.url) - self.assertEqual(response.body, self.expected_http_proxy_request_body) + assert response.status == 200 + assert response.url == request.url + assert response.body == self.expected_http_proxy_request_body http_proxy = self.getURL("").replace("http://", "") request = Request("http://example.com", meta={"proxy": http_proxy}) @@ -839,8 +836,8 @@ class HttpDownloadHandlerMock: @pytest.mark.requires_botocore -class S3AnonTestCase(unittest.TestCase): - def setUp(self): +class TestS3Anon: + def setup_method(self): crawler = get_crawler() self.s3reqh = build_from_crawler( S3DownloadHandler, @@ -854,13 +851,13 @@ class S3AnonTestCase(unittest.TestCase): def test_anon_request(self): req = Request("s3://aws-publicdatasets/") httpreq = self.download_request(req, self.spider) - self.assertEqual(hasattr(self.s3reqh, "anon"), True) - self.assertEqual(self.s3reqh.anon, True) - self.assertEqual(httpreq.url, "http://aws-publicdatasets.s3.amazonaws.com/") + assert hasattr(self.s3reqh, "anon") + assert self.s3reqh.anon + assert httpreq.url == "http://aws-publicdatasets.s3.amazonaws.com/" @pytest.mark.requires_botocore -class S3TestCase(unittest.TestCase): +class TestS3: download_handler_cls: type = S3DownloadHandler # test use same example keys than amazon developer guide @@ -870,7 +867,7 @@ class S3TestCase(unittest.TestCase): AWS_ACCESS_KEY_ID = "0PN5J17HBGZHT7JJ3X82" AWS_SECRET_ACCESS_KEY = "uV3F3YluFJax1cknvbcGwgjvx4QpvB+leU8dUj2o" - def setUp(self): + def setup_method(self): crawler = get_crawler() s3reqh = build_from_crawler( S3DownloadHandler, @@ -897,17 +894,13 @@ class S3TestCase(unittest.TestCase): yield def test_extra_kw(self): - try: - crawler = get_crawler() + crawler = get_crawler() + with pytest.raises((TypeError, NotConfigured)): build_from_crawler( S3DownloadHandler, crawler, extra_kw=True, ) - except Exception as e: - self.assertIsInstance(e, (TypeError, NotConfigured)) - else: - raise AssertionError def test_request_signing1(self): # gets an object from the johnsmith bucket. @@ -915,9 +908,9 @@ class S3TestCase(unittest.TestCase): req = Request("s3://johnsmith/photos/puppy.jpg", headers={"Date": date}) with self._mocked_date(date): httpreq = self.download_request(req, self.spider) - self.assertEqual( - httpreq.headers["Authorization"], - b"AWS 0PN5J17HBGZHT7JJ3X82:xXjDGYUmKxnwqr5KXNPGldn5LbA=", + assert ( + httpreq.headers["Authorization"] + == b"AWS 0PN5J17HBGZHT7JJ3X82:xXjDGYUmKxnwqr5KXNPGldn5LbA=" ) def test_request_signing2(self): @@ -934,9 +927,9 @@ class S3TestCase(unittest.TestCase): ) with self._mocked_date(date): httpreq = self.download_request(req, self.spider) - self.assertEqual( - httpreq.headers["Authorization"], - b"AWS 0PN5J17HBGZHT7JJ3X82:hcicpDDvL9SsO6AkvxqmIWkmOuQ=", + assert ( + httpreq.headers["Authorization"] + == b"AWS 0PN5J17HBGZHT7JJ3X82:hcicpDDvL9SsO6AkvxqmIWkmOuQ=" ) def test_request_signing3(self): @@ -952,9 +945,9 @@ class S3TestCase(unittest.TestCase): ) with self._mocked_date(date): httpreq = self.download_request(req, self.spider) - self.assertEqual( - httpreq.headers["Authorization"], - b"AWS 0PN5J17HBGZHT7JJ3X82:jsRt/rhG+Vtp88HrYL706QhE4w4=", + assert ( + httpreq.headers["Authorization"] + == b"AWS 0PN5J17HBGZHT7JJ3X82:jsRt/rhG+Vtp88HrYL706QhE4w4=" ) def test_request_signing4(self): @@ -963,9 +956,9 @@ class S3TestCase(unittest.TestCase): req = Request("s3://johnsmith/?acl", method="GET", headers={"Date": date}) with self._mocked_date(date): httpreq = self.download_request(req, self.spider) - self.assertEqual( - httpreq.headers["Authorization"], - b"AWS 0PN5J17HBGZHT7JJ3X82:thdUi9VAkzhkniLj96JIrOPGi0g=", + assert ( + httpreq.headers["Authorization"] + == b"AWS 0PN5J17HBGZHT7JJ3X82:thdUi9VAkzhkniLj96JIrOPGi0g=" ) def test_request_signing6(self): @@ -991,9 +984,9 @@ class S3TestCase(unittest.TestCase): ) with self._mocked_date(date): httpreq = self.download_request(req, self.spider) - self.assertEqual( - httpreq.headers["Authorization"], - b"AWS 0PN5J17HBGZHT7JJ3X82:C0FlOtU8Ylb9KDTpZqYkZPX91iI=", + assert ( + httpreq.headers["Authorization"] + == b"AWS 0PN5J17HBGZHT7JJ3X82:C0FlOtU8Ylb9KDTpZqYkZPX91iI=" ) def test_request_signing7(self): @@ -1006,13 +999,13 @@ class S3TestCase(unittest.TestCase): ) with self._mocked_date(date): httpreq = self.download_request(req, self.spider) - self.assertEqual( - httpreq.headers["Authorization"], - b"AWS 0PN5J17HBGZHT7JJ3X82:+CfvG8EZ3YccOrRVMXNaK2eKZmM=", + assert ( + httpreq.headers["Authorization"] + == b"AWS 0PN5J17HBGZHT7JJ3X82:+CfvG8EZ3YccOrRVMXNaK2eKZmM=" ) -class BaseFTPTestCase(unittest.TestCase): +class TestFTPBase(unittest.TestCase): username = "scrapy" password = "passwd" req_meta = {"ftp_user": username, "ftp_password": password} @@ -1068,10 +1061,10 @@ class BaseFTPTestCase(unittest.TestCase): d = self.download_handler.download_request(request, None) def _test(r): - self.assertEqual(r.status, 200) - self.assertEqual(r.body, b"I have the power!") - self.assertEqual(r.headers, {b"Local Filename": [b""], b"Size": [b"17"]}) - self.assertIsNone(r.protocol) + assert r.status == 200 + assert r.body == b"I have the power!" + assert r.headers == {b"Local Filename": [b""], b"Size": [b"17"]} + assert r.protocol is None return self._add_test_callbacks(d, _test) @@ -1083,9 +1076,9 @@ class BaseFTPTestCase(unittest.TestCase): d = self.download_handler.download_request(request, None) def _test(r): - self.assertEqual(r.status, 200) - self.assertEqual(r.body, b"Moooooooooo power!") - self.assertEqual(r.headers, {b"Local Filename": [b""], b"Size": [b"18"]}) + assert r.status == 200 + assert r.body == b"Moooooooooo power!" + assert r.headers == {b"Local Filename": [b""], b"Size": [b"18"]} return self._add_test_callbacks(d, _test) @@ -1096,7 +1089,7 @@ class BaseFTPTestCase(unittest.TestCase): d = self.download_handler.download_request(request, None) def _test(r): - self.assertEqual(r.status, 404) + assert r.status == 404 return self._add_test_callbacks(d, _test) @@ -1111,12 +1104,10 @@ class BaseFTPTestCase(unittest.TestCase): d = self.download_handler.download_request(request, None) def _test(r): - self.assertEqual(r.body, fname_bytes) - self.assertEqual( - r.headers, {b"Local Filename": [fname_bytes], b"Size": [b"17"]} - ) - self.assertTrue(local_fname.exists()) - self.assertEqual(local_fname.read_bytes(), b"I have the power!") + assert r.body == fname_bytes + assert r.headers == {b"Local Filename": [fname_bytes], b"Size": [b"17"]} + assert local_fname.exists() + assert local_fname.read_bytes() == b"I have the power!" local_fname.unlink() return self._add_test_callbacks(d, _test) @@ -1131,7 +1122,7 @@ class BaseFTPTestCase(unittest.TestCase): d = self.download_handler.download_request(request, None) def _test(r): - self.assertEqual(type(r), response_class) + assert type(r) is response_class # pylint: disable=unidiomatic-typecheck local_fname.unlink() return self._add_test_callbacks(d, _test) @@ -1143,7 +1134,7 @@ class BaseFTPTestCase(unittest.TestCase): return self._test_response_class("html-file-without-extension", HtmlResponse) -class FTPTestCase(BaseFTPTestCase): +class TestFTP(TestFTPBase): def test_invalid_credentials(self): if self.reactor_pytest == "asyncio" and sys.platform == "win32": raise unittest.SkipTest( @@ -1157,12 +1148,12 @@ class FTPTestCase(BaseFTPTestCase): d = self.download_handler.download_request(request, None) def _test(r): - self.assertEqual(r.type, ConnectionLost) + assert r.type == ConnectionLost return self._add_test_callbacks(d, errback=_test) -class AnonymousFTPTestCase(BaseFTPTestCase): +class TestAnonymousFTP(TestFTPBase): username = "anonymous" req_meta = {} @@ -1188,7 +1179,7 @@ class AnonymousFTPTestCase(BaseFTPTestCase): shutil.rmtree(self.directory) -class DataURITestCase(unittest.TestCase): +class TestDataURI(unittest.TestCase): def setUp(self): crawler = get_crawler() self.download_handler = build_from_crawler(DataURIDownloadHandler, crawler) @@ -1199,44 +1190,44 @@ class DataURITestCase(unittest.TestCase): uri = "data:,A%20brief%20note" def _test(response): - self.assertEqual(response.url, uri) - self.assertFalse(response.headers) + assert response.url == uri + assert not response.headers request = Request(uri) return self.download_request(request, self.spider).addCallback(_test) def test_default_mediatype_encoding(self): def _test(response): - self.assertEqual(response.text, "A brief note") - self.assertEqual(type(response), responsetypes.from_mimetype("text/plain")) - self.assertEqual(response.encoding, "US-ASCII") + assert response.text == "A brief note" + assert type(response) is responsetypes.from_mimetype("text/plain") # pylint: disable=unidiomatic-typecheck + assert response.encoding == "US-ASCII" request = Request("data:,A%20brief%20note") return self.download_request(request, self.spider).addCallback(_test) def test_default_mediatype(self): def _test(response): - self.assertEqual(response.text, "\u038e\u03a3\u038e") - self.assertEqual(type(response), responsetypes.from_mimetype("text/plain")) - self.assertEqual(response.encoding, "iso-8859-7") + assert response.text == "\u038e\u03a3\u038e" + assert type(response) is responsetypes.from_mimetype("text/plain") # pylint: disable=unidiomatic-typecheck + assert response.encoding == "iso-8859-7" request = Request("data:;charset=iso-8859-7,%be%d3%be") return self.download_request(request, self.spider).addCallback(_test) def test_text_charset(self): def _test(response): - self.assertEqual(response.text, "\u038e\u03a3\u038e") - self.assertEqual(response.body, b"\xbe\xd3\xbe") - self.assertEqual(response.encoding, "iso-8859-7") + assert response.text == "\u038e\u03a3\u038e" + assert response.body == b"\xbe\xd3\xbe" + assert response.encoding == "iso-8859-7" request = Request("data:text/plain;charset=iso-8859-7,%be%d3%be") return self.download_request(request, self.spider).addCallback(_test) def test_mediatype_parameters(self): def _test(response): - self.assertEqual(response.text, "\u038e\u03a3\u038e") - self.assertEqual(type(response), responsetypes.from_mimetype("text/plain")) - self.assertEqual(response.encoding, "utf-8") + assert response.text == "\u038e\u03a3\u038e" + assert type(response) is responsetypes.from_mimetype("text/plain") # pylint: disable=unidiomatic-typecheck + assert response.encoding == "utf-8" request = Request( "data:text/plain;foo=%22foo;bar%5C%22%22;" @@ -1247,14 +1238,14 @@ class DataURITestCase(unittest.TestCase): def test_base64(self): def _test(response): - self.assertEqual(response.text, "Hello, world.") + assert response.text == "Hello, world." request = Request("data:text/plain;base64,SGVsbG8sIHdvcmxkLg%3D%3D") return self.download_request(request, self.spider).addCallback(_test) def test_protocol(self): def _test(response): - self.assertIsNone(response.protocol) + assert response.protocol is None request = Request("data:,") return self.download_request(request, self.spider).addCallback(_test) diff --git a/tests/test_downloader_handlers_http2.py b/tests/test_downloader_handlers_http2.py index 17d5c2d0a..c74c09cbb 100644 --- a/tests/test_downloader_handlers_http2.py +++ b/tests/test_downloader_handlers_http2.py @@ -4,7 +4,6 @@ from unittest import mock import pytest from testfixtures import LogCapture from twisted.internet import defer, error, reactor -from twisted.trial import unittest from twisted.web import server from twisted.web.error import SchemeNotSupported from twisted.web.http import H2_ENABLED @@ -28,25 +27,25 @@ class BaseTestClasses: # A hack to prevent tests from the imported classes to run here too. # See https://stackoverflow.com/q/1323455/113586 for other ways. from tests.test_downloader_handlers import ( - Http11MockServerTestCase as Http11MockServerTestCase, + TestHttp11MockServer as TestHttp11MockServer, ) from tests.test_downloader_handlers import ( - Http11ProxyTestCase as Http11ProxyTestCase, + TestHttp11Proxy as TestHttp11Proxy, ) from tests.test_downloader_handlers import ( - Https11CustomCiphers as Https11CustomCiphers, + TestHttps11 as TestHttps11, ) from tests.test_downloader_handlers import ( - Https11InvalidDNSId as Https11InvalidDNSId, + TestHttps11CustomCiphers as TestHttps11CustomCiphers, ) from tests.test_downloader_handlers import ( - Https11InvalidDNSPattern as Https11InvalidDNSPattern, + TestHttps11InvalidDNSId as TestHttps11InvalidDNSId, ) from tests.test_downloader_handlers import ( - Https11TestCase as Https11TestCase, + TestHttps11InvalidDNSPattern as TestHttps11InvalidDNSPattern, ) from tests.test_downloader_handlers import ( - Https11WrongHostnameTestCase as Https11WrongHostnameTestCase, + TestHttps11WrongHostname as TestHttps11WrongHostname, ) @@ -56,7 +55,7 @@ def _get_dh() -> type[DownloadHandlerProtocol]: return H2DownloadHandler -class Https2TestCase(BaseTestClasses.Https11TestCase): +class TestHttps2(BaseTestClasses.TestHttps11): scheme = "https" HTTP2_DATALOSS_SKIP_REASON = "Content-Length mismatch raises InvalidBodyLengthError" @@ -97,22 +96,22 @@ class Https2TestCase(BaseTestClasses.Https11TestCase): yield self.assertFailure(d, SchemeNotSupported) def test_download_broken_content_cause_data_loss(self, url="broken"): - raise unittest.SkipTest(self.HTTP2_DATALOSS_SKIP_REASON) + pytest.skip(self.HTTP2_DATALOSS_SKIP_REASON) def test_download_broken_chunked_content_cause_data_loss(self): - raise unittest.SkipTest(self.HTTP2_DATALOSS_SKIP_REASON) + pytest.skip(self.HTTP2_DATALOSS_SKIP_REASON) def test_download_broken_content_allow_data_loss(self, url="broken"): - raise unittest.SkipTest(self.HTTP2_DATALOSS_SKIP_REASON) + pytest.skip(self.HTTP2_DATALOSS_SKIP_REASON) def test_download_broken_chunked_content_allow_data_loss(self): - raise unittest.SkipTest(self.HTTP2_DATALOSS_SKIP_REASON) + pytest.skip(self.HTTP2_DATALOSS_SKIP_REASON) def test_download_broken_content_allow_data_loss_via_setting(self, url="broken"): - raise unittest.SkipTest(self.HTTP2_DATALOSS_SKIP_REASON) + pytest.skip(self.HTTP2_DATALOSS_SKIP_REASON) def test_download_broken_chunked_content_allow_data_loss_via_setting(self): - raise unittest.SkipTest(self.HTTP2_DATALOSS_SKIP_REASON) + pytest.skip(self.HTTP2_DATALOSS_SKIP_REASON) def test_concurrent_requests_same_domain(self): spider = Spider("foo") @@ -180,31 +179,31 @@ class Https2TestCase(BaseTestClasses.Https11TestCase): return d -class Https2WrongHostnameTestCase(BaseTestClasses.Https11WrongHostnameTestCase): +class Https2WrongHostnameTestCase(BaseTestClasses.TestHttps11WrongHostname): @property def download_handler_cls(self) -> type[DownloadHandlerProtocol]: return _get_dh() -class Https2InvalidDNSId(BaseTestClasses.Https11InvalidDNSId): +class Https2InvalidDNSId(BaseTestClasses.TestHttps11InvalidDNSId): @property def download_handler_cls(self) -> type[DownloadHandlerProtocol]: return _get_dh() -class Https2InvalidDNSPattern(BaseTestClasses.Https11InvalidDNSPattern): +class Https2InvalidDNSPattern(BaseTestClasses.TestHttps11InvalidDNSPattern): @property def download_handler_cls(self) -> type[DownloadHandlerProtocol]: return _get_dh() -class Https2CustomCiphers(BaseTestClasses.Https11CustomCiphers): +class Https2CustomCiphers(BaseTestClasses.TestHttps11CustomCiphers): @property def download_handler_cls(self) -> type[DownloadHandlerProtocol]: return _get_dh() -class Http2MockServerTestCase(BaseTestClasses.Http11MockServerTestCase): +class Http2MockServerTestCase(BaseTestClasses.TestHttp11MockServer): """HTTP 2.0 test case with MockServer""" settings_dict = { @@ -215,7 +214,7 @@ class Http2MockServerTestCase(BaseTestClasses.Http11MockServerTestCase): is_secure = True -class Https2ProxyTestCase(BaseTestClasses.Http11ProxyTestCase): +class Https2ProxyTestCase(BaseTestClasses.TestHttp11Proxy): # only used for HTTPS tests keyfile = "keys/localhost.key" certfile = "keys/localhost.crt" diff --git a/tests/test_exporters.py b/tests/test_exporters.py index 48728e078..f55cb6c97 100644 --- a/tests/test_exporters.py +++ b/tests/test_exporters.py @@ -54,11 +54,11 @@ class CustomFieldDataclass: age: int = dataclasses.field(metadata={"serializer": custom_serializer}) -class BaseItemExporterTest(unittest.TestCase): +class TestBaseItemExporter: item_class: type = MyItem custom_field_item_class: type = CustomFieldItem - def setUp(self): + def setup_method(self): self.i = self.item_class(name="John\xa3", age="22") self.output = BytesIO() self.ie = self._get_exporter() @@ -72,7 +72,7 @@ class BaseItemExporterTest(unittest.TestCase): def _assert_expected_item(self, exported_dict): for k, v in exported_dict.items(): exported_dict[k] = to_unicode(v) - self.assertEqual(self.i, self.item_class(**exported_dict)) + assert self.i == self.item_class(**exported_dict) def _get_nonstring_types_item(self): return { @@ -105,45 +105,40 @@ class BaseItemExporterTest(unittest.TestCase): def test_serialize_field(self): a = ItemAdapter(self.i) res = self.ie.serialize_field(a.get_field_meta("name"), "name", a["name"]) - self.assertEqual(res, "John\xa3") + assert res == "John\xa3" res = self.ie.serialize_field(a.get_field_meta("age"), "age", a["age"]) - self.assertEqual(res, "22") + assert res == "22" def test_fields_to_export(self): ie = self._get_exporter(fields_to_export=["name"]) - self.assertEqual( - list(ie._get_serialized_fields(self.i)), [("name", "John\xa3")] - ) + assert list(ie._get_serialized_fields(self.i)) == [("name", "John\xa3")] ie = self._get_exporter(fields_to_export=["name"], encoding="latin-1") _, name = next(iter(ie._get_serialized_fields(self.i))) assert isinstance(name, str) - self.assertEqual(name, "John\xa3") + assert name == "John\xa3" ie = self._get_exporter(fields_to_export={"name": "名稱"}) - self.assertEqual( - list(ie._get_serialized_fields(self.i)), [("名稱", "John\xa3")] - ) + assert list(ie._get_serialized_fields(self.i)) == [("名稱", "John\xa3")] def test_field_custom_serializer(self): i = self.custom_field_item_class(name="John\xa3", age="22") a = ItemAdapter(i) ie = self._get_exporter() - self.assertEqual( - ie.serialize_field(a.get_field_meta("name"), "name", a["name"]), "John\xa3" - ) - self.assertEqual( - ie.serialize_field(a.get_field_meta("age"), "age", a["age"]), "24" + assert ( + ie.serialize_field(a.get_field_meta("name"), "name", a["name"]) + == "John\xa3" ) + assert ie.serialize_field(a.get_field_meta("age"), "age", a["age"]) == "24" -class BaseItemExporterDataclassTest(BaseItemExporterTest): +class TestBaseItemExporterDataclass(TestBaseItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class PythonItemExporterTest(BaseItemExporterTest): +class TestPythonItemExporter(TestBaseItemExporter): def _get_exporter(self, **kwargs): return PythonItemExporter(**kwargs) @@ -157,16 +152,13 @@ class PythonItemExporterTest(BaseItemExporterTest): i3 = self.item_class(name="Jesus", age=i2) ie = self._get_exporter() exported = ie.export_item(i3) - self.assertEqual(type(exported), dict) - self.assertEqual( - exported, - { - "age": {"age": {"age": "22", "name": "Joseph"}, "name": "Maria"}, - "name": "Jesus", - }, - ) - self.assertEqual(type(exported["age"]), dict) - self.assertEqual(type(exported["age"]["age"]), dict) + assert isinstance(exported, dict) + assert exported == { + "age": {"age": {"age": "22", "name": "Joseph"}, "name": "Maria"}, + "name": "Jesus", + } + assert isinstance(exported["age"], dict) + assert isinstance(exported["age"]["age"], dict) def test_export_list(self): i1 = self.item_class(name="Joseph", age="22") @@ -174,15 +166,12 @@ class PythonItemExporterTest(BaseItemExporterTest): i3 = self.item_class(name="Jesus", age=[i2]) ie = self._get_exporter() exported = ie.export_item(i3) - self.assertEqual( - exported, - { - "age": [{"age": [{"age": "22", "name": "Joseph"}], "name": "Maria"}], - "name": "Jesus", - }, - ) - self.assertEqual(type(exported["age"][0]), dict) - self.assertEqual(type(exported["age"][0]["age"][0]), dict) + assert exported == { + "age": [{"age": [{"age": "22", "name": "Joseph"}], "name": "Maria"}], + "name": "Jesus", + } + assert isinstance(exported["age"][0], dict) + assert isinstance(exported["age"][0]["age"][0], dict) def test_export_item_dict_list(self): i1 = self.item_class(name="Joseph", age="22") @@ -190,29 +179,26 @@ class PythonItemExporterTest(BaseItemExporterTest): i3 = self.item_class(name="Jesus", age=[i2]) ie = self._get_exporter() exported = ie.export_item(i3) - self.assertEqual( - exported, - { - "age": [{"age": [{"age": "22", "name": "Joseph"}], "name": "Maria"}], - "name": "Jesus", - }, - ) - self.assertEqual(type(exported["age"][0]), dict) - self.assertEqual(type(exported["age"][0]["age"][0]), dict) + assert exported == { + "age": [{"age": [{"age": "22", "name": "Joseph"}], "name": "Maria"}], + "name": "Jesus", + } + assert isinstance(exported["age"][0], dict) + assert isinstance(exported["age"][0]["age"][0], dict) def test_nonstring_types_item(self): item = self._get_nonstring_types_item() ie = self._get_exporter() exported = ie.export_item(item) - self.assertEqual(exported, item) + assert exported == item -class PythonItemExporterDataclassTest(PythonItemExporterTest): +class TestPythonItemExporterDataclass(TestPythonItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class PprintItemExporterTest(BaseItemExporterTest): +class TestPprintItemExporter(TestBaseItemExporter): def _get_exporter(self, **kwargs): return PprintItemExporter(self.output, **kwargs) @@ -222,12 +208,12 @@ class PprintItemExporterTest(BaseItemExporterTest): ) -class PprintItemExporterDataclassTest(PprintItemExporterTest): +class TestPprintItemExporterDataclass(TestPprintItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class PickleItemExporterTest(BaseItemExporterTest): +class TestPickleItemExporter(TestBaseItemExporter): def _get_exporter(self, **kwargs): return PickleItemExporter(self.output, **kwargs) @@ -245,8 +231,8 @@ class PickleItemExporterTest(BaseItemExporterTest): ie.finish_exporting() del ie # See the first “del self.ie” in this file for context. f.seek(0) - self.assertEqual(self.item_class(**pickle.load(f)), i1) - self.assertEqual(self.item_class(**pickle.load(f)), i2) + assert self.item_class(**pickle.load(f)) == i1 + assert self.item_class(**pickle.load(f)) == i2 def test_nonstring_types_item(self): item = self._get_nonstring_types_item() @@ -256,15 +242,15 @@ class PickleItemExporterTest(BaseItemExporterTest): ie.export_item(item) ie.finish_exporting() del ie # See the first “del self.ie” in this file for context. - self.assertEqual(pickle.loads(fp.getvalue()), item) + assert pickle.loads(fp.getvalue()) == item -class PickleItemExporterDataclassTest(PickleItemExporterTest): +class TestPickleItemExporterDataclass(TestPickleItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class MarshalItemExporterTest(BaseItemExporterTest): +class TestMarshalItemExporter(TestBaseItemExporter): def _get_exporter(self, **kwargs): self.output = tempfile.TemporaryFile() return MarshalItemExporter(self.output, **kwargs) @@ -283,15 +269,15 @@ class MarshalItemExporterTest(BaseItemExporterTest): ie.finish_exporting() del ie # See the first “del self.ie” in this file for context. fp.seek(0) - self.assertEqual(marshal.load(fp), item) + assert marshal.load(fp) == item -class MarshalItemExporterDataclassTest(MarshalItemExporterTest): +class TestMarshalItemExporterDataclass(TestMarshalItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class CsvItemExporterTest(BaseItemExporterTest): +class TestCsvItemExporter(TestBaseItemExporter): def _get_exporter(self, **kwargs): self.output = tempfile.TemporaryFile() return CsvItemExporter(self.output, **kwargs) @@ -303,7 +289,7 @@ class CsvItemExporterTest(BaseItemExporterTest): for line in to_unicode(csv).splitlines(True) ] - return self.assertEqual(split_csv(first), split_csv(second), msg=msg) + assert split_csv(first) == split_csv(second), msg def _check_output(self): self.output.seek(0) @@ -406,12 +392,12 @@ class CsvItemExporterTest(BaseItemExporterTest): ) -class CsvItemExporterDataclassTest(CsvItemExporterTest): +class TestCsvItemExporterDataclass(TestCsvItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class XmlItemExporterTest(BaseItemExporterTest): +class TestXmlItemExporter(TestBaseItemExporter): def _get_exporter(self, **kwargs): return XmlItemExporter(self.output, **kwargs) @@ -426,7 +412,7 @@ class XmlItemExporterTest(BaseItemExporterTest): doc = lxml.etree.fromstring(xmlcontent) return xmltuple(doc) - return self.assertEqual(xmlsplit(first), xmlsplit(second), msg) + assert xmlsplit(first) == xmlsplit(second), msg def assertExportResult(self, item, expected_value): fp = BytesIO() @@ -517,12 +503,12 @@ class XmlItemExporterTest(BaseItemExporterTest): ) -class XmlItemExporterDataclassTest(XmlItemExporterTest): +class TestXmlItemExporterDataclass(TestXmlItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class JsonLinesItemExporterTest(BaseItemExporterTest): +class TestJsonLinesItemExporter(TestBaseItemExporter): _expected_nested: Any = { "name": "Jesus", "age": {"name": "Maria", "age": {"name": "Joseph", "age": "22"}}, @@ -533,7 +519,7 @@ class JsonLinesItemExporterTest(BaseItemExporterTest): def _check_output(self): exported = json.loads(to_unicode(self.output.getvalue().strip())) - self.assertEqual(exported, ItemAdapter(self.i).asdict()) + assert exported == ItemAdapter(self.i).asdict() def test_nested_item(self): i1 = self.item_class(name="Joseph", age="22") @@ -544,7 +530,7 @@ class JsonLinesItemExporterTest(BaseItemExporterTest): self.ie.finish_exporting() del self.ie # See the first “del self.ie” in this file for context. exported = json.loads(to_unicode(self.output.getvalue())) - self.assertEqual(exported, self._expected_nested) + assert exported == self._expected_nested def test_extra_keywords(self): self.ie = self._get_exporter(sort_keys=True) @@ -561,23 +547,23 @@ class JsonLinesItemExporterTest(BaseItemExporterTest): del self.ie # See the first “del self.ie” in this file for context. exported = json.loads(to_unicode(self.output.getvalue())) item["time"] = str(item["time"]) - self.assertEqual(exported, item) + assert exported == item -class JsonLinesItemExporterDataclassTest(JsonLinesItemExporterTest): +class TestJsonLinesItemExporterDataclass(TestJsonLinesItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class JsonItemExporterTest(JsonLinesItemExporterTest): - _expected_nested = [JsonLinesItemExporterTest._expected_nested] +class TestJsonItemExporter(TestJsonLinesItemExporter): + _expected_nested = [TestJsonLinesItemExporter._expected_nested] def _get_exporter(self, **kwargs): return JsonItemExporter(self.output, **kwargs) def _check_output(self): exported = json.loads(to_unicode(self.output.getvalue().strip())) - self.assertEqual(exported, [ItemAdapter(self.i).asdict()]) + assert exported == [ItemAdapter(self.i).asdict()] def assertTwoItemsExported(self, item): self.ie.start_exporting() @@ -586,9 +572,7 @@ class JsonItemExporterTest(JsonLinesItemExporterTest): self.ie.finish_exporting() del self.ie # See the first “del self.ie” in this file for context. exported = json.loads(to_unicode(self.output.getvalue())) - self.assertEqual( - exported, [ItemAdapter(item).asdict(), ItemAdapter(item).asdict()] - ) + assert exported == [ItemAdapter(item).asdict(), ItemAdapter(item).asdict()] def test_two_items(self): self.assertTwoItemsExported(self.i) @@ -609,7 +593,7 @@ class JsonItemExporterTest(JsonLinesItemExporterTest): self.ie.export_item(i3) self.ie.finish_exporting() exported = json.loads(to_unicode(self.output.getvalue())) - self.assertEqual(exported, [dict(i1), dict(i3)]) + assert exported == [dict(i1), dict(i3)] def test_nested_item(self): i1 = self.item_class(name="Joseph\xa3", age="22") @@ -624,7 +608,7 @@ class JsonItemExporterTest(JsonLinesItemExporterTest): "name": "Jesus", "age": {"name": "Maria", "age": ItemAdapter(i1).asdict()}, } - self.assertEqual(exported, [expected]) + assert exported == [expected] def test_nested_dict_item(self): i1 = {"name": "Joseph\xa3", "age": "22"} @@ -636,7 +620,7 @@ class JsonItemExporterTest(JsonLinesItemExporterTest): del self.ie # See the first “del self.ie” in this file for context. exported = json.loads(to_unicode(self.output.getvalue())) expected = {"name": "Jesus", "age": {"name": "Maria", "age": i1}} - self.assertEqual(exported, [expected]) + assert exported == [expected] def test_nonstring_types_item(self): item = self._get_nonstring_types_item() @@ -646,10 +630,10 @@ class JsonItemExporterTest(JsonLinesItemExporterTest): del self.ie # See the first “del self.ie” in this file for context. exported = json.loads(to_unicode(self.output.getvalue())) item["time"] = str(item["time"]) - self.assertEqual(exported, [item]) + assert exported == [item] -class JsonItemExporterToBytesTest(BaseItemExporterTest): +class TestJsonItemExporterToBytes(TestBaseItemExporter): def _get_exporter(self, **kwargs): kwargs["encoding"] = "latin" return JsonItemExporter(self.output, **kwargs) @@ -665,18 +649,18 @@ class JsonItemExporterToBytesTest(BaseItemExporterTest): self.ie.export_item(i3) self.ie.finish_exporting() exported = json.loads(to_unicode(self.output.getvalue(), encoding="latin")) - self.assertEqual(exported, [dict(i1), dict(i3)]) + assert exported == [dict(i1), dict(i3)] -class JsonItemExporterDataclassTest(JsonItemExporterTest): +class TestJsonItemExporterDataclass(TestJsonItemExporter): item_class = MyDataClass custom_field_item_class = CustomFieldDataclass -class CustomExporterItemTest(unittest.TestCase): +class TestCustomExporterItem: item_class: type = MyItem - def setUp(self): + def setup_method(self): if self.item_class is None: raise unittest.SkipTest("item class is None") @@ -691,17 +675,13 @@ class CustomExporterItemTest(unittest.TestCase): a = ItemAdapter(i) ie = CustomItemExporter() - self.assertEqual( - ie.serialize_field(a.get_field_meta("name"), "name", a["name"]), "John" - ) - self.assertEqual( - ie.serialize_field(a.get_field_meta("age"), "age", a["age"]), "23" - ) + assert ie.serialize_field(a.get_field_meta("name"), "name", a["name"]) == "John" + assert ie.serialize_field(a.get_field_meta("age"), "age", a["age"]) == "23" i2 = {"name": "John", "age": "22"} - self.assertEqual(ie.serialize_field({}, "name", i2["name"]), "John") - self.assertEqual(ie.serialize_field({}, "age", i2["age"]), "23") + assert ie.serialize_field({}, "name", i2["name"]) == "John" + assert ie.serialize_field({}, "age", i2["age"]) == "23" -class CustomExporterDataclassTest(CustomExporterItemTest): +class TestCustomExporterDataclass(TestCustomExporterItem): item_class = MyDataClass diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py index 8e008ab98..44cd10ec3 100644 --- a/tests/test_feedexport.py +++ b/tests/test_feedexport.py @@ -88,7 +88,7 @@ def mock_google_cloud_storage() -> tuple[Any, Any, Any]: return (client_mock, bucket_mock, blob_mock) -class FileFeedStorageTest(unittest.TestCase): +class TestFileFeedStorage(unittest.TestCase): def test_store_file_uri(self): path = Path(self.mktemp()).resolve() uri = path_to_file_uri(str(path)) @@ -137,14 +137,14 @@ class FileFeedStorageTest(unittest.TestCase): file = storage.open(spider) file.write(b"content") yield storage.store(file) - self.assertTrue(path.exists()) + assert path.exists() try: - self.assertEqual(path.read_bytes(), expected_content) + assert path.read_bytes() == expected_content finally: path.unlink() -class FTPFeedStorageTest(unittest.TestCase): +class TestFTPFeedStorage(unittest.TestCase): def get_test_spider(self, settings=None): class TestSpider(scrapy.Spider): name = "test_spider" @@ -166,9 +166,9 @@ class FTPFeedStorageTest(unittest.TestCase): return storage.store(file) def _assert_stored(self, path: Path, content): - self.assertTrue(path.exists()) + assert path.exists() try: - self.assertEqual(path.read_bytes(), content) + assert path.read_bytes() == content finally: path.unlink() @@ -216,10 +216,10 @@ class FTPFeedStorageTest(unittest.TestCase): # RFC3986: 3.2.1. User Information pw_quoted = quote(string.punctuation, safe="") st = FTPFeedStorage(f"ftp://foo:{pw_quoted}@example.com/some_path", {}) - self.assertEqual(st.password, string.punctuation) + assert st.password == string.punctuation -class BlockingFeedStorageTest(unittest.TestCase): +class TestBlockingFeedStorage: def get_test_spider(self, settings=None): class TestSpider(scrapy.Spider): name = "test_spider" @@ -232,7 +232,7 @@ class BlockingFeedStorageTest(unittest.TestCase): tmp = b.open(self.get_test_spider()) tmp_path = Path(tmp.name).parent - self.assertEqual(str(tmp_path), tempfile.gettempdir()) + assert str(tmp_path) == tempfile.gettempdir() def test_temp_file(self): b = BlockingFeedStorage() @@ -241,7 +241,7 @@ class BlockingFeedStorageTest(unittest.TestCase): spider = self.get_test_spider({"FEED_TEMPDIR": str(tests_path)}) tmp = b.open(spider) tmp_path = Path(tmp.name).parent - self.assertEqual(tmp_path, tests_path) + assert tmp_path == tests_path def test_invalid_folder(self): b = BlockingFeedStorage() @@ -255,7 +255,7 @@ class BlockingFeedStorageTest(unittest.TestCase): @pytest.mark.requires_boto3 -class S3FeedStorageTest(unittest.TestCase): +class TestS3FeedStorage(unittest.TestCase): def test_parse_credentials(self): aws_credentials = { "AWS_ACCESS_KEY_ID": "settings_key", @@ -268,9 +268,9 @@ class S3FeedStorageTest(unittest.TestCase): crawler, "s3://mybucket/export.csv", ) - self.assertEqual(storage.access_key, "settings_key") - self.assertEqual(storage.secret_key, "settings_secret") - self.assertEqual(storage.session_token, "settings_token") + assert storage.access_key == "settings_key" + assert storage.secret_key == "settings_secret" + assert storage.session_token == "settings_token" # Instantiate directly storage = S3FeedStorage( "s3://mybucket/export.csv", @@ -278,17 +278,17 @@ class S3FeedStorageTest(unittest.TestCase): aws_credentials["AWS_SECRET_ACCESS_KEY"], session_token=aws_credentials["AWS_SESSION_TOKEN"], ) - self.assertEqual(storage.access_key, "settings_key") - self.assertEqual(storage.secret_key, "settings_secret") - self.assertEqual(storage.session_token, "settings_token") + assert storage.access_key == "settings_key" + assert storage.secret_key == "settings_secret" + assert storage.session_token == "settings_token" # URI priority > settings priority storage = S3FeedStorage( "s3://uri_key:uri_secret@mybucket/export.csv", aws_credentials["AWS_ACCESS_KEY_ID"], aws_credentials["AWS_SECRET_ACCESS_KEY"], ) - self.assertEqual(storage.access_key, "uri_key") - self.assertEqual(storage.secret_key, "uri_secret") + assert storage.access_key == "uri_key" + assert storage.secret_key == "uri_secret" @defer.inlineCallbacks def test_store(self): @@ -306,24 +306,23 @@ class S3FeedStorageTest(unittest.TestCase): storage.s3_client = mock.MagicMock() yield storage.store(file) - self.assertEqual( - storage.s3_client.upload_fileobj.call_args, - mock.call(Bucket=bucket, Key=key, Fileobj=file), + assert storage.s3_client.upload_fileobj.call_args == mock.call( + Bucket=bucket, Key=key, Fileobj=file ) def test_init_without_acl(self): storage = S3FeedStorage("s3://mybucket/export.csv", "access_key", "secret_key") - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.acl, None) + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.acl is None def test_init_with_acl(self): storage = S3FeedStorage( "s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl" ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.acl, "custom-acl") + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.acl == "custom-acl" def test_init_with_endpoint_url(self): storage = S3FeedStorage( @@ -332,9 +331,9 @@ class S3FeedStorageTest(unittest.TestCase): "secret_key", endpoint_url="https://example.com", ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.endpoint_url, "https://example.com") + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.endpoint_url == "https://example.com" def test_init_with_region_name(self): region_name = "ap-east-1" @@ -344,10 +343,10 @@ class S3FeedStorageTest(unittest.TestCase): "secret_key", region_name=region_name, ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.region_name, region_name) - self.assertEqual(storage.s3_client._client_config.region_name, region_name) + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.region_name == region_name + assert storage.s3_client._client_config.region_name == region_name def test_from_crawler_without_acl(self): settings = { @@ -359,9 +358,9 @@ class S3FeedStorageTest(unittest.TestCase): crawler, "s3://mybucket/export.csv", ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.acl, None) + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.acl is None def test_without_endpoint_url(self): settings = { @@ -373,9 +372,9 @@ class S3FeedStorageTest(unittest.TestCase): crawler, "s3://mybucket/export.csv", ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.endpoint_url, None) + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.endpoint_url is None def test_without_region_name(self): settings = { @@ -387,9 +386,9 @@ class S3FeedStorageTest(unittest.TestCase): crawler, "s3://mybucket/export.csv", ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.s3_client._client_config.region_name, "us-east-1") + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.s3_client._client_config.region_name == "us-east-1" def test_from_crawler_with_acl(self): settings = { @@ -402,9 +401,9 @@ class S3FeedStorageTest(unittest.TestCase): crawler, "s3://mybucket/export.csv", ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.acl, "custom-acl") + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.acl == "custom-acl" def test_from_crawler_with_endpoint_url(self): settings = { @@ -414,9 +413,9 @@ class S3FeedStorageTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings) storage = S3FeedStorage.from_crawler(crawler, "s3://mybucket/export.csv") - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.endpoint_url, "https://example.com") + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.endpoint_url == "https://example.com" def test_from_crawler_with_region_name(self): region_name = "ap-east-1" @@ -427,10 +426,10 @@ class S3FeedStorageTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings) storage = S3FeedStorage.from_crawler(crawler, "s3://mybucket/export.csv") - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.region_name, region_name) - self.assertEqual(storage.s3_client._client_config.region_name, region_name) + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.region_name == region_name + assert storage.s3_client._client_config.region_name == region_name @defer.inlineCallbacks def test_store_without_acl(self): @@ -439,9 +438,9 @@ class S3FeedStorageTest(unittest.TestCase): "access_key", "secret_key", ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.acl, None) + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.acl is None storage.s3_client = mock.MagicMock() yield storage.store(BytesIO(b"test file")) @@ -450,28 +449,28 @@ class S3FeedStorageTest(unittest.TestCase): .get("ExtraArgs", {}) .get("ACL") ) - self.assertIsNone(acl) + assert acl is None @defer.inlineCallbacks def test_store_with_acl(self): storage = S3FeedStorage( "s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl" ) - self.assertEqual(storage.access_key, "access_key") - self.assertEqual(storage.secret_key, "secret_key") - self.assertEqual(storage.acl, "custom-acl") + assert storage.access_key == "access_key" + assert storage.secret_key == "secret_key" + assert storage.acl == "custom-acl" storage.s3_client = mock.MagicMock() yield storage.store(BytesIO(b"test file")) acl = storage.s3_client.upload_fileobj.call_args[1]["ExtraArgs"]["ACL"] - self.assertEqual(acl, "custom-acl") + assert acl == "custom-acl" def test_overwrite_default(self): with LogCapture() as log: S3FeedStorage( "s3://mybucket/export.csv", "access_key", "secret_key", "custom-acl" ) - self.assertNotIn("S3 does not support appending to files", str(log)) + assert "S3 does not support appending to files" not in str(log) def test_overwrite_false(self): with LogCapture() as log: @@ -482,10 +481,10 @@ class S3FeedStorageTest(unittest.TestCase): "custom-acl", feed_options={"overwrite": False}, ) - self.assertIn("S3 does not support appending to files", str(log)) + assert "S3 does not support appending to files" in str(log) -class GCSFeedStorageTest(unittest.TestCase): +class TestGCSFeedStorage(unittest.TestCase): def test_parse_settings(self): try: from google.cloud.storage import Client # noqa: F401 @@ -543,7 +542,7 @@ class GCSFeedStorageTest(unittest.TestCase): def test_overwrite_default(self): with LogCapture() as log: GCSFeedStorage("gs://mybucket/export.csv", "myproject-123", "custom-acl") - self.assertNotIn("GCS does not support appending to files", str(log)) + assert "GCS does not support appending to files" not in str(log) def test_overwrite_false(self): with LogCapture() as log: @@ -553,10 +552,10 @@ class GCSFeedStorageTest(unittest.TestCase): "custom-acl", feed_options={"overwrite": False}, ) - self.assertIn("GCS does not support appending to files", str(log)) + assert "GCS does not support appending to files" in str(log) -class StdoutFeedStorageTest(unittest.TestCase): +class TestStdoutFeedStorage(unittest.TestCase): @defer.inlineCallbacks def test_store(self): out = BytesIO() @@ -564,20 +563,21 @@ class StdoutFeedStorageTest(unittest.TestCase): file = storage.open(scrapy.Spider("default")) file.write(b"content") yield storage.store(file) - self.assertEqual(out.getvalue(), b"content") + assert out.getvalue() == b"content" def test_overwrite_default(self): with LogCapture() as log: StdoutFeedStorage("stdout:") - self.assertNotIn( - "Standard output (stdout) storage does not support overwriting", str(log) + assert ( + "Standard output (stdout) storage does not support overwriting" + not in str(log) ) def test_overwrite_true(self): with LogCapture() as log: StdoutFeedStorage("stdout:", feed_options={"overwrite": True}) - self.assertIn( - "Standard output (stdout) storage does not support overwriting", str(log) + assert "Standard output (stdout) storage does not support overwriting" in str( + log ) @@ -639,7 +639,7 @@ class LogOnStoreFileStorage: file.close() -class FeedExportTestBase(ABC, unittest.TestCase): +class TestFeedExportBase(ABC, unittest.TestCase): class MyItem(scrapy.Item): foo = scrapy.Field() egg = scrapy.Field() @@ -769,7 +769,7 @@ class ExceptionJsonItemExporter(JsonItemExporter): raise RuntimeError("foo") -class FeedExportTest(FeedExportTestBase): +class TestFeedExport(TestFeedExportBase): @defer.inlineCallbacks def run_and_export(self, spider_cls, settings): """Run spider with specified settings; return exported data.""" @@ -812,8 +812,8 @@ class FeedExportTest(FeedExportTestBase): ) data = yield self.exported_data(items, settings) reader = csv.DictReader(to_unicode(data["csv"]).splitlines()) - self.assertEqual(reader.fieldnames, list(header)) - self.assertEqual(rows, list(reader)) + assert reader.fieldnames == list(header) + assert rows == list(reader) @defer.inlineCallbacks def assertExportedJsonLines(self, items, rows, settings=None): @@ -828,7 +828,7 @@ class FeedExportTest(FeedExportTestBase): data = yield self.exported_data(items, settings) parsed = [json.loads(to_unicode(line)) for line in data["jl"].splitlines()] rows = [{k: v for k, v in row.items() if v} for row in rows] - self.assertEqual(rows, parsed) + assert rows == parsed @defer.inlineCallbacks def assertExportedXml(self, items, rows, settings=None): @@ -844,7 +844,7 @@ class FeedExportTest(FeedExportTestBase): rows = [{k: v for k, v in row.items() if v} for row in rows] root = lxml.etree.fromstring(data["xml"]) got_rows = [{e.tag: e.text for e in it} for it in root.findall("item")] - self.assertEqual(rows, got_rows) + assert rows == got_rows @defer.inlineCallbacks def assertExportedMultiple(self, items, rows, settings=None): @@ -862,10 +862,10 @@ class FeedExportTest(FeedExportTestBase): # XML root = lxml.etree.fromstring(data["xml"]) xml_rows = [{e.tag: e.text for e in it} for it in root.findall("item")] - self.assertEqual(rows, xml_rows) + assert rows == xml_rows # JSON json_rows = json.loads(to_unicode(data["json"])) - self.assertEqual(rows, json_rows) + assert rows == json_rows @defer.inlineCallbacks def assertExportedPickle(self, items, rows, settings=None): @@ -882,7 +882,7 @@ class FeedExportTest(FeedExportTestBase): import pickle result = self._load_until_eof(data["pickle"], load_func=pickle.load) - self.assertEqual(expected, result) + assert result == expected @defer.inlineCallbacks def assertExportedMarshal(self, items, rows, settings=None): @@ -899,7 +899,7 @@ class FeedExportTest(FeedExportTestBase): import marshal result = self._load_until_eof(data["marshal"], load_func=marshal.load) - self.assertEqual(expected, result) + assert result == expected @defer.inlineCallbacks def test_stats_file_success(self): @@ -912,12 +912,8 @@ class FeedExportTest(FeedExportTestBase): } crawler = get_crawler(ItemSpider, settings) yield crawler.crawl(mockserver=self.mockserver) - self.assertIn( - "feedexport/success_count/FileFeedStorage", crawler.stats.get_stats() - ) - self.assertEqual( - crawler.stats.get_value("feedexport/success_count/FileFeedStorage"), 1 - ) + assert "feedexport/success_count/FileFeedStorage" in crawler.stats.get_stats() + assert crawler.stats.get_value("feedexport/success_count/FileFeedStorage") == 1 @defer.inlineCallbacks def test_stats_file_failed(self): @@ -934,12 +930,8 @@ class FeedExportTest(FeedExportTestBase): side_effect=KeyError("foo"), ): yield crawler.crawl(mockserver=self.mockserver) - self.assertIn( - "feedexport/failed_count/FileFeedStorage", crawler.stats.get_stats() - ) - self.assertEqual( - crawler.stats.get_value("feedexport/failed_count/FileFeedStorage"), 1 - ) + assert "feedexport/failed_count/FileFeedStorage" in crawler.stats.get_stats() + assert crawler.stats.get_value("feedexport/failed_count/FileFeedStorage") == 1 @defer.inlineCallbacks def test_stats_multiple_file(self): @@ -956,17 +948,11 @@ class FeedExportTest(FeedExportTestBase): crawler = get_crawler(ItemSpider, settings) with mock.patch.object(S3FeedStorage, "store"): yield crawler.crawl(mockserver=self.mockserver) - self.assertIn( - "feedexport/success_count/FileFeedStorage", crawler.stats.get_stats() - ) - self.assertIn( - "feedexport/success_count/StdoutFeedStorage", crawler.stats.get_stats() - ) - self.assertEqual( - crawler.stats.get_value("feedexport/success_count/FileFeedStorage"), 1 - ) - self.assertEqual( - crawler.stats.get_value("feedexport/success_count/StdoutFeedStorage"), 1 + assert "feedexport/success_count/FileFeedStorage" in crawler.stats.get_stats() + assert "feedexport/success_count/StdoutFeedStorage" in crawler.stats.get_stats() + assert crawler.stats.get_value("feedexport/success_count/FileFeedStorage") == 1 + assert ( + crawler.stats.get_value("feedexport/success_count/StdoutFeedStorage") == 1 ) @defer.inlineCallbacks @@ -993,7 +979,7 @@ class FeedExportTest(FeedExportTestBase): "FEED_STORE_EMPTY": False, } data = yield self.exported_no_data(settings) - self.assertEqual(None, data[fmt]) + assert data[fmt] is None @defer.inlineCallbacks def test_start_finish_exporting_items(self): @@ -1012,8 +998,8 @@ class FeedExportTest(FeedExportTestBase): with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot): _ = yield self.exported_data(items, settings) - self.assertFalse(listener.start_without_finish) - self.assertFalse(listener.finish_without_start) + assert not listener.start_without_finish + assert not listener.finish_without_start @defer.inlineCallbacks def test_start_finish_exporting_no_items(self): @@ -1030,8 +1016,8 @@ class FeedExportTest(FeedExportTestBase): with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot): _ = yield self.exported_data(items, settings) - self.assertFalse(listener.start_without_finish) - self.assertFalse(listener.finish_without_start) + assert not listener.start_without_finish + assert not listener.finish_without_start @defer.inlineCallbacks def test_start_finish_exporting_items_exception(self): @@ -1051,8 +1037,8 @@ class FeedExportTest(FeedExportTestBase): with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot): _ = yield self.exported_data(items, settings) - self.assertFalse(listener.start_without_finish) - self.assertFalse(listener.finish_without_start) + assert not listener.start_without_finish + assert not listener.finish_without_start @defer.inlineCallbacks def test_start_finish_exporting_no_items_exception(self): @@ -1070,8 +1056,8 @@ class FeedExportTest(FeedExportTestBase): with mock.patch("scrapy.extensions.feedexport.FeedSlot", InstrumentedFeedSlot): _ = yield self.exported_data(items, settings) - self.assertFalse(listener.start_without_finish) - self.assertFalse(listener.finish_without_start) + assert not listener.start_without_finish + assert not listener.finish_without_start @defer.inlineCallbacks def test_export_no_items_store_empty(self): @@ -1091,7 +1077,7 @@ class FeedExportTest(FeedExportTestBase): "FEED_EXPORT_INDENT": None, } data = yield self.exported_no_data(settings) - self.assertEqual(expctd, data[fmt]) + assert expctd == data[fmt] @defer.inlineCallbacks def test_export_no_items_multiple_feeds(self): @@ -1109,7 +1095,7 @@ class FeedExportTest(FeedExportTestBase): with LogCapture() as log: yield self.exported_no_data(settings) - self.assertEqual(str(log).count("Storage.store is called"), 0) + assert str(log).count("Storage.store is called") == 0 @defer.inlineCallbacks def test_export_multiple_item_classes(self): @@ -1238,7 +1224,7 @@ class FeedExportTest(FeedExportTestBase): data = yield self.exported_data(items, settings) for fmt, expected in formats.items(): - self.assertEqual(expected, data[fmt]) + assert data[fmt] == expected @defer.inlineCallbacks def test_export_based_on_custom_filters(self): @@ -1297,7 +1283,7 @@ class FeedExportTest(FeedExportTestBase): data = yield self.exported_data(items, settings) for fmt, expected in formats.items(): - self.assertEqual(expected, data[fmt]) + assert data[fmt] == expected @defer.inlineCallbacks def test_export_dicts(self): @@ -1371,7 +1357,7 @@ class FeedExportTest(FeedExportTestBase): "FEED_EXPORT_INDENT": None, } data = yield self.exported_data(items, settings) - self.assertEqual(expected, data[fmt]) + assert data[fmt] == expected formats = { "json": b'[{"foo": "Test\xd6"}]', @@ -1392,7 +1378,7 @@ class FeedExportTest(FeedExportTestBase): "FEED_EXPORT_ENCODING": "latin-1", } data = yield self.exported_data(items, settings) - self.assertEqual(expected, data[fmt]) + assert data[fmt] == expected @defer.inlineCallbacks def test_export_multiple_configs(self): @@ -1432,7 +1418,7 @@ class FeedExportTest(FeedExportTestBase): data = yield self.exported_data(items, settings) for fmt, expected in formats.items(): - self.assertEqual(expected, data[fmt]) + assert data[fmt] == expected @defer.inlineCallbacks def test_export_indentation(self): @@ -1588,7 +1574,7 @@ class FeedExportTest(FeedExportTestBase): }, } data = yield self.exported_data(items, settings) - self.assertEqual(row["expected"], data[row["format"]]) + assert data[row["format"]] == row["expected"] @defer.inlineCallbacks def test_init_exporters_storages_with_crawler(self): @@ -1600,8 +1586,8 @@ class FeedExportTest(FeedExportTestBase): }, } yield self.exported_data(items=[], settings=settings) - self.assertTrue(FromCrawlerCsvItemExporter.init_with_crawler) - self.assertTrue(FromCrawlerFileFeedStorage.init_with_crawler) + assert FromCrawlerCsvItemExporter.init_with_crawler + assert FromCrawlerFileFeedStorage.init_with_crawler @defer.inlineCallbacks def test_str_uri(self): @@ -1610,7 +1596,7 @@ class FeedExportTest(FeedExportTestBase): "FEEDS": {str(self._random_temp_filename()): {"format": "csv"}}, } data = yield self.exported_no_data(settings) - self.assertEqual(data["csv"], b"") + assert data["csv"] == b"" @defer.inlineCallbacks def test_multiple_feeds_success_logs_blocking_feed_storage(self): @@ -1631,7 +1617,7 @@ class FeedExportTest(FeedExportTestBase): print(log) for fmt in ["json", "xml", "csv"]: - self.assertIn(f"Stored {fmt} feed (2 items)", str(log)) + assert f"Stored {fmt} feed (2 items)" in str(log) @defer.inlineCallbacks def test_multiple_feeds_failing_logs_blocking_feed_storage(self): @@ -1652,7 +1638,7 @@ class FeedExportTest(FeedExportTestBase): print(log) for fmt in ["json", "xml", "csv"]: - self.assertIn(f"Error storing {fmt} feed (2 items)", str(log)) + assert f"Error storing {fmt} feed (2 items)" in str(log) @defer.inlineCallbacks def test_extend_kwargs(self): @@ -1689,7 +1675,7 @@ class FeedExportTest(FeedExportTestBase): } data = yield self.exported_data(items, settings) - self.assertEqual(row["expected"], data[feed_options["format"]]) + assert data[feed_options["format"]] == row["expected"] @defer.inlineCallbacks def test_storage_file_no_postprocessing(self): @@ -1711,7 +1697,7 @@ class FeedExportTest(FeedExportTestBase): "FEED_STORAGES": {"file": Storage}, } yield self.exported_no_data(settings) - self.assertIs(Storage.open_file, Storage.store_file) + assert Storage.open_file is Storage.store_file @defer.inlineCallbacks def test_storage_file_postprocessing(self): @@ -1741,11 +1727,11 @@ class FeedExportTest(FeedExportTestBase): "FEED_STORAGES": {"file": Storage}, } yield self.exported_no_data(settings) - self.assertIs(Storage.open_file, Storage.store_file) - self.assertFalse(Storage.file_was_closed) + assert Storage.open_file is Storage.store_file + assert not Storage.file_was_closed -class FeedPostProcessedExportsTest(FeedExportTestBase): +class TestFeedPostProcessedExports(TestFeedExportBase): items = [{"foo": "bar"}] expected = b"foo\r\nbar\r\n" @@ -1827,7 +1813,7 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): try: gzip.decompress(data[filename]) except OSError: - self.fail("Received invalid gzip data.") + pytest.fail("Received invalid gzip data.") @defer.inlineCallbacks def test_gzip_plugin_compresslevel(self): @@ -1863,8 +1849,8 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, compressed in filename_to_compressed.items(): result = gzip.decompress(data[filename]) - self.assertEqual(compressed, data[filename]) - self.assertEqual(self.expected, result) + assert compressed == data[filename] + assert result == self.expected @defer.inlineCallbacks def test_gzip_plugin_mtime(self): @@ -1898,8 +1884,8 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, compressed in filename_to_compressed.items(): result = gzip.decompress(data[filename]) - self.assertEqual(compressed, data[filename]) - self.assertEqual(self.expected, result) + assert compressed == data[filename] + assert result == self.expected @defer.inlineCallbacks def test_gzip_plugin_filename(self): @@ -1933,8 +1919,8 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, compressed in filename_to_compressed.items(): result = gzip.decompress(data[filename]) - self.assertEqual(compressed, data[filename]) - self.assertEqual(self.expected, result) + assert compressed == data[filename] + assert result == self.expected @defer.inlineCallbacks def test_lzma_plugin(self): @@ -1953,7 +1939,7 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): try: lzma.decompress(data[filename]) except lzma.LZMAError: - self.fail("Received invalid lzma data.") + pytest.fail("Received invalid lzma data.") @defer.inlineCallbacks def test_lzma_plugin_format(self): @@ -1985,8 +1971,8 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, compressed in filename_to_compressed.items(): result = lzma.decompress(data[filename]) - self.assertEqual(compressed, data[filename]) - self.assertEqual(self.expected, result) + assert compressed == data[filename] + assert result == self.expected @defer.inlineCallbacks def test_lzma_plugin_check(self): @@ -2018,8 +2004,8 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, compressed in filename_to_compressed.items(): result = lzma.decompress(data[filename]) - self.assertEqual(compressed, data[filename]) - self.assertEqual(self.expected, result) + assert compressed == data[filename] + assert result == self.expected @defer.inlineCallbacks def test_lzma_plugin_preset(self): @@ -2051,8 +2037,8 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, compressed in filename_to_compressed.items(): result = lzma.decompress(data[filename]) - self.assertEqual(compressed, data[filename]) - self.assertEqual(self.expected, result) + assert compressed == data[filename] + assert result == self.expected @defer.inlineCallbacks def test_lzma_plugin_filters(self): @@ -2075,9 +2061,9 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): } data = yield self.exported_data(self.items, settings) - self.assertEqual(compressed, data[filename]) + assert compressed == data[filename] result = lzma.decompress(data[filename]) - self.assertEqual(self.expected, result) + assert result == self.expected @defer.inlineCallbacks def test_bz2_plugin(self): @@ -2096,7 +2082,7 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): try: bz2.decompress(data[filename]) except OSError: - self.fail("Received invalid bz2 data.") + pytest.fail("Received invalid bz2 data.") @defer.inlineCallbacks def test_bz2_plugin_compresslevel(self): @@ -2128,8 +2114,8 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, compressed in filename_to_compressed.items(): result = bz2.decompress(data[filename]) - self.assertEqual(compressed, data[filename]) - self.assertEqual(self.expected, result) + assert compressed == data[filename] + assert result == self.expected @defer.inlineCallbacks def test_custom_plugin(self): @@ -2145,7 +2131,7 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): } data = yield self.exported_data(self.items, settings) - self.assertEqual(self.expected, data[filename]) + assert data[filename] == self.expected @defer.inlineCallbacks def test_custom_plugin_with_parameter(self): @@ -2163,7 +2149,7 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): } data = yield self.exported_data(self.items, settings) - self.assertEqual(expected, data[filename]) + assert data[filename] == expected @defer.inlineCallbacks def test_custom_plugin_with_compression(self): @@ -2208,7 +2194,7 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): for filename, decompressor in filename_to_decompressor.items(): result = decompressor(data[filename]) - self.assertEqual(expected, result) + assert result == expected @defer.inlineCallbacks def test_exports_compatibility_with_postproc(self): @@ -2262,10 +2248,10 @@ class FeedPostProcessedExportsTest(FeedExportTestBase): expected, result = self.items[0], marshal.loads(result) else: expected = filename_to_expected[filename] - self.assertEqual(expected, result) + assert result == expected -class BatchDeliveriesTest(FeedExportTestBase): +class TestBatchDeliveries(TestFeedExportBase): _file_mark = "_%(batch_time)s_#%(batch_id)02d_" @defer.inlineCallbacks @@ -2310,7 +2296,7 @@ class BatchDeliveriesTest(FeedExportTestBase): json.loads(to_unicode(batch_item)) for batch_item in batch.splitlines() ] expected_batch, rows = rows[:batch_size], rows[batch_size:] - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch @defer.inlineCallbacks def assertExportedCsv(self, items, header, rows, settings=None): @@ -2328,9 +2314,9 @@ class BatchDeliveriesTest(FeedExportTestBase): data = yield self.exported_data(items, settings) for batch in data["csv"]: got_batch = csv.DictReader(to_unicode(batch).splitlines()) - self.assertEqual(list(header), got_batch.fieldnames) + assert list(header) == got_batch.fieldnames expected_batch, rows = rows[:batch_size], rows[batch_size:] - self.assertEqual(expected_batch, list(got_batch)) + assert list(got_batch) == expected_batch @defer.inlineCallbacks def assertExportedXml(self, items, rows, settings=None): @@ -2351,7 +2337,7 @@ class BatchDeliveriesTest(FeedExportTestBase): root = lxml.etree.fromstring(batch) got_batch = [{e.tag: e.text for e in it} for it in root.findall("item")] expected_batch, rows = rows[:batch_size], rows[batch_size:] - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch @defer.inlineCallbacks def assertExportedMultiple(self, items, rows, settings=None): @@ -2377,13 +2363,13 @@ class BatchDeliveriesTest(FeedExportTestBase): root = lxml.etree.fromstring(batch) got_batch = [{e.tag: e.text for e in it} for it in root.findall("item")] expected_batch, xml_rows = xml_rows[:batch_size], xml_rows[batch_size:] - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch # JSON json_rows = rows.copy() for batch in data["json"]: got_batch = json.loads(batch.decode("utf-8")) expected_batch, json_rows = json_rows[:batch_size], json_rows[batch_size:] - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch @defer.inlineCallbacks def assertExportedPickle(self, items, rows, settings=None): @@ -2405,7 +2391,7 @@ class BatchDeliveriesTest(FeedExportTestBase): for batch in data["pickle"]: got_batch = self._load_until_eof(batch, load_func=pickle.load) expected_batch, rows = rows[:batch_size], rows[batch_size:] - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch @defer.inlineCallbacks def assertExportedMarshal(self, items, rows, settings=None): @@ -2427,7 +2413,7 @@ class BatchDeliveriesTest(FeedExportTestBase): for batch in data["marshal"]: got_batch = self._load_until_eof(batch, load_func=marshal.load) expected_batch, rows = rows[:batch_size], rows[batch_size:] - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch @defer.inlineCallbacks def test_export_items(self): @@ -2472,7 +2458,7 @@ class BatchDeliveriesTest(FeedExportTestBase): } data = yield self.exported_no_data(settings) data = dict(data) - self.assertEqual(0, len(data[fmt])) + assert len(data[fmt]) == 0 @defer.inlineCallbacks def test_export_no_items_store_empty(self): @@ -2496,7 +2482,7 @@ class BatchDeliveriesTest(FeedExportTestBase): } data = yield self.exported_no_data(settings) data = dict(data) - self.assertEqual(expctd, data[fmt][0]) + assert data[fmt][0] == expctd @defer.inlineCallbacks def test_export_multiple_configs(self): @@ -2552,7 +2538,7 @@ class BatchDeliveriesTest(FeedExportTestBase): data = yield self.exported_data(items, settings) for fmt, expected in formats.items(): for expected_batch, got_batch in zip(expected, data[fmt]): - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch @defer.inlineCallbacks def test_batch_item_count_feeds_setting(self): @@ -2576,7 +2562,7 @@ class BatchDeliveriesTest(FeedExportTestBase): data = yield self.exported_data(items, settings) for fmt, expected in formats.items(): for expected_batch, got_batch in zip(expected, data[fmt]): - self.assertEqual(expected_batch, got_batch) + assert got_batch == expected_batch @defer.inlineCallbacks def test_batch_path_differ(self): @@ -2598,7 +2584,7 @@ class BatchDeliveriesTest(FeedExportTestBase): "FEED_EXPORT_BATCH_ITEM_COUNT": 1, } data = yield self.exported_data(items, settings) - self.assertEqual(len(items), len(data["json"])) + assert len(items) == len(data["json"]) @defer.inlineCallbacks def test_stats_batch_file_success(self): @@ -2614,12 +2600,8 @@ class BatchDeliveriesTest(FeedExportTestBase): } crawler = get_crawler(ItemSpider, settings) yield crawler.crawl(total=2, mockserver=self.mockserver) - self.assertIn( - "feedexport/success_count/FileFeedStorage", crawler.stats.get_stats() - ) - self.assertEqual( - crawler.stats.get_value("feedexport/success_count/FileFeedStorage"), 12 - ) + assert "feedexport/success_count/FileFeedStorage" in crawler.stats.get_stats() + assert crawler.stats.get_value("feedexport/success_count/FileFeedStorage") == 12 @pytest.mark.requires_boto3 @defer.inlineCallbacks @@ -2687,13 +2669,13 @@ class BatchDeliveriesTest(FeedExportTestBase): crawler = get_crawler(TestSpider, settings) yield crawler.crawl() - self.assertEqual(len(CustomS3FeedStorage.stubs), len(items)) + assert len(CustomS3FeedStorage.stubs) == len(items) for stub in CustomS3FeedStorage.stubs[:-1]: stub.assert_no_pending_responses() # Test that the FeedExporer sends the feed_exporter_closed and feed_slot_closed signals -class FeedExporterSignalsTest(unittest.TestCase): +class TestFeedExporterSignals: items = [ {"foo": "bar1", "egg": "spam1"}, {"foo": "bar2", "egg": "spam2", "baz": "quux2"}, @@ -2754,8 +2736,8 @@ class FeedExporterSignalsTest(unittest.TestCase): self.feed_exporter_closed_signal_handler, self.feed_slot_closed_signal_handler, ) - self.assertTrue(self.feed_slot_closed_received) - self.assertTrue(self.feed_exporter_closed_received) + assert self.feed_slot_closed_received + assert self.feed_exporter_closed_received def test_feed_exporter_signals_sent_deferred(self): self.feed_exporter_closed_received = False @@ -2765,11 +2747,11 @@ class FeedExporterSignalsTest(unittest.TestCase): self.feed_exporter_closed_signal_handler_deferred, self.feed_slot_closed_signal_handler_deferred, ) - self.assertTrue(self.feed_slot_closed_received) - self.assertTrue(self.feed_exporter_closed_received) + assert self.feed_slot_closed_received + assert self.feed_exporter_closed_received -class FeedExportInitTest(unittest.TestCase): +class TestFeedExportInit: def test_unsupported_storage(self): settings = { "FEEDS": { @@ -2803,7 +2785,7 @@ class FeedExportInitTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings) exporter = FeedExporter.from_crawler(crawler) - self.assertIsInstance(exporter, FeedExporter) + assert isinstance(exporter, FeedExporter) def test_relative_pathlib_as_uri(self): settings = { @@ -2815,13 +2797,14 @@ class FeedExportInitTest(unittest.TestCase): } crawler = get_crawler(settings_dict=settings) exporter = FeedExporter.from_crawler(crawler) - self.assertIsInstance(exporter, FeedExporter) + assert isinstance(exporter, FeedExporter) -class URIParamsTest: +class TestURIParams(ABC): spider_name = "uri_params_spider" deprecated_options = False + @abstractmethod def build_settings(self, uri="file:///tmp/foobar", uri_params=None): raise NotImplementedError @@ -2850,7 +2833,7 @@ class URIParamsTest: warnings.simplefilter("error", ScrapyDeprecationWarning) feed_exporter.open_spider(spider) - self.assertEqual(feed_exporter.slots[0].uri, f"file:///tmp/{self.spider_name}") + assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" def test_none(self): def uri_params(params, spider): @@ -2866,7 +2849,7 @@ class URIParamsTest: feed_exporter.open_spider(spider) - self.assertEqual(feed_exporter.slots[0].uri, f"file:///tmp/{self.spider_name}") + assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" def test_empty_dict(self): def uri_params(params, spider): @@ -2900,7 +2883,7 @@ class URIParamsTest: warnings.simplefilter("error", ScrapyDeprecationWarning) feed_exporter.open_spider(spider) - self.assertEqual(feed_exporter.slots[0].uri, f"file:///tmp/{self.spider_name}") + assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" def test_custom_param(self): def uri_params(params, spider): @@ -2917,10 +2900,10 @@ class URIParamsTest: warnings.simplefilter("error", ScrapyDeprecationWarning) feed_exporter.open_spider(spider) - self.assertEqual(feed_exporter.slots[0].uri, f"file:///tmp/{self.spider_name}") + assert feed_exporter.slots[0].uri == f"file:///tmp/{self.spider_name}" -class URIParamsSettingTest(URIParamsTest, unittest.TestCase): +class TestURIParamsSetting(TestURIParams): deprecated_options = True def build_settings(self, uri="file:///tmp/foobar", uri_params=None): @@ -2933,7 +2916,7 @@ class URIParamsSettingTest(URIParamsTest, unittest.TestCase): } -class URIParamsFeedOptionTest(URIParamsTest, unittest.TestCase): +class TestURIParamsFeedOption(TestURIParams): deprecated_options = False def build_settings(self, uri="file:///tmp/foobar", uri_params=None): diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py index 0881bbeca..7c1b38877 100644 --- a/tests/test_http2_client_protocol.py +++ b/tests/test_http2_client_protocol.py @@ -185,7 +185,7 @@ def get_client_certificate( @skipIf(not H2_ENABLED, "HTTP/2 support in Twisted is not enabled") -class Https2ClientProtocolTestCase(TestCase): +class TestHttps2ClientProtocol(TestCase): scheme = "https" key_file = Path(__file__).parent / "keys" / "localhost.key" certificate_file = Path(__file__).parent / "keys" / "localhost.crt" @@ -277,14 +277,14 @@ class Https2ClientProtocolTestCase(TestCase): def _check_GET(self, request: Request, expected_body, expected_status): def check_response(response: Response): - self.assertEqual(response.status, expected_status) - self.assertEqual(response.body, expected_body) - self.assertEqual(response.request, request) + assert response.status == expected_status + assert response.body == expected_body + assert response.request == request content_length_header = response.headers.get("Content-Length") assert content_length_header is not None content_length = int(content_length_header) - self.assertEqual(len(response.body), content_length) + assert len(response.body) == content_length d = self.make_request(request) d.addCallback(check_response) @@ -325,35 +325,35 @@ class Https2ClientProtocolTestCase(TestCase): d = self.make_request(request) def assert_response(response: Response): - self.assertEqual(response.status, expected_status) - self.assertEqual(response.request, request) + assert response.status == expected_status + assert response.request == request content_length_header = response.headers.get("Content-Length") assert content_length_header is not None content_length = int(content_length_header) - self.assertEqual(len(response.body), content_length) + assert len(response.body) == content_length # Parse the body content_encoding_header = response.headers[b"Content-Encoding"] assert content_encoding_header is not None content_encoding = str(content_encoding_header, "utf-8") body = json.loads(str(response.body, content_encoding)) - self.assertIn("request-body", body) - self.assertIn("extra-data", body) - self.assertIn("request-headers", body) + assert "request-body" in body + assert "extra-data" in body + assert "request-headers" in body request_body = body["request-body"] - self.assertEqual(request_body, expected_request_body) + assert request_body == expected_request_body extra_data = body["extra-data"] - self.assertEqual(extra_data, expected_extra_data) + assert extra_data == expected_extra_data # Check if headers were sent successfully request_headers = body["request-headers"] for k, v in request.headers.items(): k_str = str(k, "utf-8") - self.assertIn(k_str, request_headers) - self.assertEqual(request_headers[k_str], str(v[0], "utf-8")) + assert k_str in request_headers + assert request_headers[k_str] == str(v[0], "utf-8") d.addCallback(assert_response) d.addErrback(self.fail) @@ -414,8 +414,8 @@ class Https2ClientProtocolTestCase(TestCase): request = Request(url=self.get_url("/get-data-html-large")) def assert_response(response: Response): - self.assertEqual(response.status, 499) - self.assertEqual(response.request, request) + assert response.status == 499 + assert response.request == request d = self.make_request(request) d.addCallback(assert_response) @@ -430,12 +430,12 @@ class Https2ClientProtocolTestCase(TestCase): ) def assert_cancelled_error(failure): - self.assertIsInstance(failure.value, CancelledError) + assert isinstance(failure.value, CancelledError) error_pattern = re.compile( rf"Cancelling download of {request.url}: received response " rf"size \(\d*\) larger than download max size \(1000\)" ) - self.assertEqual(len(re.findall(error_pattern, str(failure.value))), 1) + assert len(re.findall(error_pattern, str(failure.value))) == 1 d = self.make_request(request) d.addCallback(self.fail) @@ -448,14 +448,12 @@ class Https2ClientProtocolTestCase(TestCase): request = Request(url=self.get_url("/dataloss")) def assert_failure(failure: Failure): - self.assertTrue(len(failure.value.reasons) > 0) + assert len(failure.value.reasons) > 0 from h2.exceptions import InvalidBodyLengthError - self.assertTrue( - any( - isinstance(error, InvalidBodyLengthError) - for error in failure.value.reasons - ) + assert any( + isinstance(error, InvalidBodyLengthError) + for error in failure.value.reasons ) d = self.make_request(request) @@ -467,10 +465,10 @@ class Https2ClientProtocolTestCase(TestCase): request = Request(url=self.get_url("/no-content-length-header")) def assert_content_length(response: Response): - self.assertEqual(response.status, 200) - self.assertEqual(response.body, Data.NO_CONTENT_LENGTH) - self.assertEqual(response.request, request) - self.assertNotIn("Content-Length", response.headers) + assert response.status == 200 + assert response.body == Data.NO_CONTENT_LENGTH + assert response.request == request + assert "Content-Length" not in response.headers d = self.make_request(request) d.addCallback(assert_content_length) @@ -481,14 +479,12 @@ class Https2ClientProtocolTestCase(TestCase): def _check_log_warnsize(self, request, warn_pattern, expected_body): with self.assertLogs("scrapy.core.http2.stream", level="WARNING") as cm: response = yield self.make_request(request) - self.assertEqual(response.status, 200) - self.assertEqual(response.request, request) - self.assertEqual(response.body, expected_body) + assert response.status == 200 + assert response.request == request + assert response.body == expected_body # Check the warning is raised only once for this request - self.assertEqual( - sum(len(re.findall(warn_pattern, log)) for log in cm.output), 1 - ) + assert sum(len(re.findall(warn_pattern, log)) for log in cm.output) == 1 @inlineCallbacks def test_log_expected_warnsize(self): @@ -534,11 +530,11 @@ class Https2ClientProtocolTestCase(TestCase): d_list = [] def assert_inactive_stream(failure): - self.assertIsNotNone(failure.check(ResponseFailed)) + assert failure.check(ResponseFailed) is not None from scrapy.core.http2.stream import InactiveStreamClosed - self.assertTrue( - any(isinstance(e, InactiveStreamClosed) for e in failure.value.reasons) + assert any( + isinstance(e, InactiveStreamClosed) for e in failure.value.reasons ) # Send 100 request (we do not check the result) @@ -578,7 +574,7 @@ class Https2ClientProtocolTestCase(TestCase): assert content_encoding_header is not None content_encoding = str(content_encoding_header, "utf-8") data = json.loads(str(response.body, content_encoding)) - self.assertEqual(data, params) + assert data == params d = self.make_request(request) d.addCallback(assert_query_params) @@ -588,7 +584,7 @@ class Https2ClientProtocolTestCase(TestCase): def test_status_codes(self): def assert_response_status(response: Response, expected_status: int): - self.assertEqual(response.status, expected_status) + assert response.status == expected_status d_list = [] for status in [200, 404]: @@ -604,21 +600,18 @@ class Https2ClientProtocolTestCase(TestCase): request = Request(self.get_url("/status?n=200")) def assert_metadata(response: Response): - self.assertEqual(response.request, request) - self.assertIsInstance(response.certificate, Certificate) - assert response.certificate # typing - self.assertIsNotNone(response.certificate.original) - self.assertEqual( - response.certificate.getIssuer(), self.client_certificate.getIssuer() + assert response.request == request + assert isinstance(response.certificate, Certificate) + assert response.certificate.original is not None + assert ( + response.certificate.getIssuer() == self.client_certificate.getIssuer() ) - self.assertTrue( - response.certificate.getPublicKey().matches( - self.client_certificate.getPublicKey() - ) + assert response.certificate.getPublicKey().matches( + self.client_certificate.getPublicKey() ) - self.assertIsInstance(response.ip_address, IPv4Address) - self.assertEqual(str(response.ip_address), "127.0.0.1") + assert isinstance(response.ip_address, IPv4Address) + assert str(response.ip_address) == "127.0.0.1" d = self.make_request(request) d.addCallback(assert_metadata) @@ -632,11 +625,11 @@ class Https2ClientProtocolTestCase(TestCase): def assert_invalid_hostname(failure: Failure): from scrapy.core.http2.stream import InvalidHostname - self.assertIsNotNone(failure.check(InvalidHostname)) + assert failure.check(InvalidHostname) is not None error_msg = str(failure.value) - self.assertIn("localhost", error_msg) - self.assertIn("127.0.0.1", error_msg) - self.assertIn(str(request), error_msg) + assert "localhost" in error_msg + assert "127.0.0.1" in error_msg + assert str(request) in error_msg d = self.make_request(request) d.addCallback(self.fail) @@ -672,13 +665,13 @@ class Https2ClientProtocolTestCase(TestCase): from scrapy.core.http2.protocol import H2ClientProtocol if isinstance(err, TimeoutError): - self.assertIn( - f"Connection was IDLE for more than {H2ClientProtocol.IDLE_TIMEOUT}s", - str(err), + assert ( + f"Connection was IDLE for more than {H2ClientProtocol.IDLE_TIMEOUT}s" + in str(err) ) break else: - self.fail() + pytest.fail("No TimeoutError raised.") d.addCallback(self.fail) d.addErrback(assert_timeout_error) @@ -692,15 +685,15 @@ class Https2ClientProtocolTestCase(TestCase): d = self.make_request(request) def assert_request_headers(response: Response): - self.assertEqual(response.status, 200) - self.assertEqual(response.request, request) + assert response.status == 200 + assert response.request == request response_headers = json.loads(str(response.body, "utf-8")) - self.assertIsInstance(response_headers, dict) + assert isinstance(response_headers, dict) for k, v in request.headers.items(): k, v = str(k, "utf-8"), str(v[0], "utf-8") - self.assertIn(k, response_headers) - self.assertEqual(v, response_headers[k]) + assert k in response_headers + assert v == response_headers[k] d.addErrback(self.fail) d.addCallback(assert_request_headers) diff --git a/tests/test_http_cookies.py b/tests/test_http_cookies.py index 932644320..660b76d08 100644 --- a/tests/test_http_cookies.py +++ b/tests/test_http_cookies.py @@ -1,74 +1,72 @@ -from unittest import TestCase - from scrapy.http import Request, Response from scrapy.http.cookies import WrappedRequest, WrappedResponse from scrapy.utils.httpobj import urlparse_cached -class WrappedRequestTest(TestCase): - def setUp(self): +class TestWrappedRequest: + def setup_method(self): self.request = Request( "http://www.example.com/page.html", headers={"Content-Type": "text/html"} ) self.wrapped = WrappedRequest(self.request) def test_get_full_url(self): - self.assertEqual(self.wrapped.get_full_url(), self.request.url) - self.assertEqual(self.wrapped.full_url, self.request.url) + assert self.wrapped.get_full_url() == self.request.url + assert self.wrapped.full_url == self.request.url def test_get_host(self): - self.assertEqual(self.wrapped.get_host(), urlparse_cached(self.request).netloc) - self.assertEqual(self.wrapped.host, urlparse_cached(self.request).netloc) + assert self.wrapped.get_host() == urlparse_cached(self.request).netloc + assert self.wrapped.host == urlparse_cached(self.request).netloc def test_get_type(self): - self.assertEqual(self.wrapped.get_type(), urlparse_cached(self.request).scheme) - self.assertEqual(self.wrapped.type, urlparse_cached(self.request).scheme) + assert self.wrapped.get_type() == urlparse_cached(self.request).scheme + assert self.wrapped.type == urlparse_cached(self.request).scheme def test_is_unverifiable(self): - self.assertFalse(self.wrapped.is_unverifiable()) - self.assertFalse(self.wrapped.unverifiable) + assert not self.wrapped.is_unverifiable() + assert not self.wrapped.unverifiable def test_is_unverifiable2(self): self.request.meta["is_unverifiable"] = True - self.assertTrue(self.wrapped.is_unverifiable()) - self.assertTrue(self.wrapped.unverifiable) + assert self.wrapped.is_unverifiable() + assert self.wrapped.unverifiable def test_get_origin_req_host(self): - self.assertEqual(self.wrapped.origin_req_host, "www.example.com") + assert self.wrapped.origin_req_host == "www.example.com" def test_has_header(self): - self.assertTrue(self.wrapped.has_header("content-type")) - self.assertFalse(self.wrapped.has_header("xxxxx")) + assert self.wrapped.has_header("content-type") + assert not self.wrapped.has_header("xxxxx") def test_get_header(self): - self.assertEqual(self.wrapped.get_header("content-type"), "text/html") - self.assertEqual(self.wrapped.get_header("xxxxx", "def"), "def") - self.assertEqual(self.wrapped.get_header("xxxxx"), None) + assert self.wrapped.get_header("content-type") == "text/html" + assert self.wrapped.get_header("xxxxx", "def") == "def" + assert self.wrapped.get_header("xxxxx") is None wrapped = WrappedRequest( Request( "http://www.example.com/page.html", headers={"empty-binary-header": b""} ) ) - self.assertEqual(wrapped.get_header("empty-binary-header"), "") + assert wrapped.get_header("empty-binary-header") == "" def test_header_items(self): - self.assertEqual(self.wrapped.header_items(), [("Content-Type", ["text/html"])]) + assert self.wrapped.header_items() == [("Content-Type", ["text/html"])] def test_add_unredirected_header(self): self.wrapped.add_unredirected_header("hello", "world") - self.assertEqual(self.request.headers["hello"], b"world") + assert self.request.headers["hello"] == b"world" -class WrappedResponseTest(TestCase): - def setUp(self): +class TestWrappedResponse: + def setup_method(self): self.response = Response( "http://www.example.com/page.html", headers={"Content-TYpe": "text/html"} ) self.wrapped = WrappedResponse(self.response) def test_info(self): - self.assertIs(self.wrapped.info(), self.wrapped) + assert self.wrapped.info() is self.wrapped def test_get_all(self): # get_all result must be native string - self.assertEqual(self.wrapped.get_all("content-type"), ["text/html"]) + assert self.wrapped.get_all("content-type") == ["text/html"] diff --git a/tests/test_http_headers.py b/tests/test_http_headers.py index 0bbbcda46..2fcf9e83c 100644 --- a/tests/test_http_headers.py +++ b/tests/test_http_headers.py @@ -1,14 +1,13 @@ import copy -import unittest import pytest from scrapy.http import Headers -class HeadersTest(unittest.TestCase): +class TestHeaders: def assertSortedEqual(self, first, second, msg=None): - return self.assertEqual(sorted(first), sorted(second), msg) + assert sorted(first) == sorted(second), msg def test_basics(self): h = Headers({"Content-Type": "text/html", "Content-Length": 1234}) @@ -17,53 +16,53 @@ class HeadersTest(unittest.TestCase): with pytest.raises(KeyError): h["Accept"] - self.assertEqual(h.get("Accept"), None) - self.assertEqual(h.getlist("Accept"), []) + assert h.get("Accept") is None + assert h.getlist("Accept") == [] - self.assertEqual(h.get("Accept", "*/*"), b"*/*") - self.assertEqual(h.getlist("Accept", "*/*"), [b"*/*"]) - self.assertEqual( - h.getlist("Accept", ["text/html", "images/jpeg"]), - [b"text/html", b"images/jpeg"], - ) + assert h.get("Accept", "*/*") == b"*/*" + assert h.getlist("Accept", "*/*") == [b"*/*"] + assert h.getlist("Accept", ["text/html", "images/jpeg"]) == [ + b"text/html", + b"images/jpeg", + ] def test_single_value(self): h = Headers() h["Content-Type"] = "text/html" - self.assertEqual(h["Content-Type"], b"text/html") - self.assertEqual(h.get("Content-Type"), b"text/html") - self.assertEqual(h.getlist("Content-Type"), [b"text/html"]) + assert h["Content-Type"] == b"text/html" + assert h.get("Content-Type") == b"text/html" + assert h.getlist("Content-Type") == [b"text/html"] def test_multivalue(self): h = Headers() h["X-Forwarded-For"] = hlist = ["ip1", "ip2"] - self.assertEqual(h["X-Forwarded-For"], b"ip2") - self.assertEqual(h.get("X-Forwarded-For"), b"ip2") - self.assertEqual(h.getlist("X-Forwarded-For"), [b"ip1", b"ip2"]) + assert h["X-Forwarded-For"] == b"ip2" + assert h.get("X-Forwarded-For") == b"ip2" + assert h.getlist("X-Forwarded-For") == [b"ip1", b"ip2"] assert h.getlist("X-Forwarded-For") is not hlist def test_multivalue_for_one_header(self): h = Headers((("a", "b"), ("a", "c"))) - self.assertEqual(h["a"], b"c") - self.assertEqual(h.get("a"), b"c") - self.assertEqual(h.getlist("a"), [b"b", b"c"]) + assert h["a"] == b"c" + assert h.get("a") == b"c" + assert h.getlist("a") == [b"b", b"c"] def test_encode_utf8(self): h = Headers({"key": "\xa3"}, encoding="utf-8") key, val = dict(h).popitem() assert isinstance(key, bytes), key assert isinstance(val[0], bytes), val[0] - self.assertEqual(val[0], b"\xc2\xa3") + assert val[0] == b"\xc2\xa3" def test_encode_latin1(self): h = Headers({"key": "\xa3"}, encoding="latin1") key, val = dict(h).popitem() - self.assertEqual(val[0], b"\xa3") + assert val[0] == b"\xa3" def test_encode_multiple(self): h = Headers({"key": ["\xa3"]}, encoding="utf-8") key, val = dict(h).popitem() - self.assertEqual(val[0], b"\xc2\xa3") + assert val[0] == b"\xc2\xa3" def test_delete_and_contains(self): h = Headers() @@ -81,17 +80,17 @@ class HeadersTest(unittest.TestCase): h = Headers() olist = h.setdefault("X-Forwarded-For", "ip1") - self.assertEqual(h.getlist("X-Forwarded-For"), [b"ip1"]) + assert h.getlist("X-Forwarded-For") == [b"ip1"] assert h.getlist("X-Forwarded-For") is olist def test_iterables(self): idict = {"Content-Type": "text/html", "X-Forwarded-For": ["ip1", "ip2"]} h = Headers(idict) - self.assertDictEqual( - dict(h), - {b"Content-Type": [b"text/html"], b"X-Forwarded-For": [b"ip1", b"ip2"]}, - ) + assert dict(h) == { + b"Content-Type": [b"text/html"], + b"X-Forwarded-For": [b"ip1", b"ip2"], + } self.assertSortedEqual(h.keys(), [b"X-Forwarded-For", b"Content-Type"]) self.assertSortedEqual( h.items(), @@ -102,57 +101,57 @@ class HeadersTest(unittest.TestCase): def test_update(self): h = Headers() h.update({"Content-Type": "text/html", "X-Forwarded-For": ["ip1", "ip2"]}) - self.assertEqual(h.getlist("Content-Type"), [b"text/html"]) - self.assertEqual(h.getlist("X-Forwarded-For"), [b"ip1", b"ip2"]) + assert h.getlist("Content-Type") == [b"text/html"] + assert h.getlist("X-Forwarded-For") == [b"ip1", b"ip2"] def test_copy(self): h1 = Headers({"header1": ["value1", "value2"]}) h2 = copy.copy(h1) - self.assertEqual(h1, h2) - self.assertEqual(h1.getlist("header1"), h2.getlist("header1")) + assert h1 == h2 + assert h1.getlist("header1") == h2.getlist("header1") assert h1.getlist("header1") is not h2.getlist("header1") assert isinstance(h2, Headers) def test_appendlist(self): h1 = Headers({"header1": "value1"}) h1.appendlist("header1", "value3") - self.assertEqual(h1.getlist("header1"), [b"value1", b"value3"]) + assert h1.getlist("header1") == [b"value1", b"value3"] h1 = Headers() h1.appendlist("header1", "value1") h1.appendlist("header1", "value3") - self.assertEqual(h1.getlist("header1"), [b"value1", b"value3"]) + assert h1.getlist("header1") == [b"value1", b"value3"] def test_setlist(self): h1 = Headers({"header1": "value1"}) - self.assertEqual(h1.getlist("header1"), [b"value1"]) + assert h1.getlist("header1") == [b"value1"] h1.setlist("header1", [b"value2", b"value3"]) - self.assertEqual(h1.getlist("header1"), [b"value2", b"value3"]) + assert h1.getlist("header1") == [b"value2", b"value3"] def test_setlistdefault(self): h1 = Headers({"header1": "value1"}) h1.setlistdefault("header1", ["value2", "value3"]) h1.setlistdefault("header2", ["value2", "value3"]) - self.assertEqual(h1.getlist("header1"), [b"value1"]) - self.assertEqual(h1.getlist("header2"), [b"value2", b"value3"]) + assert h1.getlist("header1") == [b"value1"] + assert h1.getlist("header2") == [b"value2", b"value3"] def test_none_value(self): h1 = Headers() h1["foo"] = "bar" h1["foo"] = None h1.setdefault("foo", "bar") - self.assertEqual(h1.get("foo"), None) - self.assertEqual(h1.getlist("foo"), []) + assert h1.get("foo") is None + assert h1.getlist("foo") == [] def test_int_value(self): h1 = Headers({"hey": 5}) h1["foo"] = 1 h1.setdefault("bar", 2) h1.setlist("buz", [1, "dos", 3]) - self.assertEqual(h1.getlist("foo"), [b"1"]) - self.assertEqual(h1.getlist("bar"), [b"2"]) - self.assertEqual(h1.getlist("buz"), [b"1", b"dos", b"3"]) - self.assertEqual(h1.getlist("hey"), [b"5"]) + assert h1.getlist("foo") == [b"1"] + assert h1.getlist("bar") == [b"2"] + assert h1.getlist("buz") == [b"1", b"dos", b"3"] + assert h1.getlist("hey") == [b"5"] def test_invalid_value(self): with pytest.raises(TypeError, match="Unsupported value type"): diff --git a/tests/test_http_request.py b/tests/test_http_request.py index e5291157d..6bf0b8e3f 100644 --- a/tests/test_http_request.py +++ b/tests/test_http_request.py @@ -1,6 +1,5 @@ import json import re -import unittest import warnings import xmlrpc.client from typing import Any @@ -22,7 +21,7 @@ from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.python import to_bytes, to_unicode -class RequestTest(unittest.TestCase): +class TestRequest: request_class = Request default_method = "GET" default_headers: dict[bytes, list[bytes]] = {} @@ -40,12 +39,12 @@ class RequestTest(unittest.TestCase): r = self.request_class("http://www.example.com") assert isinstance(r.url, str) - self.assertEqual(r.url, "http://www.example.com") - self.assertEqual(r.method, self.default_method) + assert r.url == "http://www.example.com" + assert r.method == self.default_method assert isinstance(r.headers, Headers) - self.assertEqual(r.headers, self.default_headers) - self.assertEqual(r.meta, self.default_meta) + assert r.headers == self.default_headers + assert r.meta == self.default_meta meta = {"lala": "lolo"} headers = {b"caca": b"coco"} @@ -54,9 +53,9 @@ class RequestTest(unittest.TestCase): ) assert r.meta is not meta - self.assertEqual(r.meta, meta) + assert r.meta == meta assert r.headers is not headers - self.assertEqual(r.headers[b"caca"], b"coco") + assert r.headers[b"caca"] == b"coco" def test_url_scheme(self): # This test passes by not raising any (ValueError) exception @@ -83,61 +82,61 @@ class RequestTest(unittest.TestCase): r = self.request_class(url=url, headers=headers) p = self.request_class(url=url, headers=r.headers) - self.assertEqual(r.headers, p.headers) - self.assertFalse(r.headers is headers) - self.assertFalse(p.headers is r.headers) + assert r.headers == p.headers + assert r.headers is not headers + assert p.headers is not r.headers # headers must not be unicode h = Headers({"key1": "val1", "key2": "val2"}) h["newkey"] = "newval" for k, v in h.items(): - self.assertIsInstance(k, bytes) + assert isinstance(k, bytes) for s in v: - self.assertIsInstance(s, bytes) + assert isinstance(s, bytes) def test_eq(self): url = "http://www.scrapy.org" r1 = self.request_class(url=url) r2 = self.request_class(url=url) - self.assertNotEqual(r1, r2) + assert r1 != r2 set_ = set() set_.add(r1) set_.add(r2) - self.assertEqual(len(set_), 2) + assert len(set_) == 2 def test_url(self): r = self.request_class(url="http://www.scrapy.org/path") - self.assertEqual(r.url, "http://www.scrapy.org/path") + assert r.url == "http://www.scrapy.org/path" def test_url_quoting(self): r = self.request_class(url="http://www.scrapy.org/blank%20space") - self.assertEqual(r.url, "http://www.scrapy.org/blank%20space") + assert r.url == "http://www.scrapy.org/blank%20space" r = self.request_class(url="http://www.scrapy.org/blank space") - self.assertEqual(r.url, "http://www.scrapy.org/blank%20space") + assert r.url == "http://www.scrapy.org/blank%20space" def test_url_encoding(self): r = self.request_class(url="http://www.scrapy.org/price/£") - self.assertEqual(r.url, "http://www.scrapy.org/price/%C2%A3") + assert r.url == "http://www.scrapy.org/price/%C2%A3" def test_url_encoding_other(self): # encoding affects only query part of URI, not path # path part should always be UTF-8 encoded before percent-escaping r = self.request_class(url="http://www.scrapy.org/price/£", encoding="utf-8") - self.assertEqual(r.url, "http://www.scrapy.org/price/%C2%A3") + assert r.url == "http://www.scrapy.org/price/%C2%A3" r = self.request_class(url="http://www.scrapy.org/price/£", encoding="latin1") - self.assertEqual(r.url, "http://www.scrapy.org/price/%C2%A3") + assert r.url == "http://www.scrapy.org/price/%C2%A3" def test_url_encoding_query(self): r1 = self.request_class(url="http://www.scrapy.org/price/£?unit=µ") - self.assertEqual(r1.url, "http://www.scrapy.org/price/%C2%A3?unit=%C2%B5") + assert r1.url == "http://www.scrapy.org/price/%C2%A3?unit=%C2%B5" # should be same as above r2 = self.request_class( url="http://www.scrapy.org/price/£?unit=µ", encoding="utf-8" ) - self.assertEqual(r2.url, "http://www.scrapy.org/price/%C2%A3?unit=%C2%B5") + assert r2.url == "http://www.scrapy.org/price/%C2%A3?unit=%C2%B5" def test_url_encoding_query_latin1(self): # encoding is used for encoding query-string before percent-escaping; @@ -145,7 +144,7 @@ class RequestTest(unittest.TestCase): r3 = self.request_class( url="http://www.scrapy.org/price/µ?currency=£", encoding="latin1" ) - self.assertEqual(r3.url, "http://www.scrapy.org/price/%C2%B5?currency=%A3") + assert r3.url == "http://www.scrapy.org/price/%C2%B5?currency=%A3" def test_url_encoding_nonutf8_untouched(self): # percent-escaping sequences that do not match valid UTF-8 sequences @@ -164,16 +163,16 @@ class RequestTest(unittest.TestCase): # "http://www.example.org/r%C3%A9sum%C3%A9.html", which is a different # URI from "http://www.example.org/r%E9sum%E9.html". r1 = self.request_class(url="http://www.scrapy.org/price/%a3") - self.assertEqual(r1.url, "http://www.scrapy.org/price/%a3") + assert r1.url == "http://www.scrapy.org/price/%a3" r2 = self.request_class(url="http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3") - self.assertEqual(r2.url, "http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3") + assert r2.url == "http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3" r3 = self.request_class(url="http://www.scrapy.org/résumé/%a3") - self.assertEqual(r3.url, "http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3") + assert r3.url == "http://www.scrapy.org/r%C3%A9sum%C3%A9/%a3" r4 = self.request_class(url="http://www.example.org/r%E9sum%E9.html") - self.assertEqual(r4.url, "http://www.example.org/r%E9sum%E9.html") + assert r4.url == "http://www.example.org/r%E9sum%E9.html" def test_body(self): r1 = self.request_class(url="http://www.example.com/") @@ -181,19 +180,19 @@ class RequestTest(unittest.TestCase): r2 = self.request_class(url="http://www.example.com/", body=b"") assert isinstance(r2.body, bytes) - self.assertEqual(r2.encoding, "utf-8") # default encoding + assert r2.encoding == "utf-8" # default encoding r3 = self.request_class( url="http://www.example.com/", body="Price: \xa3100", encoding="utf-8" ) assert isinstance(r3.body, bytes) - self.assertEqual(r3.body, b"Price: \xc2\xa3100") + assert r3.body == b"Price: \xc2\xa3100" r4 = self.request_class( url="http://www.example.com/", body="Price: \xa3100", encoding="latin1" ) assert isinstance(r4.body, bytes) - self.assertEqual(r4.body, b"Price: \xa3100") + assert r4.body == b"Price: \xa3100" def test_copy(self): """Test Request copy""" @@ -219,25 +218,25 @@ class RequestTest(unittest.TestCase): # make sure flags list is shallow copied assert r1.flags is not r2.flags, "flags must be a shallow copy, not identical" - self.assertEqual(r1.flags, r2.flags) + assert r1.flags == r2.flags # make sure cb_kwargs dict is shallow copied assert r1.cb_kwargs is not r2.cb_kwargs, ( "cb_kwargs must be a shallow copy, not identical" ) - self.assertEqual(r1.cb_kwargs, r2.cb_kwargs) + assert r1.cb_kwargs == r2.cb_kwargs # make sure meta dict is shallow copied assert r1.meta is not r2.meta, "meta must be a shallow copy, not identical" - self.assertEqual(r1.meta, r2.meta) + assert r1.meta == r2.meta # make sure headers attribute is shallow copied assert r1.headers is not r2.headers, ( "headers must be a shallow copy, not identical" ) - self.assertEqual(r1.headers, r2.headers) - self.assertEqual(r1.encoding, r2.encoding) - self.assertEqual(r1.dont_filter, r2.dont_filter) + assert r1.headers == r2.headers + assert r1.encoding == r2.encoding + assert r1.dont_filter == r2.dont_filter # Request.body can be identical since it's an immutable object (str) @@ -258,10 +257,10 @@ class RequestTest(unittest.TestCase): hdrs = Headers(r1.headers) hdrs[b"key"] = b"value" r2 = r1.replace(method="POST", body="New body", headers=hdrs) - self.assertEqual(r1.url, r2.url) - self.assertEqual((r1.method, r2.method), ("GET", "POST")) - self.assertEqual((r1.body, r2.body), (b"", b"New body")) - self.assertEqual((r1.headers, r2.headers), (self.default_headers, hdrs)) + assert r1.url == r2.url + assert (r1.method, r2.method) == ("GET", "POST") + assert (r1.body, r2.body) == (b"", b"New body") + assert (r1.headers, r2.headers) == (self.default_headers, hdrs) # Empty attributes (which may fail if not compared properly) r3 = self.request_class( @@ -270,9 +269,9 @@ class RequestTest(unittest.TestCase): r4 = r3.replace( url="http://www.example.com/2", body=b"", meta={}, dont_filter=False ) - self.assertEqual(r4.url, "http://www.example.com/2") - self.assertEqual(r4.body, b"") - self.assertEqual(r4.meta, {}) + assert r4.url == "http://www.example.com/2" + assert r4.body == b"" + assert r4.meta == {} assert r4.dont_filter is False def test_method_always_str(self): @@ -291,32 +290,32 @@ class RequestTest(unittest.TestCase): pass r1 = self.request_class("http://example.com") - self.assertIsNone(r1.callback) - self.assertIsNone(r1.errback) + assert r1.callback is None + assert r1.errback is None r2 = self.request_class("http://example.com", callback=a_function) - self.assertIs(r2.callback, a_function) - self.assertIsNone(r2.errback) + assert r2.callback is a_function + assert r2.errback is None r3 = self.request_class("http://example.com", errback=a_function) - self.assertIsNone(r3.callback) - self.assertIs(r3.errback, a_function) + assert r3.callback is None + assert r3.errback is a_function r4 = self.request_class( url="http://example.com", callback=a_function, errback=a_function, ) - self.assertIs(r4.callback, a_function) - self.assertIs(r4.errback, a_function) + assert r4.callback is a_function + assert r4.errback is a_function r5 = self.request_class( url="http://example.com", callback=NO_CALLBACK, errback=NO_CALLBACK, ) - self.assertIs(r5.callback, NO_CALLBACK) - self.assertIs(r5.errback, NO_CALLBACK) + assert r5.callback is NO_CALLBACK + assert r5.errback is NO_CALLBACK def test_callback_and_errback_type(self): with pytest.raises(TypeError): @@ -354,53 +353,46 @@ class RequestTest(unittest.TestCase): "2%3A15&comments=' --compressed" ) r = self.request_class.from_curl(curl_command) - self.assertEqual(r.method, "POST") - self.assertEqual(r.url, "http://httpbin.org/post") - self.assertEqual( - r.body, - b"custname=John+Smith&custtel=500&custemail=jsmith%40" + assert r.method == "POST" + assert r.url == "http://httpbin.org/post" + assert ( + r.body == b"custname=John+Smith&custtel=500&custemail=jsmith%40" b"example.org&size=small&topping=cheese&topping=onion" - b"&delivery=12%3A15&comments=", - ) - self.assertEqual( - r.cookies, - { - "_gauges_unique_year": "1", - "_gauges_unique": "1", - "_gauges_unique_month": "1", - "_gauges_unique_hour": "1", - "_gauges_unique_day": "1", - }, - ) - self.assertEqual( - r.headers, - { - b"Origin": [b"http://httpbin.org"], - b"Accept-Encoding": [b"gzip, deflate"], - b"Accept-Language": [b"en-US,en;q=0.9,ru;q=0.8,es;q=0.7"], - b"Upgrade-Insecure-Requests": [b"1"], - b"User-Agent": [ - b"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537." - b"36 (KHTML, like Gecko) Ubuntu Chromium/62.0.3202" - b".75 Chrome/62.0.3202.75 Safari/537.36" - ], - b"Content-Type": [b"application /x-www-form-urlencoded"], - b"Accept": [ - b"text/html,application/xhtml+xml,application/xml;q=0." - b"9,image/webp,image/apng,*/*;q=0.8" - ], - b"Cache-Control": [b"max-age=0"], - b"Referer": [b"http://httpbin.org/forms/post"], - b"Connection": [b"keep-alive"], - }, + b"&delivery=12%3A15&comments=" ) + assert r.cookies == { + "_gauges_unique_year": "1", + "_gauges_unique": "1", + "_gauges_unique_month": "1", + "_gauges_unique_hour": "1", + "_gauges_unique_day": "1", + } + assert r.headers == { + b"Origin": [b"http://httpbin.org"], + b"Accept-Encoding": [b"gzip, deflate"], + b"Accept-Language": [b"en-US,en;q=0.9,ru;q=0.8,es;q=0.7"], + b"Upgrade-Insecure-Requests": [b"1"], + b"User-Agent": [ + b"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537." + b"36 (KHTML, like Gecko) Ubuntu Chromium/62.0.3202" + b".75 Chrome/62.0.3202.75 Safari/537.36" + ], + b"Content-Type": [b"application /x-www-form-urlencoded"], + b"Accept": [ + b"text/html,application/xhtml+xml,application/xml;q=0." + b"9,image/webp,image/apng,*/*;q=0.8" + ], + b"Cache-Control": [b"max-age=0"], + b"Referer": [b"http://httpbin.org/forms/post"], + b"Connection": [b"keep-alive"], + } def test_from_curl_with_kwargs(self): r = self.request_class.from_curl( 'curl -X PATCH "http://example.org"', method="POST", meta={"key": "value"} ) - self.assertEqual(r.method, "POST") - self.assertEqual(r.meta, {"key": "value"}) + assert r.method == "POST" + assert r.meta == {"key": "value"} def test_from_curl_ignore_unknown_options(self): # By default: it works and ignores the unknown options: --foo and -z @@ -409,7 +401,7 @@ class RequestTest(unittest.TestCase): r = self.request_class.from_curl( 'curl -X DELETE "http://example.org" --foo -z', ) - self.assertEqual(r.method, "DELETE") + assert r.method == "DELETE" # If `ignore_unknown_options` is set to `False` it raises an error with # the unknown options: --foo and -z @@ -420,17 +412,17 @@ class RequestTest(unittest.TestCase): ) -class FormRequestTest(RequestTest): +class TestFormRequest(TestRequest): request_class = FormRequest def assertQueryEqual(self, first, second, msg=None): first = to_unicode(first).split("&") second = to_unicode(second).split("&") - return self.assertEqual(sorted(first), sorted(second), msg) + assert sorted(first) == sorted(second), msg def test_empty_formdata(self): r1 = self.request_class("http://www.example.com", formdata={}) - self.assertEqual(r1.body, b"") + assert r1.body == b"" def test_formdata_overrides_querystring(self): data = (("a", "one"), ("a", "two"), ("b", "2")) @@ -438,69 +430,61 @@ class FormRequestTest(RequestTest): "http://www.example.com/?a=0&b=1&c=3#fragment", method="GET", formdata=data ).url.split("#", maxsplit=1)[0] fs = _qs(self.request_class(url, method="GET", formdata=data)) - self.assertEqual(set(fs[b"a"]), {b"one", b"two"}) - self.assertEqual(fs[b"b"], [b"2"]) - self.assertIsNone(fs.get(b"c")) + assert set(fs[b"a"]) == {b"one", b"two"} + assert fs[b"b"] == [b"2"] + assert fs.get(b"c") is None data = {"a": "1", "b": "2"} fs = _qs( self.request_class("http://www.example.com/", method="GET", formdata=data) ) - self.assertEqual(fs[b"a"], [b"1"]) - self.assertEqual(fs[b"b"], [b"2"]) + assert fs[b"a"] == [b"1"] + assert fs[b"b"] == [b"2"] def test_default_encoding_bytes(self): # using default encoding (utf-8) data = {b"one": b"two", b"price": b"\xc2\xa3 100"} r2 = self.request_class("http://www.example.com", formdata=data) - self.assertEqual(r2.method, "POST") - self.assertEqual(r2.encoding, "utf-8") + assert r2.method == "POST" + assert r2.encoding == "utf-8" self.assertQueryEqual(r2.body, b"price=%C2%A3+100&one=two") - self.assertEqual( - r2.headers[b"Content-Type"], b"application/x-www-form-urlencoded" - ) + assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_default_encoding_textual_data(self): # using default encoding (utf-8) data = {"µ one": "two", "price": "£ 100"} r2 = self.request_class("http://www.example.com", formdata=data) - self.assertEqual(r2.method, "POST") - self.assertEqual(r2.encoding, "utf-8") + assert r2.method == "POST" + assert r2.encoding == "utf-8" self.assertQueryEqual(r2.body, b"price=%C2%A3+100&%C2%B5+one=two") - self.assertEqual( - r2.headers[b"Content-Type"], b"application/x-www-form-urlencoded" - ) + assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_default_encoding_mixed_data(self): # using default encoding (utf-8) data = {"\u00b5one": b"two", b"price\xc2\xa3": "\u00a3 100"} r2 = self.request_class("http://www.example.com", formdata=data) - self.assertEqual(r2.method, "POST") - self.assertEqual(r2.encoding, "utf-8") + assert r2.method == "POST" + assert r2.encoding == "utf-8" self.assertQueryEqual(r2.body, b"%C2%B5one=two&price%C2%A3=%C2%A3+100") - self.assertEqual( - r2.headers[b"Content-Type"], b"application/x-www-form-urlencoded" - ) + assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_custom_encoding_bytes(self): data = {b"\xb5 one": b"two", b"price": b"\xa3 100"} r2 = self.request_class( "http://www.example.com", formdata=data, encoding="latin1" ) - self.assertEqual(r2.method, "POST") - self.assertEqual(r2.encoding, "latin1") + assert r2.method == "POST" + assert r2.encoding == "latin1" self.assertQueryEqual(r2.body, b"price=%A3+100&%B5+one=two") - self.assertEqual( - r2.headers[b"Content-Type"], b"application/x-www-form-urlencoded" - ) + assert r2.headers[b"Content-Type"] == b"application/x-www-form-urlencoded" def test_custom_encoding_textual_data(self): data = {"price": "£ 100"} r3 = self.request_class( "http://www.example.com", formdata=data, encoding="latin1" ) - self.assertEqual(r3.encoding, "latin1") - self.assertEqual(r3.body, b"price=%A3+100") + assert r3.encoding == "latin1" + assert r3.body == b"price=%A3+100" def test_multi_key_values(self): # using multiples values for a single key @@ -523,16 +507,14 @@ class FormRequestTest(RequestTest): response, formdata={"one": ["two", "three"], "six": "seven"} ) - self.assertEqual(req.method, "POST") - self.assertEqual( - req.headers[b"Content-type"], b"application/x-www-form-urlencoded" - ) - self.assertEqual(req.url, "http://www.example.com/this/post.php") + assert req.method == "POST" + assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" + assert req.url == "http://www.example.com/this/post.php" fs = _qs(req) - self.assertEqual(set(fs[b"test"]), {b"val1", b"val2"}) - self.assertEqual(set(fs[b"one"]), {b"two", b"three"}) - self.assertEqual(fs[b"test2"], [b"xxx"]) - self.assertEqual(fs[b"six"], [b"seven"]) + assert set(fs[b"test"]) == {b"val1", b"val2"} + assert set(fs[b"one"]) == {b"two", b"three"} + assert fs[b"test2"] == [b"xxx"] + assert fs[b"six"] == [b"seven"] def test_from_response_post_nonascii_bytes_utf8(self): response = _buildresponse( @@ -547,16 +529,14 @@ class FormRequestTest(RequestTest): response, formdata={"one": ["two", "three"], "six": "seven"} ) - self.assertEqual(req.method, "POST") - self.assertEqual( - req.headers[b"Content-type"], b"application/x-www-form-urlencoded" - ) - self.assertEqual(req.url, "http://www.example.com/this/post.php") + assert req.method == "POST" + assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" + assert req.url == "http://www.example.com/this/post.php" fs = _qs(req, to_unicode=True) - self.assertEqual(set(fs["test £"]), {"val1", "val2"}) - self.assertEqual(set(fs["one"]), {"two", "three"}) - self.assertEqual(fs["test2"], ["xxx µ"]) - self.assertEqual(fs["six"], ["seven"]) + assert set(fs["test £"]) == {"val1", "val2"} + assert set(fs["one"]) == {"two", "three"} + assert fs["test2"] == ["xxx µ"] + assert fs["six"] == ["seven"] def test_from_response_post_nonascii_bytes_latin1(self): response = _buildresponse( @@ -572,16 +552,14 @@ class FormRequestTest(RequestTest): response, formdata={"one": ["two", "three"], "six": "seven"} ) - self.assertEqual(req.method, "POST") - self.assertEqual( - req.headers[b"Content-type"], b"application/x-www-form-urlencoded" - ) - self.assertEqual(req.url, "http://www.example.com/this/post.php") + assert req.method == "POST" + assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" + assert req.url == "http://www.example.com/this/post.php" fs = _qs(req, to_unicode=True, encoding="latin1") - self.assertEqual(set(fs["test £"]), {"val1", "val2"}) - self.assertEqual(set(fs["one"]), {"two", "three"}) - self.assertEqual(fs["test2"], ["xxx µ"]) - self.assertEqual(fs["six"], ["seven"]) + assert set(fs["test £"]) == {"val1", "val2"} + assert set(fs["one"]) == {"two", "three"} + assert fs["test2"] == ["xxx µ"] + assert fs["six"] == ["seven"] def test_from_response_post_nonascii_unicode(self): response = _buildresponse( @@ -596,16 +574,14 @@ class FormRequestTest(RequestTest): response, formdata={"one": ["two", "three"], "six": "seven"} ) - self.assertEqual(req.method, "POST") - self.assertEqual( - req.headers[b"Content-type"], b"application/x-www-form-urlencoded" - ) - self.assertEqual(req.url, "http://www.example.com/this/post.php") + assert req.method == "POST" + assert req.headers[b"Content-type"] == b"application/x-www-form-urlencoded" + assert req.url == "http://www.example.com/this/post.php" fs = _qs(req, to_unicode=True) - self.assertEqual(set(fs["test £"]), {"val1", "val2"}) - self.assertEqual(set(fs["one"]), {"two", "three"}) - self.assertEqual(fs["test2"], ["xxx µ"]) - self.assertEqual(fs["six"], ["seven"]) + assert set(fs["test £"]) == {"val1", "val2"} + assert set(fs["one"]) == {"two", "three"} + assert fs["test2"] == ["xxx µ"] + assert fs["six"] == ["seven"] def test_from_response_duplicate_form_key(self): response = _buildresponse("
", url="http://www.example.com") @@ -614,8 +590,8 @@ class FormRequestTest(RequestTest): method="GET", formdata=(("foo", "bar"), ("foo", "baz")), ) - self.assertEqual(urlparse_cached(req).hostname, "www.example.com") - self.assertEqual(urlparse_cached(req).query, "foo=bar&foo=baz") + assert urlparse_cached(req).hostname == "www.example.com" + assert urlparse_cached(req).query == "foo=bar&foo=baz" def test_from_response_override_duplicate_form_key(self): response = _buildresponse( @@ -628,8 +604,8 @@ class FormRequestTest(RequestTest): response, formdata=(("two", "2"), ("two", "4")) ) fs = _qs(req) - self.assertEqual(fs[b"one"], [b"1"]) - self.assertEqual(fs[b"two"], [b"2", b"4"]) + assert fs[b"one"] == [b"1"] + assert fs[b"two"] == [b"2", b"4"] def test_from_response_extra_headers(self): response = _buildresponse( @@ -644,11 +620,9 @@ class FormRequestTest(RequestTest): formdata={"one": ["two", "three"], "six": "seven"}, headers={"Accept-Encoding": "gzip,deflate"}, ) - self.assertEqual(req.method, "POST") - self.assertEqual( - req.headers["Content-type"], b"application/x-www-form-urlencoded" - ) - self.assertEqual(req.headers["Accept-Encoding"], b"gzip,deflate") + assert req.method == "POST" + assert req.headers["Content-type"] == b"application/x-www-form-urlencoded" + assert req.headers["Accept-Encoding"] == b"gzip,deflate" def test_from_response_get(self): response = _buildresponse( @@ -662,14 +636,14 @@ class FormRequestTest(RequestTest): r1 = self.request_class.from_response( response, formdata={"one": ["two", "three"], "six": "seven"} ) - self.assertEqual(r1.method, "GET") - self.assertEqual(urlparse_cached(r1).hostname, "www.example.com") - self.assertEqual(urlparse_cached(r1).path, "/this/get.php") + assert r1.method == "GET" + assert urlparse_cached(r1).hostname == "www.example.com" + assert urlparse_cached(r1).path == "/this/get.php" fs = _qs(r1) - self.assertEqual(set(fs[b"test"]), {b"val1", b"val2"}) - self.assertEqual(set(fs[b"one"]), {b"two", b"three"}) - self.assertEqual(fs[b"test2"], [b"xxx"]) - self.assertEqual(fs[b"six"], [b"seven"]) + assert set(fs[b"test"]) == {b"val1", b"val2"} + assert set(fs[b"one"]) == {b"two", b"three"} + assert fs[b"test2"] == [b"xxx"] + assert fs[b"six"] == [b"seven"] def test_from_response_override_params(self): response = _buildresponse( @@ -680,8 +654,8 @@ class FormRequestTest(RequestTest): ) req = self.request_class.from_response(response, formdata={"two": "2"}) fs = _qs(req) - self.assertEqual(fs[b"one"], [b"1"]) - self.assertEqual(fs[b"two"], [b"2"]) + assert fs[b"one"] == [b"1"] + assert fs[b"two"] == [b"2"] def test_from_response_drop_params(self): response = _buildresponse( @@ -692,8 +666,8 @@ class FormRequestTest(RequestTest): ) req = self.request_class.from_response(response, formdata={"two": None}) fs = _qs(req) - self.assertEqual(fs[b"one"], [b"1"]) - self.assertNotIn(b"two", fs) + assert fs[b"one"] == [b"1"] + assert b"two" not in fs def test_from_response_override_method(self): response = _buildresponse( @@ -702,9 +676,9 @@ class FormRequestTest(RequestTest):