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

Add support for IoT register_certificate (#1827)

Browse source
This commit is contained in:
Adam Hodges 2019-05-25 13:30:23 -04:00 • committed by Terry Cain
commit 71f83d7e4f
4 changed files with 56 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

14
moto/iot/models.py
View file

@ -96,7 +96,7 @@ class FakeThingGroup(BaseModel):
class FakeCertificate(BaseModel):
def __init__(self, certificate_pem, status, region_name):
def __init__(self, certificate_pem, status, region_name, ca_certificate_pem=None):
m = hashlib.sha256()
m.update(str(uuid.uuid4()).encode('utf-8'))
self.certificate_id = m.hexdigest()
@ -109,12 +109,18 @@ class FakeCertificate(BaseModel):
self.transfer_data = {}
self.creation_date = time.time()
self.last_modified_date = self.creation_date
self.ca_certificate_id = None
self.ca_certificate_pem = ca_certificate_pem
if ca_certificate_pem:
m.update(str(uuid.uuid4()).encode('utf-8'))
self.ca_certificate_id = m.hexdigest()
def to_dict(self):
return {
'certificateArn': self.arn,
'certificateId': self.certificate_id,
'caCertificateId': self.ca_certificate_id,
'status': self.status,
'creationDate': self.creation_date
}
@ -410,6 +416,12 @@ class IoTBackend(BaseBackend):
def list_certificates(self):
return self.certificates.values()
def register_certificate(self, certificate_pem, ca_certificate_pem, set_as_active, status):
certificate = FakeCertificate(certificate_pem, 'ACTIVE' if set_as_active else status,
self.region_name, ca_certificate_pem)
self.certificates[certificate.certificate_id] = certificate
return certificate
def update_certificate(self, certificate_id, new_status):
cert = self.describe_certificate(certificate_id)
# TODO: validate new_status

14
moto/iot/responses.py
View file

@ -183,6 +183,20 @@ class IoTResponse(BaseResponse):
# TODO: implement pagination in the future
return json.dumps(dict(certificates=[_.to_dict() for _ in certificates]))
def register_certificate(self):
certificate_pem = self._get_param("certificatePem")
ca_certificate_pem = self._get_param("caCertificatePem")
set_as_active = self._get_bool_param("setAsActive")
status = self._get_param("status")
cert = self.iot_backend.register_certificate(
certificate_pem=certificate_pem,
ca_certificate_pem=ca_certificate_pem,
set_as_active=set_as_active,
status=status
)
return json.dumps(dict(certificateId=cert.certificate_id, certificateArn=cert.arn))
def update_certificate(self):
certificate_id = self._get_param("certificateId")
new_status = self._get_param("newStatus")