catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Implement XML responses for SNS (for Boto3)

Browse source
This commit is contained in:
Pior Bastida 2015-08-20 11:12:25 -04:00
commit 2650eab295
7 changed files with 1106 additions and 199 deletions
Download patch file Download diff file Expand all files Collapse all files

254
tests/test_sns/test_application_boto3.py Normal file
View file

@ -0,0 +1,254 @@
from __future__ import unicode_literals
import boto3
from botocore.exceptions import ClientError
from moto import mock_sns
import sure # noqa
@mock_sns
def test_create_platform_application():
conn = boto3.client('sns', region_name='us-east-1')
response = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "platform_principal",
},
)
application_arn = response['PlatformApplicationArn']
application_arn.should.equal('arn:aws:sns:us-east-1:123456789012:app/APNS/my-application')
@mock_sns
def test_get_platform_application_attributes():
conn = boto3.client('sns', region_name='us-east-1')
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "platform_principal",
},
)
arn = platform_application['PlatformApplicationArn']
attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)['Attributes']
attributes.should.equal({
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "platform_principal",
})
@mock_sns
def test_get_missing_platform_application_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn.get_platform_application_attributes.when.called_with(PlatformApplicationArn="a-fake-arn").should.throw(ClientError)
@mock_sns
def test_set_platform_application_attributes():
conn = boto3.client('sns', region_name='us-east-1')
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "platform_principal",
},
)
arn = platform_application['PlatformApplicationArn']
conn.set_platform_application_attributes(PlatformApplicationArn=arn,
Attributes={"PlatformPrincipal": "other"}
)
attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)['Attributes']
attributes.should.equal({
"PlatformCredential": "platform_credential",
"PlatformPrincipal": "other",
})
@mock_sns
def test_list_platform_applications():
conn = boto3.client('sns', region_name='us-east-1')
conn.create_platform_application(
Name="application1",
Platform="APNS",
Attributes={},
)
conn.create_platform_application(
Name="application2",
Platform="APNS",
Attributes={},
)
applications_repsonse = conn.list_platform_applications()
applications = applications_repsonse['PlatformApplications']
applications.should.have.length_of(2)
@mock_sns
def test_delete_platform_application():
conn = boto3.client('sns', region_name='us-east-1')
conn.create_platform_application(
Name="application1",
Platform="APNS",
Attributes={},
)
conn.create_platform_application(
Name="application2",
Platform="APNS",
Attributes={},
)
applications_repsonse = conn.list_platform_applications()
applications = applications_repsonse['PlatformApplications']
applications.should.have.length_of(2)
application_arn = applications[0]['PlatformApplicationArn']
conn.delete_platform_application(PlatformApplicationArn=application_arn)
applications_repsonse = conn.list_platform_applications()
applications = applications_repsonse['PlatformApplications']
applications.should.have.length_of(1)
@mock_sns
def test_create_platform_endpoint():
conn = boto3.client('sns', region_name='us-east-1')
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
)
application_arn = platform_application['PlatformApplicationArn']
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_arn.should.contain("arn:aws:sns:us-east-1:123456789012:endpoint/APNS/my-application/")
@mock_sns
def test_get_list_endpoints_by_platform_application():
conn = boto3.client('sns', region_name='us-east-1')
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
)
application_arn = platform_application['PlatformApplicationArn']
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"CustomUserData": "some data",
},
)
endpoint_arn = endpoint['EndpointArn']
endpoint_list = conn.list_endpoints_by_platform_application(
PlatformApplicationArn=application_arn
)['Endpoints']
endpoint_list.should.have.length_of(1)
endpoint_list[0]['Attributes']['CustomUserData'].should.equal('some data')
endpoint_list[0]['EndpointArn'].should.equal(endpoint_arn)
@mock_sns
def test_get_endpoint_attributes():
conn = boto3.client('sns', region_name='us-east-1')
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
)
application_arn = platform_application['PlatformApplicationArn']
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
"CustomUserData": "some data",
},
)
endpoint_arn = endpoint['EndpointArn']
attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)['Attributes']
attributes.should.equal({
"Enabled": 'false',
"CustomUserData": "some data",
})
@mock_sns
def test_get_missing_endpoint_attributes():
conn = boto3.client('sns', region_name='us-east-1')
conn.get_endpoint_attributes.when.called_with(EndpointArn="a-fake-arn").should.throw(ClientError)
@mock_sns
def test_set_endpoint_attributes():
conn = boto3.client('sns', region_name='us-east-1')
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
)
application_arn = platform_application['PlatformApplicationArn']
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
"CustomUserData": "some data",
},
)
endpoint_arn = endpoint['EndpointArn']
conn.set_endpoint_attributes(EndpointArn=endpoint_arn,
Attributes={"CustomUserData": "other data"}
)
attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)['Attributes']
attributes.should.equal({
"Enabled": 'false',
"CustomUserData": "other data",
})
@mock_sns
def test_publish_to_platform_endpoint():
conn = boto3.client('sns', region_name='us-east-1')
platform_application = conn.create_platform_application(
Name="my-application",
Platform="APNS",
Attributes={},
)
application_arn = platform_application['PlatformApplicationArn']
endpoint = conn.create_platform_endpoint(
PlatformApplicationArn=application_arn,
Token="some_unique_id",
CustomUserData="some user data",
Attributes={
"Enabled": 'false',
},
)
endpoint_arn = endpoint['EndpointArn']
conn.publish(Message="some message", MessageStructure="json", TargetArn=endpoint_arn)

