fix: Symlinks to non-existent directories, symlinks re-creating

aws_lambda_builders/actions.py

@@ -156,7 +156,12 @@ def __init__(self, source: Union[str, os.PathLike], dest: Union[str, os.PathLike
         self._dest = dest
 
     def execute(self):
+        source_path = Path(self._source)
         destination_path = Path(self._dest)
+        if not source_path.exists():
+            # Source path doesn't exist, nothing to symlink
+            LOG.debug("Source path %s does not exist, skipping generating symlink", source_path)
+            return
         if not destination_path.exists():
             os.makedirs(destination_path.parent, exist_ok=True)
         utils.create_symlink_or_copy(str(self._source), str(destination_path))

aws_lambda_builders/utils.py

@@ -198,6 +198,10 @@ def create_symlink_or_copy(source: str, destination: str) -> None:
     """Tries to create symlink, if it fails it will copy source into destination"""
     LOG.debug("Creating symlink; source: %s, destination: %s", source, destination)
     try:
+        if Path(destination).exists() and Path(destination).is_symlink():
+            # The symlink is already in place, don't try re-creating it
+            LOG.debug("Symlink between %s and %s already exists, skipping generating symlink", source, destination)
+            return
         os.symlink(Path(source).absolute(), Path(destination).absolute())
     except OSError as ex:
         LOG.warning(

tests/unit/test_actions.py

@@ -12,6 +12,7 @@
     MoveDependenciesAction,
     CleanUpAction,
     DependencyManager,
+    LinkSinglePathAction,
 )
 
 
@@ -262,3 +263,15 @@ def test_excludes_dependencies_from_source(
     @staticmethod
     def _convert_strings_to_paths(source_dest_list):
         return map(lambda item: (Path(item[0]), Path(item[1])), source_dest_list)
+
+
+class TestLinkSinglePathAction(TestCase):
+    @patch("aws_lambda_builders.actions.os.makedirs")
+    @patch("aws_lambda_builders.utils.create_symlink_or_copy")
+    def test_skips_non_existent_source(self, mock_create_symlink_or_copy, mock_makedirs):
+        src = "src/path"
+        dest = "dest/path"
+
+        LinkSinglePathAction(source=src, dest=dest).execute()
+        mock_create_symlink_or_copy.assert_not_called()
+        mock_makedirs.assert_not_called()

tests/unit/test_utils.py

@@ -1,24 +1,27 @@
 import platform
+from pathlib import Path
 
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import patch, Mock, MagicMock
 
 from aws_lambda_builders import utils
 from aws_lambda_builders.utils import decode
 
 
 class Test_create_symlink_or_copy(TestCase):
-    @patch("aws_lambda_builders.utils.Path")
     @patch("aws_lambda_builders.utils.os")
     @patch("aws_lambda_builders.utils.copytree")
-    def test_must_create_symlink_with_absolute_path(self, patched_copy_tree, pathced_os, patched_path):
+    def test_must_create_symlink_with_absolute_path(self, patched_copy_tree, patched_os):
         source_path = "source/path"
         destination_path = "destination/path"
-        utils.create_symlink_or_copy(source_path, destination_path)
 
-        pathced_os.symlink.assert_called_with(
-            patched_path(source_path).absolute(), patched_path(destination_path).absolute()
-        )
+        p = MagicMock()
+        p.return_value = False
+
+        with patch("aws_lambda_builders.utils.Path.is_symlink", p):
+            utils.create_symlink_or_copy(source_path, destination_path)
+
+        patched_os.symlink.assert_called_with(Path(source_path).absolute(), Path(destination_path).absolute())
         patched_copy_tree.assert_not_called()
 
     @patch("aws_lambda_builders.utils.Path")
@@ -34,6 +37,17 @@ def test_must_copy_if_symlink_fails(self, patched_copy_tree, pathced_os, patched
         pathced_os.symlink.assert_called_once()
         patched_copy_tree.assert_called_with(source_path, destination_path)
 
+    @patch("aws_lambda_builders.utils.Path")
+    @patch("aws_lambda_builders.utils.os")
+    @patch("aws_lambda_builders.utils.copytree")
+    def test_must_copy_if_symlink_fails(self, patched_copy_tree, pathced_os, patched_path):
+        source_path = "source/path"
+        destination_path = "destination/path"
+        utils.create_symlink_or_copy(source_path, destination_path)
+
+        pathced_os.symlink.assert_not_called()
+        patched_copy_tree.assert_not_called()
+
 
 class TestDecode(TestCase):
     def test_does_not_crash_non_utf8_encoding(self):