Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -195,20 +195,11 @@ def search(self, data):


def copytree(src, dst):
"""Modified copytree method
Note: before python3.8 there is no `dir_exists_ok` argument, therefore
this function explicitly creates one if it does not exist.
"""Copy a directory tree from src to dst, merging into dst if it already exists.

Symbolic links are preserved as symbolic links in the destination.
"""
if not os.path.exists(dst):
os.makedirs(dst)
for item in os.listdir(src):
src_item = os.path.join(src, item)
dst_item = os.path.join(dst, item)
if os.path.isdir(src_item):
# recursively call itself.
copytree(src_item, dst_item)
else:
shutil.copy2(src_item, dst_item)
shutil.copytree(src, dst, symlinks=True, dirs_exist_ok=True)
Comment thread
bnusunny marked this conversation as resolved.


def cli_exit():
Expand Down
62 changes: 14 additions & 48 deletions samcli/lib/utils/osutils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
Common OS utilities
"""

import errno
import io
import logging
import os
Expand Down Expand Up @@ -117,7 +116,7 @@ def stderr() -> io.TextIOWrapper:
def remove(path):
if path:
try:
os.remove(path)
Path(path).unlink(missing_ok=True)
except OSError:
pass

Expand All @@ -134,60 +133,27 @@ def tempfile_platform_independent():
remove(_tempfile.name)


# NOTE: Py3.8 or higher has a ``dir_exist_ok=True`` parameter to provide this functionality.
# This method can be removed if we stop supporting Py37
def copytree(source, destination, ignore=None):
"""
Similar to shutil.copytree except that it removes the limitation that the destination directory should
be present.
Recursively copy a directory tree from source to destination, allowing the destination
to already exist. Symbolic links in the source tree are preserved as symbolic links
in the destination.

Delegates to ``shutil.copytree`` with ``dirs_exist_ok=True`` and ``symlinks=True``.

:type source: str
:param source:
Path to the source folder to copy
Path to the source directory to copy
:type destination: str
:param destination:
Path to destination folder
:type ignore: function
Path to the destination directory. Will be created if it does not exist;
existing files will be overwritten by corresponding files from source.
:type ignore: callable, optional
:param ignore:
A function that returns a set of file names to ignore, given a list of available file names. Similar to the
``ignore`` property of ``shutils.copytree`` method
A callable that receives the directory being visited and a list of its contents,
and returns a set of names to ignore. See ``shutil.ignore_patterns`` for a helper.
"""

if not os.path.exists(destination):
os.makedirs(destination)

try:
# Let's try to copy the directory metadata from source to destination
shutil.copystat(source, destination)
except OSError as ex:
# Can't copy file access times in Windows
LOG.debug("Unable to copy file access times from %s to %s", source, destination, exc_info=ex)

names = os.listdir(source)
if ignore is not None:
ignored_names = ignore(source, names)
else:
ignored_names = set()

for name in names:
# Skip ignored names
if name in ignored_names:
continue

new_source = os.path.join(source, name)
new_destination = os.path.join(destination, name)

if os.path.isdir(new_source):
copytree(new_source, new_destination, ignore=ignore)
else:
try:
shutil.copy2(new_source, new_destination, follow_symlinks=False)
except OSError as e:
if e.errno != errno.EINVAL:
raise e

# Symlinks do not get copied for Windows using shutil.copy2, which is why
# they are handled separately here.
create_symlink_or_copy(new_source, new_destination)
shutil.copytree(source, destination, symlinks=True, ignore=ignore, dirs_exist_ok=True)
Comment thread
bnusunny marked this conversation as resolved.


