test_cli.py

# Copyright 2013 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 time
import signal
import os
import tempfile
import shutil

import botocore.session
from awscli.testutils import unittest, aws, BaseS3CLICommand
from awscli.testutils import temporary_file
from awscli.testutils import skip_if_windows
from awscli.testutils import random_bucket_name
from awscli.clidriver import create_clidriver


class TestBasicCommandFunctionality(unittest.TestCase):
    """
    These are a set of tests that assert high level features of
    the CLI.  They don't test anything exhaustively and they're meant as a smoke
    test to verify basic CLI functionality isn't entirely broken.
    """

    def put_object(self, bucket, key, content, extra_args=None):
        session = botocore.session.get_session()
        client = session.create_client('s3', 'us-east-1')
        client.create_bucket(Bucket=bucket, ObjectOwnership='ObjectWriter')
        time.sleep(5)
        client.delete_public_access_block(Bucket=bucket)
        self.addCleanup(client.delete_bucket, Bucket=bucket)
        call_args = {
            'Bucket': bucket,
            'Key': key, 'Body': content
        }
        if extra_args is not None:
            call_args.update(extra_args)
        client.put_object(**call_args)
        self.addCleanup(client.delete_object, Bucket=bucket, Key=key)

    def test_ec2_describe_instances(self):
        # Verify we can make a call and get output.
        p = aws('ec2 describe-instances')
        self.assertEqual(p.rc, 0)
        # We don't know what instances a user might have, but we know
        # there should at least be a Reservations key.
        self.assertIn('Reservations', p.json)

    def test_help_output(self):
        p = aws('help')
        self.assertEqual(p.rc, 0)
        self.assertIn('AWS', p.stdout)
        self.assertRegex(
            p.stdout, r'The\s+AWS\s+Command\s+Line\s+Interface')

    def test_service_help_output(self):
        p = aws('ec2 help')
        self.assertEqual(p.rc, 0)
        self.assertRegex(p.stdout, r'Amazon\s+EC2')

    def test_operation_help_output(self):
        p = aws('ec2 describe-instances help')
        self.assertEqual(p.rc, 0)
        # XXX: This is a rendering bug that needs to be fixed in bcdoc.  In
        # the RST version there are multiple spaces between certain words.
        # For now we're making the test less strict about formatting, but
        # we eventually should update this test to check exactly for
        # 'The describe-instances operation'.
        self.assertRegex(p.stdout, r'\s+Describes\s+the\s+specified\s+instances')

    def test_topic_list_help_output(self):
        p = aws('help topics')
        self.assertEqual(p.rc, 0)
        self.assertRegex(p.stdout, r'\s+AWS\s+CLI\s+Topic\s+Guide')
        self.assertRegex(
            p.stdout,
            r'\s+This\s+is\s+the\s+AWS\s+CLI\s+Topic\s+Guide'
        )

    def test_topic_help_output(self):
        p = aws('help return-codes')
        self.assertEqual(p.rc, 0)
        self.assertRegex(p.stdout, r'\s+AWS\s+CLI\s+Return\s+Codes')
        self.assertRegex(
            p.stdout,
            r'These\s+are\s+the\s+following\s+return\s+codes'
        )

    def test_operation_help_with_required_arg(self):
        p = aws('s3api get-object help')
        self.assertEqual(p.rc, 0, p.stderr)
        self.assertIn('get-object', p.stdout)

    def test_service_help_with_required_option(self):
        # In cloudsearchdomain, the --endpoint-url is required.
        # We want to make sure if you're just getting help tex
        # that we don't trigger that validation.
        p = aws('cloudsearchdomain help')
        self.assertEqual(p.rc, 0, p.stderr)
        self.assertIn('cloudsearchdomain', p.stdout)
        # And nothing on stderr about missing options.
        self.assertEqual(p.stderr, '')

    def test_operation_help_with_required_option(self):
        p = aws('cloudsearchdomain search help')
        self.assertEqual(p.rc, 0, p.stderr)
        self.assertIn('search', p.stdout)
        # And nothing on stderr about missing options.
        self.assertEqual(p.stderr, '')

    def test_help_with_warning_blocks(self):
        p = aws('bedrock-runtime invoke-model help')
        self.assertEqual(p.rc, 0, p.stderr)
        # Check text that appears in the warning block to ensure
        # the block was actually rendered.
        self.assertRegex(p.stdout, r"To\s+deny\s+all\s+inference\s+access")

    def test_param_shorthand(self):
        p = aws(
            'ec2 describe-instances --filters Name=instance-id,Values=i-123')
        self.assertEqual(p.rc, 0)
        self.assertIn('Reservations', p.json)

    def test_param_json(self):
        p = aws(
            'ec2 describe-instances --filters '
            '\'{"Name": "instance-id", "Values": ["i-123"]}\'')
        self.assertEqual(p.rc, 0, p.stdout + p.stderr)
        self.assertIn('Reservations', p.json)

    def test_param_with_bad_json(self):
        p = aws(
            'ec2 describe-instances --filters '
            '\'{"Name": "bad-filter", "Values": ["i-123"]}\'')
        self.assertEqual(p.rc, 255)
        self.assertIn("The filter 'bad-filter' is invalid", p.stderr,
                      "stdout: %s, stderr: %s" % (p.stdout, p.stderr))

    def test_param_with_file(self):
        d = tempfile.mkdtemp()
        self.addCleanup(os.rmdir, d)
        param_file = os.path.abspath(os.path.join(d, 'params.json'))
        with open(param_file, 'w') as f:
            f.write('[{"Name": "instance-id", "Values": ["i-123"]}]')
        self.addCleanup(os.remove, param_file)
        p = aws('ec2 describe-instances --filters file://%s' % param_file)
        self.assertEqual(p.rc, 0)
        self.assertIn('Reservations', p.json)

    def test_streaming_output_operation(self):
        d = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, d)
        bucket_name = random_bucket_name('clistream')

        self.put_object(bucket=bucket_name, key='foobar',
                        content='foobar contents')
        p = aws('s3api get-object --bucket %s --key foobar %s' % (
            bucket_name, os.path.join(d, 'foobar')))
        self.assertEqual(p.rc, 0)
        with open(os.path.join(d, 'foobar')) as f:
            contents = f.read()
        self.assertEqual(contents, 'foobar contents')

    def test_no_sign_request(self):
        d = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, d)

        env_vars = os.environ.copy()
        env_vars['AWS_ACCESS_KEY_ID'] = 'foo'
        env_vars['AWS_SECRET_ACCESS_KEY'] = 'bar'

        bucket_name = random_bucket_name('nosign')
        self.put_object(bucket_name, 'foo', content='bar',
                        extra_args={'ACL': 'public-read-write'})

        p = aws('s3api get-object --bucket %s --key foo %s' % (
            bucket_name, os.path.join(d, 'foo')), env_vars=env_vars)
        # Should have credential issues.
        self.assertEqual(p.rc, 255)

        p = aws('s3api get-object --bucket %s --key foo '
                '%s --no-sign-request' % (bucket_name, os.path.join(d, 'foo')),
                env_vars=env_vars)

        # Should be able to download the file when not signing.
        self.assertEqual(p.rc, 0)

        with open(os.path.join(d, 'foo')) as f:
            contents = f.read()
        self.assertEqual(contents, 'bar')

    def test_no_paginate_arg(self):
        d = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, d)
        bucket_name = random_bucket_name('nopaginate')

        self.put_object(bucket=bucket_name, key='foobar',
                        content='foobar contents')
        p = aws('s3api list-objects --bucket %s --no-paginate' % bucket_name)
        self.assertEqual(p.rc, 0, p.stdout + p.stderr)

        p = aws('s3api list-objects --bucket %s' % bucket_name)
        self.assertEqual(p.rc, 0, p.stdout + p.stderr)

    def test_top_level_options_debug(self):
        p = aws('ec2 describe-instances --debug')
        self.assertEqual(p.rc, 0)
        self.assertIn('DEBUG', p.stderr)

    def test_make_requests_to_other_region(self):
        p = aws('ec2 describe-instances --region us-west-2')
        self.assertEqual(p.rc, 0)
        self.assertIn('Reservations', p.json)

    def test_help_usage_top_level(self):
        p = aws('')
        self.assertIn('usage: aws [options] <command> '
                      '<subcommand> [<subcommand> ...] [parameters]', p.stderr)
        self.assertIn('aws: error', p.stderr)

    def test_help_usage_service_level(self):
        p = aws('ec2')
        self.assertIn('usage: aws [options] <command> '
                      '<subcommand> [<subcommand> ...] [parameters]', p.stderr)
        # python3: aws: error: the following arguments are required: operation
        # python2: aws: error: too few arguments
        # We don't care too much about the specific error message, as long
        # as it says we have a parse error.
        self.assertIn('aws: error', p.stderr)

    def test_help_usage_operation_level(self):
        p = aws('ec2 start-instances')
        self.assertIn('usage: aws [options] <command> '
                      '<subcommand> [<subcommand> ...] [parameters]', p.stderr)

    def test_unknown_argument(self):
        p = aws('ec2 describe-instances --filterss')
        self.assertEqual(p.rc, 255)
        self.assertIn('Unknown options: --filterss', p.stderr)

    def test_table_output(self):
        p = aws('ec2 describe-instances --output table --color off')
        # We're not testing the specifics of table output, we just want
        # to make sure the output looks like a table using some heuristics.
        # If this prints JSON instead of a table, for example, this test
        # should fail.
        self.assertEqual(p.rc, 0, p.stderr)
        self.assertIn('-----', p.stdout)
        self.assertIn('+-', p.stdout)
        self.assertIn('DescribeInstances', p.stdout)

    def test_version(self):
        p = aws('--version')
        self.assertEqual(p.rc, 0)
        # The version is wrote to standard out for Python 3.4 and
        # standard error for other Python versions.
        version_output = p.stderr.startswith('aws-cli') or \
            p.stdout.startswith('aws-cli')
        self.assertTrue(version_output, p.stderr)

    def test_traceback_printed_when_debug_on(self):
        p = aws('ec2 describe-instances --filters BADKEY=foo --debug')
        self.assertIn('Traceback (most recent call last):', p.stderr, p.stderr)
        # Also should see DEBUG statements:
        self.assertIn('DEBUG', p.stderr, p.stderr)

    def test_leftover_args_in_operation(self):
        p = aws('ec2 describe-instances BADKEY=foo')
        self.assertEqual(p.rc, 255)
        self.assertIn("Unknown option", p.stderr, p.stderr)

    def test_json_param_parsing(self):
        # This is convered by unit tests in botocore, but this is a sanity
        # check that we get a json response from a json service.
        p = aws('swf list-domains --registration-status REGISTERED')
        self.assertEqual(p.rc, 0)
        self.assertIsInstance(p.json, dict)

        p = aws('dynamodb list-tables')
        self.assertEqual(p.rc, 0)
        self.assertIsInstance(p.json, dict)

    def test_pagination_with_text_output(self):
        p = aws('iam list-users --output text')
        self.assertEqual(p.rc, 0)

    def test_bad_lc_ctype_env_var_is_handled(self):
        # Test for bad LC_CTYPE on Mac OS X.
        base_env_vars = os.environ.copy()
        base_env_vars['LC_CTYPE'] = 'UTF-8'
        p = aws('iam list-users', env_vars=base_env_vars)
        self.assertEqual(p.rc, 0)

    def test_error_msg_with_no_region_configured(self):
        environ = os.environ.copy()
        try:
            del environ['AWS_DEFAULT_REGION']
        except KeyError:
            pass
        environ['AWS_CONFIG_FILE'] = 'nowhere-foo'
        p = aws('ec2 describe-instances', env_vars=environ)
        self.assertIn('must specify a region', p.stderr)

    @skip_if_windows('Ctrl-C not supported on windows.')
    def test_ctrl_c_does_not_print_traceback(self):
        # Relying on the fact that this generally takes
        # more than 1 second to complete.
        process = aws('ec2 describe-images', wait_for_finish=False)
        time.sleep(1)
        process.send_signal(signal.SIGINT)
        stdout, stderr = process.communicate()
        self.assertNotIn(b'Traceback', stdout)
        self.assertNotIn(b'Traceback', stderr)


