catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

IoT: Added Rules implementation (#3552)

Browse source 
* feat(iot): Added IoT Rules implementation

* fix(iot): Split IoT Rules tests into multiple test cases

* fix(iot): Updated rule creation to work with python 2

Co-authored-by: Szymon Zmilczak <szymon.zmilczak@machiq.com>
This commit is contained in:
Szymon Zmilczak 2021-01-14 15:52:23 +01:00 • committed by GitHub
commit c8151e1bb4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 307 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

163
tests/test_iot/test_iot.py
View file

@ -1991,3 +1991,166 @@ def test_list_job_executions_for_thing():
job_execution["executionSummaries"][0].should.have.key("jobId").which.should.equal(
job_id
)
class TestTopicRules:
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",
}
@mock_iot
def test_topic_rule_create(self):
client = boto3.client("iot", region_name="ap-northeast-1")
client.create_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
# duplicated rule name
with pytest.raises(ClientError) as ex:
client.create_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
error_code = ex.value.response["Error"]["Code"]
error_code.should.equal("ResourceAlreadyExistsException")
@mock_iot
def test_topic_rule_list(self):
client = boto3.client("iot", region_name="ap-northeast-1")
# empty response
res = client.list_topic_rules()
res.should.have.key("rules").which.should.have.length_of(0)
client.create_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
client.create_topic_rule(ruleName="my-rule-2", topicRulePayload=self.payload)
res = client.list_topic_rules()
res.should.have.key("rules").which.should.have.length_of(2)
for rule, name in zip(res["rules"], [self.name, "my-rule-2"]):
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(
self.payload["ruleDisabled"]
)
rule.should.have.key("topicPattern").which.should.equal("topic/*")
@mock_iot
def test_topic_rule_get(self):
client = boto3.client("iot", region_name="ap-northeast-1")
# no such rule
with pytest.raises(ClientError) as ex:
client.get_topic_rule(ruleName=self.name)
error_code = ex.value.response["Error"]["Code"]
error_code.should.equal("ResourceNotFoundException")
client.create_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
rule = client.get_topic_rule(ruleName=self.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(self.payload["actions"])
rrule.should.have.key("awsIotSqlVersion").which.should.equal(
self.payload["awsIotSqlVersion"]
)
rrule.should.have.key("createdAt").which.should_not.be.none
rrule.should.have.key("description").which.should.equal(
self.payload["description"]
)
rrule.should.have.key("errorAction").which.should.equal(
self.payload["errorAction"]
)
rrule.should.have.key("ruleDisabled").which.should.equal(
self.payload["ruleDisabled"]
)
rrule.should.have.key("ruleName").which.should.equal(self.name)
rrule.should.have.key("sql").which.should.equal(self.payload["sql"])
@mock_iot
def test_topic_rule_replace(self):
client = boto3.client("iot", region_name="ap-northeast-1")
# no such rule
with pytest.raises(ClientError) as ex:
client.replace_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
error_code = ex.value.response["Error"]["Code"]
error_code.should.equal("ResourceNotFoundException")
client.create_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
payload = self.payload.copy()
payload["description"] = "new-description"
client.replace_topic_rule(
ruleName=self.name, topicRulePayload=payload,
)
rule = client.get_topic_rule(ruleName=self.name)
rule["rule"]["ruleName"].should.equal(self.name)
rule["rule"]["description"].should.equal(payload["description"])
@mock_iot
def test_topic_rule_disable(self):
client = boto3.client("iot", region_name="ap-northeast-1")
# no such rule
with pytest.raises(ClientError) as ex:
client.disable_topic_rule(ruleName=self.name)
error_code = ex.value.response["Error"]["Code"]
error_code.should.equal("ResourceNotFoundException")
client.create_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
client.disable_topic_rule(ruleName=self.name)
rule = client.get_topic_rule(ruleName=self.name)
rule["rule"]["ruleName"].should.equal(self.name)
rule["rule"]["ruleDisabled"].should.equal(True)
@mock_iot
def test_topic_rule_enable(self):
client = boto3.client("iot", region_name="ap-northeast-1")
# no such rule
with pytest.raises(ClientError) as ex:
client.enable_topic_rule(ruleName=self.name)
error_code = ex.value.response["Error"]["Code"]
error_code.should.equal("ResourceNotFoundException")
payload = self.payload.copy()
payload["ruleDisabled"] = True
client.create_topic_rule(ruleName=self.name, topicRulePayload=payload)
client.enable_topic_rule(ruleName=self.name)
rule = client.get_topic_rule(ruleName=self.name)
rule["rule"]["ruleName"].should.equal(self.name)
rule["rule"]["ruleDisabled"].should.equal(False)
@mock_iot
def test_topic_rule_delete(self):
client = boto3.client("iot", region_name="ap-northeast-1")
# no such rule
with pytest.raises(ClientError) as ex:
client.delete_topic_rule(ruleName=self.name)
error_code = ex.value.response["Error"]["Code"]
error_code.should.equal("ResourceNotFoundException")
client.create_topic_rule(ruleName=self.name, topicRulePayload=self.payload)
client.enable_topic_rule(ruleName=self.name)
client.delete_topic_rule(ruleName=self.name)
res = client.list_topic_rules()
res.should.have.key("rules").which.should.have.length_of(0)