Fix : SNS get attributes on FIFO topic (#4444)

This commit is contained in:
Ranjithkumar Krishnan 2021-10-20 14:11:40 +02:00 committed by GitHub
parent 46131e0340
commit 4d4e82a7f7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 65 additions and 0 deletions

View File

@ -53,6 +53,8 @@ class Topic(CloudFormationModel):
sns_backend.region_name, self.account_id, name sns_backend.region_name, self.account_id, name
) )
self._tags = {} self._tags = {}
self.fifo_topic = "false"
self.content_based_deduplication = "false"
def publish(self, message, subject=None, message_attributes=None): def publish(self, message, subject=None, message_attributes=None):
message_id = str(uuid.uuid4()) message_id = str(uuid.uuid4())

View File

@ -170,6 +170,11 @@ class SNSResponse(BaseResponse):
} }
if topic.kms_master_key_id: if topic.kms_master_key_id:
attributes["KmsMasterKeyId"] = topic.kms_master_key_id attributes["KmsMasterKeyId"] = topic.kms_master_key_id
if topic.fifo_topic == "true":
attributes["FifoTopic"] = topic.fifo_topic
attributes[
"ContentBasedDeduplication"
] = topic.content_based_deduplication
response = { response = {
"GetTopicAttributesResponse": { "GetTopicAttributesResponse": {
"GetTopicAttributesResult": {"Attributes": attributes}, "GetTopicAttributesResult": {"Attributes": attributes},
@ -836,6 +841,16 @@ GET_TOPIC_ATTRIBUTES_TEMPLATE = """<GetTopicAttributesResponse xmlns="http://sns
<value>{{ topic.kms_master_key_id }}</value> <value>{{ topic.kms_master_key_id }}</value>
</entry> </entry>
{% endif %} {% endif %}
{% if topic.fifo_topic == 'true' %}
<entry>
<key>FifoTopic</key>
<value>{{ topic.fifo_topic }}</value>
</entry>
<entry>
<key>ContentBasedDeduplication</key>
<value>{{ topic.content_based_deduplication }}</value>
</entry>
{% endif %}
</Attributes> </Attributes>
</GetTopicAttributesResult> </GetTopicAttributesResult>
<ResponseMetadata> <ResponseMetadata>

View File

@ -570,3 +570,51 @@ def test_topic_kms_master_key_id_attribute():
resp = client.get_topic_attributes(TopicArn=topic_arn) resp = client.get_topic_attributes(TopicArn=topic_arn)
resp["Attributes"].should.have.key("KmsMasterKeyId") resp["Attributes"].should.have.key("KmsMasterKeyId")
resp["Attributes"]["KmsMasterKeyId"].should.equal("key-id") resp["Attributes"]["KmsMasterKeyId"].should.equal("key-id")
@mock_sns
def test_topic_fifo_get_attributes():
client = boto3.client("sns", region_name="us-east-1")
resp = client.create_topic(
Name="test-topic-fifo-get-attr.fifo", Attributes={"FifoTopic": "true"}
)
topic_arn = resp["TopicArn"]
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes.should.have.key("FifoTopic")
attributes.should.have.key("ContentBasedDeduplication")
attributes["FifoTopic"].should.equal("true")
attributes["ContentBasedDeduplication"].should.equal("false")
client.set_topic_attributes(
TopicArn=topic_arn,
AttributeName="ContentBasedDeduplication",
AttributeValue="true",
)
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes["ContentBasedDeduplication"].should.equal("true")
@mock_sns
def test_topic_get_attributes():
client = boto3.client("sns", region_name="us-east-1")
resp = client.create_topic(Name="test-topic-get-attr")
topic_arn = resp["TopicArn"]
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes.should_not.have.key("FifoTopic")
attributes.should_not.have.key("ContentBasedDeduplication")
@mock_sns
def test_topic_get_attributes_with_fifo_false():
client = boto3.client("sns", region_name="us-east-1")
resp = client.create_topic(
Name="test-topic-get-attr-with-fifo-false", Attributes={"FifoTopic": "false"}
)
topic_arn = resp["TopicArn"]
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes.should_not.have.key("FifoTopic")
attributes.should_not.have.key("ContentBasedDeduplication")