diff --git a/moto/ses/models.py b/moto/ses/models.py index e066dedc6..b88086c21 100644 --- a/moto/ses/models.py +++ b/moto/ses/models.py @@ -43,6 +43,8 @@ class SESFeedback(BaseModel): FEEDBACK_BOUNCE_MSG = {"test": "bounce"} FEEDBACK_COMPLAINT_MSG = {"test": "complaint"} + FORWARDING_ENABLED = "feedback_forwarding_enabled" + @staticmethod def generate_message(msg_type): msg = dict(COMMON_MAIL) @@ -268,6 +270,17 @@ class SESBackend(BaseBackend): def get_send_quota(self): return SESQuota(self.sent_message_count) + def get_identity_notification_attributes(self, identities): + response = {} + for identity in identities: + response[identity] = self.sns_topics.get(identity, {}) + return response + + def set_identity_feedback_forwarding_enabled(self, identity, enabled): + 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): identity_sns_topics = self.sns_topics.get(identity, {}) if sns_topic is None: diff --git a/moto/ses/responses.py b/moto/ses/responses.py index 4376eeb41..73bc45398 100644 --- a/moto/ses/responses.py +++ b/moto/ses/responses.py @@ -118,6 +118,19 @@ class EmailResponse(BaseResponse): template = self.response_template(GET_SEND_QUOTA_RESPONSE) return template.render(quota=quota) + def get_identity_notification_attributes(self): + identities = self._get_params()["Identities"] + identities = ses_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): + identity = self._get_param("Identity") + enabled = self._get_bool_param("ForwardingEnabled") + ses_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): identity = self.querystring.get("Identity")[0] @@ -335,6 +348,38 @@ GET_SEND_QUOTA_RESPONSE = """ + + + {% for identity, config in identities.items() %} + + {{ identity }} + + false + false + false + {% if config.get("feedback_forwarding_enabled", True) == False %} + false + {% else %} + true + {% endif %} + + + {% endfor %} + + + + 46c90cfc-9055-4b84-96e3-4d6a309a8b9b + +""" + +SET_IDENTITY_FORWARDING_ENABLED_RESPONSE = """ + + + 47e0ef1a-9bf2-11e1-9279-0100e8cf109a + +""" + SET_IDENTITY_NOTIFICATION_TOPIC_RESPONSE = """ diff --git a/tests/test_ses/test_ses_sns_boto3.py b/tests/test_ses/test_ses_sns_boto3.py index b24ddeb8a..5d57d19ff 100644 --- a/tests/test_ses/test_ses_sns_boto3.py +++ b/tests/test_ses/test_ses_sns_boto3.py @@ -128,3 +128,60 @@ def test_sns_feedback_delivery_raw_email(): __test_sns_feedback__( SESFeedback.SUCCESS_ADDR, SESFeedback.DELIVERY, raw_email=True ) + + +@mock_ses +def test_get_identity_notification_attributes_default_values(): + ses = boto3.client("ses", region_name="us-east-1") + ses.verify_domain_identity(Domain="example.com") + ses.verify_email_identity(EmailAddress="test@example.com") + + resp = ses.get_identity_notification_attributes( + Identities=["test@example.com", "another@example.com"] + )["NotificationAttributes"] + resp.should.have.length_of(2) + resp.should.have.key("test@example.com") + resp.should.have.key("another@example.com") + resp["test@example.com"].should.have.key("ForwardingEnabled").equal(True) + resp["test@example.com"].should.have.key( + "HeadersInBounceNotificationsEnabled" + ).equal(False) + resp["test@example.com"].should.have.key( + "HeadersInComplaintNotificationsEnabled" + ).equal(False) + resp["test@example.com"].should.have.key( + "HeadersInDeliveryNotificationsEnabled" + ).equal(False) + resp["test@example.com"].shouldnt.have.key("BounceTopic") + resp["test@example.com"].shouldnt.have.key("ComplaintTopic") + resp["test@example.com"].shouldnt.have.key("DeliveryTopic") + + +@mock_ses +def test_set_identity_feedback_forwarding_enabled(): + ses = boto3.client("ses", region_name="us-east-1") + ses.verify_domain_identity(Domain="example.com") + ses.verify_email_identity(EmailAddress="test@example.com") + + resp = ses.get_identity_notification_attributes(Identities=["test@example.com"])[ + "NotificationAttributes" + ] + resp["test@example.com"].should.have.key("ForwardingEnabled").equal(True) + + ses.set_identity_feedback_forwarding_enabled( + Identity="test@example.com", ForwardingEnabled=False + ) + + resp = ses.get_identity_notification_attributes(Identities=["test@example.com"])[ + "NotificationAttributes" + ] + resp["test@example.com"].should.have.key("ForwardingEnabled").equal(False) + + ses.set_identity_feedback_forwarding_enabled( + Identity="test@example.com", ForwardingEnabled=True + ) + + resp = ses.get_identity_notification_attributes(Identities=["test@example.com"])[ + "NotificationAttributes" + ] + resp["test@example.com"].should.have.key("ForwardingEnabled").equal(True)