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

Linting

Browse Source
This commit is contained in:
Bert Blommers 2020-05-04 09:24:46 +01:00
parent 440213f854
commit 353bc08ac2
5 changed files with 80 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
moto/ses/exceptions.py
View File

@ -8,15 +8,20 @@ class MessageRejectedError(RESTError):
def __init__(self, message):
super(MessageRejectedError, self).__init__("MessageRejected", message)
class ConfigurationSetDoesNotExist(RESTError):
code = 400
def __init__(self, message):
super(ConfigurationSetDoesNotExist, self).__init__("ConfigurationSetDoesNotExist", message)
super(ConfigurationSetDoesNotExist, self).__init__(
"ConfigurationSetDoesNotExist", message
)
class EventDestinationAlreadyExists(RESTError):
code = 400
def __init__(self, message):
super(EventDestinationAlreadyExists, self).__init__("EventDestinationAlreadyExists", message)
super(EventDestinationAlreadyExists, self).__init__(
"EventDestinationAlreadyExists", message
)

12
moto/ses/models.py
View File

@ -6,7 +6,11 @@ from email.utils import parseaddr
from moto.core import BaseBackend, BaseModel
from moto.sns.models import sns_backends
from .exceptions import MessageRejectedError,ConfigurationSetDoesNotExist,EventDestinationAlreadyExists
from .exceptions import (
MessageRejectedError,
ConfigurationSetDoesNotExist,
EventDestinationAlreadyExists,
)
from .utils import get_random_message_id
from .feedback import COMMON_MAIL, BOUNCE, COMPLAINT, DELIVERY
@ -123,7 +127,7 @@ class SESBackend(BaseBackend):
if recipient_count > RECIPIENT_LIMIT:
raise MessageRejectedError("Too many recipients.")
if not self._is_verified_address(source):
self.rejected_messages_count+=1
self.rejected_messages_count += 1
raise MessageRejectedError("Email address not verified %s" % source)
self.__process_sns_feedback__(source, destinations, region)
@ -248,7 +252,9 @@ class SESBackend(BaseBackend):
self.config_set[configuration_set_name] = 1
return {}
def create_configuration_set_event_destination(self,configuration_set_name, event_destination):
def create_configuration_set_event_destination(
self, configuration_set_name, event_destination
):
if self.config_set.get(configuration_set_name) is None:
raise ConfigurationSetDoesNotExist("Invalid Configuration Set Name.")

36
moto/ses/responses.py
View File

