Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

SNS:Topic: Add option to update/delete via CF (#4392)

Browse Source
This commit is contained in:
Bert Blommers 2021-10-11 21:56:39 +00:00 committed by GitHub
parent 0d0354438e
commit dac33becbf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 149 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
moto/sns/models.py
View File

@ -110,6 +110,33 @@ class Topic(CloudFormationModel):
)
return topic
@classmethod
def update_from_cloudformation_json(
cls, original_resource, new_resource_name, cloudformation_json, region_name
):
cls.delete_from_cloudformation_json(
original_resource.name, cloudformation_json, region_name
)
return cls.create_from_cloudformation_json(
new_resource_name, cloudformation_json, region_name
)
@classmethod
def delete_from_cloudformation_json(
cls, resource_name, cloudformation_json, region_name
):
sns_backend = sns_backends[region_name]
properties = cloudformation_json["Properties"]
topic_name = properties.get(cls.cloudformation_name_type()) or resource_name
topic_arn = make_arn_for_topic(
DEFAULT_ACCOUNT_ID, topic_name, sns_backend.region_name
)
subscriptions, _ = sns_backend.list_subscriptions(topic_arn)
for subscription in subscriptions:
sns_backend.unsubscribe(subscription.arn)
sns_backend.delete_topic(topic_arn)
def _create_default_topic_policy(self, region_name, account_id, name):
return {
"Version": "2008-10-17",

1
tests/test_cloudformation/test_cloudformation_stack_integration.py
View File

@ -56,7 +56,6 @@ from tests import EXAMPLE_AMI_ID, EXAMPLE_AMI_ID2
from tests.test_cloudformation.fixtures import (
ec2_classic_eip,
fn_join,
rds_mysql_with_db_parameter_group,
rds_mysql_with_read_replica,
redshift,
route53_ec2_instance_with_public_ip,

122
tests/test_sns/test_sns_cloudformation.py
View File

@ -51,3 +51,125 @@ def test_sns_topic():
topic_name_output["OutputValue"].should.equal("my_topics")
topic_arn_output = [x for x in stack["Outputs"] if x["OutputKey"] == "topic_arn"][0]
topic_arn_output["OutputValue"].should.equal(topic_arn)
@mock_cloudformation
@mock_sns
def test_sns_update_topic():
dummy_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"MySNSTopic": {
"Type": "AWS::SNS::Topic",
"Properties": {
"Subscription": [
{"Endpoint": "https://example.com", "Protocol": "https"}
],
"TopicName": "my_topics",
},
}
},
"Outputs": {
"topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
"topic_arn": {"Value": {"Ref": "MySNSTopic"}},
},
}
sns_template_json = json.dumps(dummy_template)
cf = boto3.client("cloudformation", region_name="us-west-1")
cf.create_stack(StackName="test_stack", TemplateBody=sns_template_json)
sns = boto3.client("sns", region_name="us-west-1")
topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1)
dummy_template["Resources"]["MySNSTopic"]["Properties"]["Subscription"][0][
"Endpoint"
] = "https://example-updated.com"
sns_template_json = json.dumps(dummy_template)
cf.update_stack(StackName="test_stack", TemplateBody=sns_template_json)
topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1)
topic_arn = topics[0]["TopicArn"]
topic_arn.should.contain("my_topics")
subscriptions = sns.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1)
subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn)
subscription["Protocol"].should.equal("https")
subscription["SubscriptionArn"].should.contain(topic_arn)
subscription["Endpoint"].should.equal("https://example-updated.com")
@mock_cloudformation
@mock_sns
def test_sns_update_remove_topic():
dummy_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"MySNSTopic": {
"Type": "AWS::SNS::Topic",
"Properties": {
"Subscription": [
{"Endpoint": "https://example.com", "Protocol": "https"}
],
"TopicName": "my_topics",
},
}
},
"Outputs": {
"topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
"topic_arn": {"Value": {"Ref": "MySNSTopic"}},
},
}
sns_template_json = json.dumps(dummy_template)
cf = boto3.client("cloudformation", region_name="us-west-1")
cf.create_stack(StackName="test_stack", TemplateBody=sns_template_json)
sns = boto3.client("sns", region_name="us-west-1")
topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1)
dummy_template["Resources"].pop("MySNSTopic")
dummy_template.pop("Outputs")
sns_template_json = json.dumps(dummy_template)
cf.update_stack(StackName="test_stack", TemplateBody=sns_template_json)
topics = sns.list_topics()["Topics"]
topics.should.have.length_of(0)
@mock_cloudformation
@mock_sns
def test_sns_delete_topic():
dummy_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"MySNSTopic": {
"Type": "AWS::SNS::Topic",
"Properties": {
"Subscription": [
{"Endpoint": "https://example.com", "Protocol": "https"}
],
"TopicName": "my_topics",
},
}
},
"Outputs": {
"topic_name": {"Value": {"Fn::GetAtt": ["MySNSTopic", "TopicName"]}},
"topic_arn": {"Value": {"Ref": "MySNSTopic"}},
},
}
sns_template_json = json.dumps(dummy_template)
cf = boto3.client("cloudformation", region_name="us-west-1")
cf.create_stack(StackName="test_stack", TemplateBody=sns_template_json)
sns = boto3.client("sns", region_name="us-west-1")
topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1)
cf.delete_stack(StackName="test_stack")
topics = sns.list_topics()["Topics"]
topics.should.have.length_of(0)