diff --git a/samcli/hook_packages/terraform/copy_terraform_built_artifacts.py b/samcli/hook_packages/terraform/copy_terraform_built_artifacts.py index add87f60eb..b6f6062e06 100644 --- a/samcli/hook_packages/terraform/copy_terraform_built_artifacts.py +++ b/samcli/hook_packages/terraform/copy_terraform_built_artifacts.py @@ -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) def cli_exit(): diff --git a/samcli/lib/utils/osutils.py b/samcli/lib/utils/osutils.py index 4887de2eea..74cf67199b 100644 --- a/samcli/lib/utils/osutils.py +++ b/samcli/lib/utils/osutils.py @@ -2,7 +2,6 @@ Common OS utilities """ -import errno import io import logging import os @@ -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 @@ -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) def convert_files_to_unix_line_endings(path: str, target_files: Optional[List[str]] = None) -> None: diff --git a/tests/unit/lib/utils/test_osutils.py b/tests/unit/lib/utils/test_osutils.py index 3d7fe3c47a..7fdcb2d71a 100644 --- a/tests/unit/lib/utils/test_osutils.py +++ b/tests/unit/lib/utils/test_osutils.py @@ -3,7 +3,9 @@ """ import os +import shutil import sys +import tempfile from unittest import TestCase from unittest.mock import patch, Mock @@ -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):