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

IOT improvements (#4848)

Browse Source
This commit is contained in:
Bert Blommers 2022-02-10 08:02:40 -01:00 committed by GitHub
parent f8be8ce2b8
commit b28d763c08
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 562 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
IMPLEMENTATION_COVERAGE.md
View File

@ -2987,7 +2987,7 @@
## iot
<details>
<summary>30% implemented</summary>
<summary>33% implemented</summary>
- [ ] accept_certificate_transfer
- [ ] add_thing_to_billing_group
@ -3008,7 +3008,7 @@
- [ ] create_audit_suppression
- [ ] create_authorizer
- [ ] create_billing_group
- [ ] create_certificate_from_csr
- [X] create_certificate_from_csr
- [ ] create_custom_metric
- [ ] create_dimension
- [X] create_domain_configuration
@ -3037,7 +3037,7 @@
- [ ] delete_audit_suppression
- [ ] delete_authorizer
- [ ] delete_billing_group
- [ ] delete_ca_certificate
- [X] delete_ca_certificate
- [X] delete_certificate
- [ ] delete_custom_metric
- [ ] delete_dimension
@ -3072,7 +3072,7 @@
- [ ] describe_audit_task
- [ ] describe_authorizer
- [ ] describe_billing_group
- [ ] describe_ca_certificate
- [X] describe_ca_certificate
- [X] describe_certificate
- [ ] describe_custom_metric
- [ ] describe_default_authorizer
@ -3115,7 +3115,7 @@
- [ ] get_percentiles
- [X] get_policy
- [X] get_policy_version
- [ ] get_registration_code
- [X] get_registration_code
- [ ] get_statistics
- [X] get_topic_rule
- [ ] get_topic_rule_destination
@ -3131,7 +3131,7 @@
- [ ] list_billing_groups
- [ ] list_ca_certificates
- [X] list_certificates
- [ ] list_certificates_by_ca
- [X] list_certificates_by_ca
- [ ] list_custom_metrics
- [ ] list_detect_mitigation_actions_executions
- [ ] list_detect_mitigation_actions_tasks
@ -3176,7 +3176,7 @@
- [ ] list_v2_logging_levels
- [ ] list_violation_events
- [ ] put_verification_state_on_violation
- [ ] register_ca_certificate
- [X] register_ca_certificate
- [X] register_certificate
- [X] register_certificate_without_ca
- [ ] register_thing
@ -3184,7 +3184,7 @@
- [ ] remove_thing_from_billing_group
- [X] remove_thing_from_thing_group
- [X] replace_topic_rule
- [ ] search_index
- [X] search_index
- [ ] set_default_authorizer
- [X] set_default_policy_version
- [ ] set_logging_options
@ -3204,7 +3204,7 @@
- [ ] update_audit_suppression
- [ ] update_authorizer
- [ ] update_billing_group
- [ ] update_ca_certificate
- [X] update_ca_certificate
- [X] update_certificate
- [ ] update_custom_metric
- [ ] update_dimension
@ -5555,4 +5555,4 @@
- workspaces
- workspaces-web
- xray
</details>
</details>

40
docs/docs/services/iot.rst
View File

@ -44,7 +44,7 @@ iot
- [ ] create_audit_suppression
- [ ] create_authorizer
- [ ] create_billing_group
- [ ] create_certificate_from_csr
- [X] create_certificate_from_csr
- [ ] create_custom_metric
- [ ] create_dimension
- [X] create_domain_configuration
@ -73,7 +73,7 @@ iot
- [ ] delete_audit_suppression
- [ ] delete_authorizer
- [ ] delete_billing_group
- [ ] delete_ca_certificate
- [X] delete_ca_certificate
- [X] delete_certificate
- [ ] delete_custom_metric
- [ ] delete_dimension
@ -108,7 +108,7 @@ iot
- [ ] describe_audit_task
- [ ] describe_authorizer
- [ ] describe_billing_group
- [ ] describe_ca_certificate
- [X] describe_ca_certificate
- [X] describe_certificate
- [ ] describe_custom_metric
- [ ] describe_default_authorizer
@ -151,7 +151,7 @@ iot
- [ ] get_percentiles
- [X] get_policy
- [X] get_policy_version
- [ ] get_registration_code
- [X] get_registration_code
- [ ] get_statistics
- [X] get_topic_rule
- [ ] get_topic_rule_destination
@ -167,7 +167,15 @@ iot
- [ ] list_billing_groups
- [ ] list_ca_certificates
- [X] list_certificates
- [ ] list_certificates_by_ca
Pagination is not yet implemented
- [X] list_certificates_by_ca
Pagination is not yet implemented
- [ ] list_custom_metrics
- [ ] list_detect_mitigation_actions_executions
- [ ] list_detect_mitigation_actions_tasks
@ -200,6 +208,10 @@ iot
- [ ] list_targets_for_security_profile
- [X] list_thing_groups
- [X] list_thing_groups_for_thing
Pagination is not yet implemented
- [X] list_thing_principals
- [ ] list_thing_registration_task_reports
- [ ] list_thing_registration_tasks
@ -207,12 +219,16 @@ iot
- [X] list_things
- [ ] list_things_in_billing_group
- [X] list_things_in_thing_group
The recursive-parameter is not yet implemented
- [ ] list_topic_rule_destinations
- [X] list_topic_rules
- [ ] list_v2_logging_levels
- [ ] list_violation_events
- [ ] put_verification_state_on_violation
- [ ] register_ca_certificate
- [X] register_ca_certificate
- [X] register_certificate
- [X] register_certificate_without_ca
- [ ] register_thing
@ -220,7 +236,11 @@ iot
- [ ] remove_thing_from_billing_group
- [X] remove_thing_from_thing_group
- [X] replace_topic_rule
- [ ] search_index
- [X] search_index
Pagination is not yet implemented. Only basic search queries are supported for now.
- [ ] set_default_authorizer
- [X] set_default_policy_version
- [ ] set_logging_options
@ -240,7 +260,11 @@ iot
- [ ] update_audit_suppression
- [ ] update_authorizer
- [ ] update_billing_group
- [ ] update_ca_certificate
- [X] update_ca_certificate
The newAutoRegistrationStatus and removeAutoRegistration-parameters are not yet implemented
- [X] update_certificate
- [ ] update_custom_metric
- [ ] update_dimension

11
moto/iot/exceptions.py
View File

@ -1,3 +1,5 @@
import json
from moto.core.exceptions import JsonRESTError
@ -50,11 +52,18 @@ class DeleteConflictException(IoTClientError):
class ResourceAlreadyExistsException(IoTClientError):
def __init__(self, msg):
def __init__(self, msg, resource_id, resource_arn):
self.code = 409
super().__init__(
"ResourceAlreadyExistsException", msg or "The resource already exists."
)
self.description = json.dumps(
{
"message": self.message,
"resourceId": resource_id,
"resourceArn": resource_arn,
}
)
class VersionsLimitExceededException(IoTClientError):

205
moto/iot/models.py
View File

@ -5,7 +5,12 @@ import string
import time
import uuid
from collections import OrderedDict
from datetime import datetime
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization, hashes
from datetime import datetime, timedelta
from .utils import PAGINATION_MODEL
@ -39,6 +44,15 @@ class FakeThing(BaseModel):
# for iot-data
self.thing_shadow = None
def matches(self, query_string):
if query_string.startswith("thingName:"):
qs = query_string[10:].replace("*", ".*").replace("?", ".")
return re.search(f"^{qs}$", self.thing_name)
if query_string.startswith("attributes."):
k, v = query_string[11:].split(":")
return self.attributes.get(k) == v
return query_string in self.thing_name
def to_dict(self, include_default_client_id=False):
obj = {
"thingName": self.thing_name,
@ -133,26 +147,21 @@ class FakeThingGroup(BaseModel):
class FakeCertificate(BaseModel):
def __init__(self, certificate_pem, status, region_name, ca_certificate_pem=None):
def __init__(self, certificate_pem, status, region_name, ca_certificate_id=None):
m = hashlib.sha256()
m.update(certificate_pem.encode("utf-8"))
self.certificate_id = m.hexdigest()
self.arn = "arn:aws:iot:%s:1:cert/%s" % (region_name, self.certificate_id)
self.arn = f"arn:aws:iot:{region_name}:{ACCOUNT_ID}:cert/{self.certificate_id}"
self.certificate_pem = certificate_pem
self.status = status
# TODO: must adjust
self.owner = "1"
self.owner = ACCOUNT_ID
self.transfer_data = {}
self.creation_date = time.time()
self.last_modified_date = self.creation_date
self.validity_not_before = time.time() - 86400
self.validity_not_after = time.time() + 86400
self.ca_certificate_id = None
self.ca_certificate_pem = ca_certificate_pem
if ca_certificate_pem:
m.update(ca_certificate_pem.encode("utf-8"))
self.ca_certificate_id = m.hexdigest()
self.ca_certificate_id = ca_certificate_id
def to_dict(self):
return {
@ -185,6 +194,17 @@ class FakeCertificate(BaseModel):
}
class FakeCaCertificate(FakeCertificate):
def __init__(self, ca_certificate, status, region_name, registration_config):
super().__init__(
certificate_pem=ca_certificate,
status=status,
region_name=region_name,
ca_certificate_id=None,
)
self.registration_config = registration_config
class FakePolicy(BaseModel):
def __init__(self, name, document, region_name, default_version_id="1"):
self.name = name
@ -398,22 +418,20 @@ class FakeEndpoint(BaseModel):
"operation: Endpoint type %s not recognized." % endpoint_type
)
self.region_name = region_name
data_identifier = random_string(14)
identifier = random_string(14).lower()
if endpoint_type == "iot:Data":
self.endpoint = "{i}.iot.{r}.amazonaws.com".format(
i=data_identifier, r=self.region_name
i=identifier, r=self.region_name
)
elif "iot:Data-ATS" in endpoint_type:
self.endpoint = "{i}-ats.iot.{r}.amazonaws.com".format(
i=data_identifier, r=self.region_name
i=identifier, r=self.region_name
)
elif "iot:CredentialProvider" in endpoint_type:
identifier = random_string(14)
self.endpoint = "{i}.credentials.iot.{r}.amazonaws.com".format(
i=identifier, r=self.region_name
)
elif "iot:Jobs" in endpoint_type:
identifier = random_string(14)
self.endpoint = "{i}.jobs.iot.{r}.amazonaws.com".format(
i=identifier, r=self.region_name
)
@ -550,6 +568,7 @@ class IoTBackend(BaseBackend):
self.job_executions = OrderedDict()
self.thing_types = OrderedDict()
self.thing_groups = OrderedDict()
self.ca_certificates = OrderedDict()
self.certificates = OrderedDict()
self.policies = OrderedDict()
self.principal_policies = OrderedDict()
@ -577,6 +596,48 @@ class IoTBackend(BaseBackend):
policy_supported=False,
)
def create_certificate_from_csr(self, csr, set_as_active):
cert = x509.load_pem_x509_csr(csr.encode("utf-8"), default_backend())
pem = self._generate_certificate_pem(
domain_name="example.com", subject=cert.subject
)
return self.register_certificate(
pem, ca_certificate_pem=None, set_as_active=set_as_active, status="INACTIVE"
)
def _generate_certificate_pem(self, domain_name, subject):
sans = set()
sans.add(domain_name)
sans = [x509.DNSName(item) for item in sans]
key = rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend()
)
issuer = x509.Name(
[ # C = US, O = Moto, OU = Server CA 1B, CN = Moto
x509.NameAttribute(x509.NameOID.COUNTRY_NAME, "US"),
x509.NameAttribute(x509.NameOID.ORGANIZATION_NAME, "Moto"),
x509.NameAttribute(
x509.NameOID.ORGANIZATIONAL_UNIT_NAME, "Server CA 1B"
),
x509.NameAttribute(x509.NameOID.COMMON_NAME, "Moto"),
]
)
cert = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(datetime.utcnow())
.not_valid_after(datetime.utcnow() + timedelta(days=365))
.add_extension(x509.SubjectAlternativeName(sans), critical=False)
.sign(key, hashes.SHA512(), default_backend())
)
return cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
def create_thing(self, thing_name, thing_type_name, attribute_payload):
thing_types = self.list_thing_types()
thing_type = None
@ -779,18 +840,27 @@ class IoTBackend(BaseBackend):
self.certificates[certificate.certificate_id] = certificate
return certificate, key_pair
def delete_ca_certificate(self, certificate_id):
cert = self.describe_ca_certificate(certificate_id)
self._validation_delete(cert)
del self.ca_certificates[certificate_id]
def delete_certificate(self, certificate_id):
cert = self.describe_certificate(certificate_id)
self._validation_delete(cert)
del self.certificates[certificate_id]
def _validation_delete(self, cert):
if cert.status == "ACTIVE":
raise CertificateStateException(
"Certificate must be deactivated (not ACTIVE) before deletion.",
certificate_id,
cert.certificate_id,
)
certs = [
k[0]
for k, v in self.principal_things.items()
if self._get_principal(k[0]).certificate_id == certificate_id
if self._get_principal(k[0]).certificate_id == cert.certificate_id
]
if len(certs) > 0:
raise DeleteConflictException(
@ -800,7 +870,7 @@ class IoTBackend(BaseBackend):
certs = [
k[0]
for k, v in self.principal_policies.items()
if self._get_principal(k[0]).certificate_id == certificate_id
if self._get_principal(k[0]).certificate_id == cert.certificate_id
]
if len(certs) > 0:
raise DeleteConflictException(
@ -808,7 +878,10 @@ class IoTBackend(BaseBackend):
% certs[0]
)
del self.certificates[certificate_id]
def describe_ca_certificate(self, certificate_id):
if certificate_id not in self.ca_certificates:
raise ResourceNotFoundException()
return self.ca_certificates[certificate_id]
def describe_certificate(self, certificate_id):
certs = [
@ -818,36 +891,92 @@ class IoTBackend(BaseBackend):
raise ResourceNotFoundException()
return certs[0]
def get_registration_code(self):
return str(uuid.uuid4())
def list_certificates(self):
"""
Pagination is not yet implemented
"""
return self.certificates.values()
def __raise_if_certificate_already_exists(self, certificate_id):
def list_certificates_by_ca(self, ca_certificate_id):
"""
Pagination is not yet implemented
"""
return [
cert
for cert in self.certificates.values()
if cert.ca_certificate_id == ca_certificate_id
]
def __raise_if_certificate_already_exists(self, certificate_id, certificate_arn):
if certificate_id in self.certificates:
raise ResourceAlreadyExistsException(
"The certificate is already provisioned or registered"
"The certificate is already provisioned or registered",
certificate_id,
certificate_arn,
)
def register_ca_certificate(
self,
ca_certificate,
verification_certificate,
set_as_active,
registration_config,
):
certificate = FakeCaCertificate(
ca_certificate=ca_certificate,
status="ACTIVE" if set_as_active else "INACTIVE",
region_name=self.region_name,
registration_config=registration_config,
)
self.ca_certificates[certificate.certificate_id] = certificate
return certificate
def _find_ca_certificate(self, ca_certificate_pem):
for ca_cert in self.ca_certificates.values():
if ca_cert.certificate_pem == ca_certificate_pem:
return ca_cert.certificate_id
return None
def register_certificate(
self, certificate_pem, ca_certificate_pem, set_as_active, status
):
ca_certificate_id = self._find_ca_certificate(ca_certificate_pem)
certificate = FakeCertificate(
certificate_pem,
"ACTIVE" if set_as_active else status,
self.region_name,
ca_certificate_pem,
ca_certificate_id,
)
self.__raise_if_certificate_already_exists(
certificate.certificate_id, certificate_arn=certificate.arn
)
self.__raise_if_certificate_already_exists(certificate.certificate_id)
self.certificates[certificate.certificate_id] = certificate
return certificate
def register_certificate_without_ca(self, certificate_pem, status):
certificate = FakeCertificate(certificate_pem, status, self.region_name)
self.__raise_if_certificate_already_exists(certificate.certificate_id)
self.__raise_if_certificate_already_exists(
certificate.certificate_id, certificate_arn=certificate.arn
)
self.certificates[certificate.certificate_id] = certificate
return certificate
def update_ca_certificate(self, certificate_id, new_status, config):
"""
The newAutoRegistrationStatus and removeAutoRegistration-parameters are not yet implemented
"""
cert = self.describe_ca_certificate(certificate_id)
if new_status is not None:
cert.status = new_status
if config is not None:
cert.registration_config = config
def update_certificate(self, certificate_id, new_status):
cert = self.describe_certificate(certificate_id)
# TODO: validate new_status
@ -1200,10 +1329,16 @@ class IoTBackend(BaseBackend):
del thing_group.things[thing.arn]
def list_things_in_thing_group(self, thing_group_name, recursive):
"""
The recursive-parameter is not yet implemented
"""
thing_group = self.describe_thing_group(thing_group_name)
return thing_group.things.values()
def list_thing_groups_for_thing(self, thing_name):
"""
Pagination is not yet implemented
"""
thing = self.describe_thing(thing_name)
all_thing_groups = self.list_thing_groups(None, None, None)
ret = []
@ -1434,7 +1569,9 @@ class IoTBackend(BaseBackend):
def create_topic_rule(self, rule_name, sql, **kwargs):
if rule_name in self.rules:
raise ResourceAlreadyExistsException("Rule with given name already exists")
raise ResourceAlreadyExistsException(
"Rule with given name already exists", "", self.rules[rule_name].arn
)
result = re.search(r"FROM\s+([^\s]*)", sql)
topic = result.group(1).strip("'") if result else None
self.rules[rule_name] = FakeRule(
@ -1476,7 +1613,13 @@ class IoTBackend(BaseBackend):
):
if domain_configuration_name in self.domain_configurations:
raise ResourceAlreadyExistsException(
"Domain configuration with given name already exists."
"Domain configuration with given name already exists.",
self.domain_configurations[
domain_configuration_name
].domain_configuration_name,
self.domain_configurations[
domain_configuration_name
].domain_configuration_arn,
)
self.domain_configurations[domain_configuration_name] = FakeDomainConfiguration(
self.region_name,
@ -1523,5 +1666,15 @@ class IoTBackend(BaseBackend):
domain_configuration.authorizer_config = None
return domain_configuration
def search_index(self, query_string):
"""
Pagination is not yet implemented. Only basic search queries are supported for now.
"""
things = [
thing for thing in self.things.values() if thing.matches(query_string)
]
groups = []
return [t.to_dict() for t in things], groups
iot_backends = BackendDict(IoTBackend, "iot")

74
moto/iot/responses.py
View File

@ -12,6 +12,20 @@ class IoTResponse(BaseResponse):
def iot_backend(self):
return iot_backends[self.region]
def create_certificate_from_csr(self):
certificate_signing_request = self._get_param("certificateSigningRequest")
set_as_active = self._get_param("setAsActive")
cert = self.iot_backend.create_certificate_from_csr(
certificate_signing_request, set_as_active=set_as_active
)
return json.dumps(
{
"certificateId": cert.certificate_id,
"certificateArn": cert.arn,
"certificatePem": cert.certificate_pem,
}
)
def create_thing(self):
thing_name = self._get_param("thingName")
thing_type_name = self._get_param("thingTypeName")
@ -87,7 +101,7 @@ class IoTResponse(BaseResponse):
return json.dumps(thing_type.to_dict())
def describe_endpoint(self):
endpoint_type = self._get_param("endpointType")
endpoint_type = self._get_param("endpointType", "iot:Data-ATS")
endpoint = self.iot_backend.describe_endpoint(endpoint_type=endpoint_type)
return json.dumps(endpoint.to_dict())
@ -303,11 +317,28 @@ class IoTResponse(BaseResponse):
)
)
def delete_ca_certificate(self):
certificate_id = self.path.split("/")[-1]
self.iot_backend.delete_ca_certificate(certificate_id=certificate_id)
return json.dumps(dict())
def delete_certificate(self):
certificate_id = self._get_param("certificateId")
self.iot_backend.delete_certificate(certificate_id=certificate_id)
return json.dumps(dict())
def describe_ca_certificate(self):
certificate_id = self.path.split("/")[-1]
certificate = self.iot_backend.describe_ca_certificate(
certificate_id=certificate_id
)
return json.dumps(
{
"certificateDescription": certificate.to_description_dict(),
"registrationConfig": certificate.registration_config,
}
)
def describe_certificate(self):
certificate_id = self._get_param("certificateId")
certificate = self.iot_backend.describe_certificate(
@ -317,14 +348,38 @@ class IoTResponse(BaseResponse):
dict(certificateDescription=certificate.to_description_dict())
)
def get_registration_code(self):
code = self.iot_backend.get_registration_code()
return json.dumps(dict(registrationCode=code))
def list_certificates(self):
# page_size = self._get_int_param("pageSize")
# marker = self._get_param("marker")
# ascending_order = self._get_param("ascendingOrder")
certificates = self.iot_backend.list_certificates()
# TODO: implement pagination in the future
return json.dumps(dict(certificates=[_.to_dict() for _ in certificates]))
def list_certificates_by_ca(self):
ca_certificate_id = self._get_param("caCertificateId")
certificates = self.iot_backend.list_certificates_by_ca(ca_certificate_id)
return json.dumps(dict(certificates=[_.to_dict() for _ in certificates]))
def register_ca_certificate(self):
ca_certificate = self._get_param("caCertificate")
verification_certificate = self._get_param("verificationCertificate")
set_as_active = self._get_bool_param("setAsActive")
registration_config = self._get_param("registrationConfig")
cert = self.iot_backend.register_ca_certificate(
ca_certificate=ca_certificate,
verification_certificate=verification_certificate,
set_as_active=set_as_active,
registration_config=registration_config,
)
return json.dumps(
dict(certificateId=cert.certificate_id, certificateArn=cert.arn)
)
def register_certificate(self):
certificate_pem = self._get_param("certificatePem")
ca_certificate_pem = self._get_param("caCertificatePem")
@ -352,6 +407,15 @@ class IoTResponse(BaseResponse):
dict(certificateId=cert.certificate_id, certificateArn=cert.arn)
)
def update_ca_certificate(self):
certificate_id = self.path.split("/")[-1]
new_status = self._get_param("newStatus")
config = self._get_param("registrationConfig")
self.iot_backend.update_ca_certificate(
certificate_id=certificate_id, new_status=new_status, config=config
)
return json.dumps(dict())
def update_certificate(self):
certificate_id = self._get_param("certificateId")
new_status = self._get_param("newStatus")
@ -639,7 +703,6 @@ class IoTResponse(BaseResponse):
thing_name=thing_name
)
next_token = None
# TODO: implement pagination in the future
return json.dumps(dict(thingGroups=thing_groups, nextToken=next_token))
def update_thing_groups_for_thing(self):
@ -733,3 +796,8 @@ class IoTResponse(BaseResponse):
remove_authorizer_config=self._get_bool_param("removeAuthorizerConfig"),
)
return json.dumps(domain_configuration.to_dict())
def search_index(self):
query = self._get_param("queryString")
things, groups = self.iot_backend.search_index(query)
return json.dumps({"things": things, "thingGroups": groups})

