From 65611082be073018a898ae795a0a7458cf2ac9f2 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Tue, 25 Apr 2023 11:42:08 +0000 Subject: [PATCH] Techdebt: MyPy SES (#6252) --- moto/ses/exceptions.py | 30 ++-- moto/ses/models.py | 229 ++++++++++++++++++------------- moto/ses/responses.py | 162 ++++++++++++---------- moto/ses/template.py | 11 +- moto/ses/utils.py | 16 ++- setup.cfg | 2 +- tests/test_ses/test_ses_utils.py | 9 +- 7 files changed, 261 insertions(+), 198 deletions(-) diff --git a/moto/ses/exceptions.py b/moto/ses/exceptions.py index 1d39b8e5b..1c6eb9860 100644 --- a/moto/ses/exceptions.py +++ b/moto/ses/exceptions.py @@ -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.", diff --git a/moto/ses/models.py b/moto/ses/models.py index dd9c94ae6..f91328fa9 100644 --- a/moto/ses/models.py +++ b/moto/ses/models.py @@ -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") diff --git a/moto/ses/responses.py b/moto/ses/responses.py index f36aa7601..6e1d9694a 100644 --- a/moto/ses/responses.py +++ b/moto/ses/responses.py @@ -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( diff --git a/moto/ses/template.py b/moto/ses/template.py index 5c82c6091..a153cf0db 100644 --- a/moto/ses/template.py +++ b/moto/ses/template.py @@ -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, []): diff --git a/moto/ses/utils.py b/moto/ses/utils.py index 4f3a7eb43..0f28cba3b 100644 --- a/moto/ses/utils.py +++ b/moto/ses/utils.py @@ -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 diff --git a/setup.cfg b/setup.cfg index b98e4f7ec..a7008125c 100644 --- a/setup.cfg +++ b/setup.cfg @@ -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 diff --git a/tests/test_ses/test_ses_utils.py b/tests/test_ses/test_ses_utils.py index f319983cd..06ce5f464 100644 --- a/tests/test_ses/test_ses_utils.py +++ b/tests/test_ses/test_ses_utils.py @@ -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)