Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

IOT: create_thing()/describe_thing() now returns the thingId (#7081)

Browse Source
This commit is contained in:
Bert Blommers 2023-11-30 21:27:31 -01:00 committed by GitHub
parent 4c8096feb2
commit 4e1ea0bb6f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 37 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
moto/iot/models.py
View File

@ -43,6 +43,7 @@ class FakeThing(BaseModel):
region_name: str,
):
self.region_name = region_name
self.thing_id = str(random.uuid4())
self.thing_name = thing_name
self.thing_type = thing_type
self.attributes = attributes
@ -68,6 +69,7 @@ class FakeThing(BaseModel):
self,
include_default_client_id: bool = False,
include_connectivity: bool = False,
include_thing_id: bool = False,
) -> Dict[str, Any]:
obj = {
"thingName": self.thing_name,
@ -84,6 +86,8 @@ class FakeThing(BaseModel):
"connected": True,
"timestamp": time.mktime(utcnow().timetuple()),
}
if include_thing_id:
obj["thingId"] = self.thing_id
return obj
@ -697,7 +701,7 @@ class IoTBackend(BaseBackend):
thing_name: str,
thing_type_name: str,
attribute_payload: Optional[Dict[str, Any]],
) -> Tuple[str, str]:
) -> FakeThing:
thing_types = self.list_thing_types()
thing_type = None
if thing_type_name:
@ -723,7 +727,7 @@ class IoTBackend(BaseBackend):
thing_name, thing_type, attributes, self.account_id, self.region_name
)
self.things[thing.arn] = thing
return thing.thing_name, thing.arn
return thing
def create_thing_type(
self, thing_type_name: str, thing_type_properties: Dict[str, Any]
@ -1128,9 +1132,15 @@ class IoTBackend(BaseBackend):
del self.principal_policies[k]
def list_attached_policies(self, target: str) -> List[FakePolicy]:
"""
Pagination is not yet implemented
"""
return [v[1] for k, v in self.principal_policies.items() if k[0] == target]
def list_policies(self) -> Iterable[FakePolicy]:
"""
Pagination is not yet implemented
"""
return self.policies.values()
def get_policy(self, policy_name: str) -> FakePolicy:
@ -1273,12 +1283,18 @@ class IoTBackend(BaseBackend):
del self.principal_policies[k]
def list_principal_policies(self, principal_arn: str) -> List[FakePolicy]:
"""
Pagination is not yet implemented
"""
policies = [
v[1] for k, v in self.principal_policies.items() if k[0] == principal_arn
]
return policies
def list_policy_principals(self, policy_name: str) -> List[str]:
"""
Pagination is not yet implemented
"""
# this action is deprecated
# https://docs.aws.amazon.com/iot/latest/apireference/API_ListTargetsForPolicy.html
# should use ListTargetsForPolicy instead
@ -1288,6 +1304,9 @@ class IoTBackend(BaseBackend):
return principals
def list_targets_for_policy(self, policy_name: str) -> List[str]:
"""
Pagination is not yet implemented
"""
# This behaviour is different to list_policy_principals which will just return an empty list
if policy_name not in self.policies:
raise ResourceNotFoundException("Policy not found")

40
moto/iot/responses.py
View File

@ -34,12 +34,14 @@ class IoTResponse(BaseResponse):
thing_name = self._get_param("thingName")
thing_type_name = self._get_param("thingTypeName")
attribute_payload = self._get_param("attributePayload")
thing_name, thing_arn = self.iot_backend.create_thing(
thing = self.iot_backend.create_thing(
thing_name=thing_name,
thing_type_name=thing_type_name,
attribute_payload=attribute_payload,
)
return json.dumps(dict(thingName=thing_name, thingArn=thing_arn))
return json.dumps(
dict(thingName=thing.thing_name, thingArn=thing.arn, thingId=thing.thing_id)
)
def create_thing_type(self) -> str:
thing_type_name = self._get_param("thingTypeName")
@ -95,7 +97,9 @@ class IoTResponse(BaseResponse):
def describe_thing(self) -> str:
thing_name = self._get_param("thingName")
thing = self.iot_backend.describe_thing(thing_name=thing_name)
return json.dumps(thing.to_dict(include_default_client_id=True))
return json.dumps(
thing.to_dict(include_default_client_id=True, include_thing_id=True)
)
def describe_thing_type(self) -> str:
thing_type_name = self._get_param("thingTypeName")
@ -415,12 +419,8 @@ class IoTResponse(BaseResponse):
return json.dumps(policy.to_dict_at_creation())
def list_policies(self) -> str:
# marker = self._get_param("marker")
# page_size = self._get_int_param("pageSize")
# ascending_order = self._get_param("ascendingOrder")
policies = self.iot_backend.list_policies()
# TODO: implement pagination in the future
return json.dumps(dict(policies=[_.to_dict() for _ in policies]))
def get_policy(self) -> str:
@ -492,14 +492,8 @@ class IoTResponse(BaseResponse):
def list_attached_policies(self) -> str:
principal = self._get_param("target")
# marker = self._get_param("marker")
# page_size = self._get_int_param("pageSize")
policies = self.iot_backend.list_attached_policies(target=principal)
# TODO: implement pagination in the future
next_marker = None
return json.dumps(
dict(policies=[_.to_dict() for _ in policies], nextMarker=next_marker)
)
return json.dumps(dict(policies=[_.to_dict() for _ in policies]))
def attach_principal_policy(self) -> str:
policy_name = self._get_param("policyName")
@ -525,31 +519,19 @@ class IoTResponse(BaseResponse):
def list_principal_policies(self) -> str:
principal = self.headers.get("x-amzn-iot-principal")
# marker = self._get_param("marker")
# page_size = self._get_int_param("pageSize")
# ascending_order = self._get_param("ascendingOrder")
policies = self.iot_backend.list_principal_policies(principal_arn=principal)
# TODO: implement pagination in the future
next_marker = None
return json.dumps(
dict(policies=[_.to_dict() for _ in policies], nextMarker=next_marker)
)
return json.dumps(dict(policies=[_.to_dict() for _ in policies]))
def list_policy_principals(self) -> str:
policy_name = self.headers.get("x-amzn-iot-policy")
# marker = self._get_param("marker")
# page_size = self._get_int_param("pageSize")
# ascending_order = self._get_param("ascendingOrder")
principals = self.iot_backend.list_policy_principals(policy_name=policy_name)
# TODO: implement pagination in the future
next_marker = None
return json.dumps(dict(principals=principals, nextMarker=next_marker))
return json.dumps(dict(principals=principals))
def list_targets_for_policy(self) -> str:
"""https://docs.aws.amazon.com/iot/latest/apireference/API_ListTargetsForPolicy.html"""
policy_name = self._get_param("policyName")
principals = self.iot_backend.list_targets_for_policy(policy_name=policy_name)
return json.dumps(dict(targets=principals, nextMarker=None))
return json.dumps(dict(targets=principals))
def attach_thing_principal(self) -> str:
thing_name = self._get_param("thingName")

8
tests/test_iot/test_iot_things.py
View File

@ -11,12 +11,13 @@ def test_create_thing():
thing = client.create_thing(thingName=name)
assert thing["thingName"] == name
assert "thingArn" in thing
assert thing["thingArn"] is not None
assert thing["thingId"] is not None
res = client.list_things()
assert len(res["things"]) == 1
assert res["things"][0]["thingName"] is not None
assert res["things"][0]["thingArn"] is not None
assert res["things"][0]["thingName"] == name
assert res["things"][0]["thingArn"] == thing["thingArn"]
@mock_iot
@ -105,6 +106,7 @@ def test_describe_thing():
client.update_thing(thingName=name, attributePayload={"attributes": {"k1": "v1"}})
thing = client.describe_thing(thingName=name)
assert thing["thingId"] is not None
assert thing["thingName"] == name
assert "defaultClientId" in thing
assert thing["attributes"] == {"k1": "v1"}