4
moto/iotdata/models.py
View File

@ -142,6 +142,7 @@ class IoTDataPlaneBackend(BaseBackend):
def __init__(self, region_name=None):
super().__init__()
self.region_name = region_name
self.published_payloads = list()
def reset(self):
region_name = self.region_name
@ -199,8 +200,7 @@ class IoTDataPlaneBackend(BaseBackend):
return thing.thing_shadow
def publish(self, topic, qos, payload):
# do nothing because client won't know about the result
return None
self.published_payloads.append((topic, payload))
iotdata_backends = BackendDict(IoTDataPlaneBackend, "iot-data")

5
moto/iotdata/responses.py
View File

@ -41,8 +41,7 @@ class IoTDataPlaneResponse(BaseResponse):
return self.call_action()
def publish(self):
topic = self._get_param("topic")
topic = self._get_param("target")
qos = self._get_int_param("qos")
payload = self._get_param("payload")
self.iotdata_backend.publish(topic=topic, qos=qos, payload=payload)
self.iotdata_backend.publish(topic=topic, qos=qos, payload=self.body)
return json.dumps(dict())

2
tests/terraform-tests.success.txt
View File

@ -72,6 +72,8 @@ TestAccAWSIAMGroupPolicy
TestAccAWSIAMGroupPolicyAttachment
TestAccAWSIAMRole
TestAccAWSIAMUserPolicy
TestAccAWSIotEndpointDataSource
TestAccAWSIotThing
TestAccAWSIPRanges
TestAccAWSKinesisStream
TestAccAWSKmsAlias

