added unit test

Author: jonife

tests/model/eventsources/test_mq_event_source.py

@@ -1,5 +1,6 @@
 from unittest import TestCase
 from samtranslator.model.eventsources.pull import MQ
+from samtranslator.model.exceptions import InvalidEventException
 
 
 class MQEventSource(TestCase):
@@ -40,3 +41,12 @@ def test_get_policy_statements(self):
             }
         ]
         self.assertEqual(policy_statements, expected_policy_document)
+
+    def test_must_validate_secrets_manager_kms_key_id(self):
+        self.mq_event_source.SourceAccessConfigurations = [{"Type": "BASIC_AUTH", "URI": "SECRET_URI"}]
+        self.mq_event_source.Broker = "BROKER_ARN"
+        self.mq_event_source.SecretsManagerKmsKeyId = ["1abc23d4-567f-8ab9-cde0-1fab234c5d67"]
+        error_message = "(None, 'Provided SecretsManagerKmsKeyId should be of type str.')"
+        with self.assertRaises(InvalidEventException) as error:
+            self.mq_event_source.get_policy_statements()
+        self.assertEqual(error_message, str(error.exception))

tests/model/eventsources/test_self_managed_kafka_event_source.py

@@ -294,3 +294,19 @@ def test_must_raise_for_wrong_source_access_configurations_uri(self):
             self.kafka_event_source.SourceAccessConfigurations = config
             with self.assertRaises(InvalidEventException):
                 self.kafka_event_source.get_policy_statements()
+
+    def test_must_validate_secrets_manager_kms_key_id(self):
+        self.kafka_event_source.SourceAccessConfigurations = [
+            {"Type": "SASL_SCRAM_256_AUTH", "URI": "SECRET_URI"},
+            {"Type": "VPC_SUBNET", "URI": "SECRET_URI"},
+            {"Type": "VPC_SECURITY_GROUP", "URI": "SECRET_URI"},
+        ]
+        self.kafka_event_source.Topics = ["Topics"]
+        self.kafka_event_source.KafkaBootstrapServers = ["endpoint1", "endpoint2"]
+        self.kafka_event_source.Enabled = True
+        self.kafka_event_source.BatchSize = 1
+        self.kafka_event_source.SecretsManagerKmsKeyId = ["1abc23d4-567f-8ab9-cde0-1fab234c5d67"]
+        error_message = "(None, 'Provided SecretsManagerKmsKeyId should be of type str.')"
+        with self.assertRaises(InvalidEventException) as error:
+            self.kafka_event_source.get_policy_statements()
+        self.assertEqual(error_message, str(error.exception))