Refactor http/ftp URL tests, use urlpase_cached + add more tests

scrapy/downloadermiddlewares/auth.py

@@ -1,19 +1,38 @@
 """
-HTTP basic auth downloader middleware
+HTTP/FTP Authorization downloader middleware
 
 See documentation in docs/topics/downloader-middleware.rst
 """
+from six.moves.urllib.parse import urlunparse
 
 from w3lib.http import basic_auth_header
 
 from scrapy import signals
+from scrapy.utils.httpobj import urlparse_cached
 
-from six.moves.urllib.parse import urlparse
+
+def credstrip_url(parsed_url):
+    """Strip username and password from an urlparse'd URL"""
+    return urlunparse((
+        parsed_url.scheme,
+        parsed_url.netloc.split('@')[-1],
+        parsed_url.path,
+        parsed_url.params,
+        parsed_url.query,
+        parsed_url.fragment))
 
 
 class AuthMiddleware(object):
-    """Set Basic HTTP Authorization header
-    (http_user and http_pass spider class attributes)"""
+    """
+    Populate authorization credentials for HTTP and FTP requests.
+
+    For http(s):// requests, set Basic HTTP Authorization header,
+    either from http_user and http_pass spider attributes,
+    or from URL netloc parsing.
+
+    Also handle FTP credentials from ftp://user:password@... URLs,
+    populating request's meta accordingly.
+    """
 
     @classmethod
     def from_crawler(cls, crawler):
@@ -28,19 +47,33 @@ class AuthMiddleware(object):
             self.auth = basic_auth_header(usr, pwd)
 
     def process_request(self, request, spider):
-        auth = getattr(self, 'auth', None)
-        if auth and 'Authorization' not in request.headers:
-            request.headers['Authorization'] = auth
+        url = urlparse_cached(request)
+        if url.scheme.startswith('http'):
+            # do not override Auth header set priorly
+            if 'Authorization' in request.headers:
+                return
 
-        # credentials from url are supposed to override spider settings
-        url = urlparse(request.url)
-        if url.username and url.password:
-            if url.scheme.startswith('ftp'):
-                request.meta['ftp_user'] = url.username
-                request.meta['ftp_password'] = url.password
-            elif url.scheme.startswith('http'):
-                request.headers['Authorization'] = basic_auth_header(url.username, url.password)
+            new_url = None
+            # credentials from URL override spider attributes
+            if url.username or url.password:
+                auth = basic_auth_header(url.username, url.password)
 
-            # no credentials in new url
-            new_url = url.scheme + '://' + request.url.split('@')[-1]
-            return request.replace(url=new_url)
+                # no credentials in new url
+                new_url = credstrip_url(url)
+
+            else:
+                auth = getattr(self, 'auth', None)
+
+            if auth:
+                request.headers['Authorization'] = auth
+                if new_url:
+                    return request.replace(url=new_url)
+
+        elif url.scheme.startswith('ftp'):
+            if url.username or url.password:
+                # priorly set credentials take precedence
+                request.meta.setdefault('ftp_user', url.username)
+                request.meta.setdefault('ftp_password', url.password)
+
+                # no credentials in new url
+                return request.replace(url=credstrip_url(url))

tests/test_downloadermiddleware_auth.py

@@ -10,6 +10,32 @@ class TestSpider(Spider):
     http_pass = 'bar'
 
 
+class NoAuthTestSpider(Spider):
+    """A test spider that does not set http auth atttributes"""
+
+
+class AuthMiddlewareNoAuthTest(unittest.TestCase):
+
+    def setUp(self):
+        self.mw = AuthMiddleware()
+        self.spider = NoAuthTestSpider('bar')
+        self.mw.spider_opened(self.spider)
+
+    def tearDown(self):
+        del self.mw
+
+    def test_no_auth_http(self):
+        req = Request('http://scrapytest.org/')
+        assert self.mw.process_request(req, self.spider) is None
+        self.assertNotIn('Authorization', req.headers)
+
+    def test_no_auth_ftp(self):
+        req = Request('ftp://scrapytest.org/')
+        assert self.mw.process_request(req, self.spider) is None
+        self.assertNotIn('ftp_user', req.meta)
+        self.assertNotIn('ftp_password', req.meta)
+
+
 class AuthMiddlewareTest(unittest.TestCase):
 
     def setUp(self):
@@ -20,6 +46,9 @@ class AuthMiddlewareTest(unittest.TestCase):
     def tearDown(self):
         del self.mw
 
+
+class AuthMiddlewareHttpAuthTest(AuthMiddlewareTest):
+
     def test_auth(self):
         req = Request('http://scrapytest.org/')
         assert self.mw.process_request(req, self.spider) is None