176
tests/test_iot/test_iot_ca_certificates.py Normal file
View File

@ -0,0 +1,176 @@
import boto3
import pytest
from botocore.exceptions import ClientError
from moto import mock_iot
@mock_iot
def test_register_ca_certificate_simple():
client = boto3.client("iot", region_name="us-east-1")
resp = client.register_ca_certificate(
caCertificate="ca_certificate",
verificationCertificate="verification_certificate",
)
resp.should.have.key("certificateArn")
resp.should.have.key("certificateId")
@mock_iot
def test_describe_ca_certificate_unknown():
client = boto3.client("iot", region_name="us-east-2")
with pytest.raises(ClientError) as exc:
client.describe_ca_certificate(certificateId="a" * 70)
err = exc.value.response["Error"]
err["Code"].should.equal("ResourceNotFoundException")
@mock_iot
def test_describe_ca_certificate_simple():
client = boto3.client("iot", region_name="us-east-1")
resp = client.register_ca_certificate(
caCertificate="ca_certificate",
verificationCertificate="verification_certificate",
)
describe = client.describe_ca_certificate(certificateId=resp["certificateId"])
describe.should.have.key("certificateDescription")
description = describe["certificateDescription"]
description.should.have.key("certificateArn").equals(resp["certificateArn"])
description.should.have.key("certificateId").equals(resp["certificateId"])
description.should.have.key("status").equals("INACTIVE")
description.should.have.key("certificatePem").equals("ca_certificate")
@mock_iot
def test_describe_ca_certificate_advanced():
client = boto3.client("iot", region_name="us-east-1")
resp = client.register_ca_certificate(
caCertificate="ca_certificate",
verificationCertificate="verification_certificate",
setAsActive=True,
registrationConfig={
"templateBody": "template_b0dy",
"roleArn": "aws:iot:arn:role/asdfqwerwe",
},
)
describe = client.describe_ca_certificate(certificateId=resp["certificateId"])
describe.should.have.key("certificateDescription")
description = describe["certificateDescription"]
description.should.have.key("certificateArn").equals(resp["certificateArn"])
description.should.have.key("certificateId").equals(resp["certificateId"])
description.should.have.key("status").equals("ACTIVE")
description.should.have.key("certificatePem").equals("ca_certificate")
describe.should.have.key("registrationConfig")
config = describe["registrationConfig"]
config.should.have.key("templateBody").equals("template_b0dy")
config.should.have.key("roleArn").equals("aws:iot:arn:role/asdfqwerwe")
@mock_iot
def test_list_certificates_by_ca():
client = boto3.client("iot", region_name="us-east-1")
# create ca
ca_cert = client.register_ca_certificate(
caCertificate="ca_certificate",
verificationCertificate="verification_certificate",
)
ca_cert_id = ca_cert["certificateId"]
# list certificates should be empty at first
certs = client.list_certificates_by_ca(caCertificateId=ca_cert_id)
certs.should.have.key("certificates").equals([])
# create one certificate
cert1 = client.register_certificate(
certificatePem="pem" * 20, caCertificatePem="ca_certificate", setAsActive=False
)
# list certificates should return this
certs = client.list_certificates_by_ca(caCertificateId=ca_cert_id)
certs.should.have.key("certificates").length_of(1)
certs["certificates"][0]["certificateId"].should.equal(cert1["certificateId"])
# create another certificate, without ca
client.register_certificate(
certificatePem="pam" * 20,
caCertificatePem="unknown_ca_certificate",
setAsActive=False,
)
# list certificate should still only return the first certificate
certs = client.list_certificates_by_ca(caCertificateId=ca_cert_id)
certs.should.have.key("certificates").length_of(1)
@mock_iot
def test_delete_ca_certificate():
client = boto3.client("iot", region_name="us-east-1")
cert_id = client.register_ca_certificate(
caCertificate="ca_certificate",
verificationCertificate="verification_certificate",
)["certificateId"]
client.delete_ca_certificate(certificateId=cert_id)
with pytest.raises(ClientError) as exc:
client.describe_ca_certificate(certificateId=cert_id)
err = exc.value.response["Error"]
err["Code"].should.equal("ResourceNotFoundException")
@mock_iot
def test_update_ca_certificate__status():
client = boto3.client("iot", region_name="us-east-1")
cert_id = client.register_ca_certificate(
caCertificate="ca_certificate",
verificationCertificate="verification_certificate",
registrationConfig={"templateBody": "tb", "roleArn": "my:old_and_busted:arn"},
)["certificateId"]
client.update_ca_certificate(certificateId=cert_id, newStatus="DISABLE")
cert = client.describe_ca_certificate(certificateId=cert_id)
cert_desc = cert["certificateDescription"]
cert_desc.should.have.key("status").which.should.equal("DISABLE")
cert.should.have.key("registrationConfig").equals(
{"roleArn": "my:old_and_busted:arn", "templateBody": "tb"}
)
@mock_iot
def test_update_ca_certificate__config():
client = boto3.client("iot", region_name="us-east-1")
cert_id = client.register_ca_certificate(
caCertificate="ca_certificate",
verificationCertificate="verification_certificate",
)["certificateId"]
client.update_ca_certificate(
certificateId=cert_id,
registrationConfig={"templateBody": "tb", "roleArn": "my:new_and_fancy:arn"},
)
cert = client.describe_ca_certificate(certificateId=cert_id)
cert_desc = cert["certificateDescription"]
cert_desc.should.have.key("status").which.should.equal("INACTIVE")
cert.should.have.key("registrationConfig").equals(
{"roleArn": "my:new_and_fancy:arn", "templateBody": "tb"}
)
@mock_iot
def test_get_registration_code():
client = boto3.client("iot", region_name="us-west-1")
resp = client.get_registration_code()
resp.should.have.key("registrationCode")