89
tests/test_sns/test_publishing_boto3.py Normal file
View file

@ -0,0 +1,89 @@
from __future__ import unicode_literals
from six.moves.urllib.parse import parse_qs
import boto3
from freezegun import freeze_time
import httpretty
import sure # noqa
from moto import mock_sns, mock_sqs
@mock_sqs
@mock_sns
def test_publish_to_sqs():
conn = boto3.client('sns', region_name='us-east-1')
conn.create_topic(Name="some-topic")
response = conn.list_topics()
topic_arn = response["Topics"][0]['TopicArn']
sqs_conn = boto3.resource('sqs', region_name='us-east-1')
sqs_conn.create_queue(QueueName="test-queue")
conn.subscribe(TopicArn=topic_arn,
Protocol="sqs",
Endpoint="arn:aws:sqs:us-east-1:123456789012:test-queue")
conn.publish(TopicArn=topic_arn, Message="my message")
queue = sqs_conn.get_queue_by_name(QueueName="test-queue")
messages = queue.receive_messages(MaxNumberOfMessages=1)
messages[0].body.should.equal('my message')
@mock_sqs
@mock_sns
def test_publish_to_sqs_in_different_region():
conn = boto3.client('sns', region_name='us-west-1')
conn.create_topic(Name="some-topic")
response = conn.list_topics()
topic_arn = response["Topics"][0]['TopicArn']
sqs_conn = boto3.resource('sqs', region_name='us-west-2')
sqs_conn.create_queue(QueueName="test-queue")
conn.subscribe(TopicArn=topic_arn,
Protocol="sqs",
Endpoint="arn:aws:sqs:us-west-2:123456789012:test-queue")
conn.publish(TopicArn=topic_arn, Message="my message")
queue = sqs_conn.get_queue_by_name(QueueName="test-queue")
messages = queue.receive_messages(MaxNumberOfMessages=1)
messages[0].body.should.equal('my message')
@freeze_time("2013-01-01")
@mock_sns
def test_publish_to_http():
httpretty.HTTPretty.register_uri(
method="POST",
uri="http://example.com/foobar",
)
conn = boto3.client('sns', region_name='us-east-1')
conn.create_topic(Name="some-topic")
response = conn.list_topics()
topic_arn = response["Topics"][0]['TopicArn']
conn.subscribe(TopicArn=topic_arn,
Protocol="http",
Endpoint="http://example.com/foobar")
response = conn.publish(TopicArn=topic_arn, Message="my message", Subject="my subject")
message_id = response['MessageId']
last_request = httpretty.last_request()
last_request.method.should.equal("POST")
parse_qs(last_request.body.decode('utf-8')).should.equal({
"Type": ["Notification"],
"MessageId": [message_id],
"TopicArn": ["arn:aws:sns:{0}:123456789012:some-topic".format(conn._client_config.region_name)],
"Subject": ["my subject"],
"Message": ["my message"],
"Timestamp": ["2013-01-01T00:00:00.000Z"],
"SignatureVersion": ["1"],
"Signature": ["EXAMPLElDMXvB8r9R83tGoNn0ecwd5UjllzsvSvbItzfaMpN2nk5HVSw7XnOn/49IkxDKz8YrlH2qJXj2iZB0Zo2O71c4qQk1fMUDi3LGpij7RCW7AW9vYYsSqIKRnFS94ilu7NFhUzLiieYr4BKHpdTmdD6c0esKEYBpabxDSc="],
"SigningCertURL": ["https://sns.us-east-1.amazonaws.com/SimpleNotificationService-f3ecfb7224c7233fe7bb5f59f96de52f.pem"],
"UnsubscribeURL": ["https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:123456789012:some-topic:2bcfbf39-05c3-41de-beaa-fcfcc21c8f55"],
})

14
tests/test_sns/test_server.py
View file

