Enhancement : SES - Added create-receipt-rule-set, create-receipt-rul… (#3059)

* Enhancement : SES - Added create-receipt-rule-set, create-receipt-rule functionalities.

* Linting

Co-authored-by: usmankb <usman@krazybee.com>
Co-authored-by: Bert Blommers <info@bertblommers.nl>

moto/ses/exceptions.py

@@ -41,3 +41,26 @@ class TemplateDoesNotExist(RESTError):
 
     def __init__(self, message):
         super(TemplateDoesNotExist, self).__init__("TemplateDoesNotExist", message)
+
+
+class RuleSetNameAlreadyExists(RESTError):
+    code = 400
+
+    def __init__(self, message):
+        super(RuleSetNameAlreadyExists, self).__init__(
+            "RuleSetNameAlreadyExists", message
+        )
+
+
+class RuleAlreadyExists(RESTError):
+    code = 400
+
+    def __init__(self, message):
+        super(RuleAlreadyExists, self).__init__("RuleAlreadyExists", message)
+
+
+class RuleSetDoesNotExist(RESTError):
+    code = 400
+
+    def __init__(self, message):
+        super(RuleSetDoesNotExist, self).__init__("RuleSetDoesNotExist", message)

moto/ses/models.py

@@ -12,6 +12,9 @@ from .exceptions import (
     EventDestinationAlreadyExists,
     TemplateNameAlreadyExists,
     TemplateDoesNotExist,
+    RuleSetNameAlreadyExists,
+    RuleSetDoesNotExist,
+    RuleAlreadyExists,
 )
 from .utils import get_random_message_id
 from .feedback import COMMON_MAIL, BOUNCE, COMPLAINT, DELIVERY
@@ -94,6 +97,7 @@ class SESBackend(BaseBackend):
         self.config_set_event_destination = {}
         self.event_destinations = {}
         self.templates = {}
+        self.receipt_rule_set = {}
 
     def _is_verified_address(self, source):
         _, address = parseaddr(source)
@@ -294,5 +298,19 @@ class SESBackend(BaseBackend):
     def list_templates(self):
         return list(self.templates.values())
 
+    def create_receipt_rule_set(self, rule_set_name):
+        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):
+        rule_set = self.receipt_rule_set.get(rule_set_name)
+        if rule_set is None:
+            raise RuleSetDoesNotExist("Invalid Rule Set Name.")
+        if rule in rule_set:
+            raise RuleAlreadyExists("Duplicate Rule Name.")
+        rule_set.append(rule)
+        self.receipt_rule_set[rule_set_name] = rule_set
+
 
 ses_backend = SESBackend()

moto/ses/responses.py

@@ -199,6 +199,19 @@ class EmailResponse(BaseResponse):
         template = self.response_template(LIST_TEMPLATES)
         return template.render(templates=email_templates)
 
