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

Create a Command class for the ssm backend.

Browse Source 
This class will make it easier to keep track of commands in a list for
the SSM backend later on when we implement the ListCommands API call.
This commit is contained in:
Mike Liu 2018-04-25 16:22:10 -04:00
parent 21a264c337
commit a974a3dfe4
1 changed files with 84 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
moto/ssm/models.py
View File

@ -58,11 +58,74 @@ class Parameter(BaseModel):
return r
MAX_TIMEOUT_SECONDS = 3600
class Command(BaseModel):
def __init__(self, comment='', document_name='', timeout_seconds=MAX_TIMEOUT_SECONDS,
instance_ids=[], max_concurrency='', max_errors='',
notification_config={}, output_s3_bucket_name='',
output_s3_key_prefix='', output_s3_region='', parameters={},
service_role_arn='', targets=[]):
self.error_count = 0
self.completed_count = len(instance_ids)
self.target_count = len(instance_ids)
self.command_id = str(uuid.uuid4())
self.status = 'Success'
self.status_details = 'Details placeholder'
now = datetime.datetime.now()
self.requested_date_time = now.isoformat()
expires_after = now + datetime.timedelta(0, timeout_seconds)
self.expires_after = expires_after.isoformat()
self.comment = comment
self.document_name = document_name
self.instance_ids = instance_ids
self.max_concurrency = max_concurrency
self.max_errors = max_errors
self.notification_config = notification_config
self.output_s3_bucket_name = output_s3_bucket_name
self.output_s3_key_prefix = output_s3_key_prefix
self.output_s3_region = output_s3_region
self.parameters = parameters
self.service_role_arn = service_role_arn
self.targets = targets
def response_object(self):
r = {
'CommandId': self.command_id,
'Comment': self.comment,
'CompletedCount': self.completed_count,
'DocumentName': self.document_name,
'ErrorCount': self.error_count,
'ExpiresAfter': self.expires_after,
'InstanceIds': self.instance_ids,
'MaxConcurrency': self.max_concurrency,
'MaxErrors': self.max_errors,
'NotificationConfig': self.notification_config,
'OutputS3Region': self.output_s3_region,
'OutputS3BucketName': self.output_s3_bucket_name,
'OutputS3KeyPrefix': self.output_s3_key_prefix,
'Parameters': self.parameters,
'RequestedDateTime': self.requested_date_time,
'ServiceRole': self.service_role_arn,
'Status': self.status,
'StatusDetails': self.status_details,
'TargetCount': self.target_count,
'Targets': self.targets,
}
return r
class SimpleSystemManagerBackend(BaseBackend):
def __init__(self):
self._parameters = {}
self._resource_tags = defaultdict(lambda: defaultdict(dict))
self._commands = []
def delete_parameter(self, name):
try:
@ -142,36 +205,28 @@ class SimpleSystemManagerBackend(BaseBackend):
return self._resource_tags[resource_type][resource_id]
def send_command(self, **kwargs):
instances = kwargs.get('InstanceIds', [])
now = datetime.datetime.now()
expires_after = now + datetime.timedelta(0, int(kwargs.get('TimeoutSeconds', 3600)))
return {
'Command': {
'CommandId': str(uuid.uuid4()),
'DocumentName': kwargs['DocumentName'],
'Comment': kwargs.get('Comment'),
'ExpiresAfter': expires_after.isoformat(),
'Parameters': kwargs['Parameters'],
'InstanceIds': kwargs['InstanceIds'],
'Targets': kwargs.get('targets'),
'RequestedDateTime': now.isoformat(),
'Status': 'Success',
'StatusDetails': 'string',
'OutputS3Region': kwargs.get('OutputS3Region'),
'OutputS3BucketName': kwargs.get('OutputS3BucketName'),
'OutputS3KeyPrefix': kwargs.get('OutputS3KeyPrefix'),
'MaxConcurrency': 'string',
'MaxErrors': 'string',
'TargetCount': len(instances),
'CompletedCount': len(instances),
'ErrorCount': 0,
'ServiceRole': kwargs.get('ServiceRoleArn'),
'NotificationConfig': {
command = Command(
comment=kwargs.get('Comment', ''),
document_name=kwargs.get('DocumentName'),
timeout_seconds=kwargs.get('TimeoutSeconds', 3600),
instance_ids=kwargs.get('InstanceIds', []),
max_concurrency=kwargs.get('MaxConcurrency', '50'),
max_errors=kwargs.get('MaxErrors', '0'),
notification_config=kwargs.get('NotificationConfig', {
'NotificationArn': 'string',
'NotificationEvents': ['Success'],
'NotificationType': 'Command'
}
}
}),
output_s3_bucket_name=kwargs.get('OutputS3BucketName', ''),
output_s3_key_prefix=kwargs.get('OutputS3KeyPrefix', ''),
output_s3_region=kwargs.get('OutputS3Region', ''),
parameters=kwargs.get('Parameters', {}),
service_role_arn=kwargs.get('ServiceRoleArn', ''),
targets=kwargs.get('Targets', []))
self._commands.append(command)
return {
'Command': command.response_object()
}