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

Techdebt: Replace sure with regular assertions in SNS (#6621)

Browse Source
This commit is contained in:
kbalk 2023-08-10 14:07:41 -04:00 committed by GitHub
parent 38bd1af4d3
commit e9f5a64f0f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 819 additions and 838 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

128
tests/test_sns/test_application_boto3.py
View File

@ -1,10 +1,10 @@
import boto3 import boto3
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from moto import mock_sns
import sure # noqa # pylint: disable=unused-import
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
import pytest import pytest
from moto import mock_sns
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@mock_sns @mock_sns
def test_create_platform_application(): def test_create_platform_application():
@ -18,7 +18,7 @@ def test_create_platform_application():
}, },
) )
application_arn = response["PlatformApplicationArn"] application_arn = response["PlatformApplicationArn"]
application_arn.should.equal( assert application_arn == (
f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:app/APNS/my-application" f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:app/APNS/my-application"
) )
@ -38,20 +38,17 @@ def test_get_platform_application_attributes():
attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)[ attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)[
"Attributes" "Attributes"
] ]
attributes.should.equal( assert attributes == {
{ "PlatformCredential": "platform_credential",
"PlatformCredential": "platform_credential", "PlatformPrincipal": "platform_principal",
"PlatformPrincipal": "platform_principal", }
}
)
@mock_sns @mock_sns
def test_get_missing_platform_application_attributes(): def test_get_missing_platform_application_attributes():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
conn.get_platform_application_attributes.when.called_with( with pytest.raises(ClientError):
PlatformApplicationArn="a-fake-arn" conn.get_platform_application_attributes(PlatformApplicationArn="a-fake-arn")
).should.throw(ClientError)
@mock_sns @mock_sns
@ -72,7 +69,7 @@ def test_set_platform_application_attributes():
attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)[ attributes = conn.get_platform_application_attributes(PlatformApplicationArn=arn)[
"Attributes" "Attributes"
] ]
attributes.should.equal( assert attributes == (
{"PlatformCredential": "platform_credential", "PlatformPrincipal": "other"} {"PlatformCredential": "platform_credential", "PlatformPrincipal": "other"}
) )
@ -89,7 +86,7 @@ def test_list_platform_applications():
applications_response = conn.list_platform_applications() applications_response = conn.list_platform_applications()
applications = applications_response["PlatformApplications"] applications = applications_response["PlatformApplications"]
applications.should.have.length_of(2) assert len(applications) == 2
@mock_sns @mock_sns
@ -104,14 +101,14 @@ def test_delete_platform_application():
applications_response = conn.list_platform_applications() applications_response = conn.list_platform_applications()
applications = applications_response["PlatformApplications"] applications = applications_response["PlatformApplications"]
applications.should.have.length_of(2) assert len(applications) == 2
application_arn = applications[0]["PlatformApplicationArn"] application_arn = applications[0]["PlatformApplicationArn"]
conn.delete_platform_application(PlatformApplicationArn=application_arn) conn.delete_platform_application(PlatformApplicationArn=application_arn)
applications_response = conn.list_platform_applications() applications_response = conn.list_platform_applications()
applications = applications_response["PlatformApplications"] applications = applications_response["PlatformApplications"]
applications.should.have.length_of(1) assert len(applications) == 1
@mock_sns @mock_sns
@ -130,8 +127,9 @@ def test_create_platform_endpoint():
) )
endpoint_arn = endpoint["EndpointArn"] endpoint_arn = endpoint["EndpointArn"]
endpoint_arn.should.contain( assert (
f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:endpoint/APNS/my-application/" f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:endpoint/APNS/my-application/"
in endpoint_arn
) )
@ -150,12 +148,13 @@ def test_create_duplicate_platform_endpoint():
Attributes={"Enabled": "false"}, Attributes={"Enabled": "false"},
) )
conn.create_platform_endpoint.when.called_with( with pytest.raises(ClientError):
PlatformApplicationArn=application_arn, conn.create_platform_endpoint(
Token="some_unique_id", PlatformApplicationArn=application_arn,
CustomUserData="some user data", Token="some_unique_id",
Attributes={"Enabled": "true"}, CustomUserData="some user data",
).should.throw(ClientError) Attributes={"Enabled": "true"},
)
@mock_sns @mock_sns
@ -182,7 +181,7 @@ def test_create_duplicate_platform_endpoint_with_same_attributes():
) )
endpoint_arn = endpoint["EndpointArn"] endpoint_arn = endpoint["EndpointArn"]
endpoint_arn.should.equal(created_endpoint_arn) assert endpoint_arn == created_endpoint_arn
@mock_sns @mock_sns
@ -205,9 +204,9 @@ def test_get_list_endpoints_by_platform_application():
PlatformApplicationArn=application_arn PlatformApplicationArn=application_arn
)["Endpoints"] )["Endpoints"]
endpoint_list.should.have.length_of(1) assert len(endpoint_list) == 1
endpoint_list[0]["Attributes"]["CustomUserData"].should.equal("some data") assert endpoint_list[0]["Attributes"]["CustomUserData"] == "some data"
endpoint_list[0]["EndpointArn"].should.equal(endpoint_arn) assert endpoint_list[0]["EndpointArn"] == endpoint_arn
@mock_sns @mock_sns
@ -227,7 +226,7 @@ def test_get_endpoint_attributes():
endpoint_arn = endpoint["EndpointArn"] endpoint_arn = endpoint["EndpointArn"]
attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"] attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"]
attributes.should.equal( assert attributes == (
{"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "some data"} {"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "some data"}
) )
@ -235,21 +234,23 @@ def test_get_endpoint_attributes():
@mock_sns @mock_sns
def test_get_non_existent_endpoint_attributes(): def test_get_non_existent_endpoint_attributes():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
endpoint_arn = "arn:aws:sns:us-east-1:123456789012:endpoint/APNS/my-application/c1f76c42-192a-4e75-b04f-a9268ce2abf3" endpoint_arn = (
"arn:aws:sns:us-east-1:123456789012:endpoint/APNS/my-application"
"/c1f76c42-192a-4e75-b04f-a9268ce2abf3"
)
with pytest.raises(conn.exceptions.NotFoundException) as excinfo: with pytest.raises(conn.exceptions.NotFoundException) as excinfo:
conn.get_endpoint_attributes(EndpointArn=endpoint_arn) conn.get_endpoint_attributes(EndpointArn=endpoint_arn)
error = excinfo.value.response["Error"] error = excinfo.value.response["Error"]
error["Type"].should.equal("Sender") assert error["Type"] == "Sender"
error["Code"].should.equal("NotFound") assert error["Code"] == "NotFound"
error["Message"].should.equal("Endpoint does not exist") assert error["Message"] == "Endpoint does not exist"
@mock_sns @mock_sns
def test_get_missing_endpoint_attributes(): def test_get_missing_endpoint_attributes():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
conn.get_endpoint_attributes.when.called_with( with pytest.raises(ClientError):
EndpointArn="a-fake-arn" conn.get_endpoint_attributes(EndpointArn="a-fake-arn")
).should.throw(ClientError)
@mock_sns @mock_sns
@ -272,7 +273,7 @@ def test_set_endpoint_attributes():
EndpointArn=endpoint_arn, Attributes={"CustomUserData": "other data"} EndpointArn=endpoint_arn, Attributes={"CustomUserData": "other data"}
) )
attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"] attributes = conn.get_endpoint_attributes(EndpointArn=endpoint_arn)["Attributes"]
attributes.should.equal( assert attributes == (
{"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "other data"} {"Token": "some_unique_id", "Enabled": "false", "CustomUserData": "other data"}
) )
@ -291,15 +292,25 @@ def test_delete_endpoint():
Attributes={"Enabled": "true"}, Attributes={"Enabled": "true"},
) )
conn.list_endpoints_by_platform_application(PlatformApplicationArn=application_arn)[ assert (
"Endpoints" len(
].should.have.length_of(1) conn.list_endpoints_by_platform_application(
PlatformApplicationArn=application_arn
)["Endpoints"]
)
== 1
)
conn.delete_endpoint(EndpointArn=endpoint["EndpointArn"]) conn.delete_endpoint(EndpointArn=endpoint["EndpointArn"])
conn.list_endpoints_by_platform_application(PlatformApplicationArn=application_arn)[ assert (
"Endpoints" len(
].should.have.length_of(0) conn.list_endpoints_by_platform_application(
PlatformApplicationArn=application_arn
)["Endpoints"]
)
== 0
)
@mock_sns @mock_sns
@ -341,9 +352,10 @@ def test_publish_to_disabled_platform_endpoint():
endpoint_arn = endpoint["EndpointArn"] endpoint_arn = endpoint["EndpointArn"]
conn.publish.when.called_with( with pytest.raises(ClientError):
Message="some message", MessageStructure="json", TargetArn=endpoint_arn conn.publish(
).should.throw(ClientError) Message="some message", MessageStructure="json", TargetArn=endpoint_arn
)
@mock_sns @mock_sns
@ -355,11 +367,11 @@ def test_set_sms_attributes():
) )
response = conn.get_sms_attributes() response = conn.get_sms_attributes()
response.should.contain("attributes") assert "attributes" in response
response["attributes"].should.contain("DefaultSMSType") assert "DefaultSMSType" in response["attributes"]
response["attributes"].should.contain("test") assert "test" in response["attributes"]
response["attributes"]["DefaultSMSType"].should.equal("Transactional") assert response["attributes"]["DefaultSMSType"] == "Transactional"
response["attributes"]["test"].should.equal("test") assert response["attributes"]["test"] == "test"
@mock_sns @mock_sns
@ -371,10 +383,10 @@ def test_get_sms_attributes_filtered():
) )
response = conn.get_sms_attributes(attributes=["DefaultSMSType"]) response = conn.get_sms_attributes(attributes=["DefaultSMSType"])
response.should.contain("attributes") assert "attributes" in response
response["attributes"].should.contain("DefaultSMSType") assert "DefaultSMSType" in response["attributes"]
response["attributes"].should_not.contain("test") assert "test" not in response["attributes"]
response["attributes"]["DefaultSMSType"].should.equal("Transactional") assert response["attributes"]["DefaultSMSType"] == "Transactional"
@mock_sns @mock_sns
@ -395,6 +407,6 @@ def test_delete_endpoints_of_delete_app():
with pytest.raises(conn.exceptions.NotFoundException) as excinfo: with pytest.raises(conn.exceptions.NotFoundException) as excinfo:
conn.get_endpoint_attributes(EndpointArn=endpoint_arn) conn.get_endpoint_attributes(EndpointArn=endpoint_arn)
error = excinfo.value.response["Error"] error = excinfo.value.response["Error"]
error["Type"].should.equal("Sender") assert error["Type"] == "Sender"
error["Code"].should.equal("NotFound") assert error["Code"] == "NotFound"
error["Message"].should.equal("Endpoint does not exist") assert error["Message"] == "Endpoint does not exist"