+    def create_receipt_rule_set(self):
+        rule_set_name = self._get_param("RuleSetName")
+        ses_backend.create_receipt_rule_set(rule_set_name)
+        template = self.response_template(CREATE_RECEIPT_RULE_SET)
+        return template.render()
+
+    def create_receipt_rule(self):
+        rule_set_name = self._get_param("RuleSetName")
+        rule = self._get_dict_param("Rule")
+        ses_backend.create_receipt_rule(rule_set_name, rule)
+        template = self.response_template(CREATE_RECEIPT_RULE)
+        return template.render()
+
 
 VERIFY_EMAIL_IDENTITY = """<VerifyEmailIdentityResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
   <VerifyEmailIdentityResult/>
@@ -385,3 +398,17 @@ LIST_TEMPLATES = """<ListTemplatesResponse xmlns="http://ses.amazonaws.com/doc/2
         <RequestId>47e0ef1a-9bf2-11e1-9279-0100e8cf12ba</RequestId>
     </ResponseMetadata>
 </ListTemplatesResponse>"""
+
+CREATE_RECEIPT_RULE_SET = """<CreateReceiptRuleSetResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
+  <CreateReceiptRuleSetResult/>
+  <ResponseMetadata>
+    <RequestId>47e0ef1a-9bf2-11e1-9279-01ab88cf109a</RequestId>
+  </ResponseMetadata>
+</CreateReceiptRuleSetResponse>"""
+
+CREATE_RECEIPT_RULE = """<CreateReceiptRuleResponse xmlns="http://ses.amazonaws.com/doc/2010-12-01/">
+  <CreateReceiptRuleResult/>
+  <ResponseMetadata>
+    <RequestId>15e0ef1a-9bf2-11e1-9279-01ab88cf109a</RequestId>
+  </ResponseMetadata>
+</CreateReceiptRuleResponse>"""

tests/test_ses/test_ses_boto3.py

@@ -300,6 +300,118 @@ def test_create_configuration_set():
     ex.exception.response["Error"]["Code"].should.equal("EventDestinationAlreadyExists")
 
 
+@mock_ses
+def test_create_receipt_rule_set():
+    conn = boto3.client("ses", region_name="us-east-1")
+    result = conn.create_receipt_rule_set(RuleSetName="testRuleSet")
+
+    result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
+
+    with assert_raises(ClientError) as ex:
+        conn.create_receipt_rule_set(RuleSetName="testRuleSet")
+
+    ex.exception.response["Error"]["Code"].should.equal("RuleSetNameAlreadyExists")
+
+
+@mock_ses
+def test_create_receipt_rule():
+    conn = boto3.client("ses", region_name="us-east-1")
+    rule_set_name = "testRuleSet"
+    conn.create_receipt_rule_set(RuleSetName=rule_set_name)
+
+    result = conn.create_receipt_rule(
+        RuleSetName=rule_set_name,
+        Rule={
+            "Name": "testRule",
+            "Enabled": False,
+            "TlsPolicy": "Optional",
+            "Recipients": ["string"],
+            "Actions": [
+                {
+                    "S3Action": {
+                        "TopicArn": "string",
+                        "BucketName": "string",
+                        "ObjectKeyPrefix": "string",
+                        "KmsKeyArn": "string",
+                    },
+                    "BounceAction": {
+                        "TopicArn": "string",
+                        "SmtpReplyCode": "string",
+                        "StatusCode": "string",
+                        "Message": "string",
+                        "Sender": "string",
+                    },
+                }
+            ],
+            "ScanEnabled": False,
+        },
+    )
+
+    result["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
+
+    with assert_raises(ClientError) as ex:
+        conn.create_receipt_rule(
+            RuleSetName=rule_set_name,
+            Rule={
+                "Name": "testRule",
+                "Enabled": False,
+                "TlsPolicy": "Optional",
+                "Recipients": ["string"],
+                "Actions": [
+                    {
+                        "S3Action": {
+                            "TopicArn": "string",
+                            "BucketName": "string",
+                            "ObjectKeyPrefix": "string",
+                            "KmsKeyArn": "string",
+                        },
+                        "BounceAction": {
+                            "TopicArn": "string",
+                            "SmtpReplyCode": "string",
+                            "StatusCode": "string",
+                            "Message": "string",
+                            "Sender": "string",
+                        },
+                    }
+                ],
+                "ScanEnabled": False,
+            },
+        )
+
+    ex.exception.response["Error"]["Code"].should.equal("RuleAlreadyExists")
+
+    with assert_raises(ClientError) as ex:
+        conn.create_receipt_rule(
+            RuleSetName="InvalidRuleSetaName",
+            Rule={
+                "Name": "testRule",
+                "Enabled": False,
+                "TlsPolicy": "Optional",
+                "Recipients": ["string"],
+                "Actions": [
+                    {
+                        "S3Action": {
+                            "TopicArn": "string",
+                            "BucketName": "string",
+                            "ObjectKeyPrefix": "string",
+                            "KmsKeyArn": "string",
+                        },
+                        "BounceAction": {
+                            "TopicArn": "string",
+                            "SmtpReplyCode": "string",
+                            "StatusCode": "string",
+                            "Message": "string",
+                            "Sender": "string",
+                        },
+                    }
+                ],
+                "ScanEnabled": False,
+            },
+        )
+
+    ex.exception.response["Error"]["Code"].should.equal("RuleSetDoesNotExist")
+
+
 @mock_ses
 def test_create_ses_template():
     conn = boto3.client("ses", region_name="us-east-1")