Move _prepare_download to network.download

src/pip/_internal/network/download.py

@@ -1,12 +1,25 @@
 """Download files with progress indicators.
 """
+import logging
+
+from pip._vendor.requests.models import CONTENT_CHUNK_SIZE
+
+from pip._internal.models.index import PyPI
+from pip._internal.network.cache import is_from_cache
+from pip._internal.network.utils import response_chunks
+from pip._internal.utils.misc import format_size, redact_auth_from_url
 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
+from pip._internal.utils.ui import DownloadProgressProvider
 
 if MYPY_CHECK_RUNNING:
-    from typing import Optional
+    from typing import Iterable, Optional
 
     from pip._vendor.requests.models import Response
 
+    from pip._internal.models.link import Link
+
+logger = logging.getLogger(__name__)
+
 
 def _get_http_response_size(resp):
     # type: (Response) -> Optional[int]
@@ -14,3 +27,47 @@ def _get_http_response_size(resp):
         return int(resp.headers['content-length'])
     except (ValueError, KeyError, TypeError):
         return None
+
+
+def _prepare_download(
+    resp,  # type: Response
+    link,  # type: Link
+    progress_bar  # type: str
+):
+    # type: (...) -> Iterable[bytes]
+    total_length = _get_http_response_size(resp)
+
+    if link.netloc == PyPI.file_storage_domain:
+        url = link.show_url
+    else:
+        url = link.url_without_fragment
+
+    logged_url = redact_auth_from_url(url)
+
+    if total_length:
+        logged_url = '{} ({})'.format(logged_url, format_size(total_length))
+
+    if is_from_cache(resp):
+        logger.info("Using cached %s", logged_url)
+    else:
+        logger.info("Downloading %s", logged_url)
+
+    if logger.getEffectiveLevel() > logging.INFO:
+        show_progress = False
+    elif is_from_cache(resp):
+        show_progress = False
+    elif not total_length:
+        show_progress = True
+    elif total_length > (40 * 1000):
+        show_progress = True
+    else:
+        show_progress = False
+
+    chunks = response_chunks(resp, CONTENT_CHUNK_SIZE)
+
+    if not show_progress:
+        return chunks
+
+    return DownloadProgressProvider(
+        progress_bar, max=total_length
+    )(chunks)

src/pip/_internal/operations/prepare.py

@@ -13,7 +13,7 @@ import shutil
 import sys
 
 from pip._vendor import requests
