From d34dd5b08ae017828bda835e9d749bd77011b17b Mon Sep 17 00:00:00 2001 From: Michael van Tellingen Date: Mon, 2 May 2016 14:34:51 +0200 Subject: [PATCH] Implement support for sns delete_endpoint() --- moto/sns/models.py | 6 ++++++ moto/sns/responses.py | 24 +++++++++++++++++++++ tests/test_sns/test_application.py | 34 ++++++++++++++++++++++++++++++ 3 files changed, 64 insertions(+) diff --git a/moto/sns/models.py b/moto/sns/models.py index fe9683d7a..1369ed070 100644 --- a/moto/sns/models.py +++ b/moto/sns/models.py @@ -252,6 +252,12 @@ class SNSBackend(BaseBackend): endpoint.attributes.update(attributes) return endpoint + def delete_endpoint(self, arn): + try: + del self.platform_endpoints[arn] + except KeyError: + raise SNSNotFoundError("Endpoint with arn {0} not found".format(arn)) + sns_backends = {} for region in boto.sns.regions(): diff --git a/moto/sns/responses.py b/moto/sns/responses.py index 96efff767..9a20dbcb5 100644 --- a/moto/sns/responses.py +++ b/moto/sns/responses.py @@ -421,6 +421,23 @@ class SNSResponse(BaseResponse): template = self.response_template(SET_ENDPOINT_ATTRIBUTES_TEMPLATE) return template.render() + def delete_endpoint(self): + arn = self._get_param('EndpointArn') + self.backend.delete_endpoint(arn) + + if self.request_json: + return json.dumps({ + "DeleteEndpointResponse": { + "ResponseMetadata": { + "RequestId": "384bc68d-3775-12df-8963-01868b7c937f", + } + } + }) + + template = self.response_template(DELETE_ENDPOINT_TEMPLATE) + return template.render() + + CREATE_TOPIC_TEMPLATE = """ @@ -624,6 +641,13 @@ SET_ENDPOINT_ATTRIBUTES_TEMPLATE = """ + + c1d2b191-353c-5a5f-8969-fbdd3900afa8 + +""" + + SET_PLATFORM_APPLICATION_ATTRIBUTES_TEMPLATE = """ cf577bcc-b3dc-5463-88f1-3180b9412395 diff --git a/tests/test_sns/test_application.py b/tests/test_sns/test_application.py index 87d0316d6..790e3f8ff 100644 --- a/tests/test_sns/test_application.py +++ b/tests/test_sns/test_application.py @@ -222,6 +222,40 @@ def test_set_endpoint_attributes(): }) +@mock_sns +def test_delete_endpoint(): + conn = boto.connect_sns() + platform_application = conn.create_platform_application( + name="my-application", + platform="APNS", + ) + application_arn = platform_application['CreatePlatformApplicationResponse']['CreatePlatformApplicationResult']['PlatformApplicationArn'] + + endpoint = conn.create_platform_endpoint( + platform_application_arn=application_arn, + token="some_unique_id", + custom_user_data="some user data", + attributes={ + "Enabled": False, + "CustomUserData": "some data", + }, + ) + endpoint_arn = endpoint['CreatePlatformEndpointResponse']['CreatePlatformEndpointResult']['EndpointArn'] + + endpoint_list = conn.list_endpoints_by_platform_application( + platform_application_arn=application_arn + )['ListEndpointsByPlatformApplicationResponse']['ListEndpointsByPlatformApplicationResult']['Endpoints'] + + endpoint_list.should.have.length_of(1) + + conn.delete_endpoint(endpoint_arn) + + endpoint_list = conn.list_endpoints_by_platform_application( + platform_application_arn=application_arn + )['ListEndpointsByPlatformApplicationResponse']['ListEndpointsByPlatformApplicationResult']['Endpoints'] + endpoint_list.should.have.length_of(0) + + @mock_sns def test_publish_to_platform_endpoint(): conn = boto.connect_sns()