remove _copy_source_tree and friends

src/pip/_internal/operations/prepare.py

@@ -24,10 +24,9 @@
     PreviousBuildDirError,
     VcsHashUnsupported,
 )
-from pip._internal.utils.filesystem import copy2_fixed
 from pip._internal.utils.hashes import MissingHashes
 from pip._internal.utils.logging import indent_log
-from pip._internal.utils.misc import display_path, hide_url, path_to_display
+from pip._internal.utils.misc import display_path, hide_url
 from pip._internal.utils.temp_dir import TempDirectory
 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
 from pip._internal.utils.unpacking import unpack_file
@@ -128,59 +127,6 @@ def get_http_url(
     return File(from_path, content_type)
 
 
-def _copy2_ignoring_special_files(src, dest):
-    # type: (str, str) -> None
-    """Copying special files is not supported, but as a convenience to users
-    we skip errors copying them. This supports tools that may create e.g.
-    socket files in the project source directory.
-    """
-    try:
-        copy2_fixed(src, dest)
-    except shutil.SpecialFileError as e:
-        # SpecialFileError may be raised due to either the source or
-        # destination. If the destination was the cause then we would actually
-        # care, but since the destination directory is deleted prior to
-        # copy we ignore all of them assuming it is caused by the source.
-        logger.warning(
-            "Ignoring special file error '%s' encountered copying %s to %s.",
-            str(e),
-            path_to_display(src),
-            path_to_display(dest),
-        )
-
-
-def _copy_source_tree(source, target):
-    # type: (str, str) -> None
-    target_abspath = os.path.abspath(target)
-    target_basename = os.path.basename(target_abspath)
-    target_dirname = os.path.dirname(target_abspath)
-
-    def ignore(d, names):
-        # type: (str, List[str]) -> List[str]
-        skipped = []  # type: List[str]
-        if d == source:
-            # Pulling in those directories can potentially be very slow,
-            # exclude the following directories if they appear in the top
-            # level dir (and only it).
-            # See discussion at https://github.com/pypa/pip/pull/6770
-            skipped += ['.tox', '.nox']
-        if os.path.abspath(d) == target_dirname:
-            # Prevent an infinite recursion if the target is in source.
-            # This can happen when TMPDIR is set to ${PWD}/...
-            # and we copy PWD to TMPDIR.
-            skipped += [target_basename]
-        return skipped
-
-    kwargs = dict(ignore=ignore, symlinks=True)  # type: CopytreeKwargs
-
-    if not PY2:
-        # Python 2 does not support copy_function, so we only ignore
-        # errors on special file copy in Python 3.
-        kwargs['copy_function'] = _copy2_ignoring_special_files
-
-    shutil.copytree(source, target, **kwargs)
-
-
 def get_file_url(
     link,  # type: Link
     download_dir=None,  # type: Optional[str]

src/pip/_internal/utils/filesystem.py

@@ -2,8 +2,6 @@
 import os
 import os.path
 import random
-import shutil
-import stat
 import sys
 from contextlib import contextmanager
 from tempfile import NamedTemporaryFile
@@ -54,36 +52,6 @@ def check_path_owner(path):
     return False  # assume we don't own the path
 
 
-def copy2_fixed(src, dest):
-    # type: (str, str) -> None
-    """Wrap shutil.copy2() but map errors copying socket files to
-    SpecialFileError as expected.
-
-    See also https://bugs.python.org/issue37700.
-    """
-    try:
-        shutil.copy2(src, dest)
-    except (OSError, IOError):
-        for f in [src, dest]:
-            try:
-                is_socket_file = is_socket(f)
-            except OSError:
-                # An error has already occurred. Another error here is not
-                # a problem and we can ignore it.
-                pass
-            else:
-                if is_socket_file:
-                    raise shutil.SpecialFileError(
-                        "`{f}` is a socket".format(**locals()))
-
-        raise
-
-
-def is_socket(path):
-    # type: (str) -> bool
-    return stat.S_ISSOCK(os.lstat(path).st_mode)
-
-
 @contextmanager
 def adjacent_tmp_file(path, **kwargs):
     # type: (str, **Any) -> Iterator[NamedTemporaryFileResult]

tests/functional/test_install.py

@@ -2,7 +2,6 @@
 import glob
 import os
 import re
-import shutil
 import ssl
 import sys
 import textwrap
@@ -29,7 +28,6 @@
     skip_if_python2,
     windows_workaround_7667,
 )
-from tests.lib.filesystem import make_socket_file
 from tests.lib.local_repos import local_checkout
 from tests.lib.path import Path
 from tests.lib.server import (
@@ -576,30 +574,6 @@ def test_install_from_local_directory_with_symlinks_to_directories(
     assert egg_info_folder in result.files_created, str(result)
 
 
-@pytest.mark.skipif("sys.platform == 'win32' or sys.version_info < (3,)")
-def test_install_from_local_directory_with_socket_file(script, data, tmpdir):
-    """
-    Test installing from a local directory containing a socket file.
-    """
-    egg_info_file = (
-        script.site_packages /
-        "FSPkg-0.1.dev0-py{pyversion}.egg-info".format(**globals())
-    )
-    package_folder = script.site_packages / "fspkg"
-    to_copy = data.packages.joinpath("FSPkg")
-    to_install = tmpdir.joinpath("src")
-
-    shutil.copytree(to_copy, to_install)
-    # Socket file, should be ignored.
-    socket_file_path = os.path.join(to_install, "example")
-    make_socket_file(socket_file_path)
-
-    result = script.pip("install", "--verbose", to_install)
-    assert package_folder in result.files_created, str(result.stdout)
-    assert egg_info_file in result.files_created, str(result)
-    assert str(socket_file_path) in result.stderr
-
-
 def test_install_from_local_directory_with_no_setup_py(script, data):
     """
     Test installing from a local directory with no 'setup.py'.

tests/unit/test_operations_prepare.py

@@ -10,18 +10,9 @@
 from pip._internal.models.link import Link
 from pip._internal.network.download import Downloader
 from pip._internal.network.session import PipSession
-from pip._internal.operations.prepare import (
-    _copy_source_tree,
-    _download_http_url,
-    unpack_url,
-)
+from pip._internal.operations.prepare import _download_http_url, unpack_url
 from pip._internal.utils.hashes import Hashes
 from pip._internal.utils.urls import path_to_url
-from tests.lib.filesystem import (
-    get_filelist,
-    make_socket_file,
-    make_unreadable_file,
-)
 from tests.lib.path import Path
 from tests.lib.requests_mocks import MockResponse
 
@@ -101,76 +92,6 @@ def clean_project(tmpdir_factory, data):
     return new_project_dir
 
 
-def test_copy_source_tree(clean_project, tmpdir):
-    target = tmpdir.joinpath("target")
-    expected_files = get_filelist(clean_project)
-    assert len(expected_files) == 3
-
-    _copy_source_tree(clean_project, target)
-
-    copied_files = get_filelist(target)
-    assert expected_files == copied_files
-
-
-@pytest.mark.skipif("sys.platform == 'win32' or sys.version_info < (3,)")
-def test_copy_source_tree_with_socket(clean_project, tmpdir, caplog):
-    target = tmpdir.joinpath("target")
-    expected_files = get_filelist(clean_project)
-    socket_path = str(clean_project.joinpath("aaa"))
-    make_socket_file(socket_path)
-
-    _copy_source_tree(clean_project, target)
-
-    copied_files = get_filelist(target)
-    assert expected_files == copied_files
-
-    # Warning should have been logged.
-    assert len(caplog.records) == 1
-    record = caplog.records[0]
-    assert record.levelname == 'WARNING'
-    assert socket_path in record.message
-
-
-@pytest.mark.skipif("sys.platform == 'win32' or sys.version_info < (3,)")
-def test_copy_source_tree_with_socket_fails_with_no_socket_error(
-    clean_project, tmpdir
-):
-    target = tmpdir.joinpath("target")
-    expected_files = get_filelist(clean_project)
-    make_socket_file(clean_project.joinpath("aaa"))
-    unreadable_file = clean_project.joinpath("bbb")
-    make_unreadable_file(unreadable_file)
-
-    with pytest.raises(shutil.Error) as e:
-        _copy_source_tree(clean_project, target)
-
-    errored_files = [err[0] for err in e.value.args[0]]
-    assert len(errored_files) == 1
-    assert unreadable_file in errored_files
-
-    copied_files = get_filelist(target)
-    # All files without errors should have been copied.
-    assert expected_files == copied_files
-
-
-def test_copy_source_tree_with_unreadable_dir_fails(clean_project, tmpdir):
-    target = tmpdir.joinpath("target")
-    expected_files = get_filelist(clean_project)
-    unreadable_file = clean_project.joinpath("bbb")
-    make_unreadable_file(unreadable_file)
-
-    with pytest.raises(shutil.Error) as e:
-        _copy_source_tree(clean_project, target)
-
-    errored_files = [err[0] for err in e.value.args[0]]
-    assert len(errored_files) == 1
-    assert unreadable_file in errored_files
-
-    copied_files = get_filelist(target)
-    # All files without errors should have been copied.
-    assert expected_files == copied_files
-
-
 class Test_unpack_url(object):
 
     def prep(self, tmpdir, data):