73
tests/test_sns/test_publish_batch.py
View File

@ -1,9 +1,9 @@
import boto3
import json import json
import sure # noqa # pylint: disable=unused-import
import boto3
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
import pytest import pytest
from moto import mock_sns, mock_sqs from moto import mock_sns, mock_sqs
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@ -17,8 +17,8 @@ def test_publish_batch_unknown_topic():
PublishBatchRequestEntries=[{"Id": "id_1", "Message": "1"}], PublishBatchRequestEntries=[{"Id": "id_1", "Message": "1"}],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("NotFound") assert err["Code"] == "NotFound"
err["Message"].should.equal("Topic does not exist") assert err["Message"] == "Topic does not exist"
@mock_sns @mock_sns
@ -34,10 +34,8 @@ def test_publish_batch_too_many_items():
], ],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("TooManyEntriesInBatchRequest") assert err["Code"] == "TooManyEntriesInBatchRequest"
err["Message"].should.equal( assert err["Message"] == "The batch request contains more entries than permissible."
"The batch request contains more entries than permissible."
)
@mock_sns @mock_sns
@ -53,9 +51,9 @@ def test_publish_batch_non_unique_ids():
], ],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("BatchEntryIdsNotDistinct") assert err["Code"] == "BatchEntryIdsNotDistinct"
err["Message"].should.equal( assert (
"Two or more batch entries in the request have the same Id." err["Message"] == "Two or more batch entries in the request have the same Id."
) )
@ -73,8 +71,8 @@ def test_publish_batch_fifo_without_message_group_id():
PublishBatchRequestEntries=[{"Id": "id_2", "Message": "2"}], PublishBatchRequestEntries=[{"Id": "id_2", "Message": "2"}],
) )
err = exc.value.response["Error"] err = exc.value.response["Error"]
err["Code"].should.equal("InvalidParameter") assert err["Code"] == "InvalidParameter"
err["Message"].should.equal( assert err["Message"] == (
"Invalid parameter: The MessageGroupId parameter is required for FIFO topics" "Invalid parameter: The MessageGroupId parameter is required for FIFO topics"
) )
@ -90,20 +88,21 @@ def test_publish_batch_standard_with_message_group_id():
] ]
resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries) resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries)
resp.should.have.key("Successful").length_of(2) assert len(resp["Successful"]) == 2
for message_status in resp["Successful"]: for message_status in resp["Successful"]:
message_status.should.have.key("MessageId") assert "MessageId" in message_status
[m["Id"] for m in resp["Successful"]].should.equal(["id_1", "id_3"]) assert [m["Id"] for m in resp["Successful"]] == ["id_1", "id_3"]
resp.should.have.key("Failed").length_of(1) assert len(resp["Failed"]) == 1
resp["Failed"][0].should.equal( assert resp["Failed"][0] == {
{ "Id": "id_2",
"Id": "id_2", "Code": "InvalidParameter",
"Code": "InvalidParameter", "Message": (
"Message": "Invalid parameter: MessageGroupId Reason: The request includes MessageGroupId parameter that is not valid for this topic type", "Invalid parameter: MessageGroupId Reason: The request includes "
"SenderFault": True, "MessageGroupId parameter that is not valid for this topic type"
} ),
) "SenderFault": True,
}
@mock_sns @mock_sns
@ -129,22 +128,22 @@ def test_publish_batch_to_sqs():
resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries) resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries)
resp.should.have.key("Successful").length_of(3) assert len(resp["Successful"]) == 3
messages = queue.receive_messages(MaxNumberOfMessages=3) messages = queue.receive_messages(MaxNumberOfMessages=3)
messages.should.have.length_of(3) assert len(messages) == 3
messages = [json.loads(m.body) for m in messages] messages = [json.loads(m.body) for m in messages]
for m in messages: for message in messages:
for key in list(m.keys()): for key in list(message.keys()):
if key not in ["Message", "Subject", "MessageAttributes"]: if key not in ["Message", "Subject", "MessageAttributes"]:
del m[key] del message[key]
messages.should.contain({"Message": "1"}) assert {"Message": "1"} in messages
messages.should.contain({"Message": "2", "Subject": "subj2"}) assert {"Message": "2", "Subject": "subj2"} in messages
messages.should.contain( assert (
{"Message": "3", "MessageAttributes": {"a": {"Type": "String", "Value": "v"}}} {"Message": "3", "MessageAttributes": {"a": {"Type": "String", "Value": "v"}}}
) ) in messages
@mock_sqs @mock_sqs
@ -173,7 +172,7 @@ def test_publish_batch_to_sqs_raw():
] ]
resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries) resp = client.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries)
resp.should.have.key("Successful").length_of(2) assert len(resp["Successful"]) == 2
received = queue.receive_messages( received = queue.receive_messages(
MaxNumberOfMessages=10, MessageAttributeNames=["All"] MaxNumberOfMessages=10, MessageAttributeNames=["All"]
@ -181,5 +180,5 @@ def test_publish_batch_to_sqs_raw():
messages = [(message.body, message.message_attributes) for message in received] messages = [(message.body, message.message_attributes) for message in received]
messages.should.contain(("foo", None)) assert ("foo", None) in messages
messages.should.contain(("bar", {"a": {"StringValue": "v", "DataType": "String"}})) assert ("bar", {"a": {"StringValue": "v", "DataType": "String"}}) in messages

680
tests/test_sns/test_publishing_boto3.py
View File

File diff suppressed because it is too large Load Diff

20
tests/test_sns/test_server.py
View File

@ -1,26 +1,20 @@
"""Test the different server responses."""
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
import sure # noqa # pylint: disable=unused-import
import moto.server as server import moto.server as server
"""
Test the different server responses
"""
def test_sns_server_get(): def test_sns_server_get():
backend = server.create_backend_app("sns") backend = server.create_backend_app("sns")
test_client = backend.test_client() test_client = backend.test_client()
topic_data = test_client.action_data("CreateTopic", Name="testtopic") topic_data = test_client.action_data("CreateTopic", Name="testtopic")
topic_data.should.contain("CreateTopicResult") assert "CreateTopicResult" in topic_data
topic_data.should.contain( assert (
f"<TopicArn>arn:aws:sns:us-east-1:{ACCOUNT_ID}:testtopic</TopicArn>" f"<TopicArn>arn:aws:sns:us-east-1:{ACCOUNT_ID}:testtopic</TopicArn>"
) ) in topic_data
topics_data = test_client.action_data("ListTopics") topics_data = test_client.action_data("ListTopics")
topics_data.should.contain("ListTopicsResult") assert "ListTopicsResult" in topics_data
topic_data.should.contain( assert (
f"<TopicArn>arn:aws:sns:us-east-1:{ACCOUNT_ID}:testtopic</TopicArn>" f"<TopicArn>arn:aws:sns:us-east-1:{ACCOUNT_ID}:testtopic</TopicArn>"
) ) in topics_data

46
tests/test_sns/test_sns_cloudformation.py
View File

@ -1,6 +1,6 @@
import boto3
import json import json
import sure # noqa # pylint: disable=unused-import
import boto3
import pytest import pytest
from moto import mock_cloudformation, mock_sns from moto import mock_cloudformation, mock_sns
@ -16,25 +16,25 @@ def test_sns_topic():
sns = boto3.client("sns", region_name="us-west-1") sns = boto3.client("sns", region_name="us-west-1")
topics = sns.list_topics()["Topics"] topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1) assert len(topics) == 1
topic_arn = topics[0]["TopicArn"] topic_arn = topics[0]["TopicArn"]
topic_arn.should.contain("my_topics") assert "my_topics" in topic_arn
subscriptions = sns.list_subscriptions()["Subscriptions"] subscriptions = sns.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1) assert len(subscriptions) == 1
subscription = subscriptions[0] subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn) assert subscription["TopicArn"] == topic_arn
subscription["Protocol"].should.equal("https") assert subscription["Protocol"] == "https"
subscription["SubscriptionArn"].should.contain(topic_arn) assert topic_arn in subscription["SubscriptionArn"]
subscription["Endpoint"].should.equal("https://example.com") assert subscription["Endpoint"] == "https://example.com"
stack = cf.describe_stacks(StackName="test_stack")["Stacks"][0] stack = cf.describe_stacks(StackName="test_stack")["Stacks"][0]
topic_name_output = [x for x in stack["Outputs"] if x["OutputKey"] == "topic_name"][ topic_name_output = [x for x in stack["Outputs"] if x["OutputKey"] == "topic_name"][
0 0
] ]
topic_name_output["OutputValue"].should.equal("my_topics") assert topic_name_output["OutputValue"] == "my_topics"
topic_arn_output = [x for x in stack["Outputs"] if x["OutputKey"] == "topic_arn"][0] topic_arn_output = [x for x in stack["Outputs"] if x["OutputKey"] == "topic_arn"][0]
topic_arn_output["OutputValue"].should.equal(topic_arn) assert topic_arn_output["OutputValue"] == topic_arn
@mock_cloudformation @mock_cloudformation
@ -47,7 +47,7 @@ def test_sns_update_topic():
sns = boto3.client("sns", region_name="us-west-1") sns = boto3.client("sns", region_name="us-west-1")
topics = sns.list_topics()["Topics"] topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1) assert len(topics) == 1
dummy_template["Resources"]["MySNSTopic"]["Properties"]["Subscription"][0][ dummy_template["Resources"]["MySNSTopic"]["Properties"]["Subscription"][0][
"Endpoint" "Endpoint"
@ -56,17 +56,17 @@ def test_sns_update_topic():
cf.update_stack(StackName="test_stack", TemplateBody=sns_template_json) cf.update_stack(StackName="test_stack", TemplateBody=sns_template_json)
topics = sns.list_topics()["Topics"] topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1) assert len(topics) == 1
topic_arn = topics[0]["TopicArn"] topic_arn = topics[0]["TopicArn"]
topic_arn.should.contain("my_topics") assert "my_topics" in topic_arn
subscriptions = sns.list_subscriptions()["Subscriptions"] subscriptions = sns.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1) assert len(subscriptions) == 1
subscription = subscriptions[0] subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn) assert subscription["TopicArn"] == topic_arn
subscription["Protocol"].should.equal("https") assert subscription["Protocol"] == "https"
subscription["SubscriptionArn"].should.contain(topic_arn) assert topic_arn in subscription["SubscriptionArn"]
subscription["Endpoint"].should.equal("https://example-updated.com") assert subscription["Endpoint"] == "https://example-updated.com"
@mock_cloudformation @mock_cloudformation
@ -80,7 +80,7 @@ def test_sns_update_remove_topic(with_properties):
sns = boto3.client("sns", region_name="us-west-1") sns = boto3.client("sns", region_name="us-west-1")
topics = sns.list_topics()["Topics"] topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1) assert len(topics) == 1
dummy_template["Resources"].pop("MySNSTopic") dummy_template["Resources"].pop("MySNSTopic")
dummy_template.pop("Outputs") dummy_template.pop("Outputs")
@ -88,7 +88,7 @@ def test_sns_update_remove_topic(with_properties):
cf.update_stack(StackName="test_stack", TemplateBody=sns_template_json) cf.update_stack(StackName="test_stack", TemplateBody=sns_template_json)
topics = sns.list_topics()["Topics"] topics = sns.list_topics()["Topics"]
topics.should.have.length_of(0) assert len(topics) == 0
@mock_cloudformation @mock_cloudformation
@ -101,12 +101,12 @@ def test_sns_delete_topic(with_properties):
sns = boto3.client("sns", region_name="us-west-1") sns = boto3.client("sns", region_name="us-west-1")
topics = sns.list_topics()["Topics"] topics = sns.list_topics()["Topics"]
topics.should.have.length_of(1) assert len(topics) == 1
cf.delete_stack(StackName="test_stack") cf.delete_stack(StackName="test_stack")
topics = sns.list_topics()["Topics"] topics = sns.list_topics()["Topics"]
topics.should.have.length_of(0) assert len(topics) == 0
def get_template(with_properties): def get_template(with_properties):

