feat(iot): Added IoT Rules implementation

IMPLEMENTATION_COVERAGE.md

@@ -4406,7 +4406,7 @@
 - [X] create_thing
 - [X] create_thing_group
 - [X] create_thing_type
-- [ ] create_topic_rule
+- [X] create_topic_rule
 - [ ] create_topic_rule_destination
 - [ ] delete_account_audit_configuration
 - [ ] delete_authorizer
@@ -4432,7 +4432,7 @@
 - [X] delete_thing
 - [X] delete_thing_group
 - [X] delete_thing_type
-- [ ] delete_topic_rule
+- [X] delete_topic_rule
 - [ ] delete_topic_rule_destination
 - [ ] delete_v2_logging_level
 - [ ] deprecate_thing_type
@@ -4467,8 +4467,8 @@
 - [X] detach_principal_policy
 - [ ] detach_security_profile
 - [X] detach_thing_principal
-- [ ] disable_topic_rule
-- [ ] enable_topic_rule
+- [X] disable_topic_rule
+- [X] enable_topic_rule
 - [ ] get_cardinality
 - [ ] get_effective_policies
 - [ ] get_indexing_configuration
@@ -4480,7 +4480,7 @@
 - [X] get_policy_version
 - [ ] get_registration_code
 - [ ] get_statistics
-- [ ] get_topic_rule
+- [X] get_topic_rule
 - [ ] get_topic_rule_destination
 - [ ] get_v2_logging_options
 - [ ] list_active_violations
@@ -4528,7 +4528,7 @@
 - [ ] list_things_in_billing_group
 - [X] list_things_in_thing_group
 - [ ] list_topic_rule_destinations
-- [ ] list_topic_rules
+- [X] list_topic_rules
 - [ ] list_v2_logging_levels
 - [ ] list_violation_events
 - [ ] register_ca_certificate
@@ -4538,7 +4538,7 @@
 - [ ] reject_certificate_transfer
 - [ ] remove_thing_from_billing_group
 - [X] remove_thing_from_thing_group
-- [ ] replace_topic_rule
+- [X] replace_topic_rule
 - [ ] search_index
 - [ ] set_default_authorizer
 - [X] set_default_policy_version

moto/iot/models.py

@@ -425,6 +425,57 @@ def to_dict(self):
         return obj
 
 
+class FakeRule(BaseModel):
+    def __init__(
+        self,
+        rule_name,
+        description,
+        created_at,
+        rule_disabled,
+        topic_pattern,
+        actions,
+        error_action,
+        sql,
+        aws_iot_sql_version,
+        region_name,
+    ):
+        self.region_name = region_name
+        self.rule_name = rule_name
+        self.description = description or ""
+        self.created_at = created_at
+        self.rule_disabled = bool(rule_disabled)
+        self.topic_pattern = topic_pattern
+        self.actions = actions or []
+        self.error_action = error_action or {}
+        self.sql = sql
+        self.aws_iot_sql_version = aws_iot_sql_version or "2016-03-23"
+        self.arn = "arn:aws:iot:%s:1:rule/%s" % (self.region_name, rule_name)
+
+    def to_get_dict(self):
+        return {
+            "rule": {
+                "actions": self.actions,
+                "awsIotSqlVersion": self.aws_iot_sql_version,
+                "createdAt": self.created_at,
+                "description": self.description,
+                "errorAction": self.error_action,
+                "ruleDisabled": self.rule_disabled,
+                "ruleName": self.rule_name,
+                "sql": self.sql,
+            },
+            "ruleArn": self.arn,
+        }
+
+    def to_dict(self):
+        return {
+            "ruleName": self.rule_name,
+            "createdAt": self.created_at,
+            "ruleArn": self.arn,
+            "ruleDisabled": self.rule_disabled,
+            "topicPattern": self.topic_pattern,
+        }
+
+
 class IoTBackend(BaseBackend):
     def __init__(self, region_name=None):
         super(IoTBackend, self).__init__()
@@ -438,6 +489,7 @@ def __init__(self, region_name=None):
         self.policies = OrderedDict()
         self.principal_policies = OrderedDict()
         self.principal_things = OrderedDict()
+        self.rules = OrderedDict()
         self.endpoint = None
 
     def reset(self):
