test_commands.py

#!/usr/bin/env python
# Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
#     http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.

import copy
import unittest

from awscli.testutils import mock, BaseAWSHelpOutputTest, BaseAWSCommandParamsTest

from awscli.customizations.datapipeline import convert_described_objects
from awscli.customizations.datapipeline import ListRunsCommand


API_DESCRIBE_OBJECTS = [
    {"fields": [
        {
            "key": "@componentParent",
            "refValue": "S3Input"
        },
        {
            "key": "@scheduledStartTime",
            "stringValue": "2013-08-19T20:00:00"
        },
        {
            "key": "parent",
            "refValue": "S3Input"
        },
        {
            "key": "@sphere",
            "stringValue": "INSTANCE"
        },
        {
            "key": "type",
            "stringValue": "S3DataNode"
        },
        {
            "key": "@version",
            "stringValue": "1"
        },
        {
            "key": "@status",
            "stringValue": "FINISHED"
        },
        {
            "key": "@actualEndTime",
            "stringValue": "2014-02-19T19:44:44"
        },
        {
            "key": "@actualStartTime",
            "stringValue": "2014-02-19T19:44:43"
        },
        {
            "key": "output",
            "refValue": "@MyCopyActivity_2013-08-19T20:00:00"
        },
        {
            "key": "@scheduledEndTime",
            "stringValue": "2013-08-19T21:00:00"
        }
    ],
     "id": "@S3Input_2013-08-19T20:00:00",
     "name": "@S3Input_2013-08-19T20:00:00"
    },
    {"fields": [
        {
            "key": "@componentParent",
            "refValue": "MyEC2Resource"
        },
        {
            "key": "@resourceId",
            "stringValue": "i-12345"
        },
        {
            "key": "@scheduledStartTime",
            "stringValue": "2013-08-19T23:00:00"
        },
        {
            "key": "parent",
            "refValue": "MyEC2Resource"
        },
        {
            "key": "@sphere",
            "stringValue": "INSTANCE"
        },
        {
            "key": "@attemptCount",
            "stringValue": "1"
        },
        {
            "key": "type",
            "stringValue": "Ec2Resource"
        },
        {
            "key": "@version",
            "stringValue": "1"
        },
        {
            "key": "@status",
            "stringValue": "CREATING"
        },
        {
            "key": "input",
            "refValue": "@MyCopyActivity_2013-08-19T23:00:00"
        },
        {
            "key": "@triesLeft",
            "stringValue": "2"
        },
        {
            "key": "@actualStartTime",
            "stringValue": "2014-02-19T19:59:45"
        },
        {
            "key": "@headAttempt",
            "refValue": "@MyEC2Resource_2013-08-19T23:00:00_Attempt=1"
        },
        {
            "key": "@scheduledEndTime",
            "stringValue": "2013-08-20T00:00:00"
        }
    ],
     "id": "@MyEC2Resource_2013-08-19T23:00:00",
     "name": "@MyEC2Resource_2013-08-19T23:00:00"
    }
]

JSON_FORMATTER_PATH = 'awscli.formatter.JSONFormatter.__call__'
LIST_FORMATTER_PATH = 'awscli.customizations.datapipeline.listrunsformatter.ListRunsFormatter.__call__'  # noqa


class TestConvertObjects(unittest.TestCase):

    def test_convert_described_objects(self):
        converted = convert_described_objects(API_DESCRIBE_OBJECTS)
        self.assertEqual(len(converted), 2)
        # This comes from a "refValue" value.
        self.assertEqual(converted[0]['@componentParent'], 'S3Input')
        # Should also merge in @id and name.
        self.assertEqual(converted[0]['@id'], "@S3Input_2013-08-19T20:00:00")
        self.assertEqual(converted[0]['name'], "@S3Input_2013-08-19T20:00:00")
        # This comes from a "stringValue" value.
        self.assertEqual(converted[0]['@sphere'], "INSTANCE")

    def test_convert_objects_are_sorted(self):
        describe_objects = copy.deepcopy(API_DESCRIBE_OBJECTS)
        # Change the existing @scheduledStartTime from
        # 20:00:00 to 23:59:00
        describe_objects[0]['fields'][1]['stringValue'] = (
            "2013-08-19T23:59:00")
        converted = convert_described_objects(
            describe_objects,
            sort_key_func=lambda x: (x['@scheduledStartTime'], x['name']))
        self.assertEqual(converted[0]['@scheduledStartTime'],
                         '2013-08-19T23:00:00')
        self.assertEqual(converted[1]['@scheduledStartTime'],
                         '2013-08-19T23:59:00')


class FakeParsedArgs(object):
    def __init__(self, **kwargs):
        self.endpoint_url = None
        self.region = None
        self.verify_ssl = None
        self.output = None
        self.query = None
        self.__dict__.update(kwargs)


class TestCommandsRunProperly(BaseAWSCommandParamsTest):
    def setUp(self):
        super(TestCommandsRunProperly, self).setUp()
        self.query_objects = mock.Mock()
        self.describe_objects = mock.Mock()
        self.client = mock.Mock()
        self.client.get_paginator.return_value = self.query_objects
        self.client.describe_objects = self.describe_objects

        self.driver.session = mock.Mock()
        self.driver.session.emit_first_non_none_response.return_value = None
        self.driver.session.create_client.return_value = self.client
        self.query_objects.paginate.return_value.build_full_result.\
            return_value = {'ids': ['object-ids']}
        self.describe_objects.return_value = \
            {'pipelineObjects': API_DESCRIBE_OBJECTS}
        self.expected_response = convert_described_objects(
            API_DESCRIBE_OBJECTS,
            sort_key_func=lambda x: (x['@scheduledStartTime'], x['name']))

    def test_list_runs(self):
        command = ListRunsCommand(self.driver.session)
        command(['--pipeline-id', 'my-pipeline-id'],
                parsed_globals=FakeParsedArgs(region='us-east-1'))
        self.assertTrue(self.query_objects.paginate.called)
        self.describe_objects.assert_called_with(
            pipelineId='my-pipeline-id', objectIds=['object-ids'])

    @mock.patch(JSON_FORMATTER_PATH)
    @mock.patch(LIST_FORMATTER_PATH)
    def test_list_runs_formatter_explicit_choice(
            self, list_formatter, json_formatter):
        command = ListRunsCommand(self.driver.session)
        command(['--pipeline-id', 'my-pipeline-id'],
                parsed_globals=FakeParsedArgs(
                    region='us-east-1', output='json'))
        json_formatter.assert_called_once_with(
            'list-runs', self.expected_response)
        self.assertFalse(list_formatter.called)

    @mock.patch(JSON_FORMATTER_PATH)
    @mock.patch(LIST_FORMATTER_PATH)
    def test_list_runs_formatter_implicit_choice(
            self, list_formatter, json_formatter):

        command = ListRunsCommand(self.driver.session)
        command(['--pipeline-id', 'my-pipeline-id'],
                parsed_globals=FakeParsedArgs(region='us-east-1'))
        list_formatter.assert_called_once_with(
            'list-runs', self.expected_response)
        self.assertFalse(json_formatter.called)


class TestHelpOutput(BaseAWSHelpOutputTest):
    def test_list_runs_help_output(self):
        self.driver.main(['datapipeline', 'get-pipeline-definition', 'help'])
        self.assert_contains('pipeline definition')
        # The previous API docs should not be in the output
        self.assert_not_contains('pipelineObjects')