diff --git a/contract-tests/images/applications/botocore/botocore_server.py b/contract-tests/images/applications/botocore/botocore_server.py index d1736d56c..0575f4d88 100644 --- a/contract-tests/images/applications/botocore/botocore_server.py +++ b/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() diff --git a/contract-tests/tests/test/amazon/botocore/botocore_test.py b/contract-tests/tests/test/amazon/botocore/botocore_test.py index b2821a8b6..c8a346f5e 100644 --- a/contract-tests/tests/test/amazon/botocore/botocore_test.py +++ b/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",