Techdebt: MyPy SES (#6252)
This commit is contained in:
Bert Blommers
GitHub
parent
fe7b15c9f9
commit
65611082be
@ -4,98 +4,98 @@ from moto.core.exceptions import RESTError
|
||||
class MessageRejectedError(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("MessageRejected", message)
|
||||
|
||||
|
||||
class ConfigurationSetDoesNotExist(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("ConfigurationSetDoesNotExist", message)
|
||||
|
||||
|
||||
class ConfigurationSetAlreadyExists(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("ConfigurationSetAlreadyExists", message)
|
||||
|
||||
|
||||
class EventDestinationAlreadyExists(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("EventDestinationAlreadyExists", message)
|
||||
|
||||
|
||||
class TemplateNameAlreadyExists(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("TemplateNameAlreadyExists", message)
|
||||
|
||||
|
||||
class ValidationError(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("ValidationError", message)
|
||||
|
||||
|
||||
class InvalidParameterValue(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("InvalidParameterValue", message)
|
||||
|
||||
|
||||
class InvalidRenderingParameterException:
|
||||
class InvalidRenderingParameterException(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("InvalidRenderingParameterException", message)
|
||||
|
||||
|
||||
class TemplateDoesNotExist(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("TemplateDoesNotExist", message)
|
||||
|
||||
|
||||
class RuleSetNameAlreadyExists(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("RuleSetNameAlreadyExists", message)
|
||||
|
||||
|
||||
class RuleAlreadyExists(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("RuleAlreadyExists", message)
|
||||
|
||||
|
||||
class RuleSetDoesNotExist(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("RuleSetDoesNotExist", message)
|
||||
|
||||
|
||||
class RuleDoesNotExist(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, message):
|
||||
def __init__(self, message: str):
|
||||
super().__init__("RuleDoesNotExist", message)
|
||||
|
||||
|
||||
class MissingRenderingAttributeException(RESTError):
|
||||
code = 400
|
||||
|
||||
def __init__(self, var):
|
||||
def __init__(self, var: str):
|
||||
super().__init__(
|
||||
"MissingRenderingAttributeException",
|
||||
f"Attribute '{var}' is not present in the rendering data.",
|
||||
|
||||
@ -5,7 +5,7 @@ from email.mime.base import MIMEBase
|
||||
from email.utils import parseaddr
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.encoders import encode_7or8bit
|
||||
from typing import Mapping
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from moto.core import BaseBackend, BackendDict, BaseModel
|
||||
from moto.sns.models import sns_backends
|
||||
@ -48,8 +48,8 @@ class SESFeedback(BaseModel):
|
||||
FORWARDING_ENABLED = "feedback_forwarding_enabled"
|
||||
|
||||
@staticmethod
|
||||
def generate_message(account_id, msg_type):
|
||||
msg = dict(COMMON_MAIL)
|
||||
def generate_message(account_id: str, msg_type: str) -> Dict[str, Any]: # type: ignore[misc]
|
||||
msg: Dict[str, Any] = dict(COMMON_MAIL)
|
||||
msg["mail"]["sendingAccountId"] = account_id
|
||||
if msg_type == SESFeedback.BOUNCE:
|
||||
msg["bounce"] = BOUNCE
|
||||
@ -62,7 +62,14 @@ class SESFeedback(BaseModel):
|
||||
|
||||
|
||||
class Message(BaseModel):
|
||||
def __init__(self, message_id, source, subject, body, destinations):
|
||||
def __init__(
|
||||
self,
|
||||
message_id: str,
|
||||
source: str,
|
||||
subject: str,
|
||||
body: str,
|
||||
destinations: Dict[str, List[str]],
|
||||
):
|
||||
self.id = message_id
|
||||
self.source = source
|
||||
self.subject = subject
|
||||
@ -71,7 +78,14 @@ class Message(BaseModel):
|
||||
|
||||
|
||||
class TemplateMessage(BaseModel):
|
||||
def __init__(self, message_id, source, template, template_data, destinations):
|
||||
def __init__(
|
||||
self,
|
||||
message_id: str,
|
||||
source: str,
|
||||
template: List[str],
|
||||
template_data: List[str],
|
||||
destinations: Any,
|
||||
):
|
||||
self.id = message_id
|
||||
self.source = source
|
||||
self.template = template
|
||||
@ -80,7 +94,14 @@ class TemplateMessage(BaseModel):
|
||||
|
||||
|
||||
class BulkTemplateMessage(BaseModel):
|
||||
def __init__(self, message_ids, source, template, template_data, destinations):
|
||||
def __init__(
|
||||
self,
|
||||
message_ids: List[str],
|
||||
source: str,
|
||||
template: List[str],
|
||||
template_data: List[str],
|
||||
destinations: Any,
|
||||
):
|
||||
self.ids = message_ids
|
||||
self.source = source
|
||||
self.template = template
|
||||
@ -89,7 +110,9 @@ class BulkTemplateMessage(BaseModel):
|
||||
|
||||
|
||||
class RawMessage(BaseModel):
|
||||
def __init__(self, message_id, source, destinations, raw_data):
|
||||
def __init__(
|
||||
self, message_id: str, source: str, destinations: List[str], raw_data: str
|
||||
):
|
||||
self.id = message_id
|
||||
self.source = source
|
||||
self.destinations = destinations
|
||||
@ -97,11 +120,11 @@ class RawMessage(BaseModel):
|
||||
|
||||
|
||||
class SESQuota(BaseModel):
|
||||
def __init__(self, sent):
|
||||
def __init__(self, sent: int):
|
||||
self.sent = sent
|
||||
|
||||
@property
|
||||
def sent_past_24(self):
|
||||
def sent_past_24(self) -> int:
|
||||
return self.sent
|
||||
|
||||
|
||||
@ -121,23 +144,23 @@ class SESBackend(BaseBackend):
|
||||
Note that, as this is an internal API, the exact format may differ per versions.
|
||||
"""
|
||||
|
||||
def __init__(self, region_name, account_id):
|
||||
def __init__(self, region_name: str, account_id: str):
|
||||
super().__init__(region_name, account_id)
|
||||
self.addresses = []
|
||||
self.email_addresses = []
|
||||
self.domains = []
|
||||
self.sent_messages = []
|
||||
self.addresses: List[str] = []
|
||||
self.email_addresses: List[str] = []
|
||||
self.domains: List[str] = []
|
||||
self.sent_messages: List[Any] = []
|
||||
self.sent_message_count = 0
|
||||
self.rejected_messages_count = 0
|
||||
self.sns_topics = {}
|
||||
self.config_set = {}
|
||||
self.config_set_event_destination = {}
|
||||
self.event_destinations = {}
|
||||
self.identity_mail_from_domains = {}
|
||||
self.templates = {}
|
||||
self.receipt_rule_set = {}
|
||||
self.sns_topics: Dict[str, Dict[str, Any]] = {}
|
||||
self.config_set: Dict[str, int] = {}
|
||||
self.config_set_event_destination: Dict[str, Dict[str, Any]] = {}
|
||||
self.event_destinations: Dict[str, int] = {}
|
||||
self.identity_mail_from_domains: Dict[str, Dict[str, Any]] = {}
|
||||
self.templates: Dict[str, Dict[str, str]] = {}
|
||||
self.receipt_rule_set: Dict[str, List[Dict[str, Any]]] = {}
|
||||
|
||||
def _is_verified_address(self, source):
|
||||
def _is_verified_address(self, source: str) -> bool:
|
||||
_, address = parseaddr(source)
|
||||
if address in self.addresses:
|
||||
return True
|
||||
@ -146,32 +169,34 @@ class SESBackend(BaseBackend):
|
||||
_, host = address.split("@", 1)
|
||||
return host in self.domains
|
||||
|
||||
def verify_email_identity(self, address):
|
||||
def verify_email_identity(self, address: str) -> None:
|
||||
_, address = parseaddr(address)
|
||||
if address not in self.addresses:
|
||||
self.addresses.append(address)
|
||||
|
||||
def verify_email_address(self, address):
|
||||
def verify_email_address(self, address: str) -> None:
|
||||
_, address = parseaddr(address)
|
||||
self.email_addresses.append(address)
|
||||
|
||||
def verify_domain(self, domain):
|
||||
def verify_domain(self, domain: str) -> None:
|
||||
if domain.lower() not in self.domains:
|
||||
self.domains.append(domain.lower())
|
||||
|
||||
def list_identities(self):
|
||||
def list_identities(self) -> List[str]:
|
||||
return self.domains + self.addresses
|
||||
|
||||
def list_verified_email_addresses(self):
|
||||
def list_verified_email_addresses(self) -> List[str]:
|
||||
return self.email_addresses
|
||||
|
||||
def delete_identity(self, identity):
|
||||
def delete_identity(self, identity: str) -> None:
|
||||
if "@" in identity:
|
||||
self.addresses.remove(identity)
|
||||
else:
|
||||
self.domains.remove(identity)
|
||||
|
||||
def send_email(self, source, subject, body, destinations, region):
|
||||
def send_email(
|
||||
self, source: str, subject: str, body: str, destinations: Dict[str, List[str]]
|
||||
) -> Message:
|
||||
recipient_count = sum(map(len, destinations.values()))
|
||||
if recipient_count > RECIPIENT_LIMIT:
|
||||
raise MessageRejectedError("Too many recipients.")
|
||||
@ -182,11 +207,11 @@ class SESBackend(BaseBackend):
|
||||
address for addresses in destinations.values() for address in addresses
|
||||
]
|
||||
for address in [source, *destination_addresses]:
|
||||
valid, msg = is_valid_address(address)
|
||||
if not valid:
|
||||
msg = is_valid_address(address)
|
||||
if msg is not None:
|
||||
raise InvalidParameterValue(msg)
|
||||
|
||||
self.__process_sns_feedback__(source, destinations, region)
|
||||
self.__process_sns_feedback__(source, destinations)
|
||||
|
||||
message_id = get_random_message_id()
|
||||
message = Message(message_id, source, subject, body, destinations)
|
||||
@ -195,14 +220,18 @@ class SESBackend(BaseBackend):
|
||||
return message
|
||||
|
||||
def send_bulk_templated_email(
|
||||
self, source, template, template_data, destinations, region
|
||||
):
|
||||
self,
|
||||
source: str,
|
||||
template: List[str],
|
||||
template_data: List[str],
|
||||
destinations: List[Dict[str, Dict[str, List[str]]]],
|
||||
) -> BulkTemplateMessage:
|
||||
recipient_count = len(destinations)
|
||||
if recipient_count > RECIPIENT_LIMIT:
|
||||
raise MessageRejectedError("Too many destinations.")
|
||||
|
||||
total_recipient_count = sum(
|
||||
map(lambda d: sum(map(len, d["Destination"].values())), destinations)
|
||||
map(lambda d: sum(map(len, d["Destination"].values())), destinations) # type: ignore
|
||||
)
|
||||
if total_recipient_count > RECIPIENT_LIMIT:
|
||||
raise MessageRejectedError("Too many destinations.")
|
||||
@ -214,7 +243,7 @@ class SESBackend(BaseBackend):
|
||||
if not self.templates.get(template[0]):
|
||||
raise TemplateDoesNotExist(f"Template ({template[0]}) does not exist")
|
||||
|
||||
self.__process_sns_feedback__(source, destinations, region)
|
||||
self.__process_sns_feedback__(source, destinations)
|
||||
|
||||
message_id = get_random_message_id()
|
||||
message = TemplateMessage(
|
||||
@ -227,8 +256,12 @@ class SESBackend(BaseBackend):
|
||||
return BulkTemplateMessage(ids, source, template, template_data, destinations)
|
||||
|
||||
def send_templated_email(
|
||||
self, source, template, template_data, destinations, region
|
||||
):
|
||||
self,
|
||||
source: str,
|
||||
template: List[str],
|
||||
template_data: List[str],
|
||||
destinations: Dict[str, List[str]],
|
||||
) -> TemplateMessage:
|
||||
recipient_count = sum(map(len, destinations.values()))
|
||||
if recipient_count > RECIPIENT_LIMIT:
|
||||
raise MessageRejectedError("Too many recipients.")
|
||||
@ -239,14 +272,14 @@ class SESBackend(BaseBackend):
|
||||
address for addresses in destinations.values() for address in addresses
|
||||
]
|
||||
for address in [source, *destination_addresses]:
|
||||
valid, msg = is_valid_address(address)
|
||||
if not valid:
|
||||
msg = is_valid_address(address)
|
||||
if msg is not None:
|
||||
raise InvalidParameterValue(msg)
|
||||
|
||||
if not self.templates.get(template[0]):
|
||||
raise TemplateDoesNotExist(f"Template ({template[0]}) does not exist")
|
||||
|
||||
self.__process_sns_feedback__(source, destinations, region)
|
||||
self.__process_sns_feedback__(source, destinations)
|
||||
|
||||
message_id = get_random_message_id()
|
||||
message = TemplateMessage(
|
||||
@ -256,7 +289,7 @@ class SESBackend(BaseBackend):
|
||||
self.sent_message_count += recipient_count
|
||||
return message
|
||||
|
||||
def __type_of_message__(self, destinations):
|
||||
def __type_of_message__(self, destinations: Any) -> Optional[str]:
|
||||
"""Checks the destination for any special address that could indicate delivery,
|
||||
complaint or bounce like in SES simulator"""
|
||||
if isinstance(destinations, list):
|
||||
@ -278,11 +311,11 @@ class SESBackend(BaseBackend):
|
||||
|
||||
return None
|
||||
|
||||
def __generate_feedback__(self, msg_type):
|
||||
def __generate_feedback__(self, msg_type: str) -> Dict[str, Any]:
|
||||
"""Generates the SNS message for the feedback"""
|
||||
return SESFeedback.generate_message(self.account_id, msg_type)
|
||||
|
||||
def __process_sns_feedback__(self, source, destinations, region):
|
||||
def __process_sns_feedback__(self, source: str, destinations: Any) -> None:
|
||||
domain = str(source)
|
||||
if "@" in domain:
|
||||
domain = domain.split("@")[1]
|
||||
@ -293,11 +326,13 @@ class SESBackend(BaseBackend):
|
||||
if sns_topic is not None:
|
||||
message = self.__generate_feedback__(msg_type)
|
||||
if message:
|
||||
sns_backends[self.account_id][region].publish(
|
||||
sns_backends[self.account_id][self.region_name].publish(
|
||||
message, arn=sns_topic
|
||||
)
|
||||
|
||||
def send_raw_email(self, source, destinations, raw_data, region):
|
||||
def send_raw_email(
|
||||
self, source: str, destinations: List[str], raw_data: str
|
||||
) -> RawMessage:
|
||||
if source is not None:
|
||||
_, source_email_address = parseaddr(source)
|
||||
if not self._is_verified_address(source_email_address):
|
||||
@ -324,33 +359,39 @@ class SESBackend(BaseBackend):
|
||||
if recipient_count > RECIPIENT_LIMIT:
|
||||
raise MessageRejectedError("Too many recipients.")
|
||||
for address in [addr for addr in [source, *destinations] if addr is not None]:
|
||||
valid, msg = is_valid_address(address)
|
||||
if not valid:
|
||||
msg = is_valid_address(address)
|
||||
if msg is not None:
|
||||
raise InvalidParameterValue(msg)
|
||||
|
||||
self.__process_sns_feedback__(source, destinations, region)
|
||||
self.__process_sns_feedback__(source, destinations)
|
||||
|
||||
self.sent_message_count += recipient_count
|
||||
message_id = get_random_message_id()
|
||||
message = RawMessage(message_id, source, destinations, raw_data)
|
||||
self.sent_messages.append(message)
|
||||
return message
|
||||
raw_message = RawMessage(message_id, source, destinations, raw_data)
|
||||
self.sent_messages.append(raw_message)
|
||||
return raw_message
|
||||
|
||||
def get_send_quota(self):
|
||||
def get_send_quota(self) -> SESQuota:
|
||||
return SESQuota(self.sent_message_count)
|
||||
|
||||
def get_identity_notification_attributes(self, identities):
|
||||
response = {}
|
||||
def get_identity_notification_attributes(
|
||||
self, identities: List[str]
|
||||
) -> Dict[str, Dict[str, Any]]:
|
||||
response: Dict[str, Dict[str, Any]] = {}
|
||||
for identity in identities:
|
||||
response[identity] = self.sns_topics.get(identity, {})
|
||||
return response
|
||||
|
||||
def set_identity_feedback_forwarding_enabled(self, identity, enabled):
|
||||
def set_identity_feedback_forwarding_enabled(
|
||||
self, identity: str, enabled: bool
|
||||
) -> None:
|
||||
identity_sns_topics = self.sns_topics.get(identity, {})
|
||||
identity_sns_topics[SESFeedback.FORWARDING_ENABLED] = enabled
|
||||
self.sns_topics[identity] = identity_sns_topics
|
||||
|
||||
def set_identity_notification_topic(self, identity, notification_type, sns_topic):
|
||||
def set_identity_notification_topic(
|
||||
self, identity: str, notification_type: str, sns_topic: Optional[str]
|
||||
) -> None:
|
||||
identity_sns_topics = self.sns_topics.get(identity, {})
|
||||
if sns_topic is None:
|
||||
del identity_sns_topics[notification_type]
|
||||
@ -359,26 +400,22 @@ class SESBackend(BaseBackend):
|
||||
|
||||
self.sns_topics[identity] = identity_sns_topics
|
||||
|
||||
return {}
|
||||
|
||||
def create_configuration_set(self, configuration_set_name):
|
||||
def create_configuration_set(self, configuration_set_name: str) -> None:
|
||||
if configuration_set_name in self.config_set:
|
||||
raise ConfigurationSetAlreadyExists(
|
||||
f"Configuration set <{configuration_set_name}> already exists"
|
||||
)
|
||||
self.config_set[configuration_set_name] = 1
|
||||
return {}
|
||||
|
||||
def describe_configuration_set(self, configuration_set_name):
|
||||
def describe_configuration_set(self, configuration_set_name: str) -> None:
|
||||
if configuration_set_name not in self.config_set:
|
||||
raise ConfigurationSetDoesNotExist(
|
||||
f"Configuration set <{configuration_set_name}> does not exist"
|
||||
)
|
||||
return {}
|
||||
|
||||
def create_configuration_set_event_destination(
|
||||
self, configuration_set_name, event_destination
|
||||
):
|
||||
self, configuration_set_name: str, event_destination: Dict[str, Any]
|
||||
) -> None:
|
||||
|
||||
if self.config_set.get(configuration_set_name) is None:
|
||||
raise ConfigurationSetDoesNotExist("Invalid Configuration Set Name.")
|
||||
@ -389,19 +426,16 @@ class SESBackend(BaseBackend):
|
||||
self.config_set_event_destination[configuration_set_name] = event_destination
|
||||
self.event_destinations[event_destination["Name"]] = 1
|
||||
|
||||
return {}
|
||||
def get_send_statistics(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"DeliveryAttempts": self.sent_message_count,
|
||||
"Rejects": self.rejected_messages_count,
|
||||
"Complaints": 0,
|
||||
"Bounces": 0,
|
||||
"Timestamp": datetime.datetime.utcnow(),
|
||||
}
|
||||
|
||||
def get_send_statistics(self):
|
||||
|
||||
statistics = {}
|
||||
statistics["DeliveryAttempts"] = self.sent_message_count
|
||||
statistics["Rejects"] = self.rejected_messages_count
|
||||
statistics["Complaints"] = 0
|
||||
statistics["Bounces"] = 0
|
||||
statistics["Timestamp"] = datetime.datetime.utcnow()
|
||||
return statistics
|
||||
|
||||
def add_template(self, template_info):
|
||||
def add_template(self, template_info: Dict[str, str]) -> None:
|
||||
template_name = template_info["template_name"]
|
||||
if not template_name:
|
||||
raise ValidationError(
|
||||
@ -418,7 +452,7 @@ class SESBackend(BaseBackend):
|
||||
raise InvalidParameterValue("The subject must be specified.")
|
||||
self.templates[template_name] = template_info
|
||||
|
||||
def update_template(self, template_info):
|
||||
def update_template(self, template_info: Dict[str, str]) -> None:
|
||||
template_name = template_info["template_name"]
|
||||
if not template_name:
|
||||
raise ValidationError(
|
||||
@ -435,15 +469,15 @@ class SESBackend(BaseBackend):
|
||||
raise InvalidParameterValue("The subject must be specified.")
|
||||
self.templates[template_name] = template_info
|
||||
|
||||
def get_template(self, template_name):
|
||||
def get_template(self, template_name: str) -> Dict[str, str]:
|
||||
if not self.templates.get(template_name, None):
|
||||
raise TemplateDoesNotExist("Invalid Template Name.")
|
||||
return self.templates[template_name]
|
||||
|
||||
def list_templates(self):
|
||||
def list_templates(self) -> List[Dict[str, str]]:
|
||||
return list(self.templates.values())
|
||||
|
||||
def render_template(self, render_data):
|
||||
def render_template(self, render_data: Dict[str, Any]) -> str:
|
||||
template_name = render_data.get("name", "")
|
||||
template = self.templates.get(template_name, None)
|
||||
if not template:
|
||||
@ -451,7 +485,7 @@ class SESBackend(BaseBackend):
|
||||
|
||||
template_data = render_data.get("data")
|
||||
try:
|
||||
template_data = json.loads(template_data)
|
||||
template_data = json.loads(template_data) # type: ignore
|
||||
except ValueError:
|
||||
raise InvalidRenderingParameterException(
|
||||
"Template rendering data is invalid"
|
||||
@ -484,12 +518,12 @@ class SESBackend(BaseBackend):
|
||||
)
|
||||
return rendered_template
|
||||
|
||||
def create_receipt_rule_set(self, rule_set_name):
|
||||
def create_receipt_rule_set(self, rule_set_name: str) -> None:
|
||||
if self.receipt_rule_set.get(rule_set_name) is not None:
|
||||
raise RuleSetNameAlreadyExists("Duplicate Receipt Rule Set Name.")
|
||||
self.receipt_rule_set[rule_set_name] = []
|
||||
|
||||
def create_receipt_rule(self, rule_set_name, rule):
|
||||
def create_receipt_rule(self, rule_set_name: str, rule: Dict[str, Any]) -> None:
|
||||
rule_set = self.receipt_rule_set.get(rule_set_name)
|
||||
if rule_set is None:
|
||||
raise RuleSetDoesNotExist("Invalid Rule Set Name.")
|
||||
@ -498,7 +532,7 @@ class SESBackend(BaseBackend):
|
||||
rule_set.append(rule)
|
||||
self.receipt_rule_set[rule_set_name] = rule_set
|
||||
|
||||
def describe_receipt_rule_set(self, rule_set_name):
|
||||
def describe_receipt_rule_set(self, rule_set_name: str) -> List[Dict[str, Any]]:
|
||||
rule_set = self.receipt_rule_set.get(rule_set_name)
|
||||
|
||||
if rule_set is None:
|
||||
@ -506,7 +540,9 @@ class SESBackend(BaseBackend):
|
||||
|
||||
return rule_set
|
||||
|
||||
def describe_receipt_rule(self, rule_set_name, rule_name):
|
||||
def describe_receipt_rule(
|
||||
self, rule_set_name: str, rule_name: str
|
||||
) -> Dict[str, Any]:
|
||||
rule_set = self.receipt_rule_set.get(rule_set_name)
|
||||
|
||||
if rule_set is None:
|
||||
@ -518,7 +554,7 @@ class SESBackend(BaseBackend):
|
||||
|
||||
raise RuleDoesNotExist("Invalid Rule Name.")
|
||||
|
||||
def update_receipt_rule(self, rule_set_name, rule):
|
||||
def update_receipt_rule(self, rule_set_name: str, rule: Dict[str, Any]) -> None:
|
||||
rule_set = self.receipt_rule_set.get(rule_set_name)
|
||||
|
||||
if rule_set is None:
|
||||
@ -532,8 +568,11 @@ class SESBackend(BaseBackend):
|
||||
raise RuleDoesNotExist(f"Rule does not exist: {rule['name']}")
|
||||
|
||||
def set_identity_mail_from_domain(
|
||||
self, identity, mail_from_domain=None, behavior_on_mx_failure=None
|
||||
):
|
||||
self,
|
||||
identity: str,
|
||||
mail_from_domain: Optional[str] = None,
|
||||
behavior_on_mx_failure: Optional[str] = None,
|
||||
) -> None:
|
||||
if identity not in (self.domains + self.addresses):
|
||||
raise InvalidParameterValue(f"Identity '{identity}' does not exist.")
|
||||
|
||||
@ -560,7 +599,9 @@ class SESBackend(BaseBackend):
|
||||
"behavior_on_mx_failure": behavior_on_mx_failure,
|
||||
}
|
||||
|
||||
def get_identity_mail_from_domain_attributes(self, identities=None):
|
||||
def get_identity_mail_from_domain_attributes(
|
||||
self, identities: Optional[List[str]] = None
|
||||
) -> Dict[str, Dict[str, str]]:
|
||||
if identities is None:
|
||||
identities = []
|
||||
|
||||
@ -573,11 +614,13 @@ class SESBackend(BaseBackend):
|
||||
|
||||
return attributes_by_identity
|
||||
|
||||
def get_identity_verification_attributes(self, identities=None):
|
||||
def get_identity_verification_attributes(
|
||||
self, identities: Optional[List[str]] = None
|
||||
) -> Dict[str, str]:
|
||||
if identities is None:
|
||||
identities = []
|
||||
|
||||
attributes_by_identity = {}
|
||||
attributes_by_identity: Dict[str, str] = {}
|
||||
for identity in identities:
|
||||
if identity in (self.domains + self.addresses):
|
||||
attributes_by_identity[identity] = "Success"
|
||||
@ -585,4 +628,4 @@ class SESBackend(BaseBackend):
|
||||
return attributes_by_identity
|
||||
|
||||
|
||||
ses_backends: Mapping[str, SESBackend] = BackendDict(SESBackend, "ses")
|
||||
ses_backends = BackendDict(SESBackend, "ses")
|
||||
|
||||
@ -1,66 +1,71 @@
|
||||
import base64
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from moto.core.responses import BaseResponse
|
||||
from .models import ses_backends
|
||||
from .models import ses_backends, SESBackend
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class EmailResponse(BaseResponse):
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
super().__init__(service_name="ses")
|
||||
|
||||
@property
|
||||
def backend(self):
|
||||
def backend(self) -> SESBackend:
|
||||
return ses_backends[self.current_account][self.region]
|
||||
|
||||
def verify_email_identity(self):
|
||||
address = self.querystring.get("EmailAddress")[0]
|
||||
def verify_email_identity(self) -> str:
|
||||
address = self.querystring.get("EmailAddress")[0] # type: ignore
|
||||
self.backend.verify_email_identity(address)
|
||||
template = self.response_template(VERIFY_EMAIL_IDENTITY)
|
||||
return template.render()
|
||||
|
||||
def verify_email_address(self):
|
||||
address = self.querystring.get("EmailAddress")[0]
|
||||
def verify_email_address(self) -> str:
|
||||
address = self.querystring.get("EmailAddress")[0] # type: ignore
|
||||
self.backend.verify_email_address(address)
|
||||
template = self.response_template(VERIFY_EMAIL_ADDRESS)
|
||||
return template.render()
|
||||
|
||||
def list_identities(self):
|
||||
def list_identities(self) -> str:
|
||||
identities = self.backend.list_identities()
|
||||
template = self.response_template(LIST_IDENTITIES_RESPONSE)
|
||||
return template.render(identities=identities)
|
||||
|
||||
def list_verified_email_addresses(self):
|
||||
def list_verified_email_addresses(self) -> str:
|
||||
email_addresses = self.backend.list_verified_email_addresses()
|
||||
template = self.response_template(LIST_VERIFIED_EMAIL_RESPONSE)
|
||||
return template.render(email_addresses=email_addresses)
|
||||
|
||||
def verify_domain_dkim(self):
|
||||
domain = self.querystring.get("Domain")[0]
|
||||
def verify_domain_dkim(self) -> str:
|
||||
domain = self.querystring.get("Domain")[0] # type: ignore
|
||||
self.backend.verify_domain(domain)
|
||||
template = self.response_template(VERIFY_DOMAIN_DKIM_RESPONSE)
|
||||
return template.render()
|
||||
|
||||
def verify_domain_identity(self):
|
||||
domain = self.querystring.get("Domain")[0]
|
||||
def verify_domain_identity(self) -> str:
|
||||
domain = self.querystring.get("Domain")[0] # type: ignore
|
||||
self.backend.verify_domain(domain)
|
||||
template = self.response_template(VERIFY_DOMAIN_IDENTITY_RESPONSE)
|
||||
return template.render()
|
||||
|
||||
def delete_identity(self):
|
||||
domain = self.querystring.get("Identity")[0]
|
||||
def delete_identity(self) -> str:
|
||||
domain = self.querystring.get("Identity")[0] # type: ignore
|
||||
self.backend.delete_identity(domain)
|
||||
template = self.response_template(DELETE_IDENTITY_RESPONSE)
|
||||
return template.render()
|
||||
|
||||
def send_email(self):
|
||||
def send_email(self) -> str:
|
||||
bodydatakey = "Message.Body.Text.Data"
|
||||
if "Message.Body.Html.Data" in self.querystring:
|
||||
bodydatakey = "Message.Body.Html.Data"
|
||||
body = self.querystring.get(bodydatakey)[0]
|
||||
source = self.querystring.get("Source")[0]
|
||||
subject = self.querystring.get("Message.Subject.Data")[0]
|
||||
destinations = {"ToAddresses": [], "CcAddresses": [], "BccAddresses": []}
|
||||
body = self.querystring.get(bodydatakey)[0] # type: ignore
|
||||
source = self.querystring.get("Source")[0] # type: ignore
|
||||
subject = self.querystring.get("Message.Subject.Data")[0] # type: ignore
|
||||
destinations: Dict[str, List[str]] = {
|
||||
"ToAddresses": [],
|
||||
"CcAddresses": [],
|
||||
"BccAddresses": [],
|
||||
}
|
||||
for dest_type in destinations:
|
||||
# consume up to 51 to allow exception
|
||||
for i in range(1, 52):
|
||||
@ -70,18 +75,20 @@ class EmailResponse(BaseResponse):
|
||||
break
|
||||
destinations[dest_type].append(address[0])
|
||||
|
||||
message = self.backend.send_email(
|
||||
source, subject, body, destinations, self.region
|
||||
)
|
||||
message = self.backend.send_email(source, subject, body, destinations)
|
||||
template = self.response_template(SEND_EMAIL_RESPONSE)
|
||||
return template.render(message=message)
|
||||
|
||||
def send_templated_email(self):
|
||||
source = self.querystring.get("Source")[0]
|
||||
template = self.querystring.get("Template")
|
||||
template_data = self.querystring.get("TemplateData")
|
||||
def send_templated_email(self) -> str:
|
||||
source = self.querystring.get("Source")[0] # type: ignore
|
||||
template: List[str] = self.querystring.get("Template") # type: ignore
|
||||
template_data: List[str] = self.querystring.get("TemplateData") # type: ignore
|
||||
|
||||
destinations = {"ToAddresses": [], "CcAddresses": [], "BccAddresses": []}
|
||||
destinations: Dict[str, List[str]] = {
|
||||
"ToAddresses": [],
|
||||
"CcAddresses": [],
|
||||
"BccAddresses": [],
|
||||
}
|
||||
for dest_type in destinations:
|
||||
# consume up to 51 to allow exception
|
||||
for i in range(1, 52):
|
||||
@ -92,13 +99,14 @@ class EmailResponse(BaseResponse):
|
||||
destinations[dest_type].append(address[0])
|
||||
|
||||
message = self.backend.send_templated_email(
|
||||
source, template, template_data, destinations, self.region
|
||||
source, template, template_data, destinations
|
||||
)
|
||||
return self.response_template(SEND_TEMPLATED_EMAIL_RESPONSE).render(
|
||||
message=message
|
||||
)
|
||||
template = self.response_template(SEND_TEMPLATED_EMAIL_RESPONSE)
|
||||
return template.render(message=message)
|
||||
|
||||
def send_bulk_templated_email(self):
|
||||
source = self.querystring.get("Source")[0]
|
||||
def send_bulk_templated_email(self) -> str:
|
||||
source = self.querystring.get("Source")[0] # type: ignore
|
||||
template = self.querystring.get("Template")
|
||||
template_data = self.querystring.get("DefaultTemplateData")
|
||||
|
||||
@ -109,7 +117,11 @@ class EmailResponse(BaseResponse):
|
||||
)
|
||||
if self.querystring.get(destination_field) is None:
|
||||
break
|
||||
destination = {"ToAddresses": [], "CcAddresses": [], "BccAddresses": []}
|
||||
destination: Dict[str, List[str]] = {
|
||||
"ToAddresses": [],
|
||||
"CcAddresses": [],
|
||||
"BccAddresses": [],
|
||||
}
|
||||
for dest_type in destination:
|
||||
# consume up to 51 to allow exception
|
||||
for j in range(1, 52):
|
||||
@ -123,18 +135,18 @@ class EmailResponse(BaseResponse):
|
||||
destinations.append({"Destination": destination})
|
||||
|
||||
message = self.backend.send_bulk_templated_email(
|
||||
source, template, template_data, destinations, self.region
|
||||
source, template, template_data, destinations # type: ignore
|
||||
)
|
||||
template = self.response_template(SEND_BULK_TEMPLATED_EMAIL_RESPONSE)
|
||||
result = template.render(message=message)
|
||||
return result
|
||||
|
||||
def send_raw_email(self):
|
||||
def send_raw_email(self) -> str:
|
||||
source = self.querystring.get("Source")
|
||||
if source is not None:
|
||||
(source,) = source
|
||||
|
||||
raw_data = self.querystring.get("RawMessage.Data")[0]
|
||||
raw_data = self.querystring.get("RawMessage.Data")[0] # type: ignore
|
||||
raw_data = base64.b64decode(raw_data)
|
||||
raw_data = raw_data.decode("utf-8")
|
||||
destinations = []
|
||||
@ -146,35 +158,33 @@ class EmailResponse(BaseResponse):
|
||||
break
|
||||
destinations.append(address[0])
|
||||
|
||||
message = self.backend.send_raw_email(
|
||||
source, destinations, raw_data, self.region
|
||||
)
|
||||
message = self.backend.send_raw_email(source, destinations, raw_data)
|
||||
template = self.response_template(SEND_RAW_EMAIL_RESPONSE)
|
||||
return template.render(message=message)
|
||||
|
||||
def get_send_quota(self):
|
||||
def get_send_quota(self) -> str:
|
||||
quota = self.backend.get_send_quota()
|
||||
template = self.response_template(GET_SEND_QUOTA_RESPONSE)
|
||||
return template.render(quota=quota)
|
||||
|
||||
def get_identity_notification_attributes(self):
|
||||
def get_identity_notification_attributes(self) -> str:
|
||||
identities = self._get_params()["Identities"]
|
||||
identities = self.backend.get_identity_notification_attributes(identities)
|
||||
template = self.response_template(GET_IDENTITY_NOTIFICATION_ATTRIBUTES)
|
||||
return template.render(identities=identities)
|
||||
|
||||
def set_identity_feedback_forwarding_enabled(self):
|
||||
def set_identity_feedback_forwarding_enabled(self) -> str:
|
||||
identity = self._get_param("Identity")
|
||||
enabled = self._get_bool_param("ForwardingEnabled")
|
||||
self.backend.set_identity_feedback_forwarding_enabled(identity, enabled)
|
||||
template = self.response_template(SET_IDENTITY_FORWARDING_ENABLED_RESPONSE)
|
||||
return template.render()
|
||||
|
||||
def set_identity_notification_topic(self):
|
||||
def set_identity_notification_topic(self) -> str:
|
||||
|
||||
identity = self.querystring.get("Identity")[0]
|
||||
not_type = self.querystring.get("NotificationType")[0]
|
||||
sns_topic = self.querystring.get("SnsTopic")
|
||||
identity = self.querystring.get("Identity")[0] # type: ignore
|
||||
not_type = self.querystring.get("NotificationType")[0] # type: ignore
|
||||
sns_topic = self.querystring.get("SnsTopic") # type: ignore
|
||||
if sns_topic:
|
||||
sns_topic = sns_topic[0]
|
||||
|
||||
@ -182,33 +192,35 @@ class EmailResponse(BaseResponse):
|
||||
template = self.response_template(SET_IDENTITY_NOTIFICATION_TOPIC_RESPONSE)
|
||||
return template.render()
|
||||
|
||||
def get_send_statistics(self):
|
||||
def get_send_statistics(self) -> str:
|
||||
statistics = self.backend.get_send_statistics()
|
||||
template = self.response_template(GET_SEND_STATISTICS)
|
||||
return template.render(all_statistics=[statistics])
|
||||
|
||||
def create_configuration_set(self):
|
||||
configuration_set_name = self.querystring.get("ConfigurationSet.Name")[0]
|
||||
def create_configuration_set(self) -> str:
|
||||
configuration_set_name = self.querystring.get("ConfigurationSet.Name")[0] # type: ignore
|
||||
self.backend.create_configuration_set(
|
||||
configuration_set_name=configuration_set_name
|
||||
)
|
||||
template = self.response_template(CREATE_CONFIGURATION_SET)
|
||||
return template.render()
|
||||
|
||||
def describe_configuration_set(self):
|
||||
configuration_set_name = self.querystring.get("ConfigurationSetName")[0]
|
||||
def describe_configuration_set(self) -> str:
|
||||
configuration_set_name = self.querystring.get("ConfigurationSetName")[0] # type: ignore
|
||||
self.backend.describe_configuration_set(configuration_set_name)
|
||||
template = self.response_template(DESCRIBE_CONFIGURATION_SET)
|
||||
return template.render(name=configuration_set_name)
|
||||
|
||||
def create_configuration_set_event_destination(self):
|
||||
def create_configuration_set_event_destination(self) -> str:
|
||||
|
||||
configuration_set_name = self._get_param("ConfigurationSetName")
|
||||
configuration_set_name = self._get_param("ConfigurationSetName") # type: ignore
|
||||
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(
|
||||
)[
|
||||
0
|
||||
] # type: ignore
|
||||
configuration_event_name = self.querystring.get("EventDestination.Name")[0] # type: ignore
|
||||
event_topic_arn = self.querystring.get( # type: ignore
|
||||
"EventDestination.SNSDestination.TopicARN"
|
||||
)[0]
|
||||
event_matching_types = self._get_multi_param(
|
||||
@ -230,7 +242,7 @@ class EmailResponse(BaseResponse):
|
||||
template = self.response_template(CREATE_CONFIGURATION_SET_EVENT_DESTINATION)
|
||||
return template.render()
|
||||
|
||||
def create_template(self):
|
||||
def create_template(self) -> str:
|
||||
template_data = self._get_dict_param("Template")
|
||||
template_info = {}
|
||||
template_info["text_part"] = template_data.get("._text_part", "")
|
||||
@ -242,7 +254,7 @@ class EmailResponse(BaseResponse):
|
||||
template = self.response_template(CREATE_TEMPLATE)
|
||||
return template.render()
|
||||
|
||||
def update_template(self):
|
||||
def update_template(self) -> str:
|
||||
template_data = self._get_dict_param("Template")
|
||||
template_info = {}
|
||||
template_info["text_part"] = template_data.get("._text_part", "")
|
||||
@ -254,43 +266,43 @@ class EmailResponse(BaseResponse):
|
||||
template = self.response_template(UPDATE_TEMPLATE)
|
||||
return template.render()
|
||||
|
||||
def get_template(self):
|
||||
def get_template(self) -> str:
|
||||
template_name = self._get_param("TemplateName")
|
||||
template_data = self.backend.get_template(template_name)
|
||||
template = self.response_template(GET_TEMPLATE)
|
||||
return template.render(template_data=template_data)
|
||||
|
||||
def list_templates(self):
|
||||
def list_templates(self) -> str:
|
||||
email_templates = self.backend.list_templates()
|
||||
template = self.response_template(LIST_TEMPLATES)
|
||||
return template.render(templates=email_templates)
|
||||
|
||||
def test_render_template(self):
|
||||
def test_render_template(self) -> str:
|
||||
render_info = self._get_dict_param("Template")
|
||||
rendered_template = self.backend.render_template(render_info)
|
||||
template = self.response_template(RENDER_TEMPLATE)
|
||||
return template.render(template=rendered_template)
|
||||
|
||||
def create_receipt_rule_set(self):
|
||||
def create_receipt_rule_set(self) -> str:
|
||||
rule_set_name = self._get_param("RuleSetName")
|
||||
self.backend.create_receipt_rule_set(rule_set_name)
|
||||
template = self.response_template(CREATE_RECEIPT_RULE_SET)
|
||||
return template.render()
|
||||
|
||||
def create_receipt_rule(self):
|
||||
def create_receipt_rule(self) -> str:
|
||||
rule_set_name = self._get_param("RuleSetName")
|
||||
rule = self._get_dict_param("Rule.")
|
||||
self.backend.create_receipt_rule(rule_set_name, rule)
|
||||
template = self.response_template(CREATE_RECEIPT_RULE)
|
||||
return template.render()
|
||||
|
||||
def describe_receipt_rule_set(self):
|
||||
def describe_receipt_rule_set(self) -> str:
|
||||
rule_set_name = self._get_param("RuleSetName")
|
||||
|
||||
rule_set = self.backend.describe_receipt_rule_set(rule_set_name)
|
||||
|
||||
for i, rule in enumerate(rule_set):
|
||||
formatted_rule = {}
|
||||
formatted_rule: Dict[str, Any] = {}
|
||||
|
||||
for k, v in rule.items():
|
||||
self._parse_param(k, v, formatted_rule)
|
||||
@ -301,13 +313,13 @@ class EmailResponse(BaseResponse):
|
||||
|
||||
return template.render(rule_set=rule_set, rule_set_name=rule_set_name)
|
||||
|
||||
def describe_receipt_rule(self):
|
||||
def describe_receipt_rule(self) -> str:
|
||||
rule_set_name = self._get_param("RuleSetName")
|
||||
rule_name = self._get_param("RuleName")
|
||||
|
||||
receipt_rule = self.backend.describe_receipt_rule(rule_set_name, rule_name)
|
||||
|
||||
rule = {}
|
||||
rule: Dict[str, Any] = {}
|
||||
|
||||
for k, v in receipt_rule.items():
|
||||
self._parse_param(k, v, rule)
|
||||
@ -315,7 +327,7 @@ class EmailResponse(BaseResponse):
|
||||
template = self.response_template(DESCRIBE_RECEIPT_RULE)
|
||||
return template.render(rule=rule)
|
||||
|
||||
def update_receipt_rule(self):
|
||||
def update_receipt_rule(self) -> str:
|
||||
rule_set_name = self._get_param("RuleSetName")
|
||||
rule = self._get_dict_param("Rule.")
|
||||
|
||||
@ -324,7 +336,7 @@ class EmailResponse(BaseResponse):
|
||||
template = self.response_template(UPDATE_RECEIPT_RULE)
|
||||
return template.render()
|
||||
|
||||
def set_identity_mail_from_domain(self):
|
||||
def set_identity_mail_from_domain(self) -> str:
|
||||
identity = self._get_param("Identity")
|
||||
mail_from_domain = self._get_param("MailFromDomain")
|
||||
behavior_on_mx_failure = self._get_param("BehaviorOnMXFailure")
|
||||
@ -336,14 +348,16 @@ class EmailResponse(BaseResponse):
|
||||
template = self.response_template(SET_IDENTITY_MAIL_FROM_DOMAIN)
|
||||
return template.render()
|
||||
|
||||
def get_identity_mail_from_domain_attributes(self):
|
||||
def get_identity_mail_from_domain_attributes(self) -> str:
|
||||
identities = self._get_multi_param("Identities.member.")
|
||||
identities = self.backend.get_identity_mail_from_domain_attributes(identities)
|
||||
attributes_by_identity = self.backend.get_identity_mail_from_domain_attributes(
|
||||
identities
|
||||
)
|
||||
template = self.response_template(GET_IDENTITY_MAIL_FROM_DOMAIN_ATTRIBUTES)
|
||||
|
||||
return template.render(identities=identities)
|
||||
return template.render(identities=attributes_by_identity)
|
||||
|
||||
def get_identity_verification_attributes(self):
|
||||
def get_identity_verification_attributes(self) -> str:
|
||||
params = self._get_params()
|
||||
identities = params.get("Identities")
|
||||
verification_attributes = self.backend.get_identity_verification_attributes(
|
||||
|
||||
@ -1,8 +1,15 @@
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from .exceptions import MissingRenderingAttributeException
|
||||
from moto.utilities.tokenizer import GenericTokenizer
|
||||
|
||||
|
||||
def parse_template(template, template_data, tokenizer=None, until=None):
|
||||
def parse_template(
|
||||
template: str,
|
||||
template_data: Dict[str, Any],
|
||||
tokenizer: Optional[GenericTokenizer] = None,
|
||||
until: Optional[str] = None,
|
||||
) -> str:
|
||||
tokenizer = tokenizer or GenericTokenizer(template)
|
||||
state = ""
|
||||
parsed = ""
|
||||
@ -27,7 +34,7 @@ def parse_template(template, template_data, tokenizer=None, until=None):
|
||||
if state == "VAR":
|
||||
if template_data.get(var_name) is None:
|
||||
raise MissingRenderingAttributeException(var_name)
|
||||
parsed += template_data.get(var_name)
|
||||
parsed += template_data.get(var_name) # type: ignore
|
||||
else:
|
||||
current_position = tokenizer.token_pos
|
||||
for item in template_data.get(var_name, []):
|
||||
|
||||
@ -1,19 +1,21 @@
|
||||
import string
|
||||
from typing import Optional
|
||||
|
||||
from email.utils import parseaddr
|
||||
from moto.moto_api._internal import mock_random as random
|
||||
|
||||
|
||||
def random_hex(length):
|
||||
def random_hex(length: int) -> str:
|
||||
return "".join(random.choice(string.ascii_lowercase) for x in range(length))
|
||||
|
||||
|
||||
def get_random_message_id():
|
||||
def get_random_message_id() -> str:
|
||||
return f"{random_hex(16)}-{random_hex(8)}-{random_hex(4)}-{random_hex(4)}-{random_hex(4)}-{random_hex(12)}-{random_hex(6)}"
|
||||
|
||||
|
||||
def is_valid_address(addr):
|
||||
def is_valid_address(addr: str) -> Optional[str]:
|
||||
_, address = parseaddr(addr)
|
||||
address = address.split("@")
|
||||
if len(address) != 2 or not address[1]:
|
||||
return False, "Missing domain"
|
||||
return True, None
|
||||
address_parts = address.split("@")
|
||||
if len(address_parts) != 2 or not address_parts[1]:
|
||||
return "Missing domain"
|
||||
return None
|
||||
|
||||
@ -239,7 +239,7 @@ disable = W,C,R,E
|
||||
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
|
||||
|
||||
[mypy]
|
||||
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s3*,moto/sagemaker,moto/secretsmanager,moto/sqs,moto/ssm,moto/scheduler
|
||||
files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s3*,moto/sagemaker,moto/secretsmanager,moto/ses,moto/sqs,moto/ssm,moto/scheduler
|
||||
show_column_numbers=True
|
||||
show_error_codes = True
|
||||
disable_error_code=abstract
|
||||
|
||||
@ -4,14 +4,11 @@ from moto.ses.utils import is_valid_address
|
||||
|
||||
|
||||
def test_is_valid_address():
|
||||
valid, msg = is_valid_address("test@example.com")
|
||||
valid.should.equal(True)
|
||||
msg = is_valid_address("test@example.com")
|
||||
msg.should.equal(None)
|
||||
|
||||
valid, msg = is_valid_address("test@")
|
||||
valid.should.equal(False)
|
||||
msg = is_valid_address("test@")
|
||||
msg.should.be.a(str)
|
||||
|
||||
valid, msg = is_valid_address("test")
|
||||
valid.should.equal(False)
|
||||
msg = is_valid_address("test")
|
||||
msg.should.be.a(str)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user