18
tests/test_iot/test_iot_certificates.py
View File

@ -19,6 +19,22 @@ def test_certificate_id_generation_deterministic():
client.delete_certificate(certificateId=cert2["certificateId"])
@mock_iot
def test_create_certificate_from_csr():
csr = "-----BEGIN CERTIFICATE REQUEST-----\nMIICijCCAXICAQAwRTELMAkGA1UEBhMCQVUxEzARBgNVBAgTClNvbWUtU3RhdGUx\nITAfBgNVBAoTGEludGVybmV0IFdpZGdpdHMgUHR5IEx0ZDCCASIwDQYJKoZIhvcN\nAQEBBQADggEPADCCAQoCggEBAMSUg2mO7mYnhvYUB55K0/ay9WLLgPjOHnbduyCv\nN+udkJaZc+A65ux9LvVo33VHDTlV2Ms9H/42on902WtuS3BNuxdXfD068CpN2lb6\nbSAeuKc6Fdu4BIP2bFYKCyejqBoOmTEhYA8bOM1Wu/pRsq1PkAmcGkvw3mlRx45E\nB2LRicWcg3YEleEBGyLYohyeMu0pnlsc7zsu5T4bwrjetdbDPVbzgu0Mf/evU9hJ\nG/IisXNxQhzPh/DTQsKZSNddZ4bmkAQrRN1nmNXD6QoxBiVyjjgKGrPnX+hz4ugm\naaN9CsOO/cad1E3C0KiI0BQCjxRb80wOpI4utz4pEcY97sUCAwEAAaAAMA0GCSqG\nSIb3DQEBBQUAA4IBAQC64L4JHvwxdxmnXT9Lv12p5CGx99d7VOXQXy29b1yH9cJ+\nFaQ2TH377uOdorSCS4bK7eje9/HLsCNgqftR0EruwSNnukF695BWN8e/AJSZe0vA\n3J/llZ6G7MWuOIhCswsOxqNnM1htu3o6ujXVrgBMeMgQy2tfylWfI7SGR6UmtLYF\nZrPaqXdkpt47ROJNCm2Oht1B0J3QEOmbIp/2XMxrfknzwH6se/CjuliiXVPYxrtO\n5hbZcRqjhugb8FWtaLirqh3Q3+1UIJ+CW0ZczsblP7DNdqqt8YQZpWVIqR64mSXV\nAjq/cupsJST9fey8chcNSTt4nKxOGs3OgXu1ftgy\n-----END CERTIFICATE REQUEST-----\n"
client = boto3.client("iot", region_name="us-east-2")
resp = client.create_certificate_from_csr(certificateSigningRequest=csr)
resp.should.have.key("certificateArn")
resp.should.have.key("certificateId")
resp.should.have.key("certificatePem")
# Can create certificate a second time
client.create_certificate_from_csr(certificateSigningRequest=csr)
client.list_certificates().should.have.key("certificates").length_of(2)
@mock_iot
def test_create_key_and_certificate():
client = boto3.client("iot", region_name="us-east-1")
@ -145,6 +161,8 @@ def test_create_certificate_validation():
client.register_certificate_without_ca(
certificatePem=cert["certificatePem"], status="ACTIVE"
)
e.value.response.should.have.key("resourceArn").equals(cert["certificateArn"])
e.value.response.should.have.key("resourceId").equals(cert["certificateId"])
e.value.response["Error"]["Message"].should.contain(
"The certificate is already provisioned or registered"
)

