0de2e55b13
* append appropriate urls when scaffolding * make dispatch for rest-api * fix dispatch for rest-json * fix moto/core/response to obtain path and body parameters * small fixes * remove unused import * fix get_int_param * Add features of things and thing-types * fix scaffold * basic crud of cert * support basic CRUD of policy * refactor * fix formatting of scaffold * support principal_pocicy * support thing_principal * update readme * escape service to handle service w/ hyphen like iot-data * escape service w/ hyphen * fix regexp to extract region from url * escape service * Implement basic iota-data feature * iot-data shadow delta * update readme * remove unused import * remove comment * fix syntax * specify region when creating boto3 client for test * use uuid for seed of generating cert id * specify region_name to iotdata client in test * specify region to boto3 client in moto response * excude iot and iotdata tests on server mode * fix handling of thingTypeName in describe-thing * test if server is up for iot
88 lines
3.8 KiB
Python
88 lines
3.8 KiB
Python
from __future__ import unicode_literals
|
|
|
|
import json
|
|
import boto3
|
|
import sure # noqa
|
|
from nose.tools import assert_raises
|
|
from botocore.exceptions import ClientError
|
|
from moto import mock_iotdata, mock_iot
|
|
|
|
|
|
@mock_iot
|
|
@mock_iotdata
|
|
def test_basic():
|
|
iot_client = boto3.client('iot', region_name='ap-northeast-1')
|
|
client = boto3.client('iot-data', region_name='ap-northeast-1')
|
|
name = 'my-thing'
|
|
raw_payload = b'{"state": {"desired": {"led": "on"}}}'
|
|
iot_client.create_thing(thingName=name)
|
|
|
|
with assert_raises(ClientError):
|
|
client.get_thing_shadow(thingName=name)
|
|
|
|
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')
|
|
|
|
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')
|
|
|
|
client.delete_thing_shadow(thingName=name)
|
|
with assert_raises(ClientError):
|
|
client.get_thing_shadow(thingName=name)
|
|
|
|
|
|
@mock_iot
|
|
@mock_iotdata
|
|
def test_update():
|
|
iot_client = boto3.client('iot', region_name='ap-northeast-1')
|
|
client = boto3.client('iot-data', region_name='ap-northeast-1')
|
|
name = 'my-thing'
|
|
raw_payload = b'{"state": {"desired": {"led": "on"}}}'
|
|
iot_client.create_thing(thingName=name)
|
|
|
|
# first 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')
|
|
|
|
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')
|
|
|
|
# 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')
|
|
|
|
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')
|