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 3. (#6700)

This commit is contained in:
Andrey Rakhmatullin 2025-03-06 23:50:14 +04:00 committed by GitHub
parent 35212ec5b0
commit 1843a4f753
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 638 additions and 720 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
tests/test_downloadermiddleware.py
View File

@ -16,7 +16,7 @@ from scrapy.utils.python import to_bytes
from scrapy.utils.test import get_crawler, get_from_asyncio_queue
class ManagerTestCase(TestCase):
class TestManagerBase(TestCase):
settings_dict = None
def setUp(self):
@ -51,14 +51,14 @@ class ManagerTestCase(TestCase):
return ret
class DefaultsTest(ManagerTestCase):
class TestDefaults(TestManagerBase):
"""Tests default behavior with default settings"""
def test_request_response(self):
req = Request("http://example.com/index.html")
resp = Response(req.url, status=200)
ret = self._download(req, resp)
self.assertTrue(isinstance(ret, Response), "Non-response returned")
assert isinstance(ret, Response), "Non-response returned"
def test_3xx_and_invalid_gzipped_body_must_redirect(self):
"""Regression test for a failure when redirecting a compressed
@ -86,11 +86,9 @@ class DefaultsTest(ManagerTestCase):
},
)
ret = self._download(request=req, response=resp)
self.assertTrue(isinstance(ret, Request), f"Not redirected: {ret!r}")
self.assertEqual(
to_bytes(ret.url),
resp.headers["Location"],
"Not redirected to location header",
assert isinstance(ret, Request), f"Not redirected: {ret!r}"
assert to_bytes(ret.url) == resp.headers["Location"], (
"Not redirected to location header"
)
def test_200_and_invalid_gzipped_body_must_fail(self):
@ -111,7 +109,7 @@ class DefaultsTest(ManagerTestCase):
self._download(request=req, response=resp)
class ResponseFromProcessRequestTest(ManagerTestCase):
class TestResponseFromProcessRequest(TestManagerBase):
"""Tests middleware returning a response from process_request."""
def test_download_func_not_called(self):
@ -130,11 +128,11 @@ class ResponseFromProcessRequestTest(ManagerTestCase):
dfd.addBoth(results.append)
self._wait(dfd)
self.assertIs(results[0], resp)
self.assertFalse(download_func.called)
assert results[0] is resp
assert not download_func.called
class ProcessRequestInvalidOutput(ManagerTestCase):
class TestProcessRequestInvalidOutput(TestManagerBase):
"""Invalid return value for process_request method should raise an exception"""
def test_invalid_process_request(self):
@ -149,11 +147,11 @@ class ProcessRequestInvalidOutput(ManagerTestCase):
dfd = self.mwman.download(download_func, req, self.spider)
results = []
dfd.addBoth(results.append)
self.assertIsInstance(results[0], Failure)
self.assertIsInstance(results[0].value, _InvalidOutput)
assert isinstance(results[0], Failure)
assert isinstance(results[0].value, _InvalidOutput)
class ProcessResponseInvalidOutput(ManagerTestCase):
class TestProcessResponseInvalidOutput(TestManagerBase):
"""Invalid return value for process_response method should raise an exception"""
def test_invalid_process_response(self):
@ -168,11 +166,11 @@ class ProcessResponseInvalidOutput(ManagerTestCase):
dfd = self.mwman.download(download_func, req, self.spider)
results = []
dfd.addBoth(results.append)
self.assertIsInstance(results[0], Failure)
self.assertIsInstance(results[0].value, _InvalidOutput)
assert isinstance(results[0], Failure)
assert isinstance(results[0].value, _InvalidOutput)
class ProcessExceptionInvalidOutput(ManagerTestCase):
class TestProcessExceptionInvalidOutput(TestManagerBase):
"""Invalid return value for process_exception method should raise an exception"""
def test_invalid_process_exception(self):
@ -190,11 +188,11 @@ class ProcessExceptionInvalidOutput(ManagerTestCase):
dfd = self.mwman.download(download_func, req, self.spider)
results = []
dfd.addBoth(results.append)
self.assertIsInstance(results[0], Failure)
self.assertIsInstance(results[0].value, _InvalidOutput)
assert isinstance(results[0], Failure)
assert isinstance(results[0].value, _InvalidOutput)
class MiddlewareUsingDeferreds(ManagerTestCase):
class TestMiddlewareUsingDeferreds(TestManagerBase):
"""Middlewares using Deferreds should work"""
def test_deferred(self):
@ -218,12 +216,12 @@ class MiddlewareUsingDeferreds(ManagerTestCase):
dfd.addBoth(results.append)
self._wait(dfd)
self.assertIs(results[0], resp)
self.assertFalse(download_func.called)
assert results[0] is resp
assert not download_func.called
@pytest.mark.usefixtures("reactor_pytest")
class MiddlewareUsingCoro(ManagerTestCase):
class TestMiddlewareUsingCoro(TestManagerBase):
"""Middlewares using asyncio coroutines should work"""
def test_asyncdef(self):
@ -242,8 +240,8 @@ class MiddlewareUsingCoro(ManagerTestCase):
dfd.addBoth(results.append)
self._wait(dfd)
self.assertIs(results[0], resp)
self.assertFalse(download_func.called)
assert results[0] is resp
assert not download_func.called
@pytest.mark.only_asyncio
def test_asyncdef_asyncio(self):
@ -262,5 +260,5 @@ class MiddlewareUsingCoro(ManagerTestCase):
dfd.addBoth(results.append)
self._wait(dfd)
self.assertIs(results[0], resp)
self.assertFalse(download_func.called)
assert results[0] is resp
assert not download_func.called

18
tests/test_downloadermiddleware_ajaxcrawlable.py
View File

@ -1,5 +1,3 @@
import unittest
import pytest
from scrapy.downloadermiddlewares.ajaxcrawl import AjaxCrawlMiddleware
@ -9,8 +7,8 @@ from scrapy.utils.test import get_crawler
@pytest.mark.filterwarnings("ignore::scrapy.exceptions.ScrapyDeprecationWarning")
class AjaxCrawlMiddlewareTest(unittest.TestCase):
def setUp(self):
class TestAjaxCrawlMiddleware:
def setup_method(self):
crawler = get_crawler(Spider, {"AJAXCRAWL_ENABLED": True})
self.spider = crawler._create_spider("foo")
self.mw = AjaxCrawlMiddleware.from_crawler(crawler)
@ -26,13 +24,13 @@ class AjaxCrawlMiddlewareTest(unittest.TestCase):
def test_non_get(self):
req, resp = self._req_resp("http://example.com/", {"method": "HEAD"})
resp2 = self.mw.process_response(req, resp, self.spider)
self.assertEqual(resp, resp2)
assert resp == resp2
def test_binary_response(self):
req = Request("http://example.com/")
resp = Response("http://example.com/", body=b"foobar\x00\x01\x02", request=req)
resp2 = self.mw.process_response(req, resp, self.spider)
self.assertIs(resp, resp2)
assert resp is resp2
def test_ajaxcrawl(self):
req, resp = self._req_resp(
@ -41,8 +39,8 @@ class AjaxCrawlMiddlewareTest(unittest.TestCase):
{"body": self._ajaxcrawlable_body()},
)
req2 = self.mw.process_response(req, resp, self.spider)
self.assertEqual(req2.url, "http://example.com/?_escaped_fragment_=")
self.assertEqual(req2.meta["foo"], "bar")
assert req2.url == "http://example.com/?_escaped_fragment_="
assert req2.meta["foo"] == "bar"
def test_ajaxcrawl_loop(self):
req, resp = self._req_resp(
@ -53,7 +51,7 @@ class AjaxCrawlMiddlewareTest(unittest.TestCase):
resp3 = self.mw.process_response(req2, resp2, self.spider)
assert isinstance(resp3, HtmlResponse), (resp3.__class__, resp3)
self.assertEqual(resp3.request.url, "http://example.com/?_escaped_fragment_=")
assert resp3.request.url == "http://example.com/?_escaped_fragment_="
assert resp3 is resp2
def test_noncrawlable_body(self):
@ -61,4 +59,4 @@ class AjaxCrawlMiddlewareTest(unittest.TestCase):
"http://example.com/", {}, {"body": b"<html></html>"}
)
resp2 = self.mw.process_response(req, resp, self.spider)
self.assertIs(resp, resp2)
assert resp is resp2

78
tests/test_downloadermiddleware_cookies.py
View File

@ -1,5 +1,4 @@
import logging
from unittest import TestCase
import pytest
from testfixtures import LogCapture
@ -53,19 +52,19 @@ def _cookies_to_set_cookie_list(cookies):
return filter(None, (_cookie_to_set_cookie_value(cookie) for cookie in cookies))
class CookiesMiddlewareTest(TestCase):
class TestCookiesMiddleware:
def assertCookieValEqual(self, first, second, msg=None):
def split_cookies(cookies):
return sorted([s.strip() for s in to_bytes(cookies).split(b";")])
return self.assertEqual(split_cookies(first), split_cookies(second), msg=msg)
assert split_cookies(first) == split_cookies(second), msg
def setUp(self):
def setup_method(self):
self.spider = Spider("foo")
self.mw = CookiesMiddleware()
self.redirect_middleware = RedirectMiddleware(settings=Settings())
def tearDown(self):
def teardown_method(self):
del self.mw
del self.redirect_middleware
@ -80,7 +79,7 @@ class CookiesMiddlewareTest(TestCase):
req2 = Request("http://scrapytest.org/sub1/")
assert self.mw.process_request(req2, self.spider) is None
self.assertEqual(req2.headers.get("Cookie"), b"C1=value1")
assert req2.headers.get("Cookie") == b"C1=value1"
def test_setting_false_cookies_enabled(self):
with pytest.raises(NotConfigured):
@ -89,12 +88,12 @@ class CookiesMiddlewareTest(TestCase):
)
def test_setting_default_cookies_enabled(self):
self.assertIsInstance(
assert isinstance(
CookiesMiddleware.from_crawler(get_crawler()), CookiesMiddleware
)
def test_setting_true_cookies_enabled(self):
self.assertIsInstance(
assert isinstance(
CookiesMiddleware.from_crawler(
get_crawler(settings_dict={"COOKIES_ENABLED": True})
),
@ -161,7 +160,7 @@ class CookiesMiddlewareTest(TestCase):
req2 = Request("http://scrapytest.org/sub1/")
assert self.mw.process_request(req2, self.spider) is None
self.assertIn("Cookie", req2.headers)
assert "Cookie" in req2.headers
def test_dont_merge_cookies(self):
# merge some cookies into jar
@ -185,12 +184,12 @@ class CookiesMiddlewareTest(TestCase):
# check that cookies are merged back
req = Request("http://scrapytest.org/mergeme")
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers.get("Cookie"), b"C1=value1")
assert req.headers.get("Cookie") == b"C1=value1"
# check that cookies are merged when dont_merge_cookies is passed as 0
req = Request("http://scrapytest.org/mergeme", meta={"dont_merge_cookies": 0})
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers.get("Cookie"), b"C1=value1")
assert req.headers.get("Cookie") == b"C1=value1"
def test_complex_cookies(self):
# merge some cookies into jar
@ -230,7 +229,7 @@ class CookiesMiddlewareTest(TestCase):
# embed C2 for scrapytest.org/bar
req = Request("http://scrapytest.org/bar")
self.mw.process_request(req, self.spider)
self.assertEqual(req.headers.get("Cookie"), b"C2=value2")
assert req.headers.get("Cookie") == b"C2=value2"
# embed nothing for scrapytest.org/baz
req = Request("http://scrapytest.org/baz")
@ -240,7 +239,7 @@ class CookiesMiddlewareTest(TestCase):
def test_merge_request_cookies(self):
req = Request("http://scrapytest.org/", cookies={"galleta": "salada"})
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers.get("Cookie"), b"galleta=salada")
assert req.headers.get("Cookie") == b"galleta=salada"
headers = {"Set-Cookie": "C1=value1; path=/"}
res = Response("http://scrapytest.org/", headers=headers)
@ -260,7 +259,7 @@ class CookiesMiddlewareTest(TestCase):
meta={"cookiejar": "store1"},
)
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers.get("Cookie"), b"galleta=salada")
assert req.headers.get("Cookie") == b"galleta=salada"
headers = {"Set-Cookie": "C1=value1; path=/"}
res = Response("http://scrapytest.org/", headers=headers, request=req)
@ -278,7 +277,7 @@ class CookiesMiddlewareTest(TestCase):
meta={"cookiejar": "store2"},
)
assert self.mw.process_request(req3, self.spider) is None
self.assertEqual(req3.headers.get("Cookie"), b"galleta=dulce")
assert req3.headers.get("Cookie") == b"galleta=dulce"
headers = {"Set-Cookie": "C2=value2; path=/"}
res2 = Response("http://scrapytest.org/", headers=headers, request=req3)
@ -302,22 +301,22 @@ class CookiesMiddlewareTest(TestCase):
req5_2 = Request("http://scrapytest.org:1104/some-redirected-path")
assert self.mw.process_request(req5_2, self.spider) is None
self.assertEqual(req5_2.headers.get("Cookie"), b"C1=value1")
assert req5_2.headers.get("Cookie") == b"C1=value1"
req5_3 = Request("http://scrapytest.org/some-redirected-path")
assert self.mw.process_request(req5_3, self.spider) is None
self.assertEqual(req5_3.headers.get("Cookie"), b"C1=value1")
assert req5_3.headers.get("Cookie") == b"C1=value1"
# skip cookie retrieval for not http request
req6 = Request("file:///scrapy/sometempfile")
assert self.mw.process_request(req6, self.spider) is None
self.assertEqual(req6.headers.get("Cookie"), None)
assert req6.headers.get("Cookie") is None
def test_local_domain(self):
request = Request("http://example-host/", cookies={"currencyCookie": "USD"})
assert self.mw.process_request(request, self.spider) is None
self.assertIn("Cookie", request.headers)
self.assertEqual(b"currencyCookie=USD", request.headers["Cookie"])
assert "Cookie" in request.headers
assert request.headers["Cookie"] == b"currencyCookie=USD"
@pytest.mark.xfail(reason="Cookie header is not currently being processed")
def test_keep_cookie_from_default_request_headers_middleware(self):
@ -474,7 +473,7 @@ class CookiesMiddlewareTest(TestCase):
request1 = Request(cookies=input_cookies, **source)
self.mw.process_request(request1, self.spider)
cookies = request1.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies1 else None)
assert cookies == (b"a=b" if cookies1 else None)
response = Response(
headers={
@ -482,21 +481,18 @@ class CookiesMiddlewareTest(TestCase):
},
**target,
)
self.assertEqual(
self.mw.process_response(request1, response, self.spider),
response,
)
assert self.mw.process_response(request1, response, self.spider) == response
request2 = self.redirect_middleware.process_response(
request1,
response,
self.spider,
)
self.assertIsInstance(request2, Request)
assert isinstance(request2, Request)
self.mw.process_request(request2, self.spider)
cookies = request2.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies2 else None)
assert cookies == (b"a=b" if cookies2 else None)
def test_cookie_redirect_same_domain(self):
self._test_cookie_redirect(
@ -573,10 +569,10 @@ class CookiesMiddlewareTest(TestCase):
response,
self.spider,
)
self.assertIsInstance(request2, Request)
assert isinstance(request2, Request)
cookies = request2.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies2 else None)
assert cookies == (b"a=b" if cookies2 else None)
def test_cookie_header_redirect_same_domain(self):
self._test_cookie_header_redirect(
@ -626,12 +622,12 @@ class CookiesMiddlewareTest(TestCase):
request1 = Request(url1, cookies=input_cookies)
self.mw.process_request(request1, self.spider)
cookies = request1.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies1 else None)
assert cookies == (b"a=b" if cookies1 else None)
request2 = Request(url2)
self.mw.process_request(request2, self.spider)
cookies = request2.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies2 else None)
assert cookies == (b"a=b" if cookies2 else None)
def test_user_set_cookie_domain_suffix_private(self):
self._test_user_set_cookie_domain_followup(
@ -692,15 +688,12 @@ class CookiesMiddlewareTest(TestCase):
"Set-Cookie": _cookies_to_set_cookie_list(input_cookies),
}
response = Response(url1, status=200, headers=headers)
self.assertEqual(
self.mw.process_response(request1, response, self.spider),
response,
)
assert self.mw.process_response(request1, response, self.spider) == response
request2 = Request(url2)
self.mw.process_request(request2, self.spider)
actual_cookies = request2.headers.get("Cookie")
self.assertEqual(actual_cookies, b"a=b" if cookies else None)
assert actual_cookies == (b"a=b" if cookies else None)
def test_server_set_cookie_domain_suffix_private(self):
self._test_server_set_cookie_domain_followup(
@ -752,30 +745,27 @@ class CookiesMiddlewareTest(TestCase):
request1 = Request(f"{from_scheme}://a.example", cookies=input_cookies)
self.mw.process_request(request1, self.spider)
cookies = request1.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies1 else None)
assert cookies == (b"a=b" if cookies1 else None)
response = Response(
f"{from_scheme}://a.example",
headers={"Location": f"{to_scheme}://a.example"},
status=301,
)
self.assertEqual(
self.mw.process_response(request1, response, self.spider),
response,
)
assert self.mw.process_response(request1, response, self.spider) == response
request2 = self.redirect_middleware.process_response(
request1,
response,
self.spider,
)
self.assertIsInstance(request2, Request)
assert isinstance(request2, Request)
cookies = request2.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies2 else None)
assert cookies == (b"a=b" if cookies2 else None)
self.mw.process_request(request2, self.spider)
cookies = request2.headers.get("Cookie")
self.assertEqual(cookies, b"a=b" if cookies3 else None)
assert cookies == (b"a=b" if cookies3 else None)
def test_cookie_redirect_secure_undefined_downgrade(self):
self._test_cookie_redirect_scheme_change(

10
tests/test_downloadermiddleware_defaultheaders.py
View File

@ -1,5 +1,3 @@
from unittest import TestCase
from scrapy.downloadermiddlewares.defaultheaders import DefaultHeadersMiddleware
from scrapy.http import Request
from scrapy.spiders import Spider
@ -7,7 +5,7 @@ from scrapy.utils.python import to_bytes
from scrapy.utils.test import get_crawler
class TestDefaultHeadersMiddleware(TestCase):
class TestDefaultHeadersMiddleware:
def get_defaults_spider_mw(self):
crawler = get_crawler(Spider)
spider = crawler._create_spider("foo")
@ -21,15 +19,15 @@ class TestDefaultHeadersMiddleware(TestCase):
defaults, spider, mw = self.get_defaults_spider_mw()
req = Request("http://www.scrapytest.org")
mw.process_request(req, spider)
self.assertEqual(req.headers, defaults)
assert req.headers == defaults
def test_update_headers(self):
defaults, spider, mw = self.get_defaults_spider_mw()
headers = {"Accept-Language": ["es"], "Test-Header": ["test"]}
bytes_headers = {b"Accept-Language": [b"es"], b"Test-Header": [b"test"]}
req = Request("http://www.scrapytest.org", headers=headers)
self.assertEqual(req.headers, bytes_headers)
assert req.headers == bytes_headers
mw.process_request(req, spider)
defaults.update(bytes_headers)
self.assertEqual(req.headers, defaults)
assert req.headers == defaults

12
tests/test_downloadermiddleware_downloadtimeout.py
View File

@ -1,12 +1,10 @@
import unittest
from scrapy.downloadermiddlewares.downloadtimeout import DownloadTimeoutMiddleware
from scrapy.http import Request
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
class DownloadTimeoutMiddlewareTest(unittest.TestCase):
class TestDownloadTimeoutMiddleware:
def get_request_spider_mw(self, settings=None):
crawler = get_crawler(Spider, settings)
spider = crawler._create_spider("foo")
@ -17,20 +15,20 @@ class DownloadTimeoutMiddlewareTest(unittest.TestCase):
req, spider, mw = self.get_request_spider_mw()
mw.spider_opened(spider)
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta.get("download_timeout"), 180)
assert req.meta.get("download_timeout") == 180
def test_string_download_timeout(self):
req, spider, mw = self.get_request_spider_mw({"DOWNLOAD_TIMEOUT": "20.1"})
mw.spider_opened(spider)
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta.get("download_timeout"), 20.1)
assert req.meta.get("download_timeout") == 20.1
def test_spider_has_download_timeout(self):
req, spider, mw = self.get_request_spider_mw()
spider.download_timeout = 2
mw.spider_opened(spider)
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta.get("download_timeout"), 2)
assert req.meta.get("download_timeout") == 2
def test_request_has_download_timeout(self):
req, spider, mw = self.get_request_spider_mw()
@ -38,4 +36,4 @@ class DownloadTimeoutMiddlewareTest(unittest.TestCase):
mw.spider_opened(spider)
req.meta["download_timeout"] = 1
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta.get("download_timeout"), 1)
assert req.meta.get("download_timeout") == 1

30
tests/test_downloadermiddleware_httpauth.py
View File

@ -1,5 +1,3 @@
import unittest
import pytest
from w3lib.http import basic_auth_header
@ -25,8 +23,8 @@ class AnyDomainSpider(Spider):
http_auth_domain = None
class HttpAuthMiddlewareLegacyTest(unittest.TestCase):
def setUp(self):
class TestHttpAuthMiddlewareLegacy:
def setup_method(self):
self.spider = LegacySpider("foo")
def test_auth(self):
@ -35,51 +33,51 @@ class HttpAuthMiddlewareLegacyTest(unittest.TestCase):
mw.spider_opened(self.spider)
class HttpAuthMiddlewareTest(unittest.TestCase):
def setUp(self):
class TestHttpAuthMiddleware:
def setup_method(self):
self.mw = HttpAuthMiddleware()
self.spider = DomainSpider("foo")
self.mw.spider_opened(self.spider)
def tearDown(self):
def teardown_method(self):
del self.mw
def test_no_auth(self):
req = Request("http://example-noauth.com/")
assert self.mw.process_request(req, self.spider) is None
self.assertNotIn("Authorization", req.headers)
assert "Authorization" not in req.headers
def test_auth_domain(self):
req = Request("http://example.com/")
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers["Authorization"], basic_auth_header("foo", "bar"))
assert req.headers["Authorization"] == basic_auth_header("foo", "bar")
def test_auth_subdomain(self):
req = Request("http://foo.example.com/")
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers["Authorization"], basic_auth_header("foo", "bar"))
assert req.headers["Authorization"] == basic_auth_header("foo", "bar")
def test_auth_already_set(self):
req = Request("http://example.com/", headers={"Authorization": "Digest 123"})
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers["Authorization"], b"Digest 123")
assert req.headers["Authorization"] == b"Digest 123"
class HttpAuthAnyMiddlewareTest(unittest.TestCase):
def setUp(self):
class TestHttpAuthAnyMiddleware:
def setup_method(self):
self.mw = HttpAuthMiddleware()
self.spider = AnyDomainSpider("foo")
self.mw.spider_opened(self.spider)
def tearDown(self):
def teardown_method(self):
del self.mw
def test_auth(self):
req = Request("http://example.com/")
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers["Authorization"], basic_auth_header("foo", "bar"))
assert req.headers["Authorization"] == basic_auth_header("foo", "bar")
def test_auth_already_set(self):
req = Request("http://example.com/", headers={"Authorization": "Digest 123"})
assert self.mw.process_request(req, self.spider) is None
self.assertEqual(req.headers["Authorization"], b"Digest 123")
assert req.headers["Authorization"] == b"Digest 123"

51
tests/test_downloadermiddleware_httpcache.py
View File

@ -2,7 +2,6 @@ import email.utils
import shutil
import tempfile
import time
import unittest
from contextlib import contextmanager
import pytest
@ -15,11 +14,11 @@ from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
class _BaseTest(unittest.TestCase):
class TestBase:
storage_class = "scrapy.extensions.httpcache.FilesystemCacheStorage"
policy_class = "scrapy.extensions.httpcache.RFC2616Policy"
def setUp(self):
def setup_method(self):
self.yesterday = email.utils.formatdate(time.time() - 86400)
self.today = email.utils.formatdate()
self.tomorrow = email.utils.formatdate(time.time() + 86400)
@ -35,7 +34,7 @@ class _BaseTest(unittest.TestCase):
)
self.crawler.stats.open_spider(self.spider)
def tearDown(self):
def teardown_method(self):
self.crawler.stats.close_spider(self.spider, "")
shutil.rmtree(self.tmpdir)
@ -72,44 +71,42 @@ class _BaseTest(unittest.TestCase):
mw.spider_closed(self.spider)
def assertEqualResponse(self, response1, response2):
self.assertEqual(response1.url, response2.url)
self.assertEqual(response1.status, response2.status)
self.assertEqual(response1.headers, response2.headers)
self.assertEqual(response1.body, response2.body)
assert response1.url == response2.url
assert response1.status == response2.status
assert response1.headers == response2.headers
assert response1.body == response2.body
def assertEqualRequest(self, request1, request2):
self.assertEqual(request1.url, request2.url)
self.assertEqual(request1.headers, request2.headers)
self.assertEqual(request1.body, request2.body)
assert request1.url == request2.url
assert request1.headers == request2.headers
assert request1.body == request2.body
def assertEqualRequestButWithCacheValidators(self, request1, request2):
self.assertEqual(request1.url, request2.url)
assert request1.url == request2.url
assert b"If-None-Match" not in request1.headers
assert b"If-Modified-Since" not in request1.headers
assert any(
h in request2.headers for h in (b"If-None-Match", b"If-Modified-Since")
)
self.assertEqual(request1.body, request2.body)
assert request1.body == request2.body
def test_dont_cache(self):
with self._middleware() as mw:
self.request.meta["dont_cache"] = True
mw.process_response(self.request, self.response, self.spider)
self.assertEqual(
mw.storage.retrieve_response(self.spider, self.request), None
)
assert mw.storage.retrieve_response(self.spider, self.request) is None
with self._middleware() as mw:
self.request.meta["dont_cache"] = False
mw.process_response(self.request, self.response, self.spider)
if mw.policy.should_cache_response(self.response, self.request):
self.assertIsInstance(
assert isinstance(
mw.storage.retrieve_response(self.spider, self.request),
self.response.__class__,
)
class DefaultStorageTest(_BaseTest):
class TestDefaultStorage(TestBase):
def test_storage(self):
with self._storage() as storage:
request2 = self.request.copy()
@ -142,15 +139,15 @@ class DefaultStorageTest(_BaseTest):
)
storage.store_response(self.spider, self.request, response)
cached_response = storage.retrieve_response(self.spider, self.request)
self.assertIsInstance(cached_response, HtmlResponse)
assert isinstance(cached_response, HtmlResponse)
self.assertEqualResponse(response, cached_response)
class DbmStorageTest(DefaultStorageTest):
class TestDbmStorage(TestDefaultStorage):
storage_class = "scrapy.extensions.httpcache.DbmCacheStorage"
class DbmStorageWithCustomDbmModuleTest(DbmStorageTest):
class TestDbmStorageWithCustomDbmModule(TestDbmStorage):
dbm_module = "tests.mocks.dummydbm"
def _get_settings(self, **new_settings):
@ -160,16 +157,16 @@ class DbmStorageWithCustomDbmModuleTest(DbmStorageTest):
def test_custom_dbm_module_loaded(self):
# make sure our dbm module has been loaded
with self._storage() as storage:
self.assertEqual(storage.dbmodule.__name__, self.dbm_module)
assert storage.dbmodule.__name__ == self.dbm_module
class FilesystemStorageGzipTest(DefaultStorageTest):
class TestFilesystemStorageGzip(TestDefaultStorage):
def _get_settings(self, **new_settings):
new_settings.setdefault("HTTPCACHE_GZIP", True)
return super()._get_settings(**new_settings)
class DummyPolicyTest(_BaseTest):
class TestDummyPolicy(TestBase):
policy_class = "scrapy.extensions.httpcache.DummyPolicy"
def test_middleware(self):
@ -261,7 +258,7 @@ class DummyPolicyTest(_BaseTest):
assert "cached" in response.flags
class RFC2616PolicyTest(DefaultStorageTest):
class TestRFC2616Policy(TestDefaultStorage):
policy_class = "scrapy.extensions.httpcache.RFC2616Policy"
def _process_requestresponse(self, mw, request, response):
@ -357,7 +354,7 @@ class RFC2616PolicyTest(DefaultStorageTest):
assert "cached" in res2.flags
assert res2.status != 304
else:
self.assertFalse(resc)
assert not resc
assert "cached" not in res2.flags
# cache unconditionally unless response contains no-store or is a 304
@ -381,7 +378,7 @@ class RFC2616PolicyTest(DefaultStorageTest):
assert "cached" in res2.flags
assert res2.status != 304
else:
self.assertFalse(resc)
assert not resc
assert "cached" not in res2.flags
def test_cached_and_fresh(self):

88
tests/test_downloadermiddleware_httpcompression.py
View File

@ -2,7 +2,7 @@ from gzip import GzipFile
from io import BytesIO
from logging import WARNING
from pathlib import Path
from unittest import SkipTest, TestCase
from unittest import SkipTest
import pytest
from testfixtures import LogCapture
@ -51,8 +51,8 @@ FORMAT = {
}
class HttpCompressionTest(TestCase):
def setUp(self):
class TestHttpCompression:
def setup_method(self):
self.crawler = get_crawler(Spider)
self.spider = self.crawler._create_spider("scrapytest.org")
self.mw = HttpCompressionMiddleware.from_crawler(self.crawler)
@ -81,10 +81,8 @@ class HttpCompressionTest(TestCase):
return response
def assertStatsEqual(self, key, value):
self.assertEqual(
self.crawler.stats.get_value(key, spider=self.spider),
value,
str(self.crawler.stats.get_stats(self.spider)),
assert self.crawler.stats.get_value(key, spider=self.spider) == value, str(
self.crawler.stats.get_stats(self.spider)
)
def test_setting_false_compression_enabled(self):
@ -94,13 +92,13 @@ class HttpCompressionTest(TestCase):
)
def test_setting_default_compression_enabled(self):
self.assertIsInstance(
assert isinstance(
HttpCompressionMiddleware.from_crawler(get_crawler()),
HttpCompressionMiddleware,
)
def test_setting_true_compression_enabled(self):
self.assertIsInstance(
assert isinstance(
HttpCompressionMiddleware.from_crawler(
get_crawler(settings_dict={"COMPRESSION_ENABLED": True})
),
@ -111,15 +109,13 @@ class HttpCompressionTest(TestCase):
request = Request("http://scrapytest.org")
assert "Accept-Encoding" not in request.headers
self.mw.process_request(request, self.spider)
self.assertEqual(
request.headers.get("Accept-Encoding"), b", ".join(ACCEPTED_ENCODINGS)
)
assert request.headers.get("Accept-Encoding") == b", ".join(ACCEPTED_ENCODINGS)
def test_process_response_gzip(self):
response = self._getresponse("gzip")
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"gzip")
assert response.headers["Content-Encoding"] == b"gzip"
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
@ -137,7 +133,7 @@ class HttpCompressionTest(TestCase):
raise SkipTest("no brotli")
response = self._getresponse("br")
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"br")
assert response.headers["Content-Encoding"] == b"br"
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
@ -159,7 +155,7 @@ class HttpCompressionTest(TestCase):
pass
response = self._getresponse("br")
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"br")
assert response.headers["Content-Encoding"] == b"br"
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
@ -178,7 +174,7 @@ class HttpCompressionTest(TestCase):
),
)
assert newresponse is not response
self.assertEqual(newresponse.headers.getlist("Content-Encoding"), [b"br"])
assert newresponse.headers.getlist("Content-Encoding") == [b"br"]
def test_process_response_zstd(self):
try:
@ -191,7 +187,7 @@ class HttpCompressionTest(TestCase):
continue
response = self._getresponse(check_key)
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"zstd")
assert response.headers["Content-Encoding"] == b"zstd"
newresponse = self.mw.process_response(request, response, self.spider)
if raw_content is None:
raw_content = newresponse.body
@ -210,7 +206,7 @@ class HttpCompressionTest(TestCase):
pass
response = self._getresponse("zstd-static-content-size")
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"zstd")
assert response.headers["Content-Encoding"] == b"zstd"
with LogCapture(
"scrapy.downloadermiddlewares.httpcompression",
propagate=False,
@ -229,13 +225,13 @@ class HttpCompressionTest(TestCase):
),
)
assert newresponse is not response
self.assertEqual(newresponse.headers.getlist("Content-Encoding"), [b"zstd"])
assert newresponse.headers.getlist("Content-Encoding") == [b"zstd"]
def test_process_response_rawdeflate(self):
response = self._getresponse("rawdeflate")
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"deflate")
assert response.headers["Content-Encoding"] == b"deflate"
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
@ -247,7 +243,7 @@ class HttpCompressionTest(TestCase):
response = self._getresponse("zlibdeflate")
request = response.request
self.assertEqual(response.headers["Content-Encoding"], b"deflate")
assert response.headers["Content-Encoding"] == b"deflate"
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
@ -272,7 +268,7 @@ class HttpCompressionTest(TestCase):
request = response.request
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
self.assertEqual(newresponse.headers.getlist("Content-Encoding"), [b"uuencode"])
assert newresponse.headers.getlist("Content-Encoding") == [b"uuencode"]
def test_multi_compression_single_header(self):
response = self._getresponse("gzip-deflate")
@ -303,9 +299,7 @@ class HttpCompressionTest(TestCase):
),
)
assert newresponse is not response
self.assertEqual(
newresponse.headers.getlist("Content-Encoding"), [b"gzip", b"foo"]
)
assert newresponse.headers.getlist("Content-Encoding") == [b"gzip", b"foo"]
def test_multi_compression_multiple_header(self):
response = self._getresponse("gzip-deflate")
@ -322,9 +316,7 @@ class HttpCompressionTest(TestCase):
request = response.request
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
self.assertEqual(
newresponse.headers.getlist("Content-Encoding"), [b"gzip", b"foo"]
)
assert newresponse.headers.getlist("Content-Encoding") == [b"gzip", b"foo"]
def test_multi_compression_single_and_multiple_header(self):
response = self._getresponse("gzip-deflate-gzip")
@ -341,9 +333,7 @@ class HttpCompressionTest(TestCase):
request = response.request
newresponse = self.mw.process_response(request, response, self.spider)
assert newresponse is not response
self.assertEqual(
newresponse.headers.getlist("Content-Encoding"), [b"gzip", b"foo"]
)
assert newresponse.headers.getlist("Content-Encoding") == [b"gzip", b"foo"]
def test_process_response_encoding_inside_body(self):
headers = {
@ -365,8 +355,8 @@ class HttpCompressionTest(TestCase):
newresponse = self.mw.process_response(request, response, self.spider)
assert isinstance(newresponse, HtmlResponse)
self.assertEqual(newresponse.body, plainbody)
self.assertEqual(newresponse.encoding, resolve_encoding("gb2312"))
assert newresponse.body == plainbody
assert newresponse.encoding == resolve_encoding("gb2312")
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", len(plainbody))
@ -390,8 +380,8 @@ class HttpCompressionTest(TestCase):
newresponse = self.mw.process_response(request, response, self.spider)
assert isinstance(newresponse, HtmlResponse)
self.assertEqual(newresponse.body, plainbody)
self.assertEqual(newresponse.encoding, resolve_encoding("gb2312"))
assert newresponse.body == plainbody
assert newresponse.encoding == resolve_encoding("gb2312")
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", len(plainbody))
@ -413,8 +403,8 @@ class HttpCompressionTest(TestCase):
newresponse = self.mw.process_response(request, response, self.spider)
assert isinstance(newresponse, respcls)
self.assertEqual(newresponse.body, plainbody)
self.assertEqual(newresponse.encoding, resolve_encoding("gb2312"))
assert newresponse.body == plainbody
assert newresponse.encoding == resolve_encoding("gb2312")
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", len(plainbody))
@ -424,9 +414,9 @@ class HttpCompressionTest(TestCase):
request = response.request
newresponse = self.mw.process_response(request, response, self.spider)
self.assertIsNot(newresponse, response)
self.assertTrue(newresponse.body.startswith(b"<!DOCTYPE"))
self.assertNotIn("Content-Encoding", newresponse.headers)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
assert "Content-Encoding" not in newresponse.headers
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
@ -436,9 +426,9 @@ class HttpCompressionTest(TestCase):
request = response.request
newresponse = self.mw.process_response(request, response, self.spider)
self.assertIsNot(newresponse, response)
self.assertTrue(newresponse.body.startswith(b"<!DOCTYPE"))
self.assertNotIn("Content-Encoding", newresponse.headers)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
assert "Content-Encoding" not in newresponse.headers
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
@ -448,9 +438,9 @@ class HttpCompressionTest(TestCase):
request = response.request
newresponse = self.mw.process_response(request, response, self.spider)
self.assertIsNot(newresponse, response)
self.assertTrue(newresponse.body.startswith(b"<!DOCTYPE"))
self.assertNotIn("Content-Encoding", newresponse.headers)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
assert "Content-Encoding" not in newresponse.headers
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
@ -496,7 +486,7 @@ class HttpCompressionTest(TestCase):
request = Request("http://www.example.com/")
newresponse = self.mw.process_response(request, response, self.spider)
self.assertEqual(gunzip(newresponse.body), plainbody)
assert gunzip(newresponse.body) == plainbody
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 230)
@ -507,8 +497,8 @@ class HttpCompressionTest(TestCase):
request.method = "HEAD"
response = response.replace(body=None)
newresponse = self.mw.process_response(request, response, self.spider)
self.assertIs(newresponse, response)
self.assertEqual(response.body, b"")
assert newresponse is response
assert response.body == b""
self.assertStatsEqual("httpcompression/response_count", None)
self.assertStatsEqual("httpcompression/response_bytes", None)

166
tests/test_downloadermiddleware_httpproxy.py
View File

@ -1,7 +1,6 @@
import os
import pytest
from twisted.trial.unittest import TestCase
from scrapy.downloadermiddlewares.httpproxy import HttpProxyMiddleware
from scrapy.exceptions import NotConfigured
@ -12,13 +11,13 @@ from scrapy.utils.test import get_crawler
spider = Spider("foo")
class TestHttpProxyMiddleware(TestCase):
class TestHttpProxyMiddleware:
failureException = AssertionError # type: ignore[assignment]
def setUp(self):
def setup_method(self):
self._oldenv = os.environ.copy()
def tearDown(self):
def teardown_method(self):
os.environ = self._oldenv
def test_not_enabled(self):
@ -33,8 +32,8 @@ class TestHttpProxyMiddleware(TestCase):
for url in ("http://e.com", "https://e.com", "file:///tmp/a"):
req = Request(url)
assert mw.process_request(req, spider) is None
self.assertEqual(req.url, url)
self.assertEqual(req.meta, {})
assert req.url == url
assert req.meta == {}
def test_environment_proxies(self):
os.environ["http_proxy"] = http_proxy = "https://proxy.for.http:3128"
@ -49,32 +48,32 @@ class TestHttpProxyMiddleware(TestCase):
]:
req = Request(url)
assert mw.process_request(req, spider) is None
self.assertEqual(req.url, url)
self.assertEqual(req.meta.get("proxy"), proxy)
assert req.url == url
assert req.meta.get("proxy") == proxy
def test_proxy_precedence_meta(self):
os.environ["http_proxy"] = "https://proxy.com"
mw = HttpProxyMiddleware()
req = Request("http://scrapytest.org", meta={"proxy": "https://new.proxy:3128"})
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta, {"proxy": "https://new.proxy:3128"})
assert req.meta == {"proxy": "https://new.proxy:3128"}
def test_proxy_auth(self):
os.environ["http_proxy"] = "https://user:pass@proxy:3128"
mw = HttpProxyMiddleware()
req = Request("http://scrapytest.org")
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(req.headers.get("Proxy-Authorization"), b"Basic dXNlcjpwYXNz")
assert req.meta["proxy"] == "https://proxy:3128"
assert req.headers.get("Proxy-Authorization") == b"Basic dXNlcjpwYXNz"
# proxy from request.meta
req = Request(
"http://scrapytest.org",
meta={"proxy": "https://username:password@proxy:3128"},
)
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(
req.headers.get("Proxy-Authorization"), b"Basic dXNlcm5hbWU6cGFzc3dvcmQ="
assert req.meta["proxy"] == "https://proxy:3128"
assert (
req.headers.get("Proxy-Authorization") == b"Basic dXNlcm5hbWU6cGFzc3dvcmQ="
)
def test_proxy_auth_empty_passwd(self):
@ -82,15 +81,15 @@ class TestHttpProxyMiddleware(TestCase):
mw = HttpProxyMiddleware()
req = Request("http://scrapytest.org")
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(req.headers.get("Proxy-Authorization"), b"Basic dXNlcjo=")
assert req.meta["proxy"] == "https://proxy:3128"
assert req.headers.get("Proxy-Authorization") == b"Basic dXNlcjo="
# proxy from request.meta
req = Request(
"http://scrapytest.org", meta={"proxy": "https://username:@proxy:3128"}
)
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(req.headers.get("Proxy-Authorization"), b"Basic dXNlcm5hbWU6")
assert req.meta["proxy"] == "https://proxy:3128"
assert req.headers.get("Proxy-Authorization") == b"Basic dXNlcm5hbWU6"
def test_proxy_auth_encoding(self):
# utf-8 encoding
@ -98,33 +97,31 @@ class TestHttpProxyMiddleware(TestCase):
mw = HttpProxyMiddleware(auth_encoding="utf-8")
req = Request("http://scrapytest.org")
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(req.headers.get("Proxy-Authorization"), b"Basic bcOhbjpwYXNz")
assert req.meta["proxy"] == "https://proxy:3128"
assert req.headers.get("Proxy-Authorization") == b"Basic bcOhbjpwYXNz"
# proxy from request.meta
req = Request(
"http://scrapytest.org", meta={"proxy": "https://\u00fcser:pass@proxy:3128"}
)
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(
req.headers.get("Proxy-Authorization"), b"Basic w7xzZXI6cGFzcw=="
)
assert req.meta["proxy"] == "https://proxy:3128"
assert req.headers.get("Proxy-Authorization") == b"Basic w7xzZXI6cGFzcw=="
# default latin-1 encoding
mw = HttpProxyMiddleware(auth_encoding="latin-1")
req = Request("http://scrapytest.org")
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(req.headers.get("Proxy-Authorization"), b"Basic beFuOnBhc3M=")
assert req.meta["proxy"] == "https://proxy:3128"
assert req.headers.get("Proxy-Authorization") == b"Basic beFuOnBhc3M="
# proxy from request.meta, latin-1 encoding
req = Request(
"http://scrapytest.org", meta={"proxy": "https://\u00fcser:pass@proxy:3128"}
)
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta["proxy"], "https://proxy:3128")
self.assertEqual(req.headers.get("Proxy-Authorization"), b"Basic /HNlcjpwYXNz")
assert req.meta["proxy"] == "https://proxy:3128"
assert req.headers.get("Proxy-Authorization") == b"Basic /HNlcjpwYXNz"
def test_proxy_already_seted(self):
os.environ["http_proxy"] = "https://proxy.for.http:3128"
@ -157,7 +154,7 @@ class TestHttpProxyMiddleware(TestCase):
os.environ["no_proxy"] = "*"
req = Request("http://noproxy.com", meta={"proxy": "http://proxy.com"})
assert mw.process_request(req, spider) is None
self.assertEqual(req.meta, {"proxy": "http://proxy.com"})
assert req.meta == {"proxy": "http://proxy.com"}
def test_no_proxy_invalid_values(self):
os.environ["no_proxy"] = "/var/run/docker.sock"
@ -172,8 +169,8 @@ class TestHttpProxyMiddleware(TestCase):
assert middleware.process_request(request, spider) is None
request.meta["proxy"] = "https://example.com"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta["proxy"] == "https://example.com"
assert b"Proxy-Authorization" not in request.headers
def test_add_proxy_with_credentials(self):
middleware = HttpProxyMiddleware()
@ -181,15 +178,12 @@ class TestHttpProxyMiddleware(TestCase):
assert middleware.process_request(request, spider) is None
request.meta["proxy"] = "https://user1:password1@example.com"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
assert request.meta["proxy"] == "https://example.com"
encoded_credentials = middleware._basic_auth_header(
"user1",
"password1",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
def test_remove_proxy_without_credentials(self):
middleware = HttpProxyMiddleware()
@ -200,8 +194,8 @@ class TestHttpProxyMiddleware(TestCase):
assert middleware.process_request(request, spider) is None
request.meta["proxy"] = None
assert middleware.process_request(request, spider) is None
self.assertIsNone(request.meta["proxy"])
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta["proxy"] is None
assert b"Proxy-Authorization" not in request.headers
def test_remove_proxy_with_credentials(self):
middleware = HttpProxyMiddleware()
@ -212,8 +206,8 @@ class TestHttpProxyMiddleware(TestCase):
assert middleware.process_request(request, spider) is None
request.meta["proxy"] = None
assert middleware.process_request(request, spider) is None
self.assertIsNone(request.meta["proxy"])
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta["proxy"] is None
assert b"Proxy-Authorization" not in request.headers
def test_add_credentials(self):
"""If the proxy request meta switches to a proxy URL with the same
@ -228,15 +222,12 @@ class TestHttpProxyMiddleware(TestCase):
request.meta["proxy"] = "https://user1:password1@example.com"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
assert request.meta["proxy"] == "https://example.com"
encoded_credentials = middleware._basic_auth_header(
"user1",
"password1",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
def test_change_credentials(self):
"""If the proxy request meta switches to a proxy URL with different
@ -249,15 +240,12 @@ class TestHttpProxyMiddleware(TestCase):
assert middleware.process_request(request, spider) is None
request.meta["proxy"] = "https://user2:password2@example.com"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
assert request.meta["proxy"] == "https://example.com"
encoded_credentials = middleware._basic_auth_header(
"user2",
"password2",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
def test_remove_credentials(self):
"""If the proxy request meta switches to a proxy URL with the same
@ -276,21 +264,18 @@ class TestHttpProxyMiddleware(TestCase):
request.meta["proxy"] = "https://example.com"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
assert request.meta["proxy"] == "https://example.com"
encoded_credentials = middleware._basic_auth_header(
"user1",
"password1",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
request.meta["proxy"] = "https://example.com"
del request.headers[b"Proxy-Authorization"]
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta["proxy"] == "https://example.com"
assert b"Proxy-Authorization" not in request.headers
def test_change_proxy_add_credentials(self):
middleware = HttpProxyMiddleware()
@ -302,15 +287,12 @@ class TestHttpProxyMiddleware(TestCase):
request.meta["proxy"] = "https://user1:password1@example.org"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.org")
assert request.meta["proxy"] == "https://example.org"
encoded_credentials = middleware._basic_auth_header(
"user1",
"password1",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
def test_change_proxy_keep_credentials(self):
middleware = HttpProxyMiddleware()
@ -322,21 +304,18 @@ class TestHttpProxyMiddleware(TestCase):
request.meta["proxy"] = "https://user1:password1@example.org"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.org")
assert request.meta["proxy"] == "https://example.org"
encoded_credentials = middleware._basic_auth_header(
"user1",
"password1",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
# Make sure, indirectly, that _auth_proxy is updated.
request.meta["proxy"] = "https://example.com"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta["proxy"] == "https://example.com"
assert b"Proxy-Authorization" not in request.headers
def test_change_proxy_change_credentials(self):
middleware = HttpProxyMiddleware()
@ -348,15 +327,12 @@ class TestHttpProxyMiddleware(TestCase):
request.meta["proxy"] = "https://user2:password2@example.org"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.org")
assert request.meta["proxy"] == "https://example.org"
encoded_credentials = middleware._basic_auth_header(
"user2",
"password2",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
def test_change_proxy_remove_credentials(self):
"""If the proxy request meta switches to a proxy URL with a different
@ -369,8 +345,8 @@ class TestHttpProxyMiddleware(TestCase):
assert middleware.process_request(request, spider) is None
request.meta["proxy"] = "https://example.org"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta, {"proxy": "https://example.org"})
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta == {"proxy": "https://example.org"}
assert b"Proxy-Authorization" not in request.headers
def test_change_proxy_remove_credentials_preremoved_header(self):
"""Corner case of proxy switch with credentials removal where the
@ -388,8 +364,8 @@ class TestHttpProxyMiddleware(TestCase):
request.meta["proxy"] = "https://example.org"
del request.headers[b"Proxy-Authorization"]
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta, {"proxy": "https://example.org"})
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta == {"proxy": "https://example.org"}
assert b"Proxy-Authorization" not in request.headers
def test_proxy_authentication_header_undefined_proxy(self):
middleware = HttpProxyMiddleware()
@ -398,8 +374,8 @@ class TestHttpProxyMiddleware(TestCase):
headers={"Proxy-Authorization": "Basic foo"},
)
assert middleware.process_request(request, spider) is None
self.assertNotIn("proxy", request.meta)
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert "proxy" not in request.meta
assert b"Proxy-Authorization" not in request.headers
def test_proxy_authentication_header_disabled_proxy(self):
middleware = HttpProxyMiddleware()
@ -409,8 +385,8 @@ class TestHttpProxyMiddleware(TestCase):
meta={"proxy": None},
)
assert middleware.process_request(request, spider) is None
self.assertIsNone(request.meta["proxy"])
self.assertNotIn(b"Proxy-Authorization", request.headers)
assert request.meta["proxy"] is None
assert b"Proxy-Authorization" not in request.headers
def test_proxy_authentication_header_proxy_without_credentials(self):
"""As long as the proxy URL in request metadata remains the same, the
@ -423,17 +399,17 @@ class TestHttpProxyMiddleware(TestCase):
meta={"proxy": "https://example.com"},
)
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
self.assertEqual(request.headers["Proxy-Authorization"], b"Basic foo")
assert request.meta["proxy"] == "https://example.com"
assert request.headers["Proxy-Authorization"] == b"Basic foo"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
self.assertEqual(request.headers["Proxy-Authorization"], b"Basic foo")
assert request.meta["proxy"] == "https://example.com"
assert request.headers["Proxy-Authorization"] == b"Basic foo"
request.headers["Proxy-Authorization"] = b"Basic bar"
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
self.assertEqual(request.headers["Proxy-Authorization"], b"Basic bar")
assert request.meta["proxy"] == "https://example.com"
assert request.headers["Proxy-Authorization"] == b"Basic bar"
def test_proxy_authentication_header_proxy_with_same_credentials(self):
middleware = HttpProxyMiddleware()
@ -447,11 +423,8 @@ class TestHttpProxyMiddleware(TestCase):
meta={"proxy": "https://user1:password1@example.com"},
)
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials,
)
assert request.meta["proxy"] == "https://example.com"
assert request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials
def test_proxy_authentication_header_proxy_with_different_credentials(self):
middleware = HttpProxyMiddleware()
@ -465,12 +438,11 @@ class TestHttpProxyMiddleware(TestCase):
meta={"proxy": "https://user2:password2@example.com"},
)
assert middleware.process_request(request, spider) is None
self.assertEqual(request.meta["proxy"], "https://example.com")
assert request.meta["proxy"] == "https://example.com"
encoded_credentials2 = middleware._basic_auth_header(
"user2",
"password2",
)
self.assertEqual(
request.headers["Proxy-Authorization"],
b"Basic " + encoded_credentials2,
assert (
request.headers["Proxy-Authorization"] == b"Basic " + encoded_credentials2
)

732
tests/test_downloadermiddleware_redirect.py
View File

File diff suppressed because it is too large Load Diff

87
tests/test_downloadermiddleware_retry.py
View File

@ -1,5 +1,4 @@
import logging
import unittest
import pytest
from testfixtures import LogCapture
@ -21,8 +20,8 @@ from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
class RetryTest(unittest.TestCase):
def setUp(self):
class TestRetry:
def setup_method(self):
self.crawler = get_crawler(Spider)
self.spider = self.crawler._create_spider("foo")
self.mw = RetryMiddleware.from_crawler(self.crawler)
@ -70,12 +69,12 @@ class RetryTest(unittest.TestCase):
# first retry
req = self.mw.process_response(req, rsp, self.spider)
assert isinstance(req, Request)
self.assertEqual(req.meta["retry_times"], 1)
assert req.meta["retry_times"] == 1
# second retry
req = self.mw.process_response(req, rsp, self.spider)
assert isinstance(req, Request)
self.assertEqual(req.meta["retry_times"], 2)
assert req.meta["retry_times"] == 2
# discard it
assert self.mw.process_response(req, rsp, self.spider) is rsp
@ -129,19 +128,19 @@ class RetryTest(unittest.TestCase):
# first retry
req = mw.process_exception(req, exception, self.spider)
assert isinstance(req, Request)
self.assertEqual(req.meta["retry_times"], 1)
assert req.meta["retry_times"] == 1
# second retry
req = mw.process_exception(req, exception, self.spider)
assert isinstance(req, Request)
self.assertEqual(req.meta["retry_times"], 2)
assert req.meta["retry_times"] == 2
# discard it
req = mw.process_exception(req, exception, self.spider)
self.assertEqual(req, None)
assert req is None
class MaxRetryTimesTest(unittest.TestCase):
class TestMaxRetryTimes:
invalid_url = "http://www.scrapytest.org/invalid_url"
def get_spider_and_middleware(self, settings=None):
@ -272,10 +271,10 @@ class MaxRetryTimesTest(unittest.TestCase):
# discard it
req = middleware.process_exception(req, exception, spider)
self.assertEqual(req, None)
assert req is None
class GetRetryRequestTest(unittest.TestCase):
class TestGetRetryRequest:
def get_spider(self, settings=None):
crawler = get_crawler(Spider, settings or {})
return crawler._create_spider("foo")
@ -288,15 +287,15 @@ class GetRetryRequestTest(unittest.TestCase):
request,
spider=spider,
)
self.assertIsInstance(new_request, Request)
self.assertNotEqual(new_request, request)
self.assertEqual(new_request.dont_filter, True)
assert isinstance(new_request, Request)
assert new_request != request
assert new_request.dont_filter
expected_retry_times = 1
self.assertEqual(new_request.meta["retry_times"], expected_retry_times)
self.assertEqual(new_request.priority, -1)
assert new_request.meta["retry_times"] == expected_retry_times
assert new_request.priority == -1
expected_reason = "unspecified"
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
self.assertEqual(spider.crawler.stats.get_value(stat), 1)
assert spider.crawler.stats.get_value(stat) == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -316,8 +315,8 @@ class GetRetryRequestTest(unittest.TestCase):
spider=spider,
max_retry_times=max_retry_times,
)
self.assertEqual(new_request, None)
self.assertEqual(spider.crawler.stats.get_value("retry/max_reached"), 1)
assert new_request is None
assert spider.crawler.stats.get_value("retry/max_reached") == 1
failure_count = max_retry_times + 1
expected_reason = "unspecified"
log.check_present(
@ -338,15 +337,15 @@ class GetRetryRequestTest(unittest.TestCase):
spider=spider,
max_retry_times=1,
)
self.assertIsInstance(new_request, Request)
self.assertNotEqual(new_request, request)
self.assertEqual(new_request.dont_filter, True)
assert isinstance(new_request, Request)
assert new_request != request
assert new_request.dont_filter
expected_retry_times = 1
self.assertEqual(new_request.meta["retry_times"], expected_retry_times)
self.assertEqual(new_request.priority, -1)
assert new_request.meta["retry_times"] == expected_retry_times
assert new_request.priority == -1
expected_reason = "unspecified"
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
self.assertEqual(spider.crawler.stats.get_value(stat), 1)
assert spider.crawler.stats.get_value(stat) == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -368,16 +367,16 @@ class GetRetryRequestTest(unittest.TestCase):
spider=spider,
max_retry_times=max_retry_times,
)
self.assertIsInstance(new_request, Request)
self.assertNotEqual(new_request, request)
self.assertEqual(new_request.dont_filter, True)
assert isinstance(new_request, Request)
assert new_request != request
assert new_request.dont_filter
expected_retry_times = index + 1
self.assertEqual(new_request.meta["retry_times"], expected_retry_times)
self.assertEqual(new_request.priority, -expected_retry_times)
assert new_request.meta["retry_times"] == expected_retry_times
assert new_request.priority == -expected_retry_times
expected_reason = "unspecified"
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
value = spider.crawler.stats.get_value(stat)
self.assertEqual(value, expected_retry_times)
assert value == expected_retry_times
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -393,8 +392,8 @@ class GetRetryRequestTest(unittest.TestCase):
spider=spider,
max_retry_times=max_retry_times,
)
self.assertEqual(new_request, None)
self.assertEqual(spider.crawler.stats.get_value("retry/max_reached"), 1)
assert new_request is None
assert spider.crawler.stats.get_value("retry/max_reached") == 1
failure_count = max_retry_times + 1
expected_reason = "unspecified"
log.check_present(
@ -419,7 +418,7 @@ class GetRetryRequestTest(unittest.TestCase):
request,
spider=spider,
)
self.assertEqual(new_request, None)
assert new_request is None
def test_max_retry_times_meta(self):
max_retry_times = 0
@ -430,7 +429,7 @@ class GetRetryRequestTest(unittest.TestCase):
request,
spider=spider,
)
self.assertEqual(new_request, None)
assert new_request is None
def test_max_retry_times_argument(self):
max_retry_times = 0
@ -442,7 +441,7 @@ class GetRetryRequestTest(unittest.TestCase):
spider=spider,
max_retry_times=max_retry_times,
)
self.assertEqual(new_request, None)
assert new_request is None
def test_priority_adjust_setting(self):
priority_adjust = 1
@ -452,7 +451,7 @@ class GetRetryRequestTest(unittest.TestCase):
request,
spider=spider,
)
self.assertEqual(new_request.priority, priority_adjust)
assert new_request.priority == priority_adjust
def test_priority_adjust_argument(self):
priority_adjust = 1
@ -463,7 +462,7 @@ class GetRetryRequestTest(unittest.TestCase):
spider=spider,
priority_adjust=priority_adjust,
)
self.assertEqual(new_request.priority, priority_adjust)
assert new_request.priority == priority_adjust
def test_log_extra_retry_success(self):
request = Request("https://example.com")
@ -498,7 +497,7 @@ class GetRetryRequestTest(unittest.TestCase):
)
expected_retry_times = 1
for stat in ("retry/count", f"retry/reason_count/{expected_reason}"):
self.assertEqual(spider.crawler.stats.get_value(stat), 1)
assert spider.crawler.stats.get_value(stat) == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -523,7 +522,7 @@ class GetRetryRequestTest(unittest.TestCase):
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
self.assertEqual(stat, 1)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -548,7 +547,7 @@ class GetRetryRequestTest(unittest.TestCase):
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
self.assertEqual(stat, 1)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -573,7 +572,7 @@ class GetRetryRequestTest(unittest.TestCase):
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
self.assertEqual(stat, 1)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -598,7 +597,7 @@ class GetRetryRequestTest(unittest.TestCase):
stat = spider.crawler.stats.get_value(
f"retry/reason_count/{expected_reason_string}"
)
self.assertEqual(stat, 1)
assert stat == 1
log.check_present(
(
"scrapy.downloadermiddlewares.retry",
@ -643,4 +642,4 @@ class GetRetryRequestTest(unittest.TestCase):
f"{stats_key}/count",
f"{stats_key}/reason_count/{expected_reason}",
):
self.assertEqual(spider.crawler.stats.get_value(stat), 1)
assert spider.crawler.stats.get_value(stat) == 1

8
tests/test_downloadermiddleware_robotstxt.py
View File

@ -15,7 +15,7 @@ from scrapy.settings import Settings
from tests.test_robotstxt_interface import rerp_available
class RobotsTxtMiddlewareTest(unittest.TestCase):
class TestRobotsTxtMiddleware(unittest.TestCase):
def setUp(self):
self.crawler = mock.MagicMock()
self.crawler.settings = Settings()
@ -242,11 +242,11 @@ Disallow: /some/randome/page.html
def assertRobotsTxtRequested(self, base_url):
calls = self.crawler.engine.download.call_args_list
request = calls[0][0][0]
self.assertEqual(request.url, f"{base_url}/robots.txt")
self.assertEqual(request.callback, NO_CALLBACK)
assert request.url == f"{base_url}/robots.txt"
assert request.callback == NO_CALLBACK
class RobotsTxtMiddlewareWithRerpTest(RobotsTxtMiddlewareTest):
class TestRobotsTxtMiddlewareWithRerp(TestRobotsTxtMiddleware):
if not rerp_available():
skip = "Rerp parser is not installed"

14
tests/test_downloadermiddleware_stats.py
View File

@ -1,5 +1,3 @@
from unittest import TestCase
from scrapy.downloadermiddlewares.stats import DownloaderStats
from scrapy.http import Request, Response
from scrapy.spiders import Spider
@ -10,8 +8,8 @@ class MyException(Exception):
pass
class TestDownloaderStats(TestCase):
def setUp(self):
class TestDownloaderStats:
def setup_method(self):
self.crawler = get_crawler(Spider)
self.spider = self.crawler._create_spider("scrapytest.org")
self.mw = DownloaderStats(self.crawler.stats)
@ -22,10 +20,8 @@ class TestDownloaderStats(TestCase):
self.res = Response("scrapytest.org", status=400)
def assertStatsEqual(self, key, value):
self.assertEqual(
self.crawler.stats.get_value(key, spider=self.spider),
value,
str(self.crawler.stats.get_stats(self.spider)),
assert self.crawler.stats.get_value(key, spider=self.spider) == value, str(
self.crawler.stats.get_stats(self.spider)
)
def test_process_request(self):
@ -44,5 +40,5 @@ class TestDownloaderStats(TestCase):
1,
)
def tearDown(self):
def teardown_method(self):
self.crawler.stats.close_spider(self.spider, "")

10
tests/test_downloadermiddleware_useragent.py
View File

@ -1,12 +1,10 @@
from unittest import TestCase
from scrapy.downloadermiddlewares.useragent import UserAgentMiddleware
from scrapy.http import Request
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
class UserAgentMiddlewareTest(TestCase):
class TestUserAgentMiddleware:
def get_spider_and_mw(self, default_useragent):
crawler = get_crawler(Spider, {"USER_AGENT": default_useragent})
spider = crawler._create_spider("foo")
@ -16,7 +14,7 @@ class UserAgentMiddlewareTest(TestCase):
spider, mw = self.get_spider_and_mw("default_useragent")
req = Request("http://scrapytest.org/")
assert mw.process_request(req, spider) is None
self.assertEqual(req.headers["User-Agent"], b"default_useragent")
assert req.headers["User-Agent"] == b"default_useragent"
def test_remove_agent(self):
# settings USER_AGENT to None should remove the user agent
@ -33,7 +31,7 @@ class UserAgentMiddlewareTest(TestCase):
mw.spider_opened(spider)
req = Request("http://scrapytest.org/")
assert mw.process_request(req, spider) is None
self.assertEqual(req.headers["User-Agent"], b"spider_useragent")
assert req.headers["User-Agent"] == b"spider_useragent"
def test_header_agent(self):
spider, mw = self.get_spider_and_mw("default_useragent")
@ -43,7 +41,7 @@ class UserAgentMiddlewareTest(TestCase):
"http://scrapytest.org/", headers={"User-Agent": "header_useragent"}
)
assert mw.process_request(req, spider) is None
self.assertEqual(req.headers["User-Agent"], b"header_useragent")
assert req.headers["User-Agent"] == b"header_useragent"
def test_no_agent(self):
spider, mw = self.get_spider_and_mw(None)