Add get_command_invocation endpoint for AWS SSM.

Users can make send_command requests and then retrieve the invocations
of those commands with get_command_invocation. Getting a command
invocation by instance and command id is supported but only the
'aws:runShellScript' plugin name is supported and only one plugin in a
document is supported.

moto/ssm/models.py

@@ -88,9 +88,9 @@ class Command(BaseModel):
         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.requested_date_time = datetime.datetime.now()
+        self.requested_date_time_iso = self.requested_date_time.isoformat()
+        expires_after = self.requested_date_time + datetime.timedelta(0, timeout_seconds)
         self.expires_after = expires_after.isoformat()
 
         self.comment = comment
@@ -106,6 +106,12 @@ class Command(BaseModel):
         self.service_role_arn = service_role_arn
         self.targets = targets
 
+        # Create invocations with a single run command plugin.
+        self.invocations = []
+        for instance_id in self.instance_ids:
+            self.invocations.append(
+                self.invocation_response(instance_id, "aws:runShellScript"))
+
     def response_object(self):
         r = {
             'CommandId': self.command_id,
@@ -122,7 +128,7 @@ class Command(BaseModel):
             'OutputS3BucketName': self.output_s3_bucket_name,
             'OutputS3KeyPrefix': self.output_s3_key_prefix,
             'Parameters': self.parameters,
-            'RequestedDateTime': self.requested_date_time,
+            'RequestedDateTime': self.requested_date_time_iso,
             'ServiceRole': self.service_role_arn,
             'Status': self.status,
             'StatusDetails': self.status_details,
@@ -132,6 +138,51 @@ class Command(BaseModel):
 
         return r
 
+    def invocation_response(self, instance_id, plugin_name):
+        # Calculate elapsed time from requested time and now. Use a hardcoded
+        # elapsed time since there is no easy way to convert a timedelta to
+        # an ISO 8601 duration string.
+        elapsed_time_iso = "PT5M"
+        elapsed_time_delta = datetime.timedelta(minutes=5)
+        end_time = self.requested_date_time + elapsed_time_delta
+
+        r = {
+            'CommandId': self.command_id,
+            'InstanceId': instance_id,
+            'Comment': self.comment,
+            'DocumentName': self.document_name,
+            'PluginName': plugin_name,
+            'ResponseCode': 0,
+            'ExecutionStartDateTime': self.requested_date_time_iso,
+            'ExecutionElapsedTime': elapsed_time_iso,
+            'ExecutionEndDateTime': end_time.isoformat(),
+            'Status': 'Success',
+            'StatusDetails': 'Success',
+            'StandardOutputContent': '',
+            'StandardOutputUrl': '',
+            'StandardErrorContent': '',
+        }
+
+        return r
+
+    def get_invocation(self, instance_id, plugin_name):
+        invocation = next(
+            (invocation for invocation in self.invocations
+                if invocation['InstanceId'] == instance_id), None)
+
+        if invocation is None:
+            raise RESTError(
+                'InvocationDoesNotExist',
+                'An error occurred (InvocationDoesNotExist) when calling the GetCommandInvocation operation')
+
+        if plugin_name is not None and invocation['PluginName'] != plugin_name:
+                raise RESTError(
+                    'InvocationDoesNotExist',
+                    'An error occurred (InvocationDoesNotExist) when calling the GetCommandInvocation operation')
+
+
+        return invocation
+
 
 class SimpleSystemManagerBackend(BaseBackend):
 
@@ -298,6 +349,18 @@ class SimpleSystemManagerBackend(BaseBackend):
             command for command in self._commands
             if instance_id in command.instance_ids]
 
+    def get_command_invocation(self, **kwargs):
+        """
+        https://docs.aws.amazon.com/systems-manager/latest/APIReference/API_GetCommandInvocation.html
+        """
+
+        command_id = kwargs.get('CommandId')
+        instance_id = kwargs.get('InstanceId')
+        plugin_name = kwargs.get('PluginName', None)
+
+        command = self.get_command_by_id(command_id)
+        return command.get_invocation(instance_id, plugin_name)
+
 
 ssm_backends = {}
 for region, ec2_backend in ec2_backends.items():

moto/ssm/responses.py

@@ -210,3 +210,8 @@ class SimpleSystemManagerResponse(BaseResponse):
         return json.dumps(
             self.ssm_backend.list_commands(**self.request_params)
         )
+
+    def get_command_invocation(self):
+        return json.dumps(
+            self.ssm_backend.get_command_invocation(**self.request_params)
+        )

tests/test_ssm/test_ssm_boto3.py

@@ -668,3 +668,43 @@ def test_list_commands():
     with assert_raises(ClientError):
         response = client.list_commands(
             CommandId=str(uuid.uuid4()))
+
+@mock_ssm
+def test_get_command_invocation():
+    client = boto3.client('ssm', region_name='us-east-1')
+
+    ssm_document = 'AWS-RunShellScript'
+    params = {'commands': ['#!/bin/bash\necho \'hello world\'']}
+
+    response = client.send_command(
+        InstanceIds=['i-123456', 'i-234567', 'i-345678'],
+        DocumentName=ssm_document,
+        Parameters=params,
+        OutputS3Region='us-east-2',
+        OutputS3BucketName='the-bucket',
+        OutputS3KeyPrefix='pref')
+
+    cmd = response['Command']
+    cmd_id = cmd['CommandId']
+
+    instance_id = 'i-345678'
+    invocation_response = client.get_command_invocation(
+        CommandId=cmd_id,
+        InstanceId=instance_id,
+        PluginName='aws:runShellScript')
+
+    invocation_response['CommandId'].should.equal(cmd_id)
+    invocation_response['InstanceId'].should.equal(instance_id)
+
+    # test the error case for an invalid instance id
+    with assert_raises(ClientError):
+        invocation_response = client.get_command_invocation(
+            CommandId=cmd_id,
+            InstanceId='i-FAKE')
+
+    # test the error case for an invalid plugin name
+    with assert_raises(ClientError):
+        invocation_response = client.get_command_invocation(
+            CommandId=cmd_id,
+            InstanceId=instance_id,
+            PluginName='FAKE')