@@ -38,10 +67,104 @@ class AuthMiddlewareTest(unittest.TestCase):
         self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=')
         self.assertEquals(new_req.url, 'http://scrapytest.org/')
 
+    def test_auth_from_https_url(self):
+        req = Request('https://username:password@scrapytest.org/')
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=')
+        self.assertEquals(new_req.url, 'https://scrapytest.org/')
+
+    def test_auth_from_http_url_no_spider_attrs(self):
+        class AnotherTestSpider(Spider):
+            pass
+        req = Request('http://username:password@scrapytest.org/')
+        new_req = self.mw.process_request(req, AnotherTestSpider('bar'))
+        assert new_req is not None
+        self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=')
+        self.assertEquals(new_req.url, 'http://scrapytest.org/')
+
+    def test_auth_from_https_url_no_spider_attrs(self):
+        class AnotherTestSpider(Spider):
+            pass
+        req = Request('https://username:password@scrapytest.org/')
+        new_req = self.mw.process_request(req, AnotherTestSpider('bar'))
+        assert new_req is not None
+        self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6cGFzc3dvcmQ=')
+        self.assertEquals(new_req.url, 'https://scrapytest.org/')
+
+    def test_auth_from_http_url_empty_pass(self):
+        req = Request('http://username:@scrapytest.org/')
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6')
+        self.assertEquals(new_req.url, 'http://scrapytest.org/')
+
+    def test_auth_from_http_url_pass_none(self):
+        req = Request('http://username@scrapytest.org/')
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(new_req.headers['Authorization'], b'Basic dXNlcm5hbWU6Tm9uZQ==')
+        self.assertEquals(new_req.url, 'http://scrapytest.org/')
+
+    def test_auth_from_http_url_empty_user(self):
+        req = Request('http://:password@scrapytest.org/')
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(new_req.headers['Authorization'], b'Basic OnBhc3N3b3Jk')
+        self.assertEquals(new_req.url, 'http://scrapytest.org/')
+
+
+class AuthMiddlewareFtpAuthTest(AuthMiddlewareTest):
+
+    def test_no_auth_from_ftp_url_meta_unchanged(self):
+        usr, pwd = 'u', 'p'
+        req = Request('ftp://scrapytest.org/',
+                      meta={"ftp_user": usr, "ftp_password": pwd})
+        assert self.mw.process_request(req, self.spider) is None
+        self.assertEquals(req.meta['ftp_user'], usr)
+        self.assertEquals(req.meta['ftp_password'], pwd)
+
+    def test_auth_from_ftp_url_meta_unchanged(self):
+        """Request's meta credentials are kept as-is,
+        but URL is stripped from credentials
+        """
+        usr, pwd = 'u', 'p'
+        req = Request('ftp://username:password@scrapytest.org/',
+                      meta={"ftp_user": usr, "ftp_password": pwd})
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(new_req.meta['ftp_user'], usr)
+        self.assertEquals(new_req.meta['ftp_password'], pwd)
+        self.assertEquals(new_req.url, 'ftp://scrapytest.org/')
+
     def test_auth_from_ftp_url(self):
         req = Request('ftp://username:password@scrapytest.org/')
         new_req = self.mw.process_request(req, self.spider)
         assert new_req is not None
-        self.assertIn('ftp_user', new_req.meta)
-        self.assertIn('ftp_password', new_req.meta)
-        self.assertNotIn('@', new_req.url)
+        self.assertEquals(req.meta['ftp_user'], 'username')
+        self.assertEquals(req.meta['ftp_password'], 'password')
+        self.assertEquals(new_req.url, 'ftp://scrapytest.org/')
+
+    def test_auth_from_ftp_url_empty_user(self):
+        req = Request('ftp://:password@scrapytest.org/')
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(req.meta['ftp_user'], '')
+        self.assertEquals(req.meta['ftp_password'], 'password')
+        self.assertEquals(new_req.url, 'ftp://scrapytest.org/')
+
+    def test_auth_from_ftp_url_empty_pass(self):
+        req = Request('ftp://username:@scrapytest.org/')
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(req.meta['ftp_user'], 'username')
+        self.assertEquals(req.meta['ftp_password'], '')
+        self.assertEquals(new_req.url, 'ftp://scrapytest.org/')
+
+    def test_auth_from_ftp_url_pass_none(self):
+        req = Request('ftp://username@scrapytest.org/')
+        new_req = self.mw.process_request(req, self.spider)
+        assert new_req is not None
+        self.assertEquals(req.meta['ftp_user'], 'username')
+        self.assertEquals(req.meta['ftp_password'], None)
+        self.assertEquals(new_req.url, 'ftp://scrapytest.org/')