SNS create_topic should only accept valid characters (#1329)
* SNS create_topic should only accept valid characters Closes #1328 * Fix flake8 errors * Updated regex to match full TopicName constraints Fixed test_server as it contained invalid TopicNames’ per constraints * fix error message for invalid topic name
This commit is contained in:
parent
abd4d09afe
commit
2ad0f2fc1c
@ -32,3 +32,11 @@ class SNSInvalidParameter(RESTError):
|
|||||||
def __init__(self, message):
|
def __init__(self, message):
|
||||||
super(SNSInvalidParameter, self).__init__(
|
super(SNSInvalidParameter, self).__init__(
|
||||||
"InvalidParameter", message)
|
"InvalidParameter", message)
|
||||||
|
|
||||||
|
|
||||||
|
class InvalidParameterValue(RESTError):
|
||||||
|
code = 400
|
||||||
|
|
||||||
|
def __init__(self, message):
|
||||||
|
super(InvalidParameterValue, self).__init__(
|
||||||
|
"InvalidParameterValue", message)
|
||||||
|
@ -7,6 +7,7 @@ import json
|
|||||||
import boto.sns
|
import boto.sns
|
||||||
import requests
|
import requests
|
||||||
import six
|
import six
|
||||||
|
import re
|
||||||
|
|
||||||
from moto.compat import OrderedDict
|
from moto.compat import OrderedDict
|
||||||
from moto.core import BaseBackend, BaseModel
|
from moto.core import BaseBackend, BaseModel
|
||||||
@ -15,7 +16,8 @@ from moto.sqs import sqs_backends
|
|||||||
from moto.awslambda import lambda_backends
|
from moto.awslambda import lambda_backends
|
||||||
|
|
||||||
from .exceptions import (
|
from .exceptions import (
|
||||||
SNSNotFoundError, DuplicateSnsEndpointError, SnsEndpointDisabled, SNSInvalidParameter
|
SNSNotFoundError, DuplicateSnsEndpointError, SnsEndpointDisabled, SNSInvalidParameter,
|
||||||
|
InvalidParameterValue
|
||||||
)
|
)
|
||||||
from .utils import make_arn_for_topic, make_arn_for_subscription
|
from .utils import make_arn_for_topic, make_arn_for_subscription
|
||||||
|
|
||||||
@ -193,6 +195,9 @@ class SNSBackend(BaseBackend):
|
|||||||
self.sms_attributes.update(attrs)
|
self.sms_attributes.update(attrs)
|
||||||
|
|
||||||
def create_topic(self, name):
|
def create_topic(self, name):
|
||||||
|
fails_constraints = not re.match(r'^[a-zA-Z0-9](?:[A-Za-z0-9_-]{0,253}[a-zA-Z0-9])?$', name)
|
||||||
|
if fails_constraints:
|
||||||
|
raise InvalidParameterValue("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.")
|
||||||
candidate_topic = Topic(name, self)
|
candidate_topic = Topic(name, self)
|
||||||
if candidate_topic.arn in self.topics:
|
if candidate_topic.arn in self.topics:
|
||||||
return self.topics[candidate_topic.arn]
|
return self.topics[candidate_topic.arn]
|
||||||
|
@ -13,12 +13,12 @@ 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="test topic")
|
topic_data = test_client.action_data("CreateTopic", Name="testtopic")
|
||||||
topic_data.should.contain("CreateTopicResult")
|
topic_data.should.contain("CreateTopicResult")
|
||||||
topic_data.should.contain(
|
topic_data.should.contain(
|
||||||
"<TopicArn>arn:aws:sns:us-east-1:123456789012:test topic</TopicArn>")
|
"<TopicArn>arn:aws:sns:us-east-1:123456789012:testtopic</TopicArn>")
|
||||||
|
|
||||||
topics_data = test_client.action_data("ListTopics")
|
topics_data = test_client.action_data("ListTopics")
|
||||||
topics_data.should.contain("ListTopicsResult")
|
topics_data.should.contain("ListTopicsResult")
|
||||||
topic_data.should.contain(
|
topic_data.should.contain(
|
||||||
"<TopicArn>arn:aws:sns:us-east-1:123456789012:test topic</TopicArn>")
|
"<TopicArn>arn:aws:sns:us-east-1:123456789012:testtopic</TopicArn>")
|
||||||
|
@ -50,17 +50,35 @@ def test_create_topic_should_be_indempodent():
|
|||||||
topic_display_name = conn.get_topic_attributes(
|
topic_display_name = conn.get_topic_attributes(
|
||||||
TopicArn=topic_arn
|
TopicArn=topic_arn
|
||||||
)['Attributes']['DisplayName']
|
)['Attributes']['DisplayName']
|
||||||
|
|
||||||
topic_display_name.should.be.equal("should_be_set")
|
topic_display_name.should.be.equal("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(
|
conn.get_topic_attributes.when.called_with(
|
||||||
TopicArn="a-fake-arn").should.throw(ClientError)
|
TopicArn="a-fake-arn").should.throw(ClientError)
|
||||||
|
|
||||||
|
@mock_sns
|
||||||
|
def test_create_topic_must_meet_constraints():
|
||||||
|
conn = boto3.client("sns", region_name="us-east-1")
|
||||||
|
common_random_chars = [':', ";", "!", "@", "|", "^", "%"]
|
||||||
|
for char in common_random_chars:
|
||||||
|
conn.create_topic.when.called_with(
|
||||||
|
Name="no%s_invalidchar" % char).should.throw(ClientError)
|
||||||
|
conn.create_topic.when.called_with(
|
||||||
|
Name="no spaces allowed").should.throw(ClientError)
|
||||||
|
|
||||||
|
|
||||||
|
@mock_sns
|
||||||
|
def test_create_topic_should_be_of_certain_length():
|
||||||
|
conn = boto3.client("sns", region_name="us-east-1")
|
||||||
|
too_short = ""
|
||||||
|
conn.create_topic.when.called_with(
|
||||||
|
Name=too_short).should.throw(ClientError)
|
||||||
|
too_long = "x" * 257
|
||||||
|
conn.create_topic.when.called_with(
|
||||||
|
Name=too_long).should.throw(ClientError)
|
||||||
|
|
||||||
|
|
||||||
@mock_sns
|
@mock_sns
|
||||||
def test_create_topic_in_multiple_regions():
|
def test_create_topic_in_multiple_regions():
|
||||||
|
Loading…
x
Reference in New Issue
Block a user