class TestCommandLineage(unittest.TestCase):
    def setUp(self):
        self.driver = create_clidriver()
        self.top_help = self.driver.create_help_command()

    def assert_lineage_names(self, ref_lineage_names):
        command_table = self.top_help.command_table
        for i, cmd_name in enumerate(ref_lineage_names):
            command = command_table[cmd_name]
            help_command = command.create_help_command()
            command_table = help_command.command_table

        actual_lineage_names = []
        for cmd in command.lineage:
            actual_lineage_names.append(cmd.name)

        # Assert the actual names of each command in a lineage is as expected.
        self.assertEqual(actual_lineage_names, ref_lineage_names)

        # Assert that ``lineage_names`` for each command is in sync with what
        # is actually in the command's ``lineage``.
        self.assertEqual(command.lineage_names, actual_lineage_names)

    def test_service_level_commands(self):
        # Check a normal unchanged service command
        self.assert_lineage_names(['ec2'])

        # Check a service that had its name changed.
        self.assert_lineage_names(['s3api'])

        # Check a couple custom service level commands.
        self.assert_lineage_names(['s3'])
        self.assert_lineage_names(['configure'])

    def test_operation_level_commands(self):
        # Check a normal unchanged service and operation command
        self.assert_lineage_names(['dynamodb', 'create-table'])

        # Check an operation commands with a service that had its name changed.
        self.assert_lineage_names(['s3api', 'list-objects'])

        # Check a custom operation level command with no
        # custom service command.
        self.assert_lineage_names(['emr', 'create-cluster'])

        # Check a couple of operation level commands that
        # are based off a custom service command
        self.assert_lineage_names(['configure', 'set'])
        self.assert_lineage_names(['s3', 'cp'])

    def test_wait_commands(self):
        self.assert_lineage_names(['ec2', 'wait'])
        self.assert_lineage_names(['ec2', 'wait', 'instance-running'])


