mirror of https://github.com/scrapy/scrapy.git
Complete test coverage, support explicit default ports
This commit is contained in:
Adrián Chaves
parent
cb88457f95
commit
0fe0109386
|
|
@ -22,9 +22,22 @@ if TYPE_CHECKING:
|
|||
from scrapy.http import Response
|
||||
|
||||
|
||||
_NO_PORT = object()
|
||||
_DEFAULT_PORTS = {
|
||||
"http": 80,
|
||||
"https": 443,
|
||||
}
|
||||
|
||||
|
||||
def _origin(request: Request) -> str:
|
||||
parsed_url = urlparse_cached(request)
|
||||
return f"{parsed_url.scheme}://{parsed_url.netloc}"
|
||||
scheme = parsed_url.scheme
|
||||
netloc = (
|
||||
parsed_url.netloc
|
||||
if parsed_url.port != _DEFAULT_PORTS[scheme]
|
||||
else parsed_url.hostname
|
||||
)
|
||||
return f"{scheme}://{netloc}"
|
||||
|
||||
|
||||
def _setdefault_auth_origin(request: Request) -> str:
|
||||
|
|
@ -59,7 +72,10 @@ class HttpAuthMiddleware:
|
|||
def process_request(
|
||||
self, request: Request, spider: Spider
|
||||
) -> Request | Response | None:
|
||||
if b"Authorization" in request.headers:
|
||||
if (
|
||||
b"Authorization" in request.headers
|
||||
or urlparse_cached(request).scheme not in _DEFAULT_PORTS
|
||||
):
|
||||
return None
|
||||
user = request.meta.get("http_user", "")
|
||||
password = request.meta.get("http_pass", "")
|
||||
|
|
|
|||
|
|
@ -1,106 +1,177 @@
|
|||
import unittest
|
||||
|
||||
from w3lib.http import basic_auth_header
|
||||
import pytest
|
||||
|
||||
from scrapy.downloadermiddlewares.httpauth import HttpAuthMiddleware
|
||||
from scrapy.http import Request
|
||||
from scrapy.spiders import Spider
|
||||
|
||||
|
||||
class TestSpiderLegacy(Spider):
|
||||
http_user = "foo"
|
||||
http_pass = "bar"
|
||||
@pytest.mark.parametrize(
|
||||
("config", "expected"),
|
||||
(
|
||||
# Baseline
|
||||
({}, None),
|
||||
# Spider attributes.
|
||||
# http_auth_domain=None allows any domain.
|
||||
(
|
||||
{"spider_attributes": {"http_user": "su", "http_auth_domain": None}},
|
||||
b"Basic c3U6",
|
||||
),
|
||||
(
|
||||
{"spider_attributes": {"http_pass": "sp", "http_auth_domain": None}},
|
||||
b"Basic OnNw",
|
||||
),
|
||||
(
|
||||
{
|
||||
"spider_attributes": {
|
||||
"http_user": "su",
|
||||
"http_pass": "sp",
|
||||
"http_auth_domain": None,
|
||||
}
|
||||
},
|
||||
b"Basic c3U6c3A=",
|
||||
),
|
||||
# http_auth_domain=domain allows only that domain and subdomains.
|
||||
(
|
||||
{"spider_attributes": {"http_user": "su", "http_auth_domain": "a.example"}},
|
||||
b"Basic c3U6",
|
||||
),
|
||||
(
|
||||
{
|
||||
"url": "https://s.a.example/a",
|
||||
"spider_attributes": {
|
||||
"http_user": "su",
|
||||
"http_auth_domain": "a.example",
|
||||
},
|
||||
},
|
||||
b"Basic c3U6",
|
||||
),
|
||||
(
|
||||
{"spider_attributes": {"http_user": "su", "http_auth_domain": "b.example"}},
|
||||
None,
|
||||
),
|
||||
# http_auth_domain must be defined if http_user or http_pass are.
|
||||
({"spider_attributes": {"http_user": "su"}}, AttributeError),
|
||||
# Request.meta.
|
||||
({"meta": {"http_user": "mu"}}, b"Basic bXU6"),
|
||||
({"meta": {"http_pass": "mp"}}, b"Basic Om1w"),
|
||||
({"meta": {"http_user": "mu", "http_pass": "mp"}}, b"Basic bXU6bXA="),
|
||||
# Request.meta["auth_origin"]=origin prevents other origins.
|
||||
#
|
||||
# Note: auth_origin is not meant to be set by users, it is set the
|
||||
# first time a request is processed by the middleware. See
|
||||
# test_origin_setdefault.
|
||||
(
|
||||
{"meta": {"auth_origin": "https://a.example", "http_user": "mu"}},
|
||||
b"Basic bXU6",
|
||||
),
|
||||
(
|
||||
{
|
||||
"url": "https://a.example:443",
|
||||
"meta": {"auth_origin": "https://a.example", "http_user": "mu"},
|
||||
},
|
||||
b"Basic bXU6",
|
||||
),
|
||||
(
|
||||
{
|
||||
"url": "https://s.a.example",
|
||||
"meta": {"auth_origin": "https://a.example", "http_user": "mu"},
|
||||
},
|
||||
None,
|
||||
),
|
||||
({"meta": {"auth_origin": "http://a.example", "http_user": "mu"}}, None),
|
||||
({"meta": {"auth_origin": "https://a.example:1", "http_user": "mu"}}, None),
|
||||
({"meta": {"auth_origin": "https://b.example", "http_user": "mu"}}, None),
|
||||
# Takes priority over spider attributes.
|
||||
(
|
||||
{
|
||||
"meta": {"http_user": "mu"},
|
||||
"spider_attributes": {
|
||||
"http_user": "su",
|
||||
"http_pass": "sp",
|
||||
"http_auth_domain": None,
|
||||
},
|
||||
},
|
||||
b"Basic bXU6",
|
||||
),
|
||||
# If the Authorization header is set, it is not modified.
|
||||
(
|
||||
{
|
||||
"headers": {"Authorization": "a"},
|
||||
"spider_attributes": {"http_user": "su", "http_auth_domain": None},
|
||||
},
|
||||
b"a",
|
||||
),
|
||||
({"headers": {"Authorization": "a"}, "meta": {"http_user": "mu"}}, b"a"),
|
||||
# If a non-HTTP request is received, nothing is done.
|
||||
(
|
||||
{
|
||||
"url": "ftp://example.com",
|
||||
"spider_attributes": {"http_user": "su", "http_auth_domain": None},
|
||||
},
|
||||
None,
|
||||
),
|
||||
({"url": "s3://example.com", "meta": {"http_user": "mu"}}, None),
|
||||
),
|
||||
)
|
||||
def test_main(config, expected):
|
||||
url = config.get("url", "https://a.example")
|
||||
headers = config.get("headers", {})
|
||||
meta = config.get("meta", {})
|
||||
spider_attributes = config.get("spider_attributes", {})
|
||||
|
||||
class TestSpider(Spider):
|
||||
pass
|
||||
|
||||
class TestSpider(Spider):
|
||||
http_user = "foo"
|
||||
http_pass = "bar"
|
||||
http_auth_domain = "example.com"
|
||||
for k, v in spider_attributes.items():
|
||||
setattr(TestSpider, k, v)
|
||||
|
||||
mw = HttpAuthMiddleware()
|
||||
spider = TestSpider("foo")
|
||||
|
||||
class TestSpiderAny(Spider):
|
||||
http_user = "foo"
|
||||
http_pass = "bar"
|
||||
http_auth_domain = None
|
||||
if isinstance(expected, type) and issubclass(expected, Exception):
|
||||
with pytest.raises(expected):
|
||||
mw.spider_opened(spider)
|
||||
return
|
||||
|
||||
|
||||
class HttpAuthMiddlewareLegacyTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.spider = TestSpiderLegacy("foo")
|
||||
|
||||
def test_auth(self):
|
||||
with self.assertRaises(AttributeError):
|
||||
mw = HttpAuthMiddleware()
|
||||
mw.spider_opened(self.spider)
|
||||
|
||||
|
||||
class HttpAuthMiddlewareTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mw = HttpAuthMiddleware()
|
||||
self.spider = TestSpider("foo")
|
||||
self.mw.spider_opened(self.spider)
|
||||
|
||||
def tearDown(self):
|
||||
del self.mw
|
||||
|
||||
def test_no_auth(self):
|
||||
req = Request("http://noauth.example/")
|
||||
assert self.mw.process_request(req, self.spider) is None
|
||||
self.assertNotIn("Authorization", 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"))
|
||||
|
||||
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"))
|
||||
|
||||
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")
|
||||
|
||||
|
||||
class HttpAuthAnyMiddlewareTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.mw = HttpAuthMiddleware()
|
||||
self.spider = TestSpiderAny("foo")
|
||||
self.mw.spider_opened(self.spider)
|
||||
|
||||
def tearDown(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"))
|
||||
|
||||
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")
|
||||
|
||||
def test_auth_already_set_with_meta(self):
|
||||
meta = {"http_user": "bar", "http_pass": "foo"}
|
||||
req = Request(
|
||||
"http://example.com/",
|
||||
headers={"Authorization": "Digest 123"},
|
||||
meta=meta,
|
||||
mw.spider_opened(spider)
|
||||
request = Request(url, headers=headers, meta=meta)
|
||||
assert mw.process_request(request, spider) is None
|
||||
if expected is None:
|
||||
assert "Authorization" not in request.headers
|
||||
else:
|
||||
assert request.headers["Authorization"] == expected, repr(
|
||||
request.headers["Authorization"]
|
||||
)
|
||||
assert self.mw.process_request(req, self.spider) is None
|
||||
self.assertEqual(req.headers["Authorization"], b"Digest 123")
|
||||
|
||||
def test_auth_meta(self):
|
||||
meta = {"http_user": "bar", "http_pass": "foo"}
|
||||
req = Request("http://example.com/", meta=meta)
|
||||
assert self.mw.process_request(req, Spider("bar")) is None
|
||||
self.assertEqual(req.headers["Authorization"], b"Basic YmFyOmZvbw==")
|
||||
|
||||
def test_auth_meta_override(self):
|
||||
meta = {"http_user": "bar", "http_pass": "foo"}
|
||||
req = Request("http://example.com/", meta=meta)
|
||||
assert self.mw.process_request(req, self.spider) is None
|
||||
self.assertEqual(req.headers["Authorization"], b"Basic YmFyOmZvbw==")
|
||||
@pytest.mark.parametrize(
|
||||
("meta", "url", "output_value"),
|
||||
(
|
||||
({}, "https://example.com/a", None),
|
||||
({"http_user": "a", "auth_origin": "foo"}, "https://example.com/a", "foo"),
|
||||
({"http_user": "a"}, "https://example.com/a", "https://example.com"),
|
||||
({"http_user": "a"}, "http://example.com/a", "http://example.com"),
|
||||
({"http_user": "a"}, "https://example.com:443/a", "https://example.com"),
|
||||
({"http_user": "a"}, "http://example.com:80/a", "http://example.com"),
|
||||
({"http_user": "a"}, "https://example.com:80/a", "https://example.com:80"),
|
||||
({"http_user": "a"}, "http://example.com:443/a", "http://example.com:443"),
|
||||
({"http_user": "a"}, "https://example.com:1234/a", "https://example.com:1234"),
|
||||
({"http_user": "a"}, "http://example.com:1234/a", "http://example.com:1234"),
|
||||
),
|
||||
)
|
||||
def test_origin_setdefault(meta, url, output_value):
|
||||
"""When request.meta is used for authorization, an auth_origin meta key is
|
||||
defined on the request if not defined already."""
|
||||
|
||||
class TestSpider(Spider):
|
||||
pass
|
||||
|
||||
mw = HttpAuthMiddleware()
|
||||
spider = TestSpider("foo")
|
||||
mw.spider_opened(spider)
|
||||
request = Request(url, meta=meta)
|
||||
assert mw.process_request(request, spider) is None
|
||||
if output_value is None:
|
||||
assert "auth_origin" not in request.meta
|
||||
else:
|
||||
assert request.meta["auth_origin"] == output_value
|
||||
|
|
|
|||
|
|
@ -14,9 +14,6 @@ class AbstractWrapper:
|
|||
self.mw = UriUserInfoMiddleware()
|
||||
self.spider = Spider("bar")
|
||||
|
||||
def tearDown(self):
|
||||
del self.mw
|
||||
|
||||
class ProtocolTestCase(BaseTestCase):
|
||||
def test_username_and_password(self):
|
||||
req = Request(f"{self.protocol}://foo:bar@example.com/")
|
||||
|
|
|
|||
Loading…
Reference in New Issue