284
tests/test_sns/test_subscriptions_boto3.py
View File

@ -1,8 +1,6 @@
import boto3
import json import json
import sure # noqa # pylint: disable=unused-import import boto3
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
import pytest import pytest
@ -22,10 +20,10 @@ def test_subscribe_sms():
arn = resp["TopicArn"] arn = resp["TopicArn"]
resp = client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="+15551234567") resp = client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="+15551234567")
resp.should.have.key("SubscriptionArn") assert "SubscriptionArn" in resp
resp = client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="+15/55-123.4567") resp = client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="+15/55-123.4567")
resp.should.have.key("SubscriptionArn") assert "SubscriptionArn" in resp
@mock_sns @mock_sns
@ -42,7 +40,7 @@ def test_double_subscription():
TopicArn=arn, Protocol="sqs", Endpoint="arn:aws:sqs:elasticmq:000000000000:foo" TopicArn=arn, Protocol="sqs", Endpoint="arn:aws:sqs:elasticmq:000000000000:foo"
) )
resp1["SubscriptionArn"].should.equal(resp2["SubscriptionArn"]) assert resp1["SubscriptionArn"] == resp2["SubscriptionArn"]
@mock_sns @mock_sns
@ -56,19 +54,29 @@ def test_subscribe_bad_sms():
# Test invalid number # Test invalid number
client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="NAA+15551234567") client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="NAA+15551234567")
except ClientError as err: except ClientError as err:
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
client.subscribe.when.called_with( with pytest.raises(ClientError) as client_err:
TopicArn=arn, Protocol="sms", Endpoint="+15--551234567" client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="+15--551234567")
).should.throw(ClientError, "Invalid SMS endpoint: +15--551234567") assert (
client_err.value.response["Error"]["Message"]
== "Invalid SMS endpoint: +15--551234567"
)
client.subscribe.when.called_with( with pytest.raises(ClientError) as client_err:
TopicArn=arn, Protocol="sms", Endpoint="+15551234567." client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="+15551234567.")
).should.throw(ClientError, "Invalid SMS endpoint: +15551234567.") assert (
client_err.value.response["Error"]["Message"]
== "Invalid SMS endpoint: +15551234567."
)
client.subscribe.when.called_with( with pytest.raises(ClientError) as client_err:
TopicArn=arn, Protocol="sms", Endpoint="/+15551234567" assert client.subscribe(TopicArn=arn, Protocol="sms", Endpoint="/+15551234567")
).should.throw(ClientError, "Invalid SMS endpoint: /+15551234567")
assert (
client_err.value.response["Error"]["Message"]
== "Invalid SMS endpoint: /+15551234567"
)
@mock_sns @mock_sns
@ -81,19 +89,19 @@ def test_creating_subscription():
conn.subscribe(TopicArn=topic_arn, Protocol="http", Endpoint="http://example.com/") conn.subscribe(TopicArn=topic_arn, Protocol="http", Endpoint="http://example.com/")
subscriptions = conn.list_subscriptions()["Subscriptions"] subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1) assert len(subscriptions) == 1
subscription = subscriptions[0] subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn) assert subscription["TopicArn"] == topic_arn
subscription["Protocol"].should.equal("http") assert subscription["Protocol"] == "http"
subscription["SubscriptionArn"].should.contain(topic_arn) assert topic_arn in subscription["SubscriptionArn"]
subscription["Endpoint"].should.equal("http://example.com/") assert subscription["Endpoint"] == "http://example.com/"
# Now unsubscribe the subscription # Now unsubscribe the subscription
conn.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"]) conn.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])
# And there should be zero subscriptions left # And there should be zero subscriptions left
subscriptions = conn.list_subscriptions()["Subscriptions"] subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(0) assert len(subscriptions) == 0
@mock_sns @mock_sns
@ -108,13 +116,13 @@ def test_unsubscribe_from_deleted_topic():
) )
subscriptions = client.list_subscriptions()["Subscriptions"] subscriptions = client.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1) assert len(subscriptions) == 1
subscription = subscriptions[0] subscription = subscriptions[0]
subscription_arn = subscription["SubscriptionArn"] subscription_arn = subscription["SubscriptionArn"]
subscription["TopicArn"].should.equal(topic_arn) assert subscription["TopicArn"] == topic_arn
subscription["Protocol"].should.equal("http") assert subscription["Protocol"] == "http"
subscription_arn.should.contain(topic_arn) assert topic_arn in subscription_arn
subscription["Endpoint"].should.equal("http://example.com/") assert subscription["Endpoint"] == "http://example.com/"
# Now delete the topic # Now delete the topic
client.delete_topic(TopicArn=topic_arn) client.delete_topic(TopicArn=topic_arn)
@ -122,17 +130,17 @@ def test_unsubscribe_from_deleted_topic():
# And there should now be 0 topics # And there should now be 0 topics
topics_json = client.list_topics() topics_json = client.list_topics()
topics = topics_json["Topics"] topics = topics_json["Topics"]
topics.should.have.length_of(0) assert len(topics) == 0
# as per the documentation deleting a topic deletes all the subscriptions # as per the documentation deleting a topic deletes all the subscriptions
subscriptions = client.list_subscriptions()["Subscriptions"] subscriptions = client.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(0) assert len(subscriptions) == 0
# Now delete hanging subscription # Now delete hanging subscription
client.unsubscribe(SubscriptionArn=subscription_arn) client.unsubscribe(SubscriptionArn=subscription_arn)
subscriptions = client.list_subscriptions()["Subscriptions"] subscriptions = client.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(0) assert len(subscriptions) == 0
# Deleting it again should not result in any error # Deleting it again should not result in any error
client.unsubscribe(SubscriptionArn=subscription_arn) client.unsubscribe(SubscriptionArn=subscription_arn)
@ -159,8 +167,8 @@ def test_getting_subscriptions_by_topic():
topic1_subscriptions = conn.list_subscriptions_by_topic(TopicArn=topic1_arn)[ topic1_subscriptions = conn.list_subscriptions_by_topic(TopicArn=topic1_arn)[
"Subscriptions" "Subscriptions"
] ]
topic1_subscriptions.should.have.length_of(1) assert len(topic1_subscriptions) == 1
topic1_subscriptions[0]["Endpoint"].should.equal("http://example1.com/") assert topic1_subscriptions[0]["Endpoint"] == "http://example1.com/"
@mock_sns @mock_sns
@ -180,26 +188,24 @@ def test_subscription_paging():
) )
all_subscriptions = conn.list_subscriptions() all_subscriptions = conn.list_subscriptions()
all_subscriptions["Subscriptions"].should.have.length_of(DEFAULT_PAGE_SIZE) assert len(all_subscriptions["Subscriptions"]) == DEFAULT_PAGE_SIZE
next_token = all_subscriptions["NextToken"] next_token = all_subscriptions["NextToken"]
next_token.should.equal(str(DEFAULT_PAGE_SIZE)) assert next_token == str(DEFAULT_PAGE_SIZE)
all_subscriptions = conn.list_subscriptions(NextToken=next_token) all_subscriptions = conn.list_subscriptions(NextToken=next_token)
all_subscriptions["Subscriptions"].should.have.length_of(int(DEFAULT_PAGE_SIZE / 3)) assert len(all_subscriptions["Subscriptions"]) == int(DEFAULT_PAGE_SIZE / 3)
all_subscriptions.shouldnt.have("NextToken") assert "NextToken" not in all_subscriptions
topic1_subscriptions = conn.list_subscriptions_by_topic(TopicArn=topic1_arn) topic1_subscriptions = conn.list_subscriptions_by_topic(TopicArn=topic1_arn)
topic1_subscriptions["Subscriptions"].should.have.length_of(DEFAULT_PAGE_SIZE) assert len(topic1_subscriptions["Subscriptions"]) == DEFAULT_PAGE_SIZE
next_token = topic1_subscriptions["NextToken"] next_token = topic1_subscriptions["NextToken"]
next_token.should.equal(str(DEFAULT_PAGE_SIZE)) assert next_token == str(DEFAULT_PAGE_SIZE)
topic1_subscriptions = conn.list_subscriptions_by_topic( topic1_subscriptions = conn.list_subscriptions_by_topic(
TopicArn=topic1_arn, NextToken=next_token TopicArn=topic1_arn, NextToken=next_token
) )
topic1_subscriptions["Subscriptions"].should.have.length_of( assert len(topic1_subscriptions["Subscriptions"]) == int(DEFAULT_PAGE_SIZE / 3)
int(DEFAULT_PAGE_SIZE / 3) assert "NextToken" not in topic1_subscriptions
)
topic1_subscriptions.shouldnt.have("NextToken")
@mock_sns @mock_sns
@ -215,17 +221,17 @@ def test_subscribe_attributes():
SubscriptionArn=resp["SubscriptionArn"] SubscriptionArn=resp["SubscriptionArn"]
) )
response.should.contain("Attributes") assert "Attributes" in response
attributes = response["Attributes"] attributes = response["Attributes"]
attributes["PendingConfirmation"].should.equal("false") assert attributes["PendingConfirmation"] == "false"
attributes["ConfirmationWasAuthenticated"].should.equal("true") assert attributes["ConfirmationWasAuthenticated"] == "true"
attributes["Endpoint"].should.equal("http://test.com") assert attributes["Endpoint"] == "http://test.com"
attributes["TopicArn"].should.equal(arn) assert attributes["TopicArn"] == arn
attributes["Protocol"].should.equal("http") assert attributes["Protocol"] == "http"
attributes["SubscriptionArn"].should.equal(resp["SubscriptionArn"]) assert attributes["SubscriptionArn"] == resp["SubscriptionArn"]
attributes["Owner"].should.equal(str(ACCOUNT_ID)) assert attributes["Owner"] == str(ACCOUNT_ID)
attributes["RawMessageDelivery"].should.equal("false") assert attributes["RawMessageDelivery"] == "false"
json.loads(attributes["EffectiveDeliveryPolicy"]).should.equal( assert json.loads(attributes["EffectiveDeliveryPolicy"]) == (
DEFAULT_EFFECTIVE_DELIVERY_POLICY DEFAULT_EFFECTIVE_DELIVERY_POLICY
) )
@ -272,28 +278,28 @@ def test_creating_subscription_with_attributes():
) )
subscriptions = conn.list_subscriptions()["Subscriptions"] subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1) assert len(subscriptions) == 1
subscription = subscriptions[0] subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn) assert subscription["TopicArn"] == topic_arn
subscription["Protocol"].should.equal("http") assert subscription["Protocol"] == "http"
subscription["SubscriptionArn"].should.contain(topic_arn) assert topic_arn in subscription["SubscriptionArn"]
subscription["Endpoint"].should.equal("http://example.com/") assert subscription["Endpoint"] == "http://example.com/"
# Test the subscription attributes have been set # Test the subscription attributes have been set
subscription_arn = subscription["SubscriptionArn"] subscription_arn = subscription["SubscriptionArn"]
attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn) attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn)
attrs["Attributes"]["RawMessageDelivery"].should.equal("true") assert attrs["Attributes"]["RawMessageDelivery"] == "true"
attrs["Attributes"]["DeliveryPolicy"].should.equal(delivery_policy) assert attrs["Attributes"]["DeliveryPolicy"] == delivery_policy
attrs["Attributes"]["FilterPolicy"].should.equal(filter_policy) assert attrs["Attributes"]["FilterPolicy"] == filter_policy
attrs["Attributes"]["SubscriptionRoleArn"].should.equal(subscription_role_arn) assert attrs["Attributes"]["SubscriptionRoleArn"] == subscription_role_arn
# Now unsubscribe the subscription # Now unsubscribe the subscription
conn.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"]) conn.unsubscribe(SubscriptionArn=subscription["SubscriptionArn"])
# And there should be zero subscriptions left # And there should be zero subscriptions left
subscriptions = conn.list_subscriptions()["Subscriptions"] subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(0) assert len(subscriptions) == 0
# invalid attr name # invalid attr name
with pytest.raises(ClientError): with pytest.raises(ClientError):
@ -319,12 +325,12 @@ def test_delete_subscriptions_on_delete_topic():
) )
subscriptions = conn.list_subscriptions()["Subscriptions"] subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1) assert len(subscriptions) == 1
conn.delete_topic(TopicArn=topic.get("TopicArn")) conn.delete_topic(TopicArn=topic.get("TopicArn"))
subscriptions = conn.list_subscriptions()["Subscriptions"] subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(0) assert len(subscriptions) == 0
@mock_sns @mock_sns
@ -337,16 +343,16 @@ def test_set_subscription_attributes():
conn.subscribe(TopicArn=topic_arn, Protocol="http", Endpoint="http://example.com/") conn.subscribe(TopicArn=topic_arn, Protocol="http", Endpoint="http://example.com/")
subscriptions = conn.list_subscriptions()["Subscriptions"] subscriptions = conn.list_subscriptions()["Subscriptions"]
subscriptions.should.have.length_of(1) assert len(subscriptions) == 1
subscription = subscriptions[0] subscription = subscriptions[0]
subscription["TopicArn"].should.equal(topic_arn) assert subscription["TopicArn"] == topic_arn
subscription["Protocol"].should.equal("http") assert subscription["Protocol"] == "http"
subscription["SubscriptionArn"].should.contain(topic_arn) assert topic_arn in subscription["SubscriptionArn"]
subscription["Endpoint"].should.equal("http://example.com/") assert subscription["Endpoint"] == "http://example.com/"
subscription_arn = subscription["SubscriptionArn"] subscription_arn = subscription["SubscriptionArn"]
attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn) attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn)
attrs.should.have.key("Attributes") assert "Attributes" in attrs
conn.set_subscription_attributes( conn.set_subscription_attributes(
SubscriptionArn=subscription_arn, SubscriptionArn=subscription_arn,
AttributeName="RawMessageDelivery", AttributeName="RawMessageDelivery",
@ -384,9 +390,9 @@ def test_set_subscription_attributes():
attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn) attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn)
attrs["Attributes"]["RawMessageDelivery"].should.equal("true") assert attrs["Attributes"]["RawMessageDelivery"] == "true"
attrs["Attributes"]["DeliveryPolicy"].should.equal(delivery_policy) assert attrs["Attributes"]["DeliveryPolicy"] == delivery_policy
attrs["Attributes"]["FilterPolicy"].should.equal(filter_policy) assert attrs["Attributes"]["FilterPolicy"] == filter_policy
filter_policy_scope = "MessageBody" filter_policy_scope = "MessageBody"
conn.set_subscription_attributes( conn.set_subscription_attributes(
@ -397,7 +403,7 @@ def test_set_subscription_attributes():
attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn) attrs = conn.get_subscription_attributes(SubscriptionArn=subscription_arn)
attrs["Attributes"]["FilterPolicyScope"].should.equal(filter_policy_scope) assert attrs["Attributes"]["FilterPolicyScope"] == filter_policy_scope
# not existing subscription # not existing subscription
with pytest.raises(ClientError): with pytest.raises(ClientError):
@ -436,8 +442,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: FilterPolicy: Filter policy is too complex" "Invalid parameter: FilterPolicy: Filter policy is too complex"
) )
@ -450,8 +456,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: FilterPolicy: Match value must be String, number, true, false, or null" "Invalid parameter: FilterPolicy: Match value must be String, number, true, false, or null"
) )
@ -464,8 +470,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: FilterPolicy: exists match pattern must be either true or false." "Invalid parameter: FilterPolicy: exists match pattern must be either true or false."
) )
@ -478,8 +484,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: FilterPolicy: Unrecognized match type error" "Invalid parameter: FilterPolicy: Unrecognized match type error"
) )
@ -492,7 +498,7 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InternalFailure") assert err.response["Error"]["Code"] == "InternalFailure"
with pytest.raises(ClientError) as err_info: with pytest.raises(ClientError) as err_info:
conn.subscribe( conn.subscribe(
@ -505,8 +511,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Value of < must be numeric\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Value of < must be numeric\n at ..."
) )
@ -523,8 +529,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Value of <= must be numeric\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Value of <= must be numeric\n at ..."
) )
@ -537,9 +543,10 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Invalid member in numeric match: ]\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: "
"Invalid member in numeric match: ]\n at ..."
) )
with pytest.raises(ClientError) as err_info: with pytest.raises(ClientError) as err_info:
@ -553,9 +560,10 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Invalid member in numeric match: 50\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Invalid "
"member in numeric match: 50\n at ..."
) )
with pytest.raises(ClientError) as err_info: with pytest.raises(ClientError) as err_info:
@ -567,8 +575,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Value of < must be numeric\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Value of < must be numeric\n at ..."
) )
@ -581,9 +589,10 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Unrecognized numeric range operator: 0\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: "
"Unrecognized numeric range operator: 0\n at ..."
) )
with pytest.raises(ClientError) as err_info: with pytest.raises(ClientError) as err_info:
@ -597,9 +606,10 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Too many elements in numeric expression\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Too many "
"elements in numeric expression\n at ..."
) )
with pytest.raises(ClientError) as err_info: with pytest.raises(ClientError) as err_info:
@ -613,8 +623,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Bad numeric range operator: >\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Bad numeric range operator: >\n at ..."
) )
@ -629,8 +639,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Bottom must be less than top\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Bottom must be less than top\n at ..."
) )
@ -645,8 +655,8 @@ def test_subscribe_invalid_filter_policy():
) )
err = err_info.value err = err_info.value
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Attributes Reason: FilterPolicy: Value of < must be numeric\n at ..." "Invalid parameter: Attributes Reason: FilterPolicy: Value of < must be numeric\n at ..."
) )
@ -660,9 +670,10 @@ def test_subscribe_invalid_filter_policy():
}, },
) )
except ClientError as err: except ClientError as err:
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: Filter policy scope MessageAttributes does not support nested filter policy" "Invalid parameter: Filter policy scope MessageAttributes does "
"not support nested filter policy"
) )
try: try:
@ -681,8 +692,8 @@ def test_subscribe_invalid_filter_policy():
Attributes={"FilterPolicy": json.dumps(filter_policy)}, Attributes={"FilterPolicy": json.dumps(filter_policy)},
) )
except ClientError as err: except ClientError as err:
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: FilterPolicy: Filter policy can not have more than 5 keys" "Invalid parameter: FilterPolicy: Filter policy can not have more than 5 keys"
) )
@ -696,8 +707,9 @@ def test_subscribe_invalid_filter_policy():
"key_d": {"key_e": ["value_one", "value_two", "value_three"]}, "key_d": {"key_e": ["value_one", "value_two", "value_three"]},
"key_f": ["value_one", "value_two", "value_three"], "key_f": ["value_one", "value_two", "value_three"],
} }
# The first array has four values in a three-level nested key, and the second has three values in a two-level # The first array has four values in a three-level nested key,
# nested key. The total combination is calculated as follows: # and the second has three values in a two-level nested key. The
# total combination is calculated as follows:
# 3 x 4 x 2 x 3 x 1 x 3 = 216 # 3 x 4 x 2 x 3 x 1 x 3 = 216
conn.subscribe( conn.subscribe(
TopicArn=topic_arn, TopicArn=topic_arn,
@ -709,8 +721,8 @@ def test_subscribe_invalid_filter_policy():
}, },
) )
except ClientError as err: except ClientError as err:
err.response["Error"]["Code"].should.equal("InvalidParameter") assert err.response["Error"]["Code"] == "InvalidParameter"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Invalid parameter: FilterPolicy: Filter policy is too complex" "Invalid parameter: FilterPolicy: Filter policy is too complex"
) )
@ -720,8 +732,8 @@ def test_check_not_opted_out():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
response = conn.check_if_phone_number_is_opted_out(phoneNumber="+447428545375") response = conn.check_if_phone_number_is_opted_out(phoneNumber="+447428545375")
response.should.contain("isOptedOut") assert "isOptedOut" in response
response["isOptedOut"].should.be(False) assert response["isOptedOut"] is False
@mock_sns @mock_sns
@ -731,8 +743,8 @@ def test_check_opted_out():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
response = conn.check_if_phone_number_is_opted_out(phoneNumber="+447428545399") response = conn.check_if_phone_number_is_opted_out(phoneNumber="+447428545399")
response.should.contain("isOptedOut") assert "isOptedOut" in response
response["isOptedOut"].should.be(True) assert response["isOptedOut"] is True
@mock_sns @mock_sns
@ -749,8 +761,8 @@ def test_list_opted_out():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
response = conn.list_phone_numbers_opted_out() response = conn.list_phone_numbers_opted_out()
response.should.contain("phoneNumbers") assert "phoneNumbers" in response
len(response["phoneNumbers"]).should.be.greater_than(0) assert len(response["phoneNumbers"]) > 0
@mock_sns @mock_sns
@ -763,8 +775,8 @@ def test_opt_in():
conn.opt_in_phone_number(phoneNumber=response["phoneNumbers"][0]) conn.opt_in_phone_number(phoneNumber=response["phoneNumbers"][0])
response = conn.list_phone_numbers_opted_out() response = conn.list_phone_numbers_opted_out()
len(response["phoneNumbers"]).should.be.greater_than(0) assert len(response["phoneNumbers"]) > 0
len(response["phoneNumbers"]).should.be.lower_than(current_len) assert len(response["phoneNumbers"]) < current_len
@mock_sns @mock_sns
@ -774,7 +786,12 @@ def test_confirm_subscription():
conn.confirm_subscription( conn.confirm_subscription(
TopicArn=response["TopicArn"], TopicArn=response["TopicArn"],
Token="2336412f37fb687f5d51e6e241d59b68c4e583a5cee0be6f95bbf97ab8d2441cf47b99e848408adaadf4c197e65f03473d53c4ba398f6abbf38ce2e8ebf7b4ceceb2cd817959bcde1357e58a2861b05288c535822eb88cac3db04f592285249971efc6484194fc4a4586147f16916692", Token=(
"2336412f37fb687f5d51e6e241d59b68c4e583a5cee0be6f95bbf97ab8d"
"2441cf47b99e848408adaadf4c197e65f03473d53c4ba398f6abbf38ce2"
"e8ebf7b4ceceb2cd817959bcde1357e58a2861b05288c535822eb88cac3"
"db04f592285249971efc6484194fc4a4586147f16916692"
),
AuthenticateOnUnsubscribe="true", AuthenticateOnUnsubscribe="true",
) )
@ -783,15 +800,18 @@ def test_confirm_subscription():
def test_get_subscription_attributes_error_not_exists(): def test_get_subscription_attributes_error_not_exists():
# given # given
client = boto3.client("sns", region_name="us-east-1") client = boto3.client("sns", region_name="us-east-1")
sub_arn = f"arn:aws:sqs:us-east-1:{ACCOUNT_ID}:test-queue:66d97e76-31e5-444f-8fa7-b60b680d0d39" sub_arn = (
f"arn:aws:sqs:us-east-1:{ACCOUNT_ID}:test-queue"
":66d97e76-31e5-444f-8fa7-b60b680d0d39"
)
# when # when
with pytest.raises(ClientError) as e: with pytest.raises(ClientError) as exc:
client.get_subscription_attributes(SubscriptionArn=sub_arn) client.get_subscription_attributes(SubscriptionArn=sub_arn)
# then # then
ex = e.value exc_value = exc.value
ex.operation_name.should.equal("GetSubscriptionAttributes") assert exc_value.operation_name == "GetSubscriptionAttributes"
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(404) assert exc_value.response["ResponseMetadata"]["HTTPStatusCode"] == 404
ex.response["Error"]["Code"].should.contain("NotFound") assert "NotFound" in exc_value.response["Error"]["Code"]
ex.response["Error"]["Message"].should.equal("Subscription does not exist") assert exc_value.response["Error"]["Message"] == "Subscription does not exist"

