Implement support for sns delete_endpoint()

This commit is contained in:
Michael van Tellingen 2016-05-02 14:34:51 +02:00
parent 9c176671d0
commit d34dd5b08a
No known key found for this signature in database
GPG Key ID: 96B67042ABC44F77
3 changed files with 64 additions and 0 deletions

View File

@ -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():

View File

@ -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 = """<CreateTopicResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<CreateTopicResult>
@ -624,6 +641,13 @@ SET_ENDPOINT_ATTRIBUTES_TEMPLATE = """<SetEndpointAttributesResponse xmlns="http
</ResponseMetadata>
</SetEndpointAttributesResponse>"""
DELETE_ENDPOINT_TEMPLATE = """<DeleteEndpointResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<ResponseMetadata>
<RequestId>c1d2b191-353c-5a5f-8969-fbdd3900afa8</RequestId>
</ResponseMetadata>
</DeleteEndpointResponse>"""
SET_PLATFORM_APPLICATION_ATTRIBUTES_TEMPLATE = """<SetPlatformApplicationAttributesResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/">
<ResponseMetadata>
<RequestId>cf577bcc-b3dc-5463-88f1-3180b9412395</RequestId>

View File

@ -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()