From 1016487c78b32fa0a67d093c3a43f39cd68aa231 Mon Sep 17 00:00:00 2001 From: Mike Liu Date: Wed, 25 Apr 2018 16:25:44 -0400 Subject: [PATCH] Implement the ListCommands API endpoint for the SSM client. Currently only supports getting commands by CommandId and InstanceIds. --- moto/ssm/models.py | 32 ++++++++++++++++++++++++++++++++ moto/ssm/responses.py | 5 +++++ 2 files changed, 37 insertions(+) diff --git a/moto/ssm/models.py b/moto/ssm/models.py index 4f1bca213..688cffcd5 100644 --- a/moto/ssm/models.py +++ b/moto/ssm/models.py @@ -3,6 +3,7 @@ from __future__ import unicode_literals from collections import defaultdict from moto.core import BaseBackend, BaseModel +from moto.core.exceptions import RESTError from moto.ec2 import ec2_backends import datetime @@ -229,6 +230,37 @@ class SimpleSystemManagerBackend(BaseBackend): 'Command': command.response_object() } + def list_commands(self, **kwargs): + """ + https://docs.aws.amazon.com/systems-manager/latest/APIReference/API_ListCommands.html + """ + commands = self._commands + + command_id = kwargs.get('CommandId', None) + if command_id: + commands = [self.get_command_by_id(command_id)] + instance_id = kwargs.get('InstanceId', None) + if instance_id: + commands = self.get_commands_by_instance_id(instance_id) + + return { + 'Commands': [command.response_object() for command in commands] + } + + def get_command_by_id(self, id): + command = next( + (command for command in self._commands if command.command_id == id), None) + + if command is None: + raise RESTError('InvalidCommandId', 'Invalid command id.') + + return command + + def get_commands_by_instance_id(self, instance_id): + return [ + command for command in self._commands + if instance_id in command.instance_ids] + ssm_backends = {} for region, ec2_backend in ec2_backends.items(): diff --git a/moto/ssm/responses.py b/moto/ssm/responses.py index d9906a82e..e86fe05e3 100644 --- a/moto/ssm/responses.py +++ b/moto/ssm/responses.py @@ -204,3 +204,8 @@ class SimpleSystemManagerResponse(BaseResponse): return json.dumps( self.ssm_backend.send_command(**self.request_params) ) + + def list_commands(self): + return json.dumps( + self.ssm_backend.list_commands(**self.request_params) + )