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

Add proper SNS paging. Closes #144, #145.

Browse source
This commit is contained in:
Steve Pulec 2014-11-29 22:37:48 -05:00
commit d24099c401
4 changed files with 90 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

37
tests/test_sns/test_subscriptions.py
View file

@ -4,6 +4,7 @@ import boto
import sure # noqa
from moto import mock_sns
from moto.sns.models import DEFAULT_PAGE_SIZE
@mock_sns
@ -48,3 +49,39 @@ def test_getting_subscriptions_by_topic():
topic1_subscriptions = conn.get_all_subscriptions_by_topic(topic1_arn)["ListSubscriptionsByTopicResponse"]["ListSubscriptionsByTopicResult"]["Subscriptions"]
topic1_subscriptions.should.have.length_of(1)
topic1_subscriptions[0]['Endpoint'].should.equal("http://example1.com/")
@mock_sns
def test_subscription_paging():
conn = boto.connect_sns()
conn.create_topic("topic1")
conn.create_topic("topic2")
topics_json = conn.get_all_topics()
topics = topics_json["ListTopicsResponse"]["ListTopicsResult"]["Topics"]
topic1_arn = topics[0]['TopicArn']
topic2_arn = topics[1]['TopicArn']
for index in range(DEFAULT_PAGE_SIZE + int(DEFAULT_PAGE_SIZE / 3)):
conn.subscribe(topic1_arn, 'email', 'email_' + str(index) + '@test.com')
conn.subscribe(topic2_arn, 'email', 'email_' + str(index) + '@test.com')
all_subscriptions = conn.get_all_subscriptions()
all_subscriptions["ListSubscriptionsResponse"]["ListSubscriptionsResult"]["Subscriptions"].should.have.length_of(DEFAULT_PAGE_SIZE)
next_token = all_subscriptions["ListSubscriptionsResponse"]["ListSubscriptionsResult"]["NextToken"]
next_token.should.equal(DEFAULT_PAGE_SIZE)
all_subscriptions = conn.get_all_subscriptions(next_token=next_token * 2)
all_subscriptions["ListSubscriptionsResponse"]["ListSubscriptionsResult"]["Subscriptions"].should.have.length_of(int(DEFAULT_PAGE_SIZE * 2 / 3))
next_token = all_subscriptions["ListSubscriptionsResponse"]["ListSubscriptionsResult"]["NextToken"]
next_token.should.equal(None)
topic1_subscriptions = conn.get_all_subscriptions_by_topic(topic1_arn)
topic1_subscriptions["ListSubscriptionsByTopicResponse"]["ListSubscriptionsByTopicResult"]["Subscriptions"].should.have.length_of(DEFAULT_PAGE_SIZE)
next_token = topic1_subscriptions["ListSubscriptionsByTopicResponse"]["ListSubscriptionsByTopicResult"]["NextToken"]
next_token.should.equal(DEFAULT_PAGE_SIZE)
topic1_subscriptions = conn.get_all_subscriptions_by_topic(topic1_arn, next_token=next_token)
topic1_subscriptions["ListSubscriptionsByTopicResponse"]["ListSubscriptionsByTopicResult"]["Subscriptions"].should.have.length_of(int(DEFAULT_PAGE_SIZE / 3))
next_token = topic1_subscriptions["ListSubscriptionsByTopicResponse"]["ListSubscriptionsByTopicResult"]["NextToken"]
next_token.should.equal(None)

23
tests/test_sns/test_topics.py
View file

@ -5,7 +5,7 @@ import six
import sure # noqa
from moto import mock_sns
from moto.sns.models import DEFAULT_TOPIC_POLICY, DEFAULT_EFFECTIVE_DELIVERY_POLICY
from moto.sns.models import DEFAULT_TOPIC_POLICY, DEFAULT_EFFECTIVE_DELIVERY_POLICY, DEFAULT_PAGE_SIZE
@mock_sns
@ -77,3 +77,24 @@ def test_topic_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 = boto.connect_sns()
for index in range(DEFAULT_PAGE_SIZE + int(DEFAULT_PAGE_SIZE / 2)):
conn.create_topic("some-topic_" + str(index))
topics_json = conn.get_all_topics()
topics_list = topics_json["ListTopicsResponse"]["ListTopicsResult"]["Topics"]
next_token = topics_json["ListTopicsResponse"]["ListTopicsResult"]["NextToken"]
len(topics_list).should.equal(DEFAULT_PAGE_SIZE)
next_token.should.equal(DEFAULT_PAGE_SIZE)
topics_json = conn.get_all_topics(next_token=next_token)
topics_list = topics_json["ListTopicsResponse"]["ListTopicsResult"]["Topics"]
next_token = topics_json["ListTopicsResponse"]["ListTopicsResult"]["NextToken"]
topics_list.should.have.length_of(int(DEFAULT_PAGE_SIZE / 2))
next_token.should.equal(None)