Techdebt: Replace sure with regular assertions in IOTData (#6619)

This commit is contained in:
Bert Blommers 2023-08-09 07:37:49 +00:00 committed by GitHub
parent 993a904ac4
commit db87597018
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 32 additions and 46 deletions

View File

@ -1,6 +1,5 @@
import json
import boto3
import sure # noqa # pylint: disable=unused-import
import pytest
from botocore.exceptions import ClientError
@ -25,22 +24,18 @@ def test_basic():
payload = json.loads(res["payload"].read())
expected_state = '{"desired": {"led": "on"}}'
payload.should.have.key("state").which.should.equal(json.loads(expected_state))
payload.should.have.key("metadata").which.should.have.key(
"desired"
).which.should.have.key("led")
payload.should.have.key("version").which.should.equal(1)
payload.should.have.key("timestamp")
assert payload["state"] == json.loads(expected_state)
assert "led" in payload["metadata"]["desired"]
assert payload["version"] == 1
assert "timestamp" in payload
res = client.get_thing_shadow(thingName=name)
payload = json.loads(res["payload"].read())
expected_state = b'{"desired": {"led": "on"}, "delta": {"led": "on"}}'
payload.should.have.key("state").which.should.equal(json.loads(expected_state))
payload.should.have.key("metadata").which.should.have.key(
"desired"
).which.should.have.key("led")
payload.should.have.key("version").which.should.equal(1)
payload.should.have.key("timestamp")
assert payload["state"] == json.loads(expected_state)
assert "led" in payload["metadata"]["desired"]
assert payload["version"] == 1
assert "timestamp" in payload
client.delete_thing_shadow(thingName=name)
with pytest.raises(ClientError):
@ -60,50 +55,42 @@ def test_update():
res = client.update_thing_shadow(thingName=name, payload=raw_payload)
payload = json.loads(res["payload"].read())
expected_state = '{"desired": {"led": "on"}}'
payload.should.have.key("state").which.should.equal(json.loads(expected_state))
payload.should.have.key("metadata").which.should.have.key(
"desired"
).which.should.have.key("led")
payload.should.have.key("version").which.should.equal(1)
payload.should.have.key("timestamp")
assert payload["state"] == json.loads(expected_state)
assert "led" in payload["metadata"]["desired"]
assert payload["version"] == 1
assert "timestamp" in payload
res = client.get_thing_shadow(thingName=name)
payload = json.loads(res["payload"].read())
expected_state = b'{"desired": {"led": "on"}, "delta": {"led": "on"}}'
payload.should.have.key("state").which.should.equal(json.loads(expected_state))
payload.should.have.key("metadata").which.should.have.key(
"desired"
).which.should.have.key("led")
payload.should.have.key("version").which.should.equal(1)
payload.should.have.key("timestamp")
assert payload["state"] == json.loads(expected_state)
assert "led" in payload["metadata"]["desired"]
assert payload["version"] == 1
assert "timestamp" in payload
# reporting new state
new_payload = b'{"state": {"reported": {"led": "on"}}}'
res = client.update_thing_shadow(thingName=name, payload=new_payload)
payload = json.loads(res["payload"].read())
expected_state = '{"reported": {"led": "on"}}'
payload.should.have.key("state").which.should.equal(json.loads(expected_state))
payload.should.have.key("metadata").which.should.have.key(
"reported"
).which.should.have.key("led")
payload.should.have.key("version").which.should.equal(2)
payload.should.have.key("timestamp")
assert payload["state"] == json.loads(expected_state)
assert "led" in payload["metadata"]["reported"]
assert payload["version"] == 2
assert "timestamp" in payload
res = client.get_thing_shadow(thingName=name)
payload = json.loads(res["payload"].read())
expected_state = b'{"desired": {"led": "on"}, "reported": {"led": "on"}}'
payload.should.have.key("state").which.should.equal(json.loads(expected_state))
payload.should.have.key("metadata").which.should.have.key(
"desired"
).which.should.have.key("led")
payload.should.have.key("version").which.should.equal(2)
payload.should.have.key("timestamp")
assert payload["state"] == json.loads(expected_state)
assert "led" in payload["metadata"]["desired"]
assert payload["version"] == 2
assert "timestamp" in payload
raw_payload = b'{"state": {"desired": {"led": "on"}}, "version": 1}'
with pytest.raises(ClientError) as ex:
client.update_thing_shadow(thingName=name, payload=raw_payload)
ex.value.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(409)
ex.value.response["Error"]["Message"].should.equal("Version conflict")
assert ex.value.response["ResponseMetadata"]["HTTPStatusCode"] == 409
assert ex.value.response["Error"]["Message"] == "Version conflict"
@mock_iotdata
@ -116,10 +103,10 @@ def test_publish():
if not settings.TEST_SERVER_MODE:
mock_backend = moto.iotdata.models.iotdata_backends[ACCOUNT_ID][region_name]
mock_backend.published_payloads.should.have.length_of(3)
mock_backend.published_payloads.should.contain(("test/topic1", b"pl1"))
mock_backend.published_payloads.should.contain(("test/topic2", b"pl2"))
mock_backend.published_payloads.should.contain(("test/topic3", b"\xbf"))
assert len(mock_backend.published_payloads) == 3
assert ("test/topic1", b"pl1") in mock_backend.published_payloads
assert ("test/topic2", b"pl2") in mock_backend.published_payloads
assert ("test/topic3", b"\xbf") in mock_backend.published_payloads
@mock_iot

View File

@ -1,7 +1,6 @@
from urllib.parse import quote
import pytest
import sure # noqa # pylint: disable=unused-import
import moto.server as server
from moto import mock_iotdata
@ -19,7 +18,7 @@ def test_iotdata_list():
# just making sure that server is up
thing_name = "nothing"
res = test_client.get(f"/things/{thing_name}/shadow")
res.status_code.should.equal(404)
assert res.status_code == 404
@pytest.mark.parametrize(
@ -38,4 +37,4 @@ def test_publish(url_encode_topic):
topic_for_path = quote(topic, safe="") if url_encode_topic else topic
result = test_client.post(f"/topics/{topic_for_path}")
result.status_code.should.equal(200)
assert result.status_code == 200