@ -1,8 +1,5 @@
from __future__ import unicode_literals
import json
import re
import sure # noqa
import moto.server as server
@ -16,9 +13,10 @@ def test_sns_server_get():
backend = server.create_backend_app("sns")
test_client = backend.test_client()
topic_data = test_client.action_json("CreateTopic", Name="test topic")
topic_arn = topic_data["CreateTopicResponse"]["CreateTopicResult"]["TopicArn"]
topics_data = test_client.action_json("ListTopics")
topics_arns = [t["TopicArn"] for t in topics_data["ListTopicsResponse"]["ListTopicsResult"]["Topics"]]
topic_data = test_client.action_data("CreateTopic", Name="test topic")
topic_data.should.contain("CreateTopicResult")
topic_data.should.contain("<TopicArn>arn:aws:sns:us-east-1:123456789012:test topic</TopicArn>")
assert topic_arn in topics_arns
topics_data = test_client.action_data("ListTopics")
topics_data.should.contain("ListTopicsResult")
topic_data.should.contain("<TopicArn>arn:aws:sns:us-east-1:123456789012:test topic</TopicArn>")

90
tests/test_sns/test_subscriptions_boto3.py Normal file
View file

@ -0,0 +1,90 @@
from __future__ import unicode_literals
import boto3
import sure # noqa
from moto import mock_sns
from moto.sns.models import DEFAULT_PAGE_SIZE
@mock_sns
def test_creating_subscription():
conn = boto3.client('sns', region_name='us-east-1')
conn.create_topic(Name="some-topic")
response = conn.list_topics()
topic_arn = response["Topics"][0]['TopicArn']
conn.subscribe(TopicArn=topic_arn,
Protocol="http",
Endpoint="http://example.com/")
subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1)
subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn)
subscription["Protocol"].should.equal("http")
subscription["SubscriptionArn"].should.contain(topic_arn)
subscription["Endpoint"].should.equal("http://example.com/")
# Now unsubscribe the subscription
conn.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])
# And there should be zero subscriptions left
subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(0)
@mock_sns
def test_getting_subscriptions_by_topic():
conn = boto3.client('sns', region_name='us-east-1')
conn.create_topic(Name="topic1")
conn.create_topic(Name="topic2")
response = conn.list_topics()
topics = response["Topics"]
topic1_arn = topics[0]['TopicArn']
topic2_arn = topics[1]['TopicArn']
conn.subscribe(TopicArn=topic1_arn,
Protocol="http",
Endpoint="http://example1.com/")
conn.subscribe(TopicArn=topic2_arn,
Protocol="http",
Endpoint="http://example2.com/")
topic1_subscriptions = conn.list_subscriptions_by_topic(TopicArn=topic1_arn)["Subscriptions"]
topic1_subscriptions.should.have.length_of(1)
topic1_subscriptions[0]['Endpoint'].should.equal("http://example1.com/")
@mock_sns
def test_subscription_paging():
conn = boto3.client('sns', region_name='us-east-1')
conn.create_topic(Name="topic1")
response = conn.list_topics()
topics = response["Topics"]
topic1_arn = topics[0]['TopicArn']
for index in range(DEFAULT_PAGE_SIZE + int(DEFAULT_PAGE_SIZE / 3)):
conn.subscribe(TopicArn=topic1_arn,
Protocol='email',
Endpoint='email_' + str(index) + '@test.com')
all_subscriptions = conn.list_subscriptions()
all_subscriptions["Subscriptions"].should.have.length_of(DEFAULT_PAGE_SIZE)
next_token = all_subscriptions["NextToken"]
next_token.should.equal(str(DEFAULT_PAGE_SIZE))
all_subscriptions = conn.list_subscriptions(NextToken=next_token)
all_subscriptions["Subscriptions"].should.have.length_of(int(DEFAULT_PAGE_SIZE / 3))
all_subscriptions.shouldnt.have("NextToken")
topic1_subscriptions = conn.list_subscriptions_by_topic(TopicArn=topic1_arn)
topic1_subscriptions["Subscriptions"].should.have.length_of(DEFAULT_PAGE_SIZE)
next_token = topic1_subscriptions["NextToken"]
next_token.should.equal(str(DEFAULT_PAGE_SIZE))
topic1_subscriptions = conn.list_subscriptions_by_topic(TopicArn=topic1_arn, NextToken=next_token)
topic1_subscriptions["Subscriptions"].should.have.length_of(int(DEFAULT_PAGE_SIZE / 3))
topic1_subscriptions.shouldnt.have("NextToken")

125
tests/test_sns/test_topics_boto3.py Normal file
View file

