fix(event_sources): Update CodePipeline event source to include optional encryption_key field and make user_parameters field optional

aws_lambda_powertools/utilities/data_classes/code_pipeline_job_event.py

@@ -16,14 +16,14 @@ def function_name(self) -> str:
         return self["FunctionName"]
 
     @property
-    def user_parameters(self) -> str:
+    def user_parameters(self) -> Optional[str]:
         """User parameters"""
-        return self["UserParameters"]
+        return self.get("UserParameters", None)
 
     @property
-    def decoded_user_parameters(self) -> Dict[str, Any]:
+    def decoded_user_parameters(self) -> Optional[Dict[str, Any]]:
         """Json Decoded user parameters"""
-        if self._json_data is None:
+        if self._json_data is None and self.user_parameters is not None:
             self._json_data = json.loads(self.user_parameters)
         return self._json_data
 
@@ -97,6 +97,16 @@ def expiration_time(self) -> Optional[int]:
         return self.get("expirationTime")
 
 
+class CodePipelineEncryptionKey(DictWrapper):
+    @property
+    def get_id(self) -> str:
+        return self["id"]
+
+    @property
+    def get_type(self) -> str:
+        return self["type"]
+
+
 class CodePipelineData(DictWrapper):
     """CodePipeline Job Data"""
 
@@ -125,6 +135,12 @@ def continuation_token(self) -> Optional[str]:
         """A continuation token if continuing job"""
         return self.get("continuationToken")
 