@@ -1275,6 +1327,47 @@ def list_job_executions_for_thing(
 
         return job_executions, next_token
 
+    def list_topic_rules(self):
+        return [r.to_dict() for r in self.rules.values()]
+
+    def get_topic_rule(self, rule_name):
+        if rule_name not in self.rules:
+            raise ResourceNotFoundException()
+        return self.rules[rule_name].to_get_dict()
+
+    def create_topic_rule(self, rule_name, sql, **kwargs):
+        if rule_name in self.rules:
+            raise ResourceAlreadyExistsException("Rule with given name already exists")
+        result = re.search(r"FROM\s+([^\s]*)", sql)
+        topic = result.group(1).strip("'") if result else None
+        self.rules[rule_name] = FakeRule(
+            rule_name=rule_name,
+            created_at=int(datetime.now().timestamp()),
+            topic_pattern=topic,
+            sql=sql,
+            region_name=self.region_name,
+            **kwargs
+        )
+
+    def replace_topic_rule(self, rule_name, **kwargs):
+        self.delete_topic_rule(rule_name)
+        self.create_topic_rule(rule_name, **kwargs)
+
+    def delete_topic_rule(self, rule_name):
+        if rule_name not in self.rules:
+            raise ResourceNotFoundException()
+        del self.rules[rule_name]
+
+    def enable_topic_rule(self, rule_name):
+        if rule_name not in self.rules:
+            raise ResourceNotFoundException()
+        self.rules[rule_name].rule_disabled = False
+
+    def disable_topic_rule(self, rule_name):
+        if rule_name not in self.rules:
+            raise ResourceNotFoundException()
+        self.rules[rule_name].rule_disabled = True
+
 
 iot_backends = {}
 for region in Session().get_available_regions("iot"):

moto/iot/responses.py

@@ -635,3 +635,47 @@ def update_thing_groups_for_thing(self):
             thing_groups_to_remove=thing_groups_to_remove,
         )
         return json.dumps(dict())
+
+    def list_topic_rules(self):
+        return json.dumps(dict(rules=self.iot_backend.list_topic_rules()))
+
+    def get_topic_rule(self):
+        return json.dumps(
+            self.iot_backend.get_topic_rule(rule_name=self._get_param("ruleName"))
+        )
+
+    def create_topic_rule(self):
+        self.iot_backend.create_topic_rule(
+            rule_name=self._get_param("ruleName"),
+            description=self._get_param("description"),
+            rule_disabled=self._get_param("ruleDisabled"),
+            actions=self._get_param("actions"),
+            error_action=self._get_param("errorAction"),
+            sql=self._get_param("sql"),
+            aws_iot_sql_version=self._get_param("awsIotSqlVersion"),
+        )
+        return json.dumps(dict())
+
+    def replace_topic_rule(self):
+        self.iot_backend.replace_topic_rule(
+            rule_name=self._get_param("ruleName"),
+            description=self._get_param("description"),
+            rule_disabled=self._get_param("ruleDisabled"),
+            actions=self._get_param("actions"),
+            error_action=self._get_param("errorAction"),
+            sql=self._get_param("sql"),
+            aws_iot_sql_version=self._get_param("awsIotSqlVersion"),
+        )
+        return json.dumps(dict())
+
+    def delete_topic_rule(self):
+        self.iot_backend.delete_topic_rule(rule_name=self._get_param("ruleName"))
+        return json.dumps(dict())
+
+    def enable_topic_rule(self):
+        self.iot_backend.enable_topic_rule(rule_name=self._get_param("ruleName"))
+        return json.dumps(dict())
+
+    def disable_topic_rule(self):
+        self.iot_backend.disable_topic_rule(rule_name=self._get_param("ruleName"))
+        return json.dumps(dict())

tests/test_iot/test_iot.py

@@ -1991,3 +1991,133 @@ def test_list_job_executions_for_thing():
     job_execution["executionSummaries"][0].should.have.key("jobId").which.should.equal(
         job_id
     )