def convert_files_to_unix_line_endings(path: str, target_files: Optional[List[str]] = None) -> None:
Expand Down
102 changes: 56 additions & 46 deletions tests/unit/lib/utils/test_osutils.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@
"""

import os
import shutil
import sys
import tempfile

from unittest import TestCase
from unittest.mock import patch, Mock
Expand Down Expand Up @@ -91,66 +93,74 @@ def test_must_delete_if_path_exist(self, patched_rmtree, patched_path):


class Test_copytree(TestCase):
@patch("samcli.lib.utils.osutils.Path")
@patch("samcli.lib.utils.osutils.os")
@patch("samcli.lib.utils.osutils.shutil.copy2")
def test_must_copytree(self, patched_copy2, patched_os, patched_path):
@patch("samcli.lib.utils.osutils.shutil.copytree")
def test_must_copytree(self, patched_copytree):
source_path = "mock-source/path"
destination_path = "mock-destination/path"
mock_path_obj = Mock()
patched_path.exists.return_value = True
patched_os.path.return_value = mock_path_obj

patched_os.path.join.side_effect = [source_path, destination_path]
patched_os.path.isdir.return_value = False
patched_os.listdir.return_value = ["mock-source-file1"]
osutils.copytree(source_path, destination_path)

patched_os.path.join.assert_called()
patched_copy2.assert_called_with(source_path, destination_path, follow_symlinks=False)
patched_copytree.assert_called_once_with(
source_path, destination_path, symlinks=True, ignore=None, dirs_exist_ok=True
)

@patch("samcli.lib.utils.osutils.Path")
@patch("samcli.lib.utils.osutils.os")
@patch("samcli.lib.utils.osutils.shutil.copy2")
def test_copytree_throws_oserror_path_exists(self, patched_copy2, patched_os, patched_path):
@patch("samcli.lib.utils.osutils.shutil.copytree")
def test_copytree_with_ignore(self, patched_copytree):
source_path = "mock-source/path"
destination_path = "mock-destination/path"
mock_path_obj = Mock()
patched_path.exists.return_value = True
patched_os.path.return_value = mock_path_obj
patched_copy2.side_effect = OSError("mock-os-error")
mock_ignore = Mock()

osutils.copytree(source_path, destination_path, ignore=mock_ignore)

patched_copytree.assert_called_once_with(
source_path, destination_path, symlinks=True, ignore=mock_ignore, dirs_exist_ok=True
)

@patch("samcli.lib.utils.osutils.shutil.copytree")
def test_copytree_throws_oserror(self, patched_copytree):
patched_copytree.side_effect = OSError("mock-os-error")

patched_os.path.join.side_effect = [source_path, destination_path]
patched_os.path.isdir.return_value = False
patched_os.listdir.return_value = ["mock-source-file1"]
with self.assertRaises(OSError):
osutils.copytree(source_path, destination_path)
osutils.copytree("mock-source/path", "mock-destination/path")

patched_os.path.join.assert_called()
patched_copy2.assert_called_with(source_path, destination_path, follow_symlinks=False)

@patch("samcli.lib.utils.osutils.create_symlink_or_copy")
@patch("samcli.lib.utils.osutils.Path")
@patch("samcli.lib.utils.osutils.os")
@patch("samcli.lib.utils.osutils.shutil.copy2")
def test_copytree_symlink_copy_error_handling(
self, patched_copy2, patched_os, patched_path, patched_create_symlink_or_copy
):
source_path = "mock-source/path"
destination_path = "mock-destination/path"
mock_path_obj = Mock()
patched_path.exists.return_value = True
patched_os.path.return_value = mock_path_obj
patched_copy2.side_effect = OSError(22, "mock-os-error")
class Test_copytree_symlinks(TestCase):
"""Integration-style tests verifying symlinks=True preserves symlinks on disk."""

patched_os.path.join.side_effect = [source_path, destination_path]
patched_os.path.isdir.return_value = False
patched_os.listdir.return_value = ["mock-source-file1"]
osutils.copytree(source_path, destination_path)
def setUp(self):
self.src = tempfile.mkdtemp()
self.dst = tempfile.mkdtemp()
# clean dst so copytree creates it
shutil.rmtree(self.dst)

def tearDown(self):
shutil.rmtree(self.src, ignore_errors=True)
shutil.rmtree(self.dst, ignore_errors=True)

def test_preserves_file_symlink(self):
real_file = os.path.join(self.src, "real.txt")
with open(real_file, "w") as f:
f.write("content")
os.symlink("real.txt", os.path.join(self.src, "link.txt"))

osutils.copytree(self.src, self.dst)

link = os.path.join(self.dst, "link.txt")
self.assertTrue(os.path.islink(link))
self.assertEqual(os.readlink(link), "real.txt")

def test_preserves_directory_symlink(self):
subdir = os.path.join(self.src, "subdir")
os.makedirs(subdir)
with open(os.path.join(subdir, "file.txt"), "w") as f:
f.write("content")
os.symlink("subdir", os.path.join(self.src, "link_dir"))

osutils.copytree(self.src, self.dst)

patched_os.path.join.assert_called()
patched_copy2.assert_called_with(source_path, destination_path, follow_symlinks=False)
patched_create_symlink_or_copy.assert_called_with(source_path, destination_path)
link = os.path.join(self.dst, "link_dir")
self.assertTrue(os.path.islink(link))
self.assertEqual(os.readlink(link), "subdir")


class Test_create_symlink_or_copy(TestCase):
Expand Down
Loading