# We're using BaseS3CLICommand because we need a service to use
# for testing the global arguments.  We're picking S3 here because
# the BaseS3CLICommand has a lot of utility functions that help
# with this.
class TestGlobalArgs(BaseS3CLICommand):

    def test_endpoint_url(self):
        p = aws('s3api list-objects --bucket dnscompat '
                '--endpoint-url http://localhost:51515 '
                '--debug')
        debug_logs = p.stderr
        original_hostname = 'dnscompat.s3.amazonaws.com'
        expected = 'localhost'
        self.assertNotIn(original_hostname, debug_logs,
                         '--endpoint-url is being ignored.')
        self.assertIn(expected, debug_logs)

    def test_no_pagination(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', contents=b'bar')
        self.put_object(bucket_name, 'foo2.txt', contents=b'bar')
        self.put_object(bucket_name, 'foo3.txt', contents=b'bar')
        p = aws('s3api list-objects --bucket %s '
                '--no-paginate --output json' % bucket_name)
        # A really simple way to check that --no-paginate was
        # honored is to see if we have all the mirrored input
        # arguments in the response json.  These normally aren't
        # present when the response is paginated.
        self.assert_no_errors(p)
        response_json = p.json
        self.assertIn('IsTruncated', response_json)
        self.assertIn('Name', response_json)

    def test_no_paginate_and_original_args(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', contents=b'bar')
        self.put_object(bucket_name, 'foo2.txt', contents=b'bar')
        self.put_object(bucket_name, 'foo3.txt', contents=b'bar')
        p = aws('s3api list-objects --bucket %s '
                '--max-keys 1 --no-paginate --output json' % bucket_name)
        self.assert_no_errors(p)
        response_json = p.json
        self.assertEqual(len(response_json['Contents']), 1)

    def test_max_items(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', contents=b'bar')
        self.put_object(bucket_name, 'foo2.txt', contents=b'bar')
        self.put_object(bucket_name, 'foo3.txt', contents=b'bar')
        p = aws('s3api list-objects --bucket %s '
                '--max-items 1 --output json' % bucket_name)
        self.assert_no_errors(p)
        response_json = p.json
        self.assertEqual(len(response_json['Contents']), 1)

    def test_query(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'foo.txt', contents=b'bar')
        p = aws('s3api list-objects --bucket %s '
                '--query Contents[].Key --output json' % bucket_name)
        self.assert_no_errors(p)
        response_json = p.json
        self.assertEqual(response_json, ['foo.txt'])

    def test_no_sign_requests(self):
        bucket_name = self.create_bucket()
        self.put_object(bucket_name, 'public', contents=b'bar',
                        extra_args={'ACL': 'public-read'})
        self.put_object(bucket_name, 'private', contents=b'bar')
        env = os.environ.copy()
        # Set the env vars to bad values so if we do actually
        # try to sign the request, we'll get an auth error.
        env['AWS_ACCESS_KEY_ID'] = 'foo'
        env['AWS_SECRET_ACCESS_KEY'] = 'bar'
        p = aws('s3api head-object --bucket %s --key public --no-sign-request'
                % bucket_name, env_vars=env)
        self.assert_no_errors(p)
        self.assertIn('ETag', p.json)

        # Should fail because we're not signing the request but the object is
        # private.
        p = aws('s3api head-object --bucket %s --key private --no-sign-request'
                % bucket_name, env_vars=env)
        self.assertEqual(p.rc, 255)

    def test_profile_arg_has_precedence_over_env_vars(self):
        # At a high level, we're going to set access_key/secret_key
        # via env vars, but ensure that a --profile <foo> results
        # in creds being retrieved from the shared creds file
        # and not from env vars.
        env_vars = os.environ.copy()
        with temporary_file('w') as f:
            env_vars.pop('AWS_PROFILE', None)
            env_vars.pop('AWS_DEFAULT_PROFILE', None)
            # 'aws configure list' only shows 4 values
            # from the credentials so we'll show
            # 4 char values.
            env_vars['AWS_ACCESS_KEY_ID'] = 'enva'
            env_vars['AWS_SECRET_ACCESS_KEY'] = 'envb'
            env_vars['AWS_SHARED_CREDENTIALS_FILE'] = f.name
            env_vars['AWS_CONFIG_FILE'] = 'does-not-exist-foo'
            f.write(
                '[from_argument]\n'
                'aws_access_key_id=proa\n'
                'aws_secret_access_key=prob\n'
            )
            f.flush()
            p = aws('configure list --profile from_argument',
                    env_vars=env_vars)
            # 1. We should see the profile name being set.
            self.assertIn('from_argument', p.stdout)
            # 2. The creds should be proa/prob, which come
            #    from the "from_argument" profile.
            self.assertIn('proa', p.stdout)
            self.assertIn('prob', p.stdout)
            self.assertIn('shared-credentials-file', p.stdout)

    def test_profile_arg_wins_over_profile_env_var(self):
        env_vars = os.environ.copy()
        with temporary_file('w') as f:
            # Remove existing profile related env vars.
            env_vars.pop('AWS_PROFILE', None)
            env_vars.pop('AWS_DEFAULT_PROFILE', None)
            env_vars['AWS_SHARED_CREDENTIALS_FILE'] = f.name
            env_vars['AWS_CONFIG_FILE'] = 'does-not-exist-foo'
            f.write(
                '[from_env_var]\n'
                'aws_access_key_id=enva\n'
                'aws_secret_access_key=envb\n'
                '\n'
                '[from_argument]\n'
                'aws_access_key_id=proa\n'
                'aws_secret_access_key=prob\n'
            )
            f.flush()
            # Now we set the current profile via env var:
            env_vars['AWS_PROFILE'] = 'from_env_var'
            # If we specify the --profile argument, that
            # value should win over the AWS_PROFILE env var.
            p = aws('configure list --profile from_argument',
                    env_vars=env_vars)
            # 1. We should see the profile name being set.
            self.assertIn('from_argument', p.stdout)
            # 2. The creds should be profa/profb, which come
            #    from the "from_argument" profile.
            self.assertIn('proa', p.stdout)
            self.assertIn('prob', p.stdout)


if __name__ == '__main__':
    unittest.main()