Merge pull request #7478 from chrahunt/refactor/make-functional-downl…

…oad-tests

Make unpack_* unit tests into functional tests

src/pip/_internal/operations/prepare.py

@@ -237,14 +237,6 @@ def unpack_file_url(
             logger.info('Link is a directory, ignoring download_dir')
         return None
 
-    # If --require-hashes is off, `hashes` is either empty, the
-    # link's embedded hash, or MissingHashes; it is required to
-    # match. If --require-hashes is on, we are satisfied by any
-    # hash in `hashes` matching: a URL-based or an option-based
-    # one; no internet-sourced hash will be in `hashes`.
-    if hashes:
-        hashes.check_against_path(link_path)
-
     # If a download dir is specified, is the file already there and valid?
     already_downloaded_path = None
     if download_dir:
@@ -257,6 +249,14 @@ def unpack_file_url(
     else:
         from_path = link_path
 
+    # If --require-hashes is off, `hashes` is either empty, the
+    # link's embedded hash, or MissingHashes; it is required to
+    # match. If --require-hashes is on, we are satisfied by any
+    # hash in `hashes` matching: a URL-based or an option-based
+    # one; no internet-sourced hash will be in `hashes`.
+    if hashes:
+        hashes.check_against_path(from_path)
+
     content_type = mimetypes.guess_type(from_path)[0]
 
     # unpack the archive to the build dir location. even when only downloading

tests/functional/test_download.py

@@ -1,11 +1,14 @@
 import os.path
 import shutil
 import textwrap
+from hashlib import sha256
 
 import pytest
 
 from pip._internal.cli.status_codes import ERROR
+from pip._internal.utils.urls import path_to_url
 from tests.lib.path import Path
+from tests.lib.server import file_response
 
 
 def fake_wheel(data, wheel_path):
@@ -720,3 +723,97 @@ def test_download_prefer_binary_when_only_tarball_exists(script, data):
         Path('scratch') / 'source-1.0.tar.gz'
         in result.files_created
     )
+
+
+@pytest.fixture(scope="session")
+def shared_script(tmpdir_factory, script_factory):
+    tmpdir = Path(str(tmpdir_factory.mktemp("download_shared_script")))
+    script = script_factory(tmpdir.joinpath("workspace"))
+    return script
+
+
+def test_download_file_url(shared_script, shared_data, tmpdir):
+    download_dir = tmpdir / 'download'
+    download_dir.mkdir()
+    downloaded_path = download_dir / 'simple-1.0.tar.gz'
+
+    simple_pkg = shared_data.packages / 'simple-1.0.tar.gz'
+
+    shared_script.pip(
+        'download',
+        '-d',
+        str(download_dir),
+        '--no-index',
+        path_to_url(str(simple_pkg)),
+    )
+
+    assert downloaded_path.exists()
+    assert simple_pkg.read_bytes() == downloaded_path.read_bytes()
+
+
+def test_download_file_url_existing_ok_download(
+    shared_script, shared_data, tmpdir
+):
+    download_dir = tmpdir / 'download'
+    download_dir.mkdir()
+    downloaded_path = download_dir / 'simple-1.0.tar.gz'
+    fake_existing_package = shared_data.packages / 'simple-2.0.tar.gz'
+    shutil.copy(str(fake_existing_package), str(downloaded_path))
+    downloaded_path_bytes = downloaded_path.read_bytes()
+    digest = sha256(downloaded_path_bytes).hexdigest()
+
+    simple_pkg = shared_data.packages / 'simple-1.0.tar.gz'
+    url = "{}#sha256={}".format(path_to_url(simple_pkg), digest)
+
+    shared_script.pip('download', '-d', str(download_dir), url)
+
+    assert downloaded_path_bytes == downloaded_path.read_bytes()
+
+
+def test_download_file_url_existing_bad_download(
+    shared_script, shared_data, tmpdir
+):
+    download_dir = tmpdir / 'download'
+    download_dir.mkdir()
+    downloaded_path = download_dir / 'simple-1.0.tar.gz'
+    fake_existing_package = shared_data.packages / 'simple-2.0.tar.gz'
+    shutil.copy(str(fake_existing_package), str(downloaded_path))
+
+    simple_pkg = shared_data.packages / 'simple-1.0.tar.gz'
+    simple_pkg_bytes = simple_pkg.read_bytes()
+    digest = sha256(simple_pkg_bytes).hexdigest()
+    url = "{}#sha256={}".format(path_to_url(simple_pkg), digest)
+
+    shared_script.pip('download', '-d', str(download_dir), url)
+
+    assert simple_pkg_bytes == downloaded_path.read_bytes()
+
+
+def test_download_http_url_bad_hash(
+    shared_script, shared_data, tmpdir, mock_server
+):
+    download_dir = tmpdir / 'download'
+    download_dir.mkdir()
+    downloaded_path = download_dir / 'simple-1.0.tar.gz'
+    fake_existing_package = shared_data.packages / 'simple-2.0.tar.gz'
+    shutil.copy(str(fake_existing_package), str(downloaded_path))
+
+    simple_pkg = shared_data.packages / 'simple-1.0.tar.gz'
+    simple_pkg_bytes = simple_pkg.read_bytes()
+    digest = sha256(simple_pkg_bytes).hexdigest()
+    mock_server.set_responses([
+        file_response(simple_pkg)
+    ])
+    mock_server.start()
+    base_address = 'http://{}:{}'.format(mock_server.host, mock_server.port)
+    url = "{}/simple-1.0.tar.gz#sha256={}".format(base_address, digest)
+
+    shared_script.pip('download', '-d', str(download_dir), url)
+
+    assert simple_pkg_bytes == downloaded_path.read_bytes()
+
+    mock_server.stop()
+    requests = mock_server.get_requests()
+    assert len(requests) == 1
+    assert requests[0]['PATH_INFO'] == '/simple-1.0.tar.gz'
+    assert requests[0]['HTTP_ACCEPT_ENCODING'] == 'identity'

tests/lib/path.py

@@ -178,6 +178,11 @@ def joinpath(self, *parts):
     def join(self, *parts):
         raise RuntimeError('Path.join is invalid, use joinpath instead.')
 
+    def read_bytes(self):
+        # type: () -> bytes
+        with open(self, "rb") as fp:
+            return fp.read()
+
     def read_text(self):
         with open(self, "r") as fp:
             return fp.read()

tests/unit/test_operations_prepare.py

@@ -1,11 +1,10 @@
-import hashlib
 import os
 import shutil
-from shutil import copy, rmtree
+from shutil import rmtree
 from tempfile import mkdtemp
 
 import pytest
-from mock import Mock, patch
+from mock import Mock
 
 from pip._internal.exceptions import HashMismatch
 from pip._internal.models.link import Link
@@ -19,7 +18,6 @@
 )
 from pip._internal.utils.hashes import Hashes
 from pip._internal.utils.urls import path_to_url
