feat: Add --resource option input validation for sync --code

samcli/commands/sync/command.py

@@ -34,7 +34,7 @@
 from samcli.lib.telemetry.metric import track_command, track_template_warnings
 from samcli.lib.warnings.sam_cli_warning import CodeDeployWarning, CodeDeployConditionWarning
 from samcli.commands.build.command import _get_mode_value_from_envvar
-from samcli.lib.sync.sync_flow_factory import SyncFlowFactory
+from samcli.lib.sync.sync_flow_factory import SyncCodeResources, SyncFlowFactory
 from samcli.lib.sync.sync_flow_executor import SyncFlowExecutor
 from samcli.lib.providers.sam_stack_provider import SamLocalStackProvider
 from samcli.lib.providers.provider import (
@@ -122,13 +122,14 @@
 @click.option(
     "--resource",
     multiple=True,
-    help="Sync code for all types of the resource.",
+    type=click.Choice(SyncCodeResources.values(), case_sensitive=True),
+    help=f"Sync code for all resources of the given resource type. Accepted values are {SyncCodeResources.values()}",
 )
 @click.option(
     "--dependency-layer/--no-dependency-layer",
     default=True,
     is_flag=True,
-    help="This option separates the dependencies of individual function into another layer, for speeding up the sync"
+    help="This option separates the dependencies of individual function into another layer, for speeding up the sync."
     "process",
 )
 @stack_name_option(required=True)  # pylint: disable=E1120

samcli/lib/deploy/deployer.py

@@ -24,7 +24,6 @@
 from typing import Dict, List, Optional
 
 import botocore
-import click
 
 from samcli.lib.deploy.utils import DeployColor
 from samcli.commands.deploy.exceptions import (
@@ -597,7 +596,7 @@ def sync(
                 self.wait_for_execute(stack_name, "CREATE", False)
                 msg = "\nStack creation succeeded. Sync infra completed.\n"
 
-            click.secho(msg, fg="green")
+            LOG.info(self._colored.green(msg))
 
             return result
         except botocore.exceptions.ClientError as ex:

samcli/lib/sync/sync_flow_factory.py

@@ -38,6 +38,35 @@
 LOG = logging.getLogger(__name__)
 
 
+class SyncCodeResources:
+    """
+    A class that records the supported resource types that can perform sync --code
+    """
+
+    _accepted_resources = [
+        AWS_SERVERLESS_FUNCTION,
+        AWS_LAMBDA_FUNCTION,
+        AWS_SERVERLESS_LAYERVERSION,
+        AWS_LAMBDA_LAYERVERSION,
+        AWS_SERVERLESS_API,
+        AWS_APIGATEWAY_RESTAPI,
+        AWS_SERVERLESS_HTTPAPI,
+        AWS_APIGATEWAY_V2_API,
+        AWS_SERVERLESS_STATEMACHINE,
+        AWS_STEPFUNCTIONS_STATEMACHINE,
+    ]
+
+    @classmethod
+    def values(cls) -> List[str]:
+        """
+        A class getter to retrieve the accepted resource list
+
+        Returns: List[str]
+            The accepted resources list
+        """
+        return cls._accepted_resources
+
+
 class SyncFlowFactory(ResourceTypeBasedFactory[SyncFlow]):  # pylint: disable=E1136
     """Factory class for SyncFlow
     Creates appropriate SyncFlow types based on stack resource types

tests/integration/sync/test_sync_code.py

@@ -100,7 +100,7 @@ def test_sync_code_function(self):
             TestSyncCodeBase.temp_dir.joinpath("function"),
         )
 
-        self.stack_resources = self._get_stacks(TestSyncCode.stack_name)
+        self.stack_resources = self._get_stacks(TestSyncCodeBase.stack_name)
         if self.dependency_layer:
             # Test update manifest
             layer_contents = self.get_dependency_layer_contents_from_arn(self.stack_resources, "python", 1)
@@ -266,6 +266,26 @@ def test_sync_code_state_machine(self):
         state_machine = self.stack_resources.get(AWS_STEPFUNCTIONS_STATEMACHINE)[0]
         self.assertEqual(self._get_sfn_response(state_machine), '"World 2"')
 
+    def test_sync_code_invalid_resource_type(self):
+        sync_command_list = self.get_sync_command_list(
+            template_file=TestSyncCodeBase.template_path,
+            code=True,
+            watch=False,
+            resource_list=["AWS::Serverless::InvalidResource"],
+            stack_name=TestSyncCodeBase.stack_name,
+            parameter_overrides="Parameter=Clarity",
+            image_repository=self.ecr_repo_name,
+            s3_prefix=self.s3_prefix,
+            kms_key_id=self.kms_key,
+            tags="integ=true clarity=yes foo_bar=baz",
+        )
+        sync_process_execute = run_command_with_input(sync_command_list, "y\n".encode())
+        self.assertEqual(sync_process_execute.process.returncode, 2)
+        self.assertIn(
+            "Invalid value for '--resource': invalid choice: AWS::Serverless::InvalidResource",
+            str(sync_process_execute.stderr),
+        )
+
 
 @skipIf(SKIP_SYNC_TESTS, "Skip sync tests in CI/CD only")
 class TestSyncCodeDotnetFunctionTemplate(TestSyncCodeBase):
@@ -315,13 +335,13 @@ class TestSyncCodeNodejsFunctionTemplate(TestSyncCodeBase):
     folder = "code"
 
     def test_sync_code_nodejs_function(self):
-        shutil.rmtree(Path(TestSyncCode.temp_dir).joinpath("nodejs_function"), ignore_errors=True)
+        shutil.rmtree(Path(TestSyncCodeBase.temp_dir).joinpath("nodejs_function"), ignore_errors=True)
         shutil.copytree(
             self.test_data_path.joinpath("code").joinpath("after").joinpath("nodejs_function"),
-            Path(TestSyncCode.temp_dir).joinpath("nodejs_function"),
+            Path(TestSyncCodeBase.temp_dir).joinpath("nodejs_function"),
         )
 
-        self.stack_resources = self._get_stacks(TestSyncCode.stack_name)
+        self.stack_resources = self._get_stacks(TestSyncCodeBase.stack_name)
         if self.dependency_layer:
             # Test update manifest
             layer_contents = self.get_dependency_layer_contents_from_arn(
@@ -336,7 +356,7 @@ def test_sync_code_nodejs_function(self):
             watch=False,
             resource_list=["AWS::Serverless::Function"],
             dependency_layer=self.dependency_layer,
-            stack_name=TestSyncCode.stack_name,
+            stack_name=TestSyncCodeBase.stack_name,
             parameter_overrides="Parameter=Clarity",
             image_repository=self.ecr_repo_name,
             s3_prefix=self.s3_prefix,
@@ -347,7 +367,7 @@ def test_sync_code_nodejs_function(self):
         self.assertEqual(sync_process_execute.process.returncode, 0)
 
         # CFN Api call here to collect all the stack resources
-        self.stack_resources = self._get_stacks(TestSyncCode.stack_name)
+        self.stack_resources = self._get_stacks(TestSyncCodeBase.stack_name)
         # Lambda Api call here, which tests both the python function and the layer
         lambda_functions = self.stack_resources.get(AWS_LAMBDA_FUNCTION)
         for lambda_function in lambda_functions:
@@ -585,7 +605,6 @@ def test_sync_code_nested_getattr_layer(self):
             s3_prefix=self.s3_prefix,
             kms_key_id=self.kms_key,
             tags="integ=true clarity=yes foo_bar=baz",
-            debug=True,
         )
         sync_process_execute = run_command_with_input(sync_command_list, "y\n".encode())
         self.assertEqual(sync_process_execute.process.returncode, 0)

tests/unit/commands/sync/test_command.py

@@ -478,7 +478,7 @@ def test_execute_code_sync_single_type_resource(
     ):
 
         resource_identifier_strings = ["Function1", "Function2"]
-        resource_types = ["Type1"]
+        resource_types = ["AWS::Serverless::Function"]
         sync_flows = [MagicMock(), MagicMock(), MagicMock()]
         sync_flow_factory_mock.return_value.create_sync_flow.side_effect = sync_flows
         get_unique_resource_ids_mock.return_value = {
@@ -508,7 +508,7 @@ def test_execute_code_sync_single_type_resource(
         self.assertEqual(sync_flow_executor_mock.return_value.add_sync_flow.call_count, 3)
 
         get_unique_resource_ids_mock.assert_called_once_with(
-            get_stacks_mock.return_value[0], resource_identifier_strings, ["Type1"]
+            get_stacks_mock.return_value[0], resource_identifier_strings, ["AWS::Serverless::Function"]
         )
 
     @patch("samcli.commands.sync.command.click")
@@ -525,7 +525,7 @@ def test_execute_code_sync_multiple_type_resource(
         click_mock,
     ):
         resource_identifier_strings = ["Function1", "Function2"]
-        resource_types = ["Type1", "Type2"]
+        resource_types = ["AWS::Serverless::Function", "AWS::Serverless::LayerVersion"]
         sync_flows = [MagicMock(), MagicMock(), MagicMock(), MagicMock()]
         sync_flow_factory_mock.return_value.create_sync_flow.side_effect = sync_flows
         get_unique_resource_ids_mock.return_value = {
@@ -559,7 +559,9 @@ def test_execute_code_sync_multiple_type_resource(
         self.assertEqual(sync_flow_executor_mock.return_value.add_sync_flow.call_count, 4)
 
         get_unique_resource_ids_mock.assert_any_call(
-            get_stacks_mock.return_value[0], resource_identifier_strings, ["Type1", "Type2"]
+            get_stacks_mock.return_value[0],
+            resource_identifier_strings,
+            ["AWS::Serverless::Function", "AWS::Serverless::LayerVersion"],
         )
 
     @patch("samcli.commands.sync.command.click")

tests/unit/lib/sync/test_sync_flow_factory.py

@@ -1,8 +1,20 @@
 from unittest import TestCase
 from unittest.mock import MagicMock, patch, Mock
 
-from samcli.lib.sync.sync_flow_factory import SyncFlowFactory
+from samcli.lib.sync.sync_flow_factory import SyncCodeResources, SyncFlowFactory
 from samcli.lib.utils.cloudformation import CloudFormationResourceSummary
+from samcli.lib.utils.resources import (
+    AWS_SERVERLESS_FUNCTION,
+    AWS_LAMBDA_FUNCTION,
+    AWS_SERVERLESS_LAYERVERSION,
+    AWS_LAMBDA_LAYERVERSION,
+    AWS_SERVERLESS_API,
+    AWS_APIGATEWAY_RESTAPI,
+    AWS_SERVERLESS_HTTPAPI,
+    AWS_APIGATEWAY_V2_API,
+    AWS_SERVERLESS_STATEMACHINE,
+    AWS_STEPFUNCTIONS_STATEMACHINE,
+)
 
 
 class TestSyncFlowFactory(TestCase):
@@ -161,3 +173,21 @@ def test_create_none_generator_sync_flow(self, get_resource_by_id_mock):
         factory._get_generator_function = get_generator_function_mock
 
         self.assertIsNone(factory.create_sync_flow(resource_identifier))
+
+
+class TestSyncCodeResources(TestCase):
+    def test_values(self):
+        output = SyncCodeResources.values()
+        expected = [
+            AWS_SERVERLESS_FUNCTION,
+            AWS_LAMBDA_FUNCTION,
+            AWS_SERVERLESS_LAYERVERSION,
+            AWS_LAMBDA_LAYERVERSION,
+            AWS_SERVERLESS_API,
+            AWS_APIGATEWAY_RESTAPI,
+            AWS_SERVERLESS_HTTPAPI,
+            AWS_APIGATEWAY_V2_API,
+            AWS_SERVERLESS_STATEMACHINE,
+            AWS_STEPFUNCTIONS_STATEMACHINE,
+        ]
+        self.assertEqual(expected, output)