Add Contract Tests for SNS

contract-tests/images/applications/botocore/botocore_server.py

@@ -52,6 +52,8 @@ def do_GET(self):
             self._handle_secretsmanager_request()
         if self.in_path("stepfunctions"):
             self._handle_stepfunctions_request()
+        if self.in_path("sns"):
+            self._handle_sns_request()
 
         self._end_request(self.main_status)
 
@@ -369,6 +371,27 @@ def _handle_stepfunctions_request(self) -> None:
         else:
             set_main_status(404)
 
+    def _handle_sns_request(self) -> None:
+        sns_client = boto3.client("sns", endpoint_url=_AWS_SDK_ENDPOINT, region_name=_AWS_REGION)
+        if self.in_path(_FAULT):
+            set_main_status(500)
+            try:
+                fault_client = boto3.client("sns", endpoint_url=_FAULT_ENDPOINT, region_name=_AWS_REGION)
+                fault_client.meta.events.register(
+                    "before-call.sns.GetTopicAttributes",
+                    lambda **kwargs: inject_500_error("GetTopicAttributes", **kwargs),
+                )
+                fault_client.get_topic_attributes(TopicArn="arn:aws:sns:us-west-2:000000000000:invalid-topic")
+            except Exception as exception:
+                print("Expected exception occurred", exception)
+        elif self.in_path("gettopicattributes/test-topic"):
+            set_main_status(200)
+            sns_client.get_topic_attributes(
+                TopicArn="arn:aws:sns:us-west-2:000000000000:test-topic",
+            )
+        else:
+            set_main_status(404)
+
     def _end_request(self, status_code: int):
         self.send_response_only(status_code)
         self.end_headers()
@@ -557,6 +580,11 @@ def prepare_aws_server() -> None:
                 Name="testSecret", SecretString="secretValue", Description="This is a test secret"
             )
 
+        # Set up SNS so tests can access a topic.
+        sns_client: BaseClient = boto3.client("sns", endpoint_url=_AWS_SDK_ENDPOINT, region_name=_AWS_REGION)
+        create_topic_response = sns_client.create_topic(Name="test-topic")
+        print("Created topic successfully:", create_topic_response)
+
         # Set up Step Functions so tests can access a state machine and activity.
         sfn_client: BaseClient = boto3.client("stepfunctions", endpoint_url=_AWS_SDK_ENDPOINT, region_name=_AWS_REGION)
         sfn_response = sfn_client.list_state_machines()

contract-tests/tests/test/amazon/botocore/botocore_test.py

@@ -47,6 +47,7 @@
 _AWS_SECRET_ARN: str = "aws.secretsmanager.secret.arn"
 _AWS_STATE_MACHINE_ARN: str = "aws.stepfunctions.state_machine.arn"
 _AWS_ACTIVITY_ARN: str = "aws.stepfunctions.activity.arn"
+_AWS_SNS_TOPIC_ARN: str = "aws.sns.topic.arn"
 
 
 # pylint: disable=too-many-public-methods,too-many-lines
@@ -86,7 +87,7 @@ def set_up_dependency_container(cls):
         cls._local_stack: LocalStackContainer = (
             LocalStackContainer(image="localstack/localstack:3.5.0")
             .with_name("localstack")
-            .with_services("s3", "sqs", "dynamodb", "kinesis", "secretsmanager", "iam", "stepfunctions")
+            .with_services("s3", "sqs", "dynamodb", "kinesis", "secretsmanager", "iam", "stepfunctions", "sns")
             .with_env("DEFAULT_REGION", "us-west-2")
             .with_kwargs(network=NETWORK_NAME, networking_config=local_stack_networking_config)
         )
@@ -752,6 +753,45 @@ def test_secretsmanager_fault(self):
             span_name="Secrets Manager.GetSecretValue",
         )
 
+    def test_sns_get_topic_attributes(self):
+        self.do_test_requests(
+            "sns/gettopicattributes/test-topic",
+            "GET",
+            200,
+            0,
+            0,
+            rpc_service="SNS",
+            remote_service="AWS::SNS",
+            remote_operation="GetTopicAttributes",
+            remote_resource_type="AWS::SNS::Topic",
+            remote_resource_identifier="test-topic",
+            cloudformation_primary_identifier="arn:aws:sns:us-west-2:000000000000:test-topic",
+            request_specific_attributes={_AWS_SNS_TOPIC_ARN: "arn:aws:sns:us-west-2:000000000000:test-topic"},
+            span_name="SNS.GetTopicAttributes",
+        )
+
+    # TODO: Add error case for sns - our test setup is not setting the http status code properly
+    # for this resource
+
+    def test_sns_fault(self):
+        self.do_test_requests(
+            "sns/fault",
+            "GET",
+            500,
+            0,
+            1,
+            rpc_service="SNS",
+            remote_service="AWS::SNS",
+            remote_operation="GetTopicAttributes",
+            remote_resource_type="AWS::SNS::Topic",
+            remote_resource_identifier="invalid-topic",
+            cloudformation_primary_identifier="arn:aws:sns:us-west-2:000000000000:invalid-topic",
+            request_specific_attributes={
+                _AWS_SNS_TOPIC_ARN: "arn:aws:sns:us-west-2:000000000000:invalid-topic",
+            },
+            span_name="SNS.GetTopicAttributes",
+        )
+
     def test_stepfunctions_describe_state_machine(self):
         self.do_test_requests(
             "stepfunctions/describestatemachine/my-state-machine",