49
tests/test_iot/test_iot_search.py Normal file
View File

@ -0,0 +1,49 @@
import boto3
import pytest
from moto import mock_iot
@mock_iot
@pytest.mark.parametrize(
"query_string,results",
[
["abc", {"abc", "abcefg", "uuuabc"}],
["thingName:abc", {"abc"}],
["thingName:ab*", {"abc", "abd", "abcefg"}],
["thingName:ab?", {"abc", "abd"}],
],
)
def test_search_things(query_string, results):
client = boto3.client("iot", region_name="ap-northeast-1")
for name in ["abc", "abd", "bbe", "abcefg", "uuuabc", "bbefg"]:
client.create_thing(thingName=name)
resp = client.search_index(queryString=query_string)
resp.should.have.key("thingGroups").equals([])
resp.should.have.key("things").length_of(len(results))
thing_names = [t["thingName"] for t in resp["things"]]
set(thing_names).should.equal(results)
@mock_iot
@pytest.mark.parametrize(
"query_string,results",
[["attributes.attr0:abc", {"abc"}], ["attributes.attr1:abc", set()]],
)
def test_search_attribute_specific_value(query_string, results):
client = boto3.client("iot", region_name="ap-northeast-1")
for idx, name in enumerate(["abc", "abd", "bbe", "abcefg", "uuuabc", "bbefg"]):
client.create_thing(
thingName=name, attributePayload={"attributes": {f"attr{idx}": name}}
)
resp = client.search_index(queryString=query_string)
resp.should.have.key("thingGroups").equals([])
resp.should.have.key("things").length_of(len(results))
thing_names = [t["thingName"] for t in resp["things"]]
set(thing_names).should.equal(results)

14
tests/test_iotdata/test_iotdata.py
View File

@ -3,7 +3,9 @@ import boto3
import sure # noqa # pylint: disable=unused-import
import pytest
from botocore.exceptions import ClientError
from moto import mock_iotdata, mock_iot
import moto.iotdata.models
from moto import mock_iotdata, mock_iot, settings
@mock_iot
@ -105,8 +107,14 @@ def test_update():
@mock_iotdata
def test_publish():
client = boto3.client("iot-data", region_name="ap-northeast-1")
client.publish(topic="test/topic", qos=1, payload=b"")
region_name = "ap-northeast-1"
client = boto3.client("iot-data", region_name=region_name)
client.publish(topic="test/topic", qos=1, payload=b"pl")
if not settings.TEST_SERVER_MODE:
mock_backend = moto.iotdata.models.iotdata_backends[region_name]
mock_backend.published_payloads.should.have.length_of(1)
mock_backend.published_payloads.should.contain(("test/topic", "pl"))
@mock_iot