Fetch the input artifact from event without using name

Author: Simon Meng

src/s3helper.py

@@ -8,8 +8,6 @@
 
 LOG = lambdalogging.getLogger(__name__)
 
-PACKAGED_TEMPLATE = 'BuildArtifact'
-
 
 def get_input_artifact(event):
     """Get the packaged SAM template from CodePipeline S3 Bucket.
@@ -30,7 +28,8 @@ def get_input_artifact(event):
     )
 
     input_artifacts = event['CodePipeline.job']['data']['inputArtifacts']
-    artifact_to_fetch = _find_artifact_in_list(input_artifacts)
+    _validate_input_artifacts(input_artifacts)
+    artifact_to_fetch = input_artifacts[0]
 
     artifact_s3_location = artifact_to_fetch['location']['s3Location']
     bucket = artifact_s3_location['bucketName']
@@ -43,25 +42,14 @@ def get_input_artifact(event):
     return _unzip_as_string(zipped_content_as_bytes)
 
 
-def _find_artifact_in_list(input_artifacts):
-    """Find the artifact named 'PackagedTemplate' in the list of artifacts.
+def _validate_input_artifacts(input_artifacts):
+    """Validate the length of input artifacts list is 1.
 
     Arguments:
         input_artifacts {dict list} -- list of input artifacts
-        https://docs.aws.amazon.com/codepipeline/latest/APIReference/API_Artifact.html
-
-    Raises:
-        RuntimeError -- Raise error if not able to find the artifact
-
-    Returns:
-        dict -- artifact to fetch from S3
-
     """
-    for artifact in input_artifacts:
-        if artifact['name'] == PACKAGED_TEMPLATE:
-            return artifact
-
-    raise RuntimeError('Unable to find the artifact with name ' + PACKAGED_TEMPLATE)
+    if len(input_artifacts) != 1:
+        raise RuntimeError('You should only have one input artifact. Please check the setting for the action.')
 
 
 def _unzip_as_string(data):

test/unit/test_constants.py

@@ -24,17 +24,6 @@ def generate_pipeline_event(input_artifacts):
 
 mock_codepipeline_event = generate_pipeline_event(
     [
-        {
-            "location": {
-                "s3Location": {
-                    "bucketName": "sample-pipeline-artifact-store-bucket",
-                    "objectKey": "sample-artifact-key1"
-                },
-                "type": "S3"
-            },
-            "revision": None,
-            "name": "NotPackagedTemplate"
-        },
         {
             "location": {
                 "s3Location": {
@@ -48,7 +37,7 @@ def generate_pipeline_event(input_artifacts):
         }
     ]
 )
-mock_codepipeline_event_no_artifact_found = generate_pipeline_event(
+mock_codepipeline_event_more_than_one_input_artifacts = generate_pipeline_event(
     [
         {
             "location": {
@@ -60,6 +49,18 @@ def generate_pipeline_event(input_artifacts):
             },
             "revision": None,
             "name": "NotPackagedTemplate"
+        },
+        {
+            "location": {
+                "s3Location": {
+                    "bucketName": "sample-pipeline-artifact-store-bucket",
+                    "objectKey": "sample-artifact-key"
+                },
+                "type": "S3"
+            },
+            "revision": None,
+            "name": "BuildArtifact"
         }
     ]
 )
+mock_codepipeline_event_no_input_artifacts = generate_pipeline_event([])

test/unit/test_handler.py

@@ -5,8 +5,7 @@
 from serverlessrepo.exceptions import S3PermissionsRequired
 
 import handler
-from test_constants import mock_codepipeline_event, mock_codepipeline_event_no_artifact_found
-from s3helper import PACKAGED_TEMPLATE
+from test_constants import mock_codepipeline_event, mock_codepipeline_event_more_than_one_input_artifacts
 
 
 @pytest.fixture
@@ -42,13 +41,13 @@ def test_publish(mock_s3helper, mock_codepipelinehelper, mock_serverlessrepo):
     )
 
 
-def test_publish_unable_to_find_artifact(mock_s3helper, mock_codepipelinehelper, mock_serverlessrepo):
-    exception_thrown = RuntimeError('Unable to find the artifact with name ' + PACKAGED_TEMPLATE)
+def test_publish_more_than_one_input_artifacts(mock_s3helper, mock_codepipelinehelper, mock_serverlessrepo):
+    exception_thrown = RuntimeError('You should only have one input artifact. Please check the setting for the action.')
     mock_s3helper.get_input_artifact.side_effect = exception_thrown
 
-    handler.publish(mock_codepipeline_event_no_artifact_found, None)
+    handler.publish(mock_codepipeline_event_more_than_one_input_artifacts, None)
 
-    mock_s3helper.get_input_artifact.assert_called_once_with(mock_codepipeline_event_no_artifact_found)
+    mock_s3helper.get_input_artifact.assert_called_once_with(mock_codepipeline_event_more_than_one_input_artifacts)
     mock_serverlessrepo.assert_not_called()
     mock_codepipelinehelper.put_job_failure.assert_called_once_with(
         'sample-codepipeline-job-id',

test/unit/test_s3helper.py

@@ -5,7 +5,11 @@
 from botocore.exceptions import ClientError
 
 import s3helper
-from test_constants import mock_codepipeline_event, mock_codepipeline_event_no_artifact_found
+from test_constants import (
+    mock_codepipeline_event,
+    mock_codepipeline_event_more_than_one_input_artifacts,
+    mock_codepipeline_event_no_input_artifacts
+)
 
 
 @pytest.fixture
@@ -39,12 +43,29 @@ def test_get_input_artifact(mock_boto3, mock_zipfile):
     )
 
 
-def test_get_input_artifact_unable_to_find_artifact(mock_boto3, mock_zipfile):
+def test_get_input_artifact_more_than_one_input_artifacts(mock_boto3, mock_zipfile):
     mock_s3 = MagicMock()
     mock_boto3.client.return_value = mock_s3
 
-    with pytest.raises(RuntimeError, match='Unable to find the artifact with name ' + s3helper.PACKAGED_TEMPLATE):
-        s3helper.get_input_artifact(mock_codepipeline_event_no_artifact_found)
+    with pytest.raises(
+        RuntimeError,
+        match='You should only have one input artifact. Please check the setting for the action.'
+    ):
+        s3helper.get_input_artifact(mock_codepipeline_event_more_than_one_input_artifacts)
+
+    mock_s3.get_object.assert_not_called()
+    mock_zipfile.assert_not_called()
+
+
+def test_get_input_artifact_no_input_artifacts(mock_boto3, mock_zipfile):
+    mock_s3 = MagicMock()
+    mock_boto3.client.return_value = mock_s3
+
+    with pytest.raises(
+        RuntimeError,
+        match='You should only have one input artifact. Please check the setting for the action.'
+    ):
+        s3helper.get_input_artifact(mock_codepipeline_event_no_input_artifacts)
 
     mock_s3.get_object.assert_not_called()
     mock_zipfile.assert_not_called()