+
+
+@mock_iot
+def test_topic_rules():
+    client = boto3.client("iot", region_name="ap-northeast-1")
+    name = "my-rule"
+    payload = {
+        "sql": "SELECT * FROM 'topic/*' WHERE something > 0",
+        "actions": [
+            {"dynamoDBv2": {"putItem": {"tableName": "my-table"}, "roleArn": "my-role"}}
+        ],
+        "errorAction": {
+            "republish": {"qos": 0, "roleArn": "my-role", "topic": "other-topic"}
+        },
+        "description": "my-description",
+        "ruleDisabled": False,
+        "awsIotSqlVersion": "2016-03-23",
+    }
+
+    # create
+    client.create_topic_rule(ruleName=name, topicRulePayload=payload)
+
+    # list
+    res = client.list_topic_rules()
+    res.should.have.key("rules").which.should.have.length_of(1)
+    for rule in res["rules"]:
+        rule.should.have.key("ruleName").which.should.equal(name)
+        rule.should.have.key("createdAt").which.should_not.be.none
+        rule.should.have.key("ruleArn").which.should_not.be.none
+        rule.should.have.key("ruleDisabled").which.should.equal(payload["ruleDisabled"])
+        rule.should.have.key("topicPattern").which.should.equal("topic/*")
+
+    # get
+    rule = client.get_topic_rule(ruleName=name)
+    rule.should.have.key("ruleArn").which.should_not.be.none
+    rule.should.have.key("rule")
+    rrule = rule["rule"]
+    rrule.should.have.key("actions").which.should.equal(payload["actions"])
+    rrule.should.have.key("awsIotSqlVersion").which.should.equal(
+        payload["awsIotSqlVersion"]
+    )
+    rrule.should.have.key("createdAt").which.should_not.be.none
+    rrule.should.have.key("description").which.should.equal(payload["description"])
+    rrule.should.have.key("errorAction").which.should.equal(payload["errorAction"])
+    rrule.should.have.key("ruleDisabled").which.should.equal(payload["ruleDisabled"])
+    rrule.should.have.key("ruleName").which.should.equal(name)
+    rrule.should.have.key("sql").which.should.equal(payload["sql"])
+
+    # replace
+    payload["description"] = "new-description"
+    client.replace_topic_rule(
+        ruleName=name, topicRulePayload=payload,
+    )
+    rule = client.get_topic_rule(ruleName=name)
+    rule["rule"]["ruleName"].should.equal(name)
+    rule["rule"]["description"].should.equal(payload["description"])
+
+    # disable
+    client.disable_topic_rule(ruleName=name)
+    rule = client.get_topic_rule(ruleName=name)
+    rule["rule"]["ruleName"].should.equal(name)
+    rule["rule"]["ruleDisabled"].should.equal(True)
+
+    # enable
+    client.enable_topic_rule(ruleName=name)
+    rule = client.get_topic_rule(ruleName=name)
+    rule["rule"]["ruleName"].should.equal(name)
+    rule["rule"]["ruleDisabled"].should.equal(False)
+
+    # delete
+    client.delete_topic_rule(ruleName=name)
+    res = client.list_topic_rules()
+    res.should.have.key("rules").which.should.have.length_of(0)
+
+    # errors
+
+    # get not existing rule
+    try:
+        client.get_topic_rule(ruleName=name)
+    except ClientError as exc:
+        error_code = exc.response["Error"]["Code"]
+        error_code.should.equal("ResourceNotFoundException")
+    else:
+        raise Exception("Should have raised error")
+
+    # replace not existing rule
+    try:
+        client.replace_topic_rule(ruleName=name, topicRulePayload=payload)
+    except ClientError as exc:
+        error_code = exc.response["Error"]["Code"]
+        error_code.should.equal("ResourceNotFoundException")
+    else:
+        raise Exception("Should have raised error")
+
+    # delete not existing rule
+    try:
+        client.delete_topic_rule(ruleName=name)
+    except ClientError as exc:
+        error_code = exc.response["Error"]["Code"]
+        error_code.should.equal("ResourceNotFoundException")
+    else:
+        raise Exception("Should have raised error")
+
+    # enable not existing rule
+    try:
+        client.enable_topic_rule(ruleName=name)
+    except ClientError as exc:
+        error_code = exc.response["Error"]["Code"]
+        error_code.should.equal("ResourceNotFoundException")
+    else:
+        raise Exception("Should have raised error")
+
+    # disable not existing rule
+    try:
+        client.disable_topic_rule(ruleName=name)
+    except ClientError as exc:
+        error_code = exc.response["Error"]["Code"]
+        error_code.should.equal("ResourceNotFoundException")
+    else:
+        raise Exception("Should have raised error")
+
+    # create duplicated rule
+    client.create_topic_rule(ruleName=name, topicRulePayload=payload)
+    try:
+        client.create_topic_rule(ruleName=name, topicRulePayload=payload)
+    except ClientError as exc:
+        error_code = exc.response["Error"]["Code"]
+        error_code.should.equal("ResourceAlreadyExistsException")
+    else:
+        raise Exception("Should have raised error")