@ -140,34 +140,42 @@ class EmailResponse(BaseResponse):
def create_configuration_set(self):
configuration_set_name = self.querystring.get("ConfigurationSet.Name")[0]
ses_backend.create_configuration_set(configuration_set_name=configuration_set_name)
ses_backend.create_configuration_set(
configuration_set_name=configuration_set_name
)
template = self.response_template(CREATE_CONFIGURATION_SET)
return template.render()
def create_configuration_set_event_destination(self):
configuration_set_name = self._get_param('ConfigurationSetName')
is_configuration_event_enabled = self.querystring.get("EventDestination.Enabled")[0]
configuration_set_name = self._get_param("ConfigurationSetName")
is_configuration_event_enabled = self.querystring.get(
"EventDestination.Enabled"
)[0]
configuration_event_name = self.querystring.get("EventDestination.Name")[0]
event_topic_arn = self.querystring.get("EventDestination.SNSDestination.TopicARN")[0]
event_matching_types = self._get_multi_param("EventDestination.MatchingEventTypes.member")
event_topic_arn = self.querystring.get(
"EventDestination.SNSDestination.TopicARN"
)[0]
event_matching_types = self._get_multi_param(
"EventDestination.MatchingEventTypes.member"
)
event_destination = {"Name":configuration_event_name,
"Enabled":is_configuration_event_enabled,
"EventMatchingTypes":event_matching_types,
"SNSDestination":event_topic_arn
}
event_destination = {
"Name": configuration_event_name,
"Enabled": is_configuration_event_enabled,
"EventMatchingTypes": event_matching_types,
"SNSDestination": event_topic_arn,
}
ses_backend.create_configuration_set_event_destination(
configuration_set_name=configuration_set_name,
event_destination=event_destination
)
configuration_set_name=configuration_set_name,
event_destination=event_destination,
)
template = self.response_template(CREATE_CONFIGURATION_SET_EVENT_DESTINATION)
return template.render()
VERIFY_EMAIL_IDENTITY = """<VerifyEmailIdentityResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
<VerifyEmailIdentityResult/>
<ResponseMetadata>

18
tests/test_ses/test_ses.py
View File

@ -144,8 +144,12 @@ def test_get_send_statistics():
# tests to verify rejects in get_send_statistics
result = conn.get_send_statistics()
reject_count = int(result["GetSendStatisticsResponse"]["SendDataPoints"][0]["Rejects"])
delivery_count = int(result["GetSendStatisticsResponse"]["SendDataPoints"][0]["DeliveryAttempts"])
reject_count = int(
result["GetSendStatisticsResponse"]["SendDataPoints"][0]["Rejects"]
)
delivery_count = int(
result["GetSendStatisticsResponse"]["SendDataPoints"][0]["DeliveryAttempts"]
)
reject_count.should.equal(1)
delivery_count.should.equal(0)
@ -157,9 +161,11 @@ def test_get_send_statistics():
# tests to delivery attempts in get_send_statistics
result = conn.get_send_statistics()
reject_count = int(result["GetSendStatisticsResponse"]["SendDataPoints"][0]["Rejects"])
delivery_count = int(result["GetSendStatisticsResponse"]["SendDataPoints"][0]["DeliveryAttempts"])
reject_count = int(
result["GetSendStatisticsResponse"]["SendDataPoints"][0]["Rejects"]
)
delivery_count = int(
result["GetSendStatisticsResponse"]["SendDataPoints"][0]["DeliveryAttempts"]
)
reject_count.should.equal(1)
delivery_count.should.equal(1)

62
tests/test_ses/test_ses_boto3.py
View File

@ -230,52 +230,50 @@ def test_send_email_notification_with_encoded_sender():
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@mock_ses
def test_create_configuration_set():
conn = boto3.client("ses", region_name="us-east-1")
conn.create_configuration_set(ConfigurationSet=dict({"Name": "test"}))
conn.create_configuration_set_event_destination(
ConfigurationSetName='test',
EventDestination={
'Name': 'snsEvent',
'Enabled': True,
'MatchingEventTypes': [
'send',
],
'SNSDestination': {
'TopicARN': 'arn:aws:sns:us-east-1:123456789012:myTopic'
}
})
ConfigurationSetName="test",
EventDestination={
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
},
)
with assert_raises(ClientError) as ex:
conn.create_configuration_set_event_destination(
ConfigurationSetName='failtest',
ConfigurationSetName="failtest",
EventDestination={
'Name': 'snsEvent',
'Enabled': True,
'MatchingEventTypes': [
'send',
],
'SNSDestination': {
'TopicARN': 'arn:aws:sns:us-east-1:123456789012:myTopic'
}
})
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
},
)
ex.exception.response["Error"]["Code"].should.equal("ConfigurationSetDoesNotExist")
with assert_raises(ClientError) as ex:
conn.create_configuration_set_event_destination(
ConfigurationSetName='test',
ConfigurationSetName="test",
EventDestination={
'Name': 'snsEvent',
'Enabled': True,
'MatchingEventTypes': [
'send',
],
'SNSDestination': {
'TopicARN': 'arn:aws:sns:us-east-1:123456789012:myTopic'
}
})
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
},
)
ex.exception.response["Error"]["Code"].should.equal("EventDestinationAlreadyExists")
ex.exception.response["Error"]["Code"].should.equal("EventDestinationAlreadyExists")