+    @property
+    def encryption_key(self) -> Optional[CodePipelineEncryptionKey]:
+        """Represents a CodePipeline encryption key"""
+        key_data = self.get("encryptionKey")
+        return CodePipelineEncryptionKey(key_data) if key_data is not None else None
+
 
 class CodePipelineJobEvent(DictWrapper):
     """AWS CodePipeline Job Event
@@ -155,12 +171,12 @@ def data(self) -> CodePipelineData:
         return CodePipelineData(self._job["data"])
 
     @property
-    def user_parameters(self) -> str:
+    def user_parameters(self) -> Optional[str]:
         """Action configuration user parameters"""
         return self.data.action_configuration.configuration.user_parameters
 
     @property
-    def decoded_user_parameters(self) -> Dict[str, Any]:
+    def decoded_user_parameters(self) -> Optional[Dict[str, Any]]:
         """Json Decoded action configuration user parameters"""
         return self.data.action_configuration.configuration.decoded_user_parameters
 

tests/events/codePipelineEventEmptyUserParameters.json

@@ -0,0 +1,32 @@
+{
+    "CodePipeline.job": {
+        "id": "11111111-abcd-1111-abcd-111111abcdef",
+        "accountId": "111111111111",
+        "data": {
+            "actionConfiguration": {
+                "configuration": {
+                    "FunctionName": "MyLambdaFunctionForAWSCodePipeline"
+                }
+            },
+            "inputArtifacts": [
+                {
+                    "name": "ArtifactName",
+                    "revision": null,
+                    "location": {
+                        "type": "S3",
+                        "s3Location": {
+                            "bucketName": "the name of the bucket configured as the pipeline artifact store in Amazon S3, for example codepipeline-us-east-2-1234567890",
+                            "objectKey": "the name of the application, for example CodePipelineDemoApplication.zip"
+                        }
+                    }
+                }
+            ],
+            "outputArtifacts": [],
+            "artifactCredentials": {
+                "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
+                "secretAccessKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+                "sessionToken": "MIICiTCCAfICCQD6m7oRw0uXOjANBgkqhkiG9w0BAQUFADCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wHhcNMTEwNDI1MjA0NTIxWhcNMTIwNDI0MjA0NTIxWjCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAMaK0dn+a4GmWIWJ21uUSfwfEvySWtC2XADZ4nB+BLYgVIk60CpiwsZ3G93vUEIO3IyNoH/f0wYK8m9TrDHudUZg3qX4waLG5M43q7Wgc/MbQITxOUSQv7c7ugFFDzQGBzZswY6786m86gpEIbb3OhjZnzcvQAaRHhdlQWIMm2nrAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAtCu4nUhVVxYUntneD9+h8Mg9q6q+auNKyExzyLwaxlAoo7TJHidbtS4J5iNmZgXL0FkbFFBjvSfpJIlJ00zbhNYS5f6GuoEDmFJl0ZxBHjJnyp378OD8uTs7fLvjx79LjSTbNYiytVbZPQUQ5Yaxu2jXnimvw3rrszlaEXAMPLE="
+            }
+        }
+    }
+}

tests/events/codePipelineEventWithEncryptionKey.json

@@ -0,0 +1,38 @@
+{
+    "CodePipeline.job": {
+        "id": "11111111-abcd-1111-abcd-111111abcdef",
+        "accountId": "111111111111",
+        "data": {
+            "actionConfiguration": {
+                "configuration": {
+                    "FunctionName": "MyLambdaFunctionForAWSCodePipeline",
+                    "UserParameters": "some-input-such-as-a-URL"
+                }
+            },
+            "inputArtifacts": [
+                {
+                    "name": "ArtifactName",
+                    "revision": null,
+                    "location": {
+                        "type": "S3",
+                        "s3Location": {
+                            "bucketName": "the name of the bucket configured as the pipeline artifact store in Amazon S3, for example codepipeline-us-east-2-1234567890",
+                            "objectKey": "the name of the application, for example CodePipelineDemoApplication.zip"
+                        }
+                    }
+                }
+            ],
+            "outputArtifacts": [],
+            "artifactCredentials": {
+                "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
+                "secretAccessKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+                "sessionToken": "MIICiTCCAfICCQD6m7oRw0uXOjANBgkqhkiG9w0BAQUFADCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wHhcNMTEwNDI1MjA0NTIxWhcNMTIwNDI0MjA0NTIxWjCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAMaK0dn+a4GmWIWJ21uUSfwfEvySWtC2XADZ4nB+BLYgVIk60CpiwsZ3G93vUEIO3IyNoH/f0wYK8m9TrDHudUZg3qX4waLG5M43q7Wgc/MbQITxOUSQv7c7ugFFDzQGBzZswY6786m86gpEIbb3OhjZnzcvQAaRHhdlQWIMm2nrAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAtCu4nUhVVxYUntneD9+h8Mg9q6q+auNKyExzyLwaxlAoo7TJHidbtS4J5iNmZgXL0FkbFFBjvSfpJIlJ00zbhNYS5f6GuoEDmFJl0ZxBHjJnyp378OD8uTs7fLvjx79LjSTbNYiytVbZPQUQ5Yaxu2jXnimvw3rrszlaEXAMPLE="
+            },
+            "continuationToken": "A continuation token if continuing job",
+            "encryptionKey": {
+                "id": "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab",
+                "type": "KMS"
+              }
+        }
+    }
+}

tests/functional/test_data_classes.py

@@ -1544,6 +1544,62 @@ def test_code_pipeline_event():
     assert artifact_credentials_dict["sessionToken"] == artifact_credentials.session_token
 
 
+def test_code_pipeline_event_with_encryption_keys():
+    event = CodePipelineJobEvent(load_event("codePipelineEventWithEncryptionKey.json"))
+
+    job = event["CodePipeline.job"]
+    assert job["id"] == event.get_id
+    assert job["accountId"] == event.account_id
+
+    data = event.data
+    assert isinstance(data, CodePipelineData)
+    assert job["data"]["continuationToken"] == data.continuation_token
+    configuration = data.action_configuration.configuration
+    assert "MyLambdaFunctionForAWSCodePipeline" == configuration.function_name
+    assert event.user_parameters == configuration.user_parameters
+    assert "some-input-such-as-a-URL" == configuration.user_parameters
+
+    input_artifacts = data.input_artifacts
+    assert len(input_artifacts) == 1
+    assert "ArtifactName" == input_artifacts[0].name
+    assert input_artifacts[0].revision is None
+    assert "S3" == input_artifacts[0].location.get_type
+
+    output_artifacts = data.output_artifacts
+    assert len(output_artifacts) == 0
+
+    artifact_credentials = data.artifact_credentials
+    artifact_credentials_dict = event["CodePipeline.job"]["data"]["artifactCredentials"]
+    assert artifact_credentials_dict["accessKeyId"] == artifact_credentials.access_key_id
+    assert artifact_credentials_dict["secretAccessKey"] == artifact_credentials.secret_access_key
+    assert artifact_credentials_dict["sessionToken"] == artifact_credentials.session_token
+
+    encryption_key = data.encryption_key
+    assert "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab" == encryption_key.get_id
+    assert "KMS" == encryption_key.get_type
+
+
+def test_code_pipeline_event_missing_user_parameters():
+    event = CodePipelineJobEvent(load_event("codePipelineEventEmptyUserParameters.json"))
+
+    assert event.data.continuation_token is None
+    configuration = event.data.action_configuration.configuration
+    decoded_params = configuration.decoded_user_parameters
+    assert decoded_params == event.decoded_user_parameters
+    assert decoded_params is None
+    assert configuration.decoded_user_parameters is None
+
+
+def test_code_pipeline_event_non_json_user_parameters():
+    event = CodePipelineJobEvent(load_event("codePipelineEvent.json"))
+
+    configuration = event.data.action_configuration.configuration
+    assert configuration.user_parameters is not None
+
+    with pytest.raises(json.decoder.JSONDecodeError):
+        configuration.decoded_user_parameters
+
+
 def test_code_pipeline_event_decoded_data():
     event = CodePipelineJobEvent(load_event("codePipelineEventData.json"))