@ -0,0 +1,125 @@
from __future__ import unicode_literals
import boto3
import six
import json
import sure # noqa
from botocore.exceptions import ClientError
from moto import mock_sns
from moto.sns.models import DEFAULT_TOPIC_POLICY, DEFAULT_EFFECTIVE_DELIVERY_POLICY, DEFAULT_PAGE_SIZE
@mock_sns
def test_create_and_delete_topic():
conn = boto3.client("sns", region_name="us-east-1")
conn.create_topic(Name="some-topic")
topics_json = conn.list_topics()
topics = topics_json["Topics"]
topics.should.have.length_of(1)
topics[0]['TopicArn'].should.equal(
"arn:aws:sns:{0}:123456789012:some-topic"
.format(conn._client_config.region_name)
)
# Delete the topic
conn.delete_topic(TopicArn=topics[0]['TopicArn'])
# And there should now be 0 topics
topics_json = conn.list_topics()
topics = topics_json["Topics"]
topics.should.have.length_of(0)
@mock_sns
def test_get_missing_topic():
conn = boto3.client("sns", region_name="us-east-1")
conn.get_topic_attributes.when.called_with(TopicArn="a-fake-arn").should.throw(ClientError)
@mock_sns
def test_create_topic_in_multiple_regions():
for region in ['us-west-1', 'us-west-2']:
conn = boto3.client("sns", region_name=region)
conn.create_topic(Name="some-topic")
list(conn.list_topics()["Topics"]).should.have.length_of(1)
@mock_sns
def test_topic_corresponds_to_region():
for region in ['us-east-1', 'us-west-2']:
conn = boto3.client("sns", region_name=region)
conn.create_topic(Name="some-topic")
topics_json = conn.list_topics()
topic_arn = topics_json["Topics"][0]['TopicArn']
topic_arn.should.equal("arn:aws:sns:{0}:123456789012:some-topic".format(region))
@mock_sns
def test_topic_attributes():
conn = boto3.client("sns", region_name="us-east-1")
conn.create_topic(Name="some-topic")
topics_json = conn.list_topics()
topic_arn = topics_json["Topics"][0]['TopicArn']
attributes = conn.get_topic_attributes(TopicArn=topic_arn)['Attributes']
attributes["TopicArn"].should.equal(
"arn:aws:sns:{0}:123456789012:some-topic"
.format(conn._client_config.region_name)
)
attributes["Owner"].should.equal('123456789012')
attributes["Policy"].should.equal(DEFAULT_TOPIC_POLICY)
attributes["DisplayName"].should.equal("")
attributes["SubscriptionsPending"].should.equal('0')
attributes["SubscriptionsConfirmed"].should.equal('0')
attributes["SubscriptionsDeleted"].should.equal('0')
attributes["DeliveryPolicy"].should.equal("")
attributes["EffectiveDeliveryPolicy"].should.equal(DEFAULT_EFFECTIVE_DELIVERY_POLICY)
# boto can't handle prefix-mandatory strings:
# i.e. unicode on Python 2 -- u"foobar"
# and bytes on Python 3 -- b"foobar"
if six.PY2:
policy = json.dumps({b"foo": b"bar"})
displayname = b"My display name"
delivery = json.dumps({b"http": {b"defaultHealthyRetryPolicy": {b"numRetries": 5}}})
else:
policy = json.dumps({u"foo": u"bar"})
displayname = u"My display name"
delivery = json.dumps({u"http": {u"defaultHealthyRetryPolicy": {u"numRetries": 5}}})
conn.set_topic_attributes(TopicArn=topic_arn,
AttributeName="Policy",
AttributeValue=policy)
conn.set_topic_attributes(TopicArn=topic_arn,
AttributeName="DisplayName",
AttributeValue=displayname)
conn.set_topic_attributes(TopicArn=topic_arn,
AttributeName="DeliveryPolicy",
AttributeValue=delivery)
attributes = conn.get_topic_attributes(TopicArn=topic_arn)['Attributes']
attributes["Policy"].should.equal('{"foo": "bar"}')
attributes["DisplayName"].should.equal("My display name")
attributes["DeliveryPolicy"].should.equal('{"http": {"defaultHealthyRetryPolicy": {"numRetries": 5}}}')
@mock_sns
def test_topic_paging():
conn = boto3.client("sns", region_name="us-east-1")
for index in range(DEFAULT_PAGE_SIZE + int(DEFAULT_PAGE_SIZE / 2)):
conn.create_topic(Name="some-topic_" + str(index))
response = conn.list_topics()
topics_list = response["Topics"]
next_token = response["NextToken"]
len(topics_list).should.equal(DEFAULT_PAGE_SIZE)
int(next_token).should.equal(DEFAULT_PAGE_SIZE)
response = conn.list_topics(NextToken=next_token)
topics_list = response["Topics"]
response.shouldnt.have("NextToken")
topics_list.should.have.length_of(int(DEFAULT_PAGE_SIZE / 2))