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

Converting tests to plain asserts, part 7. (#6710)

This commit is contained in:
Andrey Rakhmatullin 2025-03-09 23:23:51 +04:00 committed by GitHub
parent 02ed71d887
commit 380c2279b9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 512 additions and 569 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

243
tests/test_downloader_handlers.py
View File

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

41
tests/test_downloader_handlers_http2.py
View File

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

168
tests/test_exporters.py
View File

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

367
tests/test_feedexport.py
View File

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

123
tests/test_http2_client_protocol.py
View File

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

52
tests/test_http_cookies.py
View File

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

87
tests/test_http_headers.py
View File

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