423
tests/test_sns/test_topics_boto3.py
View File

@ -1,13 +1,11 @@
import boto3
import json import json
import pytest
import sure # noqa # pylint: disable=unused-import
import boto3
from botocore.exceptions import ClientError from botocore.exceptions import ClientError
from moto import mock_sns from moto import mock_sns
from moto.sns.models import DEFAULT_EFFECTIVE_DELIVERY_POLICY, DEFAULT_PAGE_SIZE from moto.sns.models import DEFAULT_EFFECTIVE_DELIVERY_POLICY, DEFAULT_PAGE_SIZE
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
import pytest
@mock_sns @mock_sns
@ -18,8 +16,8 @@ def test_create_and_delete_topic():
topics_json = conn.list_topics() topics_json = conn.list_topics()
topics = topics_json["Topics"] topics = topics_json["Topics"]
topics.should.have.length_of(1) assert len(topics) == 1
topics[0]["TopicArn"].should.equal( assert topics[0]["TopicArn"] == (
f"arn:aws:sns:{conn._client_config.region_name}:{ACCOUNT_ID}:{topic_name}" f"arn:aws:sns:{conn._client_config.region_name}:{ACCOUNT_ID}:{topic_name}"
) )
@ -32,7 +30,7 @@ def test_create_and_delete_topic():
# And there should now be 0 topics # And there should now be 0 topics
topics_json = conn.list_topics() topics_json = conn.list_topics()
topics = topics_json["Topics"] topics = topics_json["Topics"]
topics.should.have.length_of(0) assert len(topics) == 0
@mock_sns @mock_sns
@ -55,7 +53,7 @@ def test_create_topic_with_attributes():
topic_arn = topics_json["Topics"][0]["TopicArn"] topic_arn = topics_json["Topics"][0]["TopicArn"]
attributes = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"] attributes = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes["DisplayName"].should.equal("test-topic") assert attributes["DisplayName"] == "test-topic"
@mock_sns @mock_sns
@ -70,7 +68,7 @@ def test_create_topic_with_tags():
) )
topic_arn = response["TopicArn"] topic_arn = response["TopicArn"]
conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"].should.equal( assert conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"] == (
[ [
{"Key": "tag_key_1", "Value": "tag_value_1"}, {"Key": "tag_key_1", "Value": "tag_value_1"},
{"Key": "tag_key_2", "Value": "tag_value_2"}, {"Key": "tag_key_2", "Value": "tag_value_2"},
@ -88,22 +86,23 @@ def test_create_topic_should_be_indempodent():
topic_display_name = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"][ topic_display_name = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"][
"DisplayName" "DisplayName"
] ]
topic_display_name.should.be.equal("should_be_set") assert topic_display_name == "should_be_set"
# recreate topic to prove indempodentcy # recreate topic to prove indempodentcy
topic_arn = conn.create_topic(Name="some-topic")["TopicArn"] topic_arn = conn.create_topic(Name="some-topic")["TopicArn"]
topic_display_name = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"][ topic_display_name = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"][
"DisplayName" "DisplayName"
] ]
topic_display_name.should.be.equal("should_be_set") assert topic_display_name == "should_be_set"
@mock_sns @mock_sns
def test_get_missing_topic(): def test_get_missing_topic():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
conn.get_topic_attributes.when.called_with( with pytest.raises(ClientError):
TopicArn="arn:aws:sns:us-east-1:424242424242:a-fake-arn" conn.get_topic_attributes(
).should.throw(ClientError) TopicArn="arn:aws:sns:us-east-1:424242424242:a-fake-arn"
)
@mock_sns @mock_sns
@ -111,21 +110,21 @@ def test_create_topic_must_meet_constraints():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
common_random_chars = [":", ";", "!", "@", "|", "^", "%"] common_random_chars = [":", ";", "!", "@", "|", "^", "%"]
for char in common_random_chars: for char in common_random_chars:
conn.create_topic.when.called_with(Name=f"no{char}_invalidchar").should.throw( with pytest.raises(ClientError):
ClientError conn.create_topic(Name=f"no{char}_invalidchar")
) with pytest.raises(ClientError):
conn.create_topic.when.called_with(Name="no spaces allowed").should.throw( conn.create_topic(Name="no spaces allowed")
ClientError
)
@mock_sns @mock_sns
def test_create_topic_should_be_of_certain_length(): def test_create_topic_should_be_of_certain_length():
conn = boto3.client("sns", region_name="us-east-1") conn = boto3.client("sns", region_name="us-east-1")
too_short = "" too_short = ""
conn.create_topic.when.called_with(Name=too_short).should.throw(ClientError) with pytest.raises(ClientError):
conn.create_topic(Name=too_short)
too_long = "x" * 257 too_long = "x" * 257
conn.create_topic.when.called_with(Name=too_long).should.throw(ClientError) with pytest.raises(ClientError):
conn.create_topic(Name=too_long)
@mock_sns @mock_sns
@ -134,7 +133,7 @@ def test_create_topic_in_multiple_regions():
conn = boto3.client("sns", region_name=region) conn = boto3.client("sns", region_name=region)
topic_arn = conn.create_topic(Name="some-topic")["TopicArn"] topic_arn = conn.create_topic(Name="some-topic")["TopicArn"]
# We can find the topic # We can find the topic
list(conn.list_topics()["Topics"]).should.have.length_of(1) assert len(list(conn.list_topics()["Topics"])) == 1
# We can read the Topic details # We can read the Topic details
topic = boto3.resource("sns", region_name=region).Topic(topic_arn) topic = boto3.resource("sns", region_name=region).Topic(topic_arn)
@ -158,7 +157,7 @@ def test_topic_corresponds_to_region():
conn.create_topic(Name="some-topic") conn.create_topic(Name="some-topic")
topics_json = conn.list_topics() topics_json = conn.list_topics()
topic_arn = topics_json["Topics"][0]["TopicArn"] topic_arn = topics_json["Topics"][0]["TopicArn"]
topic_arn.should.equal(f"arn:aws:sns:{region}:{ACCOUNT_ID}:some-topic") assert topic_arn == f"arn:aws:sns:{region}:{ACCOUNT_ID}:some-topic"
@mock_sns @mock_sns
@ -170,42 +169,40 @@ def test_topic_attributes():
topic_arn = topics_json["Topics"][0]["TopicArn"] topic_arn = topics_json["Topics"][0]["TopicArn"]
attributes = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"] attributes = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes["TopicArn"].should.equal( assert attributes["TopicArn"] == (
f"arn:aws:sns:{conn._client_config.region_name}:{ACCOUNT_ID}:some-topic" f"arn:aws:sns:{conn._client_config.region_name}:{ACCOUNT_ID}:some-topic"
) )
attributes["Owner"].should.equal(ACCOUNT_ID) assert attributes["Owner"] == ACCOUNT_ID
json.loads(attributes["Policy"]).should.equal( assert json.loads(attributes["Policy"]) == {
{ "Version": "2008-10-17",
"Version": "2008-10-17", "Id": "__default_policy_ID",
"Id": "__default_policy_ID", "Statement": [
"Statement": [ {
{ "Effect": "Allow",
"Effect": "Allow", "Sid": "__default_statement_ID",
"Sid": "__default_statement_ID", "Principal": {"AWS": "*"},
"Principal": {"AWS": "*"}, "Action": [
"Action": [ "SNS:GetTopicAttributes",
"SNS:GetTopicAttributes", "SNS:SetTopicAttributes",
"SNS:SetTopicAttributes", "SNS:AddPermission",
"SNS:AddPermission", "SNS:RemovePermission",
"SNS:RemovePermission", "SNS:DeleteTopic",
"SNS:DeleteTopic", "SNS:Subscribe",
"SNS:Subscribe", "SNS:ListSubscriptionsByTopic",
"SNS:ListSubscriptionsByTopic", "SNS:Publish",
"SNS:Publish", "SNS:Receive",
"SNS:Receive", ],
], "Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:some-topic",
"Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:some-topic", "Condition": {"StringEquals": {"AWS:SourceOwner": ACCOUNT_ID}},
"Condition": {"StringEquals": {"AWS:SourceOwner": ACCOUNT_ID}}, }
} ],
], }
} assert attributes["DisplayName"] == ""
) assert attributes["SubscriptionsPending"] == "0"
attributes["DisplayName"].should.equal("") assert attributes["SubscriptionsConfirmed"] == "0"
attributes["SubscriptionsPending"].should.equal("0") assert attributes["SubscriptionsDeleted"] == "0"
attributes["SubscriptionsConfirmed"].should.equal("0") assert attributes["DeliveryPolicy"] == ""
attributes["SubscriptionsDeleted"].should.equal("0") assert json.loads(attributes["EffectiveDeliveryPolicy"]) == (
attributes["DeliveryPolicy"].should.equal("")
json.loads(attributes["EffectiveDeliveryPolicy"]).should.equal(
DEFAULT_EFFECTIVE_DELIVERY_POLICY DEFAULT_EFFECTIVE_DELIVERY_POLICY
) )
@ -226,9 +223,9 @@ def test_topic_attributes():
) )
attributes = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"] attributes = conn.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes["Policy"].should.equal('{"foo":"bar"}') assert attributes["Policy"] == '{"foo":"bar"}'
attributes["DisplayName"].should.equal("My display name") assert attributes["DisplayName"] == "My display name"
attributes["DeliveryPolicy"].should.equal( assert attributes["DeliveryPolicy"] == (
'{"http": {"defaultHealthyRetryPolicy": {"numRetries": 5}}}' '{"http": {"defaultHealthyRetryPolicy": {"numRetries": 5}}}'
) )
@ -243,14 +240,14 @@ def test_topic_paging():
topics_list = response["Topics"] topics_list = response["Topics"]
next_token = response["NextToken"] next_token = response["NextToken"]
len(topics_list).should.equal(DEFAULT_PAGE_SIZE) assert len(topics_list) == DEFAULT_PAGE_SIZE
int(next_token).should.equal(DEFAULT_PAGE_SIZE) assert int(next_token) == DEFAULT_PAGE_SIZE
response = conn.list_topics(NextToken=next_token) response = conn.list_topics(NextToken=next_token)
topics_list = response["Topics"] topics_list = response["Topics"]
response.shouldnt.have("NextToken") assert "NextToken" not in response
topics_list.should.have.length_of(int(DEFAULT_PAGE_SIZE / 2)) assert len(topics_list) == int(DEFAULT_PAGE_SIZE / 2)
@mock_sns @mock_sns
@ -266,69 +263,65 @@ def test_add_remove_permissions():
) )
response = client.get_topic_attributes(TopicArn=topic_arn) response = client.get_topic_attributes(TopicArn=topic_arn)
json.loads(response["Attributes"]["Policy"]).should.equal( assert json.loads(response["Attributes"]["Policy"]) == {
{ "Version": "2008-10-17",
"Version": "2008-10-17", "Id": "__default_policy_ID",
"Id": "__default_policy_ID", "Statement": [
"Statement": [ {
{ "Effect": "Allow",
"Effect": "Allow", "Sid": "__default_statement_ID",
"Sid": "__default_statement_ID", "Principal": {"AWS": "*"},
"Principal": {"AWS": "*"}, "Action": [
"Action": [ "SNS:GetTopicAttributes",
"SNS:GetTopicAttributes", "SNS:SetTopicAttributes",
"SNS:SetTopicAttributes", "SNS:AddPermission",
"SNS:AddPermission", "SNS:RemovePermission",
"SNS:RemovePermission", "SNS:DeleteTopic",
"SNS:DeleteTopic", "SNS:Subscribe",
"SNS:Subscribe", "SNS:ListSubscriptionsByTopic",
"SNS:ListSubscriptionsByTopic", "SNS:Publish",
"SNS:Publish", "SNS:Receive",
"SNS:Receive", ],
], "Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions",
"Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions", "Condition": {"StringEquals": {"AWS:SourceOwner": ACCOUNT_ID}},
"Condition": {"StringEquals": {"AWS:SourceOwner": ACCOUNT_ID}}, },
}, {
{ "Sid": "test",
"Sid": "test", "Effect": "Allow",
"Effect": "Allow", "Principal": {"AWS": "arn:aws:iam::999999999999:root"},
"Principal": {"AWS": "arn:aws:iam::999999999999:root"}, "Action": "SNS:Publish",
"Action": "SNS:Publish", "Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions",
"Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions", },
}, ],
], }
}
)
client.remove_permission(TopicArn=topic_arn, Label="test") client.remove_permission(TopicArn=topic_arn, Label="test")
response = client.get_topic_attributes(TopicArn=topic_arn) response = client.get_topic_attributes(TopicArn=topic_arn)
json.loads(response["Attributes"]["Policy"]).should.equal( assert json.loads(response["Attributes"]["Policy"]) == {
{ "Version": "2008-10-17",
"Version": "2008-10-17", "Id": "__default_policy_ID",
"Id": "__default_policy_ID", "Statement": [
"Statement": [ {
{ "Effect": "Allow",
"Effect": "Allow", "Sid": "__default_statement_ID",
"Sid": "__default_statement_ID", "Principal": {"AWS": "*"},
"Principal": {"AWS": "*"}, "Action": [
"Action": [ "SNS:GetTopicAttributes",
"SNS:GetTopicAttributes", "SNS:SetTopicAttributes",
"SNS:SetTopicAttributes", "SNS:AddPermission",
"SNS:AddPermission", "SNS:RemovePermission",
"SNS:RemovePermission", "SNS:DeleteTopic",
"SNS:DeleteTopic", "SNS:Subscribe",
"SNS:Subscribe", "SNS:ListSubscriptionsByTopic",
"SNS:ListSubscriptionsByTopic", "SNS:Publish",
"SNS:Publish", "SNS:Receive",
"SNS:Receive", ],
], "Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions",
"Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions", "Condition": {"StringEquals": {"AWS:SourceOwner": ACCOUNT_ID}},
"Condition": {"StringEquals": {"AWS:SourceOwner": ACCOUNT_ID}}, }
} ],
], }
}
)
client.add_permission( client.add_permission(
TopicArn=topic_arn, TopicArn=topic_arn,
@ -338,20 +331,18 @@ def test_add_remove_permissions():
) )
response = client.get_topic_attributes(TopicArn=topic_arn) response = client.get_topic_attributes(TopicArn=topic_arn)
json.loads(response["Attributes"]["Policy"])["Statement"][1].should.equal( assert json.loads(response["Attributes"]["Policy"])["Statement"][1] == {
{ "Sid": "test",
"Sid": "test", "Effect": "Allow",
"Effect": "Allow", "Principal": {
"Principal": { "AWS": [
"AWS": [ "arn:aws:iam::888888888888:root",
"arn:aws:iam::888888888888:root", "arn:aws:iam::999999999999:root",
"arn:aws:iam::999999999999:root", ]
] },
}, "Action": ["SNS:Publish", "SNS:Subscribe"],
"Action": ["SNS:Publish", "SNS:Subscribe"], "Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions",
"Resource": f"arn:aws:sns:us-east-1:{ACCOUNT_ID}:test-permissions", }
}
)
# deleting non existing permission should be successful # deleting non existing permission should be successful
client.remove_permission(TopicArn=topic_arn, Label="non-existing") client.remove_permission(TopicArn=topic_arn, Label="non-existing")
@ -368,30 +359,36 @@ def test_add_permission_errors():
ActionName=["Publish"], ActionName=["Publish"],
) )
client.add_permission.when.called_with( with pytest.raises(ClientError) as client_err:
TopicArn=topic_arn, client.add_permission(
Label="test", TopicArn=topic_arn,
AWSAccountId=["999999999999"], Label="test",
ActionName=["AddPermission"], AWSAccountId=["999999999999"],
).should.throw(ClientError, "Statement already exists") ActionName=["AddPermission"],
)
assert client_err.value.response["Error"]["Message"] == "Statement already exists"
client.add_permission.when.called_with( with pytest.raises(ClientError) as client_err:
TopicArn=topic_arn + "-not-existing", client.add_permission(
Label="test-2", TopicArn=topic_arn + "-not-existing",
AWSAccountId=["999999999999"], Label="test-2",
ActionName=["AddPermission"], AWSAccountId=["999999999999"],
).should.throw( ActionName=["AddPermission"],
ClientError, )
"An error occurred (NotFound) when calling the AddPermission operation: Topic does not exist", assert client_err.value.response["Error"]["Code"] == "NotFound"
assert client_err.value.response["Error"]["Message"] == "Topic does not exist"
with pytest.raises(ClientError) as client_err:
client.add_permission(
TopicArn=topic_arn,
Label="test-2",
AWSAccountId=["999999999999"],
ActionName=["NotExistingAction"],
)
assert client_err.value.response["Error"]["Message"] == (
"Policy statement action out of service scope!"
) )
client.add_permission.when.called_with(
TopicArn=topic_arn,
Label="test-2",
AWSAccountId=["999999999999"],
ActionName=["NotExistingAction"],
).should.throw(ClientError, "Policy statement action out of service scope!")
@mock_sns @mock_sns
def test_remove_permission_errors(): def test_remove_permission_errors():
@ -404,12 +401,11 @@ def test_remove_permission_errors():
ActionName=["Publish"], ActionName=["Publish"],
) )
client.remove_permission.when.called_with( with pytest.raises(ClientError) as client_err:
TopicArn=topic_arn + "-not-existing", Label="test" client.remove_permission(TopicArn=topic_arn + "-not-existing", Label="test")
).should.throw(
ClientError, assert client_err.value.response["Error"]["Code"] == "NotFound"
"An error occurred (NotFound) when calling the RemovePermission operation: Topic does not exist", assert client_err.value.response["Error"]["Message"] == "Topic does not exist"
)
@mock_sns @mock_sns
@ -421,14 +417,14 @@ def test_tag_topic():
conn.tag_resource( conn.tag_resource(
ResourceArn=topic_arn, Tags=[{"Key": "tag_key_1", "Value": "tag_value_1"}] ResourceArn=topic_arn, Tags=[{"Key": "tag_key_1", "Value": "tag_value_1"}]
) )
conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"].should.equal( assert conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"] == (
[{"Key": "tag_key_1", "Value": "tag_value_1"}] [{"Key": "tag_key_1", "Value": "tag_value_1"}]
) )
conn.tag_resource( conn.tag_resource(
ResourceArn=topic_arn, Tags=[{"Key": "tag_key_2", "Value": "tag_value_2"}] ResourceArn=topic_arn, Tags=[{"Key": "tag_key_2", "Value": "tag_value_2"}]
) )
conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"].should.equal( assert conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"] == (
[ [
{"Key": "tag_key_1", "Value": "tag_value_1"}, {"Key": "tag_key_1", "Value": "tag_value_1"},
{"Key": "tag_key_2", "Value": "tag_value_2"}, {"Key": "tag_key_2", "Value": "tag_value_2"},
@ -438,7 +434,7 @@ def test_tag_topic():
conn.tag_resource( conn.tag_resource(
ResourceArn=topic_arn, Tags=[{"Key": "tag_key_1", "Value": "tag_value_X"}] ResourceArn=topic_arn, Tags=[{"Key": "tag_key_1", "Value": "tag_value_X"}]
) )
conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"].should.equal( assert conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"] == (
[ [
{"Key": "tag_key_1", "Value": "tag_value_X"}, {"Key": "tag_key_1", "Value": "tag_value_X"},
{"Key": "tag_key_2", "Value": "tag_value_2"}, {"Key": "tag_key_2", "Value": "tag_value_2"},
@ -459,13 +455,13 @@ def test_untag_topic():
topic_arn = response["TopicArn"] topic_arn = response["TopicArn"]
conn.untag_resource(ResourceArn=topic_arn, TagKeys=["tag_key_1"]) conn.untag_resource(ResourceArn=topic_arn, TagKeys=["tag_key_1"])
conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"].should.equal( assert conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"] == (
[{"Key": "tag_key_2", "Value": "tag_value_2"}] [{"Key": "tag_key_2", "Value": "tag_value_2"}]
) )
# removing a non existing tag should not raise any error # removing a non existing tag should not raise any error
conn.untag_resource(ResourceArn=topic_arn, TagKeys=["not-existing-tag"]) conn.untag_resource(ResourceArn=topic_arn, TagKeys=["not-existing-tag"])
conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"].should.equal( assert conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"] == (
[{"Key": "tag_key_2", "Value": "tag_value_2"}] [{"Key": "tag_key_2", "Value": "tag_value_2"}]
) )
@ -477,9 +473,9 @@ def test_list_tags_for_resource_error():
Name="some-topic-with-tags", Tags=[{"Key": "tag_key_1", "Value": "tag_value_X"}] Name="some-topic-with-tags", Tags=[{"Key": "tag_key_1", "Value": "tag_value_X"}]
) )
conn.list_tags_for_resource.when.called_with( with pytest.raises(ClientError) as client_err:
ResourceArn="not-existing-topic" conn.list_tags_for_resource(ResourceArn="not-existing-topic")
).should.throw(ClientError, "Resource does not exist") assert client_err.value.response["Error"]["Message"] == "Resource does not exist"
@mock_sns @mock_sns
@ -490,22 +486,24 @@ def test_tag_resource_errors():
) )
topic_arn = response["TopicArn"] topic_arn = response["TopicArn"]
conn.tag_resource.when.called_with( with pytest.raises(ClientError) as client_err:
ResourceArn="not-existing-topic", conn.tag_resource(
Tags=[{"Key": "tag_key_1", "Value": "tag_value_1"}], ResourceArn="not-existing-topic",
).should.throw(ClientError, "Resource does not exist") Tags=[{"Key": "tag_key_1", "Value": "tag_value_1"}],
)
assert client_err.value.response["Error"]["Message"] == "Resource does not exist"
too_many_tags = [ too_many_tags = [
{"Key": f"tag_key_{i}", "Value": f"tag_value_{i}"} for i in range(51) {"Key": f"tag_key_{i}", "Value": f"tag_value_{i}"} for i in range(51)
] ]
conn.tag_resource.when.called_with( with pytest.raises(ClientError) as client_err:
ResourceArn=topic_arn, Tags=too_many_tags conn.tag_resource(ResourceArn=topic_arn, Tags=too_many_tags)
).should.throw( assert client_err.value.response["Error"]["Message"] == (
ClientError, "Could not complete request: tag quota of per resource exceeded" "Could not complete request: tag quota of per resource exceeded"
) )
# when the request fails, the tags should not be updated # when the request fails, the tags should not be updated
conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"].should.equal( assert conn.list_tags_for_resource(ResourceArn=topic_arn)["Tags"] == (
[{"Key": "tag_key_1", "Value": "tag_value_X"}] [{"Key": "tag_key_1", "Value": "tag_value_X"}]
) )
@ -517,9 +515,9 @@ def test_untag_resource_error():
Name="some-topic-with-tags", Tags=[{"Key": "tag_key_1", "Value": "tag_value_X"}] Name="some-topic-with-tags", Tags=[{"Key": "tag_key_1", "Value": "tag_value_X"}]
) )
conn.untag_resource.when.called_with( with pytest.raises(ClientError) as client_err:
ResourceArn="not-existing-topic", TagKeys=["tag_key_1"] conn.untag_resource(ResourceArn="not-existing-topic", TagKeys=["tag_key_1"])
).should.throw(ClientError, "Resource does not exist") assert client_err.value.response["Error"]["Message"] == "Resource does not exist"
@mock_sns @mock_sns
@ -534,29 +532,32 @@ def test_create_fifo_topic():
try: try:
conn.create_topic(Name="test_topic", Attributes={"FifoTopic": "true"}) conn.create_topic(Name="test_topic", Attributes={"FifoTopic": "true"})
except ClientError as err: except ClientError as err:
err.response["Error"]["Code"].should.equal("InvalidParameterValue") assert err.response["Error"]["Code"] == "InvalidParameterValue"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Fifo Topic names must end with .fifo and must be made up of only uppercase and lowercase ASCII letters, " "Fifo Topic names must end with .fifo and must be made up of only "
"numbers, underscores, and hyphens, and must be between 1 and 256 characters long." "uppercase and lowercase ASCII letters, numbers, underscores, "
"and hyphens, and must be between 1 and 256 characters long."
) )
err.response["Error"]["Type"].should.equal("Sender") assert err.response["Error"]["Type"] == "Sender"
try: try:
conn.create_topic(Name="test_topic.fifo") conn.create_topic(Name="test_topic.fifo")
except ClientError as err: except ClientError as err:
err.response["Error"]["Code"].should.equal("InvalidParameterValue") assert err.response["Error"]["Code"] == "InvalidParameterValue"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Topic names must be made up of only uppercase and lowercase ASCII letters, numbers, underscores, " "Topic names must be made up of only uppercase and lowercase "
"ASCII letters, numbers, underscores, "
"and hyphens, and must be between 1 and 256 characters long." "and hyphens, and must be between 1 and 256 characters long."
) )
try: try:
conn.create_topic(Name="topic.name.fifo", Attributes={"FifoTopic": "true"}) conn.create_topic(Name="topic.name.fifo", Attributes={"FifoTopic": "true"})
except ClientError as err: except ClientError as err:
err.response["Error"]["Code"].should.equal("InvalidParameterValue") assert err.response["Error"]["Code"] == "InvalidParameterValue"
err.response["Error"]["Message"].should.equal( assert err.response["Error"]["Message"] == (
"Fifo Topic names must end with .fifo and must be made up of only uppercase and lowercase ASCII letters, " "Fifo Topic names must end with .fifo and must be made up of only "
"numbers, underscores, and hyphens, and must be between 1 and 256 characters long." "uppercase and lowercase ASCII letters, numbers, underscores, "
"and hyphens, and must be between 1 and 256 characters long."
) )
@ -566,22 +567,22 @@ def test_topic_kms_master_key_id_attribute():
resp = client.create_topic(Name="test-sns-no-key-attr") resp = client.create_topic(Name="test-sns-no-key-attr")
topic_arn = resp["TopicArn"] topic_arn = resp["TopicArn"]
resp = client.get_topic_attributes(TopicArn=topic_arn) resp = client.get_topic_attributes(TopicArn=topic_arn)
resp["Attributes"].should_not.have.key("KmsMasterKeyId") assert "KmsMasterKeyId" not in resp["Attributes"]
client.set_topic_attributes( client.set_topic_attributes(
TopicArn=topic_arn, AttributeName="KmsMasterKeyId", AttributeValue="test-key" TopicArn=topic_arn, AttributeName="KmsMasterKeyId", AttributeValue="test-key"
) )
resp = client.get_topic_attributes(TopicArn=topic_arn) resp = client.get_topic_attributes(TopicArn=topic_arn)
resp["Attributes"].should.have.key("KmsMasterKeyId") assert "KmsMasterKeyId" in resp["Attributes"]
resp["Attributes"]["KmsMasterKeyId"].should.equal("test-key") assert resp["Attributes"]["KmsMasterKeyId"] == "test-key"
resp = client.create_topic( resp = client.create_topic(
Name="test-sns-with-key-attr", Attributes={"KmsMasterKeyId": "key-id"} Name="test-sns-with-key-attr", Attributes={"KmsMasterKeyId": "key-id"}
) )
topic_arn = resp["TopicArn"] topic_arn = resp["TopicArn"]
resp = client.get_topic_attributes(TopicArn=topic_arn) resp = client.get_topic_attributes(TopicArn=topic_arn)
resp["Attributes"].should.have.key("KmsMasterKeyId") assert "KmsMasterKeyId" in resp["Attributes"]
resp["Attributes"]["KmsMasterKeyId"].should.equal("key-id") assert resp["Attributes"]["KmsMasterKeyId"] == "key-id"
@mock_sns @mock_sns
@ -593,11 +594,11 @@ def test_topic_fifo_get_attributes():
topic_arn = resp["TopicArn"] topic_arn = resp["TopicArn"]
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"] attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes.should.have.key("FifoTopic") assert "FifoTopic" in attributes
attributes.should.have.key("ContentBasedDeduplication") assert "ContentBasedDeduplication" in attributes
attributes["FifoTopic"].should.equal("true") assert attributes["FifoTopic"] == "true"
attributes["ContentBasedDeduplication"].should.equal("false") assert attributes["ContentBasedDeduplication"] == "false"
client.set_topic_attributes( client.set_topic_attributes(
TopicArn=topic_arn, TopicArn=topic_arn,
@ -605,7 +606,7 @@ def test_topic_fifo_get_attributes():
AttributeValue="true", AttributeValue="true",
) )
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"] attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes["ContentBasedDeduplication"].should.equal("true") assert attributes["ContentBasedDeduplication"] == "true"
@mock_sns @mock_sns
@ -615,8 +616,8 @@ def test_topic_get_attributes():
topic_arn = resp["TopicArn"] topic_arn = resp["TopicArn"]
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"] attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes.should_not.have.key("FifoTopic") assert "FifoTopic" not in attributes
attributes.should_not.have.key("ContentBasedDeduplication") assert "ContentBasedDeduplication" not in attributes
@mock_sns @mock_sns
@ -628,5 +629,5 @@ def test_topic_get_attributes_with_fifo_false():
topic_arn = resp["TopicArn"] topic_arn = resp["TopicArn"]
attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"] attributes = client.get_topic_attributes(TopicArn=topic_arn)["Attributes"]
attributes.should_not.have.key("FifoTopic") assert "FifoTopic" not in attributes
attributes.should_not.have.key("ContentBasedDeduplication") assert "ContentBasedDeduplication" not in attributes

3
tests/test_sns/test_utils.py
View File

@ -1,6 +1,7 @@
from moto.sns.utils import FilterPolicyMatcher
import pytest import pytest
from moto.sns.utils import FilterPolicyMatcher
def test_filter_policy_matcher_scope_sanity_check(): def test_filter_policy_matcher_scope_sanity_check():
with pytest.raises(FilterPolicyMatcher.CheckException): with pytest.raises(FilterPolicyMatcher.CheckException):