-from pip._vendor.requests.models import CONTENT_CHUNK_SIZE, Response
+from pip._vendor.requests.models import Response
 from pip._vendor.six import PY2
 
 from pip._internal.distributions import (
@@ -28,11 +28,8 @@ from pip._internal.exceptions import (
     PreviousBuildDirError,
     VcsHashUnsupported,
 )
-from pip._internal.models.index import PyPI
-from pip._internal.network.cache import is_from_cache
-from pip._internal.network.download import _get_http_response_size
+from pip._internal.network.download import _prepare_download
 from pip._internal.network.session import PipSession
-from pip._internal.network.utils import response_chunks
 from pip._internal.utils.compat import expanduser
 from pip._internal.utils.filesystem import copy2_fixed
 from pip._internal.utils.hashes import MissingHashes
@@ -42,17 +39,14 @@ from pip._internal.utils.misc import (
     ask_path_exists,
     backup_dir,
     display_path,
-    format_size,
     hide_url,
     normalize_path,
     path_to_display,
-    redact_auth_from_url,
     rmtree,
     splitext,
 )
 from pip._internal.utils.temp_dir import TempDirectory
 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
-from pip._internal.utils.ui import DownloadProgressProvider
 from pip._internal.utils.unpacking import unpack_file
 from pip._internal.vcs import vcs
 
@@ -110,50 +104,6 @@ def unpack_vcs_link(link, location):
     vcs_backend.unpack(location, url=hide_url(link.url))
 
 
-def _prepare_download(
-    resp,  # type: Response
-    link,  # type: Link
-    progress_bar  # type: str
-):
-    # type: (...) -> Iterable[bytes]
-    total_length = _get_http_response_size(resp)
-
-    if link.netloc == PyPI.file_storage_domain:
-        url = link.show_url
-    else:
-        url = link.url_without_fragment
-
-    logged_url = redact_auth_from_url(url)
-
-    if total_length:
-        logged_url = '{} ({})'.format(logged_url, format_size(total_length))
-
-    if is_from_cache(resp):
-        logger.info("Using cached %s", logged_url)
-    else:
-        logger.info("Downloading %s", logged_url)
-
-    if logger.getEffectiveLevel() > logging.INFO:
-        show_progress = False
-    elif is_from_cache(resp):
-        show_progress = False
-    elif not total_length:
-        show_progress = True
-    elif total_length > (40 * 1000):
-        show_progress = True
-    else:
-        show_progress = False
-
-    chunks = response_chunks(resp, CONTENT_CHUNK_SIZE)
-
-    if not show_progress:
-        return chunks
-
-    return DownloadProgressProvider(
-        progress_bar, max=total_length
-    )(chunks)
-
-
 def _copy_file(filename, location, link):
     copy = True
     download_location = os.path.join(location, link.filename)

tests/unit/test_network_download.py

@@ -0,0 +1,36 @@
+import pytest
+import logging
+
+from pip._internal.models.link import Link
+from pip._internal.network.download import _prepare_download
+from tests.lib.requests_mocks import MockResponse
+
+
+@pytest.mark.parametrize("url, headers, from_cache, expected", [
+    ('http://example.com/foo.tgz', {}, False,
+        "Downloading http://example.com/foo.tgz"),
+    ('http://example.com/foo.tgz', {'content-length': 2}, False,
+        "Downloading http://example.com/foo.tgz (2 bytes)"),
+    ('http://example.com/foo.tgz', {'content-length': 2}, True,
+        "Using cached http://example.com/foo.tgz (2 bytes)"),
+    ('https://files.pythonhosted.org/foo.tgz', {}, False,
+        "Downloading foo.tgz"),
+    ('https://files.pythonhosted.org/foo.tgz', {'content-length': 2}, False,
+        "Downloading foo.tgz (2 bytes)"),
+    ('https://files.pythonhosted.org/foo.tgz', {'content-length': 2}, True,
+        "Using cached foo.tgz"),
+])
+def test_prepare_download__log(caplog, url, headers, from_cache, expected):
+    caplog.set_level(logging.INFO)
+    resp = MockResponse(b'')
+    resp.url = url
+    resp.headers = headers
+    if from_cache:
+        resp.from_cache = from_cache
+    link = Link(url)
+    _prepare_download(resp, link, progress_bar="on")
+
+    assert len(caplog.records) == 1
+    record = caplog.records[0]
+    assert record.levelname == 'INFO'
+    assert expected in record.message

tests/unit/test_operations_prepare.py

@@ -1,5 +1,4 @@
 import hashlib
-import logging
 import os
 import shutil
 import sys
@@ -16,7 +15,6 @@ from pip._internal.operations.prepare import (
     Downloader,
     _copy_source_tree,
     _download_http_url,
-    _prepare_download,
     parse_content_disposition,
     sanitize_content_filename,
     unpack_file_url,
@@ -194,36 +192,6 @@ def test_download_http_url__no_directory_traversal(tmpdir):
     assert actual == ['out_dir_file']
 
 
-@pytest.mark.parametrize("url, headers, from_cache, expected", [
-    ('http://example.com/foo.tgz', {}, False,
-        "Downloading http://example.com/foo.tgz"),
-    ('http://example.com/foo.tgz', {'content-length': 2}, False,
-        "Downloading http://example.com/foo.tgz (2 bytes)"),
-    ('http://example.com/foo.tgz', {'content-length': 2}, True,
-        "Using cached http://example.com/foo.tgz (2 bytes)"),
-    ('https://files.pythonhosted.org/foo.tgz', {}, False,
-        "Downloading foo.tgz"),
-    ('https://files.pythonhosted.org/foo.tgz', {'content-length': 2}, False,
-        "Downloading foo.tgz (2 bytes)"),
-    ('https://files.pythonhosted.org/foo.tgz', {'content-length': 2}, True,
-        "Using cached foo.tgz"),
-])
-def test_prepare_download__log(caplog, url, headers, from_cache, expected):
-    caplog.set_level(logging.INFO)
-    resp = MockResponse(b'')
-    resp.url = url
-    resp.headers = headers
-    if from_cache:
-        resp.from_cache = from_cache
-    link = Link(url)
-    _prepare_download(resp, link, progress_bar="on")
-
-    assert len(caplog.records) == 1
-    record = caplog.records[0]
-    assert record.levelname == 'INFO'
-    assert expected in record.message
-
-
 @pytest.fixture
 def clean_project(tmpdir_factory, data):
     tmpdir = Path(str(tmpdir_factory.mktemp("clean_project")))