-from tests.lib import create_file
 from tests.lib.filesystem import (
     get_filelist,
     make_socket_file,
@@ -61,50 +59,6 @@ def _fake_session_get(*args, **kwargs):
         rmtree(temp_dir)
 
 
-@patch('pip._internal.operations.prepare.unpack_file')
-def test_unpack_http_url_bad_downloaded_checksum(mock_unpack_file):
-    """
-    If already-downloaded file has bad checksum, re-download.
-    """
-    base_url = 'http://www.example.com/somepackage.tgz'
-    contents = b'downloaded'
-    download_hash = hashlib.new('sha1', contents)
-    link = Link(base_url + '#sha1=' + download_hash.hexdigest())
-
-    session = Mock()
-    session.get = Mock()
-    response = session.get.return_value = MockResponse(contents)
-    response.headers = {'content-type': 'application/x-tar'}
-    response.url = base_url
-    downloader = Downloader(session, progress_bar="on")
-
-    download_dir = mkdtemp()
-    try:
-        downloaded_file = os.path.join(download_dir, 'somepackage.tgz')
-        create_file(downloaded_file, 'some contents')
-
-        unpack_http_url(
-            link,
-            'location',
-            downloader=downloader,
-            download_dir=download_dir,
-            hashes=Hashes({'sha1': [download_hash.hexdigest()]})
-        )
-
-        # despite existence of downloaded file with bad hash, downloaded again
-        session.get.assert_called_once_with(
-            'http://www.example.com/somepackage.tgz',
-            headers={"Accept-Encoding": "identity"},
-            stream=True,
-        )
-        # cached file is replaced with newly downloaded file
-        with open(downloaded_file) as fh:
-            assert fh.read() == 'downloaded'
-
-    finally:
-        rmtree(download_dir)
-
-
 def test_download_http_url__no_directory_traversal(tmpdir):
     """
     Test that directory traversal doesn't happen on download when the
@@ -239,29 +193,6 @@ def test_unpack_file_url_no_download(self, tmpdir, data):
         assert not os.path.isfile(
             os.path.join(self.download_dir, self.dist_file))
 
-    def test_unpack_file_url_and_download(self, tmpdir, data):
-        self.prep(tmpdir, data)
-        unpack_file_url(self.dist_url, self.build_dir,
-                        download_dir=self.download_dir)
-        assert os.path.isdir(os.path.join(self.build_dir, 'simple'))
-        assert os.path.isfile(os.path.join(self.download_dir, self.dist_file))
-
-    def test_unpack_file_url_download_already_exists(self, tmpdir,
-                                                     data, monkeypatch):
-        self.prep(tmpdir, data)
-        # add in previous download (copy simple-2.0 as simple-1.0)
-        # so we can tell it didn't get overwritten
-        dest_file = os.path.join(self.download_dir, self.dist_file)
-        copy(self.dist_path2, dest_file)
-        with open(self.dist_path2, 'rb') as f:
-            dist_path2_md5 = hashlib.md5(f.read()).hexdigest()
-
-        unpack_file_url(self.dist_url, self.build_dir,
-                        download_dir=self.download_dir)
-        # our hash should be the same, i.e. not overwritten by simple-1.0 hash
-        with open(dest_file, 'rb') as f:
-            assert dist_path2_md5 == hashlib.md5(f.read()).hexdigest()
-
     def test_unpack_file_url_bad_hash(self, tmpdir, data,
                                       monkeypatch):
         """
@@ -275,37 +206,6 @@ def test_unpack_file_url_bad_hash(self, tmpdir, data,
                             self.build_dir,
                             hashes=Hashes({'md5': ['bogus']}))
 
-    def test_unpack_file_url_download_bad_hash(self, tmpdir, data,
-                                               monkeypatch):
-        """
-        Test when existing download has different hash from the file url
-        fragment
-        """
-        self.prep(tmpdir, data)
-
-        # add in previous download (copy simple-2.0 as simple-1.0 so it's wrong
-        # hash)
-        dest_file = os.path.join(self.download_dir, self.dist_file)
-        copy(self.dist_path2, dest_file)
-
-        with open(self.dist_path, 'rb') as f:
-            dist_path_md5 = hashlib.md5(f.read()).hexdigest()
-        with open(dest_file, 'rb') as f:
-            dist_path2_md5 = hashlib.md5(f.read()).hexdigest()
-
-        assert dist_path_md5 != dist_path2_md5
-
-        url = '{}#md5={}'.format(self.dist_url.url, dist_path_md5)
-        dist_url = Link(url)
-        unpack_file_url(dist_url, self.build_dir,
-                        download_dir=self.download_dir,
-                        hashes=Hashes({'md5': [dist_path_md5]}))
-
-        # confirm hash is for simple1-1.0
-        # the previous bad download has been removed
-        with open(dest_file, 'rb') as f:
-            assert hashlib.md5(f.read()).hexdigest() == dist_path_md5
-
     def test_unpack_file_url_thats_a_dir(self, tmpdir, data):
         self.prep(tmpdir, data)
         dist_path = data.packages.joinpath("FSPkg")