Support SNS subscription attributes (#1087)

* remove code for local test

* Add SNS set_subscription_attributes and get_subscription_attributes
This commit is contained in:
Toshiya Kawasaki 2017-09-08 03:19:34 +09:00 committed by Jack Danger
commit 0c3708a8e7
5 changed files with 142 additions and 3 deletions

View file

@ -13,7 +13,7 @@ from moto.core import BaseBackend, BaseModel
from moto.core.utils import iso_8601_datetime_with_milliseconds
from moto.sqs import sqs_backends
from .exceptions import (
SNSNotFoundError, DuplicateSnsEndpointError, SnsEndpointDisabled
SNSNotFoundError, DuplicateSnsEndpointError, SnsEndpointDisabled, SNSInvalidParameter
)
from .utils import make_arn_for_topic, make_arn_for_subscription
@ -76,6 +76,7 @@ class Subscription(BaseModel):
self.endpoint = endpoint
self.protocol = protocol
self.arn = make_arn_for_subscription(self.topic.arn)
self.attributes = {}
def publish(self, message, message_id):
if self.protocol == 'sqs':
@ -301,6 +302,26 @@ class SNSBackend(BaseBackend):
raise SNSNotFoundError(
"Endpoint with arn {0} not found".format(arn))
def get_subscription_attributes(self, arn):
_subscription = [_ for _ in self.subscriptions.values() if _.arn == arn]
if not _subscription:
raise SNSNotFoundError("Subscription with arn {0} not found".format(arn))
subscription = _subscription[0]
return subscription.attributes
def set_subscription_attributes(self, arn, name, value):
if name not in ['RawMessageDelivery', 'DeliveryPolicy']:
raise SNSInvalidParameter('AttributeName')
# TODO: should do validation
_subscription = [_ for _ in self.subscriptions.values() if _.arn == arn]
if not _subscription:
raise SNSNotFoundError("Subscription with arn {0} not found".format(arn))
subscription = _subscription[0]
subscription.attributes[name] = value
sns_backends = {}
for region in boto.sns.regions():