SES - setIdentityFeedbackForwardingEnabled & GetIdentityNotificationAttributes (#4445)

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:

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 = """<GetSendQuotaResponse xmlns="http://ses.amazonaws.c
   </ResponseMetadata>
 </GetSendQuotaResponse>"""
 
+GET_IDENTITY_NOTIFICATION_ATTRIBUTES = """<GetIdentityNotificationAttributesResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
+  <GetIdentityNotificationAttributesResult>
+    <NotificationAttributes>
+    {% for identity, config in identities.items() %}
+      <entry>
+        <key>{{ identity }}</key>
+        <value>
+          <HeadersInBounceNotificationsEnabled>false</HeadersInBounceNotificationsEnabled>
+          <HeadersInDeliveryNotificationsEnabled>false</HeadersInDeliveryNotificationsEnabled>
+          <HeadersInComplaintNotificationsEnabled>false</HeadersInComplaintNotificationsEnabled>
+          {% if config.get("feedback_forwarding_enabled", True) == False %}
+          <ForwardingEnabled>false</ForwardingEnabled>
+          {% else %}
+          <ForwardingEnabled>true</ForwardingEnabled>
+          {% endif %}
+        </value>
+      </entry>
+      {% endfor %}
+    </NotificationAttributes>
+  </GetIdentityNotificationAttributesResult>
+  <ResponseMetadata>
+    <RequestId>46c90cfc-9055-4b84-96e3-4d6a309a8b9b</RequestId>
+  </ResponseMetadata>
+</GetIdentityNotificationAttributesResponse>"""
+
+SET_IDENTITY_FORWARDING_ENABLED_RESPONSE = """<SetIdentityFeedbackForwardingEnabledResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
+  <SetIdentityFeedbackForwardingEnabledResult/>
+  <ResponseMetadata>
+    <RequestId>47e0ef1a-9bf2-11e1-9279-0100e8cf109a</RequestId>
+  </ResponseMetadata>
+</SetIdentityFeedbackForwardingEnabledResponse>"""
+
 SET_IDENTITY_NOTIFICATION_TOPIC_RESPONSE = """<SetIdentityNotificationTopicResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
   <SetIdentityNotificationTopicResult/>
   <ResponseMetadata>

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)