Adds support to delete old versions when successful deployment

samcli/lib/sync/flows/alias_version_sync_flow.py

@@ -23,12 +23,14 @@ class AliasVersionSyncFlow(SyncFlow):
 
     _function_identifier: str
     _alias_name: str
+    _delete_old_alias: bool
     _lambda_client: Any
 
     def __init__(
         self,
         function_identifier: str,
         alias_name: str,
+        delete_old_alias: bool,
         build_context: "BuildContext",
         deploy_context: "DeployContext",
         sync_context: "SyncContext",
@@ -64,6 +66,7 @@ def __init__(
         self._function_identifier = function_identifier
         self._alias_name = alias_name
         self._lambda_client = None
+        self._delete_old_alias = delete_old_alias
 
     @property
     def sync_state_identifier(self) -> str:
@@ -90,13 +93,18 @@ def compare_remote(self) -> bool:
 
     def sync(self) -> None:
         function_physical_id = self.get_physical_id(self._function_identifier)
+        current_alias_version = self._get_version_alias_if_exists()
         version = self._lambda_client.publish_version(FunctionName=function_physical_id).get("Version")
         self._local_sha = str_checksum(str(version), hashlib.sha256())
         LOG.debug("%sCreated new function version: %s", self.log_prefix, version)
         if version:
             self._lambda_client.update_alias(
                 FunctionName=function_physical_id, Name=self._alias_name, FunctionVersion=version
             )
+            if self._delete_old_alias and current_alias_version:
+                formatted_function_name_version = "{}:{}".format(function_physical_id, current_alias_version)
+                LOG.debug("%Deleting Old Version Alias: %s", self.log_prefix, current_alias_version)
+                self._lambda_client.delete_function(FunctionName=formatted_function_name_version)
 
     def gather_dependencies(self) -> List[SyncFlow]:
         return []
@@ -107,3 +115,13 @@ def _get_resource_api_calls(self) -> List[ResourceAPICall]:
     def _equality_keys(self) -> Any:
         """Combination of function identifier and alias name can used to identify each unique SyncFlow"""
         return self._function_identifier, self._alias_name
+
+    def _get_version_alias_if_exists(self) -> Optional[str]:
+        try:
+            return str(
+                self._lambda_client.get_alias(
+                    FunctionName=self.get_physical_id(self._function_identifier), Name=self._alias_name
+                ).get("FunctionVersion")
+            )
+        except self._lambda_client.exceptions.ResourceNotFoundException:
+            return None

samcli/lib/sync/flows/function_sync_flow.py

@@ -109,11 +109,13 @@ def gather_dependencies(self) -> List[SyncFlow]:
             raise FunctionNotFound(f"Unable to find function {self._function_identifier}")
 
         auto_publish_alias_name = function_resource.get("Properties", dict()).get("AutoPublishAlias", None)
+        auto_delete_old_alias = function_resource.get("Metadata", dict()).get("AutoDeleteOldAlias", False)
         if auto_publish_alias_name:
             sync_flows.append(
                 AliasVersionSyncFlow(
                     self._function_identifier,
                     auto_publish_alias_name,
+                    auto_delete_old_alias,
                     self._build_context,
                     self._deploy_context,
                     self._sync_context,

tests/unit/lib/sync/flows/test_alias_version_sync_flow.py

@@ -1,19 +1,17 @@
-import os
 import hashlib
-
-from samcli.lib.sync.sync_flow import SyncFlow
 from unittest import TestCase
-from unittest.mock import ANY, MagicMock, call, mock_open, patch
+from unittest.mock import MagicMock, patch
 
 from samcli.lib.sync.flows.alias_version_sync_flow import AliasVersionSyncFlow
 from samcli.lib.utils.hash import str_checksum
 
 
 class TestAliasVersionSyncFlow(TestCase):
-    def create_sync_flow(self):
+    def create_sync_flow(self, delete_old_alias=False):
         sync_flow = AliasVersionSyncFlow(
             "Function1",
             "Alias1",
+            delete_old_alias,
             build_context=MagicMock(),
             deploy_context=MagicMock(),
             sync_context=MagicMock(),
@@ -76,3 +74,25 @@ def test_local_sha(self, session_mock):
 
         sync_flow.sync()
         self.assertEqual(sync_flow._local_sha, str_checksum("2", hashlib.sha256()))
+
+    @patch("samcli.lib.sync.sync_flow.Session")
+    def test_delete_old_alias(self, session_mock):
+        sync_flow = self.create_sync_flow(True)
+
+        sync_flow.get_physical_id = MagicMock()
+        sync_flow.get_physical_id.return_value = "PhysicalFunction1"
+
+        sync_flow.set_up()
+
+        sync_flow._lambda_client.get_alias.return_value = {"FunctionVersion": "1"}
+        sync_flow._lambda_client.publish_version.return_value = {"Version": "2"}
+
+        sync_flow.sync()
+
+        sync_flow._lambda_client.publish_version.assert_called_once_with(FunctionName="PhysicalFunction1")
+        sync_flow._lambda_client.update_alias.assert_called_once_with(
+            FunctionName="PhysicalFunction1", Name="Alias1", FunctionVersion="2"
+        )
+        sync_flow._lambda_client.delete_function.assert_called_once_with(
+            FunctionName="{}:{}".format("PhysicalFunction1", "1")
+        )