| #!/usr/bin/env python |
| # Copyright (c) 2013 Amazon.com, Inc. or its affiliates. All Rights Reserved |
| # |
| # Permission is hereby granted, free of charge, to any person obtaining a |
| # copy of this software and associated documentation files (the |
| # "Software"), to deal in the Software without restriction, including |
| # without limitation the rights to use, copy, modify, merge, publish, dis- |
| # tribute, sublicense, and/or sell copies of the Software, and to permit |
| # persons to whom the Software is furnished to do so, subject to the fol- |
| # lowing conditions: |
| # |
| # The above copyright notice and this permission notice shall be included |
| # in all copies or substantial portions of the Software. |
| # |
| # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
| # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- |
| # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT |
| # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, |
| # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
| # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS |
| # IN THE SOFTWARE. |
| # |
| import json |
| from tests.unit import unittest |
| from tests.unit import AWSMockServiceTestCase |
| from mock import Mock |
| |
| from boto.sns.connection import SNSConnection |
| |
| QUEUE_POLICY = { |
| u'Policy': |
| (u'{"Version":"2008-10-17","Id":"arn:aws:sqs:us-east-1:' |
| 'idnum:testqueuepolicy/SQSDefaultPolicy","Statement":' |
| '[{"Sid":"sidnum","Effect":"Allow","Principal":{"AWS":"*"},' |
| '"Action":"SQS:GetQueueUrl","Resource":' |
| '"arn:aws:sqs:us-east-1:idnum:testqueuepolicy"}]}')} |
| |
| |
| class TestSNSConnection(AWSMockServiceTestCase): |
| connection_class = SNSConnection |
| |
| def setUp(self): |
| super(TestSNSConnection, self).setUp() |
| |
| def default_body(self): |
| return b"{}" |
| |
| def test_sqs_with_existing_policy(self): |
| self.set_http_response(status_code=200) |
| |
| queue = Mock() |
| queue.get_attributes.return_value = QUEUE_POLICY |
| queue.arn = 'arn:aws:sqs:us-east-1:idnum:queuename' |
| |
| self.service_connection.subscribe_sqs_queue('topic_arn', queue) |
| self.assert_request_parameters({ |
| 'Action': 'Subscribe', |
| 'ContentType': 'JSON', |
| 'Endpoint': 'arn:aws:sqs:us-east-1:idnum:queuename', |
| 'Protocol': 'sqs', |
| 'TopicArn': 'topic_arn', |
| 'Version': '2010-03-31', |
| }, ignore_params_values=[]) |
| |
| # Verify that the queue policy was properly updated. |
| actual_policy = json.loads(queue.set_attribute.call_args[0][1]) |
| self.assertEqual(actual_policy['Version'], '2008-10-17') |
| # A new statement should be appended to the end of the statement list. |
| self.assertEqual(len(actual_policy['Statement']), 2) |
| self.assertEqual(actual_policy['Statement'][1]['Action'], |
| 'SQS:SendMessage') |
| |
| def test_sqs_with_no_previous_policy(self): |
| self.set_http_response(status_code=200) |
| |
| queue = Mock() |
| queue.get_attributes.return_value = {} |
| queue.arn = 'arn:aws:sqs:us-east-1:idnum:queuename' |
| |
| self.service_connection.subscribe_sqs_queue('topic_arn', queue) |
| self.assert_request_parameters({ |
| 'Action': 'Subscribe', |
| 'ContentType': 'JSON', |
| 'Endpoint': 'arn:aws:sqs:us-east-1:idnum:queuename', |
| 'Protocol': 'sqs', |
| 'TopicArn': 'topic_arn', |
| 'Version': '2010-03-31', |
| }, ignore_params_values=[]) |
| actual_policy = json.loads(queue.set_attribute.call_args[0][1]) |
| # Only a single statement should be part of the policy. |
| self.assertEqual(len(actual_policy['Statement']), 1) |
| |
| def test_publish_with_positional_args(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.publish('topic', 'message', 'subject') |
| self.assert_request_parameters({ |
| 'Action': 'Publish', |
| 'TopicArn': 'topic', |
| 'Subject': 'subject', |
| 'Message': 'message', |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_publish_with_kwargs(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.publish(topic='topic', |
| message='message', |
| subject='subject') |
| self.assert_request_parameters({ |
| 'Action': 'Publish', |
| 'TopicArn': 'topic', |
| 'Subject': 'subject', |
| 'Message': 'message', |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_publish_with_target_arn(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.publish(target_arn='target_arn', |
| message='message', |
| subject='subject') |
| self.assert_request_parameters({ |
| 'Action': 'Publish', |
| 'TargetArn': 'target_arn', |
| 'Subject': 'subject', |
| 'Message': 'message', |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_create_platform_application(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.create_platform_application( |
| name='MyApp', |
| platform='APNS', |
| attributes={ |
| 'PlatformPrincipal': 'a ssl certificate', |
| 'PlatformCredential': 'a private key' |
| } |
| ) |
| self.assert_request_parameters({ |
| 'Action': 'CreatePlatformApplication', |
| 'Name': 'MyApp', |
| 'Platform': 'APNS', |
| 'Attributes.entry.1.key': 'PlatformCredential', |
| 'Attributes.entry.1.value': 'a private key', |
| 'Attributes.entry.2.key': 'PlatformPrincipal', |
| 'Attributes.entry.2.value': 'a ssl certificate', |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_set_platform_application_attributes(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.set_platform_application_attributes( |
| platform_application_arn='arn:myapp', |
| attributes={'PlatformPrincipal': 'a ssl certificate', |
| 'PlatformCredential': 'a private key'}) |
| self.assert_request_parameters({ |
| 'Action': 'SetPlatformApplicationAttributes', |
| 'PlatformApplicationArn': 'arn:myapp', |
| 'Attributes.entry.1.key': 'PlatformCredential', |
| 'Attributes.entry.1.value': 'a private key', |
| 'Attributes.entry.2.key': 'PlatformPrincipal', |
| 'Attributes.entry.2.value': 'a ssl certificate', |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_create_platform_endpoint(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.create_platform_endpoint( |
| platform_application_arn='arn:myapp', |
| token='abcde12345', |
| custom_user_data='john', |
| attributes={'Enabled': False}) |
| self.assert_request_parameters({ |
| 'Action': 'CreatePlatformEndpoint', |
| 'PlatformApplicationArn': 'arn:myapp', |
| 'Token': 'abcde12345', |
| 'CustomUserData': 'john', |
| 'Attributes.entry.1.key': 'Enabled', |
| 'Attributes.entry.1.value': False, |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_set_endpoint_attributes(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.set_endpoint_attributes( |
| endpoint_arn='arn:myendpoint', |
| attributes={'CustomUserData': 'john', |
| 'Enabled': False}) |
| self.assert_request_parameters({ |
| 'Action': 'SetEndpointAttributes', |
| 'EndpointArn': 'arn:myendpoint', |
| 'Attributes.entry.1.key': 'CustomUserData', |
| 'Attributes.entry.1.value': 'john', |
| 'Attributes.entry.2.key': 'Enabled', |
| 'Attributes.entry.2.value': False, |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_message_is_required(self): |
| self.set_http_response(status_code=200) |
| |
| with self.assertRaises(TypeError): |
| self.service_connection.publish(topic='topic', subject='subject') |
| |
| def test_publish_with_json(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.publish( |
| message=json.dumps({ |
| 'default': 'Ignored.', |
| 'GCM': { |
| 'data': 'goes here', |
| } |
| }), |
| message_structure='json', |
| subject='subject', |
| target_arn='target_arn' |
| ) |
| self.assert_request_parameters({ |
| 'Action': 'Publish', |
| 'TargetArn': 'target_arn', |
| 'Subject': 'subject', |
| 'MessageStructure': 'json', |
| }, ignore_params_values=['Version', 'ContentType', 'Message']) |
| self.assertDictEqual( |
| json.loads(self.actual_request.params["Message"]), |
| {"default": "Ignored.", "GCM": {"data": "goes here"}}) |
| |
| def test_publish_with_utf8_message(self): |
| self.set_http_response(status_code=200) |
| subject = message = u'We \u2665 utf-8'.encode('utf-8') |
| self.service_connection.publish('topic', message, subject) |
| self.assert_request_parameters({ |
| 'Action': 'Publish', |
| 'TopicArn': 'topic', |
| 'Subject': subject, |
| 'Message': message, |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| def test_publish_with_attributes(self): |
| self.set_http_response(status_code=200) |
| |
| self.service_connection.publish( |
| message=json.dumps({ |
| 'default': 'Ignored.', |
| 'GCM': { |
| 'data': 'goes here', |
| } |
| }, sort_keys=True), |
| message_structure='json', |
| subject='subject', |
| target_arn='target_arn', |
| message_attributes={ |
| 'name1': { |
| 'data_type': 'Number', |
| 'string_value': '42' |
| }, |
| 'name2': { |
| 'data_type': 'String', |
| 'string_value': 'Bob' |
| }, |
| }, |
| ) |
| self.assert_request_parameters({ |
| 'Action': 'Publish', |
| 'TargetArn': 'target_arn', |
| 'Subject': 'subject', |
| 'Message': '{"GCM": {"data": "goes here"}, "default": "Ignored."}', |
| 'MessageStructure': 'json', |
| 'MessageAttributes.entry.1.Name': 'name1', |
| 'MessageAttributes.entry.1.Value.DataType': 'Number', |
| 'MessageAttributes.entry.1.Value.StringValue': '42', |
| 'MessageAttributes.entry.2.Name': 'name2', |
| 'MessageAttributes.entry.2.Value.DataType': 'String', |
| 'MessageAttributes.entry.2.Value.StringValue': 'Bob', |
| }, ignore_params_values=['Version', 'ContentType']) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
