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

Add validity dates to IoT fakecerts

Browse source
This commit is contained in:
cătălin 2021-07-23 11:49:52 +02:00
commit e94a3e39df
2 changed files with 85 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

163
moto/iot/models.py
View file

@ -12,6 +12,7 @@ from datetime import datetime
from boto3 import Session from boto3 import Session
from moto.core import BaseBackend, BaseModel from moto.core import BaseBackend, BaseModel
from moto.utilities.utils import random_string
from .exceptions import ( from .exceptions import (
CertificateStateException, CertificateStateException,
DeleteConflictException, DeleteConflictException,
@ -21,7 +22,6 @@ from .exceptions import (
VersionConflictException, VersionConflictException,
ResourceAlreadyExistsException, ResourceAlreadyExistsException,
) )
from moto.utilities.utils import random_string
class FakeThing(BaseModel): class FakeThing(BaseModel):
@ -73,12 +73,12 @@ class FakeThingType(BaseModel):
class FakeThingGroup(BaseModel): class FakeThingGroup(BaseModel):
def __init__( def __init__(
self, self,
thing_group_name, thing_group_name,
parent_group_name, parent_group_name,
thing_group_properties, thing_group_properties,
region_name, region_name,
thing_groups, thing_groups,
): ):
self.region_name = region_name self.region_name = region_name
self.thing_group_name = thing_group_name self.thing_group_name = thing_group_name
@ -144,7 +144,8 @@ class FakeCertificate(BaseModel):
self.transfer_data = {} self.transfer_data = {}
self.creation_date = time.time() self.creation_date = time.time()
self.last_modified_date = self.creation_date 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_id = None
self.ca_certificate_pem = ca_certificate_pem self.ca_certificate_pem = ca_certificate_pem
if ca_certificate_pem: if ca_certificate_pem:
@ -174,6 +175,10 @@ class FakeCertificate(BaseModel):
"ownedBy": self.owner, "ownedBy": self.owner,
"creationDate": self.creation_date, "creationDate": self.creation_date,
"lastModifiedDate": self.last_modified_date, "lastModifiedDate": self.last_modified_date,
"validity": {
"notBefore": self.validity_not_before,
"notAfter": self.validity_not_after
},
"transferData": self.transfer_data, "transferData": self.transfer_data,
} }
@ -250,17 +255,17 @@ class FakeJob(BaseModel):
JOB_ID_REGEX = re.compile(JOB_ID_REGEX_PATTERN) JOB_ID_REGEX = re.compile(JOB_ID_REGEX_PATTERN)
def __init__( def __init__(
self, self,
job_id, job_id,
targets, targets,
document_source, document_source,
document, document,
description, description,
presigned_url_config, presigned_url_config,
target_selection, target_selection,
job_executions_rollout_config, job_executions_rollout_config,
document_parameters, document_parameters,
region_name, region_name,
): ):
if not self._job_id_matcher(self.JOB_ID_REGEX, job_id): if not self._job_id_matcher(self.JOB_ID_REGEX, job_id):
raise InvalidRequestException() raise InvalidRequestException()
@ -326,12 +331,12 @@ class FakeJob(BaseModel):
class FakeJobExecution(BaseModel): class FakeJobExecution(BaseModel):
def __init__( def __init__(
self, self,
job_id, job_id,
thing_arn, thing_arn,
status="QUEUED", status="QUEUED",
force_canceled=False, force_canceled=False,
status_details_map={}, status_details_map={},
): ):
self.job_id = job_id self.job_id = job_id
self.status = status # IN_PROGRESS | CANCELED | COMPLETED self.status = status # IN_PROGRESS | CANCELED | COMPLETED
@ -429,17 +434,17 @@ class FakeEndpoint(BaseModel):
class FakeRule(BaseModel): class FakeRule(BaseModel):
def __init__( def __init__(
self, self,
rule_name, rule_name,
description, description,
created_at, created_at,
rule_disabled, rule_disabled,
topic_pattern, topic_pattern,
actions, actions,
error_action, error_action,
sql, sql,
aws_iot_sql_version, aws_iot_sql_version,
region_name, region_name,
): ):
self.region_name = region_name self.region_name = region_name
self.rule_name = rule_name self.rule_name = rule_name
@ -539,16 +544,16 @@ class IoTBackend(BaseBackend):
return self.thing_types.values() return self.thing_types.values()
def list_things( def list_things(
self, attribute_name, attribute_value, thing_type_name, max_results, token self, attribute_name, attribute_value, thing_type_name, max_results, token
): ):
all_things = [_.to_dict() for _ in self.things.values()] all_things = [_.to_dict() for _ in self.things.values()]
if attribute_name is not None and thing_type_name is not None: if attribute_name is not None and thing_type_name is not None:
filtered_things = list( filtered_things = list(
filter( filter(
lambda elem: attribute_name in elem["attributes"] lambda elem: attribute_name in elem["attributes"]
and elem["attributes"][attribute_name] == attribute_value and elem["attributes"][attribute_name] == attribute_value
and "thingTypeName" in elem and "thingTypeName" in elem
and elem["thingTypeName"] == thing_type_name, and elem["thingTypeName"] == thing_type_name,
all_things, all_things,
) )
) )
@ -556,7 +561,7 @@ class IoTBackend(BaseBackend):
filtered_things = list( filtered_things = list(
filter( filter(
lambda elem: attribute_name in elem["attributes"] lambda elem: attribute_name in elem["attributes"]
and elem["attributes"][attribute_name] == attribute_value, and elem["attributes"][attribute_name] == attribute_value,
all_things, all_things,
) )
) )
@ -564,7 +569,7 @@ class IoTBackend(BaseBackend):
filtered_things = list( filtered_things = list(
filter( filter(
lambda elem: "thingTypeName" in elem lambda elem: "thingTypeName" in elem
and elem["thingTypeName"] == thing_type_name, and elem["thingTypeName"] == thing_type_name,
all_things, all_things,
) )
) )
@ -578,7 +583,7 @@ class IoTBackend(BaseBackend):
) )
else: else:
token = int(token) token = int(token)
things = filtered_things[token : token + max_results] things = filtered_things[token: token + max_results]
next_token = ( next_token = (
str(token + max_results) str(token + max_results)
if len(filtered_things) > token + max_results if len(filtered_things) > token + max_results
@ -624,12 +629,12 @@ class IoTBackend(BaseBackend):
del self.thing_types[thing_type.arn] del self.thing_types[thing_type.arn]
def update_thing( def update_thing(
self, self,
thing_name, thing_name,
thing_type_name, thing_type_name,
attribute_payload, attribute_payload,
expected_version, expected_version,
remove_thing_type, remove_thing_type,
): ):
# if attributes payload = {}, nothing # if attributes payload = {}, nothing
thing = self.describe_thing(thing_name) thing = self.describe_thing(thing_name)
@ -730,7 +735,7 @@ class IoTBackend(BaseBackend):
) )
def register_certificate( def register_certificate(
self, certificate_pem, ca_certificate_pem, set_as_active, status self, certificate_pem, ca_certificate_pem, set_as_active, status
): ):
certificate = FakeCertificate( certificate = FakeCertificate(
certificate_pem, certificate_pem,
@ -954,7 +959,7 @@ class IoTBackend(BaseBackend):
return thing_groups[0] return thing_groups[0]
def create_thing_group( def create_thing_group(
self, thing_group_name, parent_group_name, thing_group_properties self, thing_group_name, parent_group_name, thing_group_properties
): ):
thing_group = FakeThingGroup( thing_group = FakeThingGroup(
thing_group_name, thing_group_name,
@ -1012,7 +1017,7 @@ class IoTBackend(BaseBackend):
] ]
def update_thing_group( def update_thing_group(
self, thing_group_name, thing_group_properties, expected_version self, thing_group_name, thing_group_properties, expected_version
): ):
thing_group = self.describe_thing_group(thing_group_name) thing_group = self.describe_thing_group(thing_group_name)
if expected_version and expected_version != thing_group.version: if expected_version and expected_version != thing_group.version:
@ -1071,7 +1076,7 @@ class IoTBackend(BaseBackend):
return thing return thing
def add_thing_to_thing_group( def add_thing_to_thing_group(
self, thing_group_name, thing_group_arn, thing_name, thing_arn self, thing_group_name, thing_group_arn, thing_name, thing_arn
): ):
thing_group = self._identify_thing_group(thing_group_name, thing_group_arn) thing_group = self._identify_thing_group(thing_group_name, thing_group_arn)
thing = self._identify_thing(thing_name, thing_arn) thing = self._identify_thing(thing_name, thing_arn)
@ -1081,7 +1086,7 @@ class IoTBackend(BaseBackend):
thing_group.things[thing.arn] = thing thing_group.things[thing.arn] = thing
def remove_thing_from_thing_group( def remove_thing_from_thing_group(
self, thing_group_name, thing_group_arn, thing_name, thing_arn self, thing_group_name, thing_group_arn, thing_name, thing_arn
): ):
thing_group = self._identify_thing_group(thing_group_name, thing_group_arn) thing_group = self._identify_thing_group(thing_group_name, thing_group_arn)
thing = self._identify_thing(thing_name, thing_arn) thing = self._identify_thing(thing_name, thing_arn)
@ -1109,7 +1114,7 @@ class IoTBackend(BaseBackend):
return ret return ret
def update_thing_groups_for_thing( def update_thing_groups_for_thing(
self, thing_name, thing_groups_to_add, thing_groups_to_remove self, thing_name, thing_groups_to_add, thing_groups_to_remove
): ):
thing = self.describe_thing(thing_name) thing = self.describe_thing(thing_name)
for thing_group_name in thing_groups_to_add: for thing_group_name in thing_groups_to_add:
@ -1124,16 +1129,16 @@ class IoTBackend(BaseBackend):
) )
def create_job( def create_job(
self, self,
job_id, job_id,
targets, targets,
document_source, document_source,
document, document,
description, description,
presigned_url_config, presigned_url_config,
target_selection, target_selection,
job_executions_rollout_config, job_executions_rollout_config,
document_parameters, document_parameters,
): ):
job = FakeJob( job = FakeJob(
job_id, job_id,
@ -1192,13 +1197,13 @@ class IoTBackend(BaseBackend):
return self.jobs[job_id] return self.jobs[job_id]
def list_jobs( def list_jobs(
self, self,
status, status,
target_selection, target_selection,
max_results, max_results,
token, token,
thing_group_name, thing_group_name,
thing_group_id, thing_group_id,
): ):
# TODO: implement filters # TODO: implement filters
all_jobs = [_.to_dict() for _ in self.jobs.values()] all_jobs = [_.to_dict() for _ in self.jobs.values()]
@ -1209,7 +1214,7 @@ class IoTBackend(BaseBackend):
next_token = str(max_results) if len(filtered_jobs) > max_results else None next_token = str(max_results) if len(filtered_jobs) > max_results else None
else: else:
token = int(token) token = int(token)
jobs = filtered_jobs[token : token + max_results] jobs = filtered_jobs[token: token + max_results]
next_token = ( next_token = (
str(token + max_results) str(token + max_results)
if len(filtered_jobs) > token + max_results if len(filtered_jobs) > token + max_results
@ -1225,15 +1230,15 @@ class IoTBackend(BaseBackend):
raise ResourceNotFoundException() raise ResourceNotFoundException()
if job_execution is None or ( if job_execution is None or (
execution_number is not None execution_number is not None
and job_execution.execution_number != execution_number and job_execution.execution_number != execution_number
): ):
raise ResourceNotFoundException() raise ResourceNotFoundException()
return job_execution return job_execution
def cancel_job_execution( def cancel_job_execution(
self, job_id, thing_name, force, expected_version, status_details self, job_id, thing_name, force, expected_version, status_details
): ):
job_execution = self.job_executions[(job_id, thing_name)] job_execution = self.job_executions[(job_id, thing_name)]
@ -1288,7 +1293,7 @@ class IoTBackend(BaseBackend):
next_token = str(max_results) if len(job_executions) > max_results else None next_token = str(max_results) if len(job_executions) > max_results else None
else: else:
token = int(token) token = int(token)
job_executions = job_executions[token : token + max_results] job_executions = job_executions[token: token + max_results]
next_token = ( next_token = (
str(token + max_results) str(token + max_results)
if len(job_executions) > token + max_results if len(job_executions) > token + max_results
@ -1298,7 +1303,7 @@ class IoTBackend(BaseBackend):
return job_executions, next_token return job_executions, next_token
def list_job_executions_for_thing( def list_job_executions_for_thing(
self, thing_name, status, max_results, next_token self, thing_name, status, max_results, next_token
): ):
job_executions = [ job_executions = [
self.job_executions[je].to_dict() self.job_executions[je].to_dict()
@ -1320,7 +1325,7 @@ class IoTBackend(BaseBackend):
next_token = str(max_results) if len(job_executions) > max_results else None next_token = str(max_results) if len(job_executions) > max_results else None
else: else:
token = int(token) token = int(token)
job_executions = job_executions[token : token + max_results] job_executions = job_executions[token: token + max_results]
next_token = ( next_token = (
str(token + max_results) str(token + max_results)
if len(job_executions) > token + max_results if len(job_executions) > token + max_results

2
moto/iot/responses.py
View file

@ -49,7 +49,7 @@ class IoTResponse(BaseResponse):
next_token = str(max_results) if len(thing_types) > max_results else None next_token = str(max_results) if len(thing_types) > max_results else None
else: else:
token = int(previous_next_token) token = int(previous_next_token)
result = thing_types[token : token + max_results] result = thing_types[token: token + max_results]
next_token = ( next_token = (
str(token + max_results) str(token + max_results)
if len(thing_types) > token + max_results if len(thing_types) > token + max_results