# Copyright 2018 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 from functools import partial import pytest from awscli.autocomplete import parser from awscli.testutils import unittest from tests.unit.autocomplete import InMemoryIndex # This models an 'aws ec2 stop-instances' command # along with the 'region', 'endpoint-url', and 'debug' global params. SAMPLE_MODEL = InMemoryIndex( # This format is intended to match the structure you'd get from # querying the index db to minimize any parity issues between # the tests and the real indexer. index={ 'arg_names': { '': { 'aws': ['region', 'endpoint-url', 'debug'], }, 'aws.ec2': { 'stop-instances': ['instance-ids', 'foo-arg'], }, 'aws.s3': { 'cp': ['bucket', 'key'], }, 'aws.s3api': { 'get-object': ['outfile', 'bucket', 'key'], }, 'aws.logs': {'tail': ['group_name', 'filter-pattern']}, }, 'command_names': { '': [('aws', None)], 'aws': [ ('ec2', None), ('logs', None), ('s3api', None), ('s3', None), ], 'aws.ec2': [('stop-instances', None)], 'aws.logs': [('tail', None)], 'aws.s3': [('cp', None)], 'aws.s3api': [('get-object', None)], }, 'arg_data': { '': { 'aws': { 'debug': ( 'debug', 'boolean', 'aws', '', None, False, False, ), 'endpoint-url': ( 'endpoint-url', 'string', 'aws', '', None, False, False, ), 'region': ( 'region', 'string', 'aws', '', None, False, False, ), } }, 'aws.ec2': { 'stop-instances': { 'instance-ids': ( 'instance-ids', 'string', 'stop-instances', 'aws.ec2.', '*', False, False, ), 'foo-arg': ( 'foo-arg', 'string', 'stop-instances', 'aws.ec2', None, False, False, ), 'positional': ( 'positional', 'string', 'stop-instances', 'aws.ec2', None, True, False, ), } }, 'aws.logs': { 'tail': { 'group_name': ( 'group_name', 'string', 'tail', 'aws.logs.', None, True, False, ), 'filter-pattern': ( 'filter-pattern', 'string', 'tail', 'aws.logs', None, False, False, ), } }, 'aws.s3': { 'cp': { 'bucket': ( 'bucket', 'string', 'cp', 'aws.s3.', None, False, False, ), 'key': ( 'key', 'string', 'cp', 'aws.s3.', None, False, False, ), } }, 'aws.s3api': { 'get-object': { 'outfile': ( 'outfile', 'string', 'get-object', 'aws.s3api.', None, False, False, ), 'bucket': ( 'bucket', 'string', 'get-object', 'aws.s3api.', None, False, False, ), 'key': ( 'key', 'string', 'get-object', 'aws.s3api.', None, False, False, ), } }, }, }, ) @pytest.mark.parametrize( 'command_line', [ ( 'aws ec2 stop-instances ' '--instance-ids i-123 i-124 --foo-arg value --debug ' '--endpoint-url https://foo ' ), ( 'aws --debug ec2 stop-instances ' '--instance-ids i-123 i-124 --foo-arg value ' '--endpoint-url https://foo ' ), ( 'aws --endpoint-url https://foo --debug ec2 stop-instances ' '--instance-ids i-123 i-124 --foo-arg value ' ), ( 'aws ec2 --debug --endpoint-url https://foo stop-instances ' '--instance-ids i-123 i-124 --foo-arg value ' ), ( 'aws ec2 stop-instances --debug --endpoint-url https://foo ' '--instance-ids i-123 i-124 --foo-arg value ' ), ( 'aws ec2 --endpoint-url https://foo stop-instances --debug ' '--instance-ids i-123 i-124 --foo-arg value ' ), ], ) def test_can_handle_arbitrary_ordering(command_line): # This test verifies that parse to the same result given a command # with its arguments in varying orders. For more detailed testing, # see TestCanParseCLICommand below. expected = parser.ParsedResult( current_command='stop-instances', global_params={'debug': True, 'endpoint-url': 'https://foo'}, parsed_params={'instance-ids': ['i-123', 'i-124'], 'foo-arg': 'value'}, lineage=['aws', 'ec2'], current_fragment='', ) _assert_parses_to(command_line, expected) def _assert_parses_to(command_line, expected): p = parser.CLIParser(SAMPLE_MODEL) result = p.parse(command_line) for key, value in vars(expected).items(): actual = getattr(result, key) assert getattr(result, key) == value, '%r != %r for attribute: %r' % ( actual, value, key, ) def _generate_command_chunks(): command_line = ( 'aws ec2 stop-instances --instance-ids i-123 i-124 ' '--foo-arg value --debug --endpoint-url https://foo' ) return [command_line[:i] for i in range(1, len(command_line))] @pytest.mark.parametrize("chunk", _generate_command_chunks()) def test_properties_of_unparsed_results(chunk): # The parser should never raise an exception. If it can't # understand something it should still return a ParsedResult # with the parts it doesn't understand addded to the unparsed_items # attribute. The ParsedResult should always have some basic invariants # we can verify which are called out in the tests below. # This test ensures that at every single slice of the full command_line # we always produce a sensical ParsedResult. cli_parser = parser.CLIParser(SAMPLE_MODEL) _assert_parsed_properties(chunk, cli_parser) def _assert_parsed_properties(chunk, cli_parser): result = cli_parser.parse(chunk) assert isinstance(result, parser.ParsedResult) if chunk[-1].isspace(): # If there's a space as the last char, then we should have # a current_fragment of an empty string. This results in # all results being returned from the prefix match in the # auto-completer. assert result.current_fragment == '' elif result.current_fragment is not None: # The current_fragment, if not None is always the last part # of the command line. assert chunk.endswith(result.current_fragment) class TestCanParseCLICommand(unittest.TestCase): def setUp(self): self.cli_parser = self.create_parser() def create_parser(self, return_first_command_match=False): return parser.CLIParser( SAMPLE_MODEL, return_first_command_match=return_first_command_match ) def assert_parsed_results_equal(self, actual, **expected): # Asserts that every kwargs in expected matches what was actually # parsed. for key, value in expected.items(): actual_value = getattr(actual, key) self.assertEqual( actual_value, value, '%r != %r (attr: %r)' % (actual_value, value, key), ) def assert_parsed_s3api_result_correct(self, result): self.assert_parsed_results_equal( result, current_command='get-object', parsed_params={ 'outfile': 'filename', 'bucket': 'myBucket', 'key': 'foo', }, lineage=['aws', 's3api'], unparsed_items=[], ) def test_parsed_result_not_equal(self): self.assertFalse(parser.ParsedResult(current_command='ec2') == 'ec2') def test_can_handle_empty_string_for_command(self): result = self.cli_parser.parse('') self.assert_parsed_results_equal( result, current_command=None, global_params={}, parsed_params={}, lineage=[], unparsed_items=[], ) def test_result_if_aws_exe_alias(self): result = self.cli_parser.parse('aws2 ec2') self.assert_parsed_results_equal( result, current_command='ec2', global_params={}, parsed_params={}, lineage=['aws'], unparsed_items=[], ) def test_can_parse_operation_command_accepts_single_value_arg(self): result = self.cli_parser.parse('aws ec2 stop-instances --foo-arg bar ') self.assert_parsed_results_equal( result, current_command='stop-instances', global_params={}, parsed_params={'foo-arg': 'bar'}, lineage=['aws', 'ec2'], ) def test_can_parse_operation_command_with_param(self): result = self.cli_parser.parse( 'aws ec2 stop-instances --instance-ids i-1 ' ) self.assert_parsed_results_equal( result, current_command='stop-instances', global_params={}, parsed_params={'instance-ids': ['i-1']}, lineage=['aws', 'ec2'], ) def test_can_parse_bool_param(self): result = self.cli_parser.parse( 'aws --debug ec2 stop-instances --instance-ids i-1 ' ) self.assert_parsed_results_equal( result, current_command='stop-instances', global_params={'debug': True}, parsed_params={'instance-ids': ['i-1']}, lineage=['aws', 'ec2'], ) def test_can_parse_bool_param_in_any_location(self): result = self.cli_parser.parse( 'aws ec2 stop-instances --instance-ids i-1 --debug' ) self.assert_parsed_results_equal( result, current_command='stop-instances', global_params={'debug': True}, parsed_params={'instance-ids': ['i-1']}, lineage=['aws', 'ec2'], ) def test_can_parse_operation_command(self): result = self.cli_parser.parse('aws ec2 stop-instances') self.assert_parsed_results_equal( result, current_command='stop-instances', global_params={}, parsed_params={}, lineage=['aws', 'ec2'], ) def test_can_parse_service_command(self): result = self.cli_parser.parse('aws ec2') self.assert_parsed_results_equal( result, current_command='ec2', parsed_params={}, lineage=['aws'], ) def test_can_parse_aws_command(self): result = self.cli_parser.parse('aws') self.assert_parsed_results_equal( result, current_command='aws', parsed_params={}, global_params={}, lineage=[], ) def test_ignores_unknown_args(self): result = self.cli_parser.parse( 'aws ec2 stop-instances --unknown-arg bar' ) self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={}, lineage=['aws', 'ec2'], ) def test_can_parse_if_there_longer_command(self): result = self.cli_parser.parse('aws s3') self.assert_parsed_results_equal( result, current_command='aws', parsed_params={}, global_params={}, current_fragment='s3', lineage=[], ) def test_parse_shortest_command_if_first_wins(self): parser = self.create_parser(return_first_command_match=True) result = parser.parse('aws s3') self.assert_parsed_results_equal( result, current_command='s3', parsed_params={}, global_params={}, lineage=['aws'], ) def test_can_proceed_if_there_longer_command(self): result = self.cli_parser.parse('aws s3 ') self.assert_parsed_results_equal( result, current_command='s3', global_params={}, current_fragment='', lineage=['aws'], ) def test_can_consume_one_or_more_nargs(self): result = self.cli_parser.parse( 'aws ec2 stop-instances --instance-ids i-1 i-2 i-3 ' ) self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={'instance-ids': ['i-1', 'i-2', 'i-3']}, lineage=['aws', 'ec2'], ) def test_can_consume_zero_or_one_nargs(self): model = copy.deepcopy(SAMPLE_MODEL) nargs_one_or_more = '?' model.index['arg_data']['aws.ec2']['stop-instances']['foo-arg'] = ( 'foo-arg', 'string', 'stop-instances', 'aws.ec2', nargs_one_or_more, False, False, ) p = parser.CLIParser(model) self.assertEqual( p.parse('aws ec2 stop-instances --foo-arg --debug').parsed_params[ 'foo-arg' ], None, ) self.assertEqual( p.parse( 'aws ec2 stop-instances --foo-arg bar --debug' ).parsed_params['foo-arg'], 'bar', ) def test_truncates_line_based_on_location(self): # The 22nd index cuts off right after `stop-instances`. result = self.cli_parser.parse( 'aws ec2 stop-instances --instance-ids i-1 i-2 i-3', 22 ) # We should not have parsed the 'instance-ids'. self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={}, lineage=['aws', 'ec2'], ) def test_preserves_current_word(self): result = self.cli_parser.parse('aws ec2 stop-') self.assert_parsed_results_equal( result, current_command='ec2', parsed_params={}, global_params={}, lineage=['aws'], current_fragment='stop-', ) def test_word_not_preserved_if_not_adjacent_to_word(self): result = self.cli_parser.parse('aws ec2 stop- ') self.assert_parsed_results_equal( result, current_command='ec2', parsed_params={}, global_params={}, lineage=['aws'], current_fragment='', unparsed_items=['stop-'], ) def test_current_fragment_populated_on_work_break(self): result = self.cli_parser.parse('aws ec2 ') self.assert_parsed_results_equal( result, current_command='ec2', parsed_params={}, global_params={}, lineage=['aws'], current_fragment='', ) def test_current_fragment_can_be_option(self): result = self.cli_parser.parse('aws ec2 stop-instances --inst') # We should not have parsed the 'instance-ids'. self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={}, lineage=['aws', 'ec2'], current_fragment='--inst', ) def test_option_not_preserved_when_space_separated(self): result = self.cli_parser.parse('aws ec2 stop-instances --inst ') self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={}, lineage=['aws', 'ec2'], current_fragment='', unparsed_items=['--inst'], ) def test_can_have_unparsed_option_with_current_fragment(self): result = self.cli_parser.parse('aws ec2 stop-instances --inst foo') self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={}, lineage=['aws', 'ec2'], current_fragment='foo', unparsed_items=['--inst'], ) def test_unknown_option_does_not_consume_arg(self): # In this case we're unlikely to offer any helpful # auto-completion, but we still need to decided where # we should put the 'foo' value. I think it makes the # most sense to put this under "unparsed_items". result = self.cli_parser.parse('aws ec2 stop-instances --inst foo ') self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={}, lineage=['aws', 'ec2'], current_fragment='', unparsed_items=['--inst', 'foo'], ) def test_can_handle_multiple_unknown_options(self): result = self.cli_parser.parse('aws ec2 stop-instances --inst --foo ') self.assert_parsed_results_equal( result, current_command='stop-instances', parsed_params={}, lineage=['aws', 'ec2'], current_fragment='', unparsed_items=['--inst', '--foo'], ) def test_can_handle_unparsed_values(self): result = self.cli_parser.parse('aws ec stop-insta ') self.assert_parsed_results_equal( result, current_command='aws', current_fragment='', unparsed_items=['ec', 'stop-insta'], ) def test_can_handle_error_cases_gracefully(self): # The --foo-arg has nargs of None so this is an invalid command. # It would normally generate a parser error. We should make sure # we handle this gracefully. result = self.cli_parser.parse( 'aws ec2 stop-instances --foo-arg a b --' ) self.assert_parsed_results_equal( result, current_command='stop-instances', lineage=['aws', 'ec2'], current_fragment=None, unparsed_items=['b', '--'], ) def test_current_fragment_populated_for_param_value(self): result = self.cli_parser.parse( 'aws ec2 stop-instances --instance-ids i-' ) self.assert_parsed_results_equal( result, current_command='stop-instances', current_param='instance-ids', parsed_params={'instance-ids': None}, lineage=['aws', 'ec2'], current_fragment='i-', unparsed_items=[], ) def test_params_populated_unless_last_value(self): result = self.cli_parser.parse( 'aws ec2 stop-instances --instance-ids i-1 i-' ) self.assert_parsed_results_equal( result, current_command='stop-instances', current_param='instance-ids', # Here i-1 is in the parsed params because we've already moved # on from that word, however, 'i-' is not added to the parsed # params dict because we're still parsing that value. It goes in # the ``current_fragment`` attribute. parsed_params={'instance-ids': ['i-1']}, lineage=['aws', 'ec2'], current_fragment='i-', unparsed_items=[], ) def test_params_stop_populate_when_dashes_entered(self): result = self.cli_parser.parse( 'aws ec2 stop-instances --instance-ids i-1 --' ) self.assert_parsed_results_equal( result, current_command='stop-instances', current_param=None, parsed_params={'instance-ids': ['i-1']}, lineage=['aws', 'ec2'], current_fragment='--', unparsed_items=[], ) def test_curent_fragment_works_for_global_params(self): result = self.cli_parser.parse( 'aws --endpoint-url http', ) self.assert_parsed_results_equal( result, current_command='aws', current_param='endpoint-url', # Here i-1 is in the parsed params because we've already moved # on from that word, however, 'i-' is not added to the parsed # params dict because we're still parsing that value. It goes in # the ``current_fragment`` attribute. parsed_params={}, lineage=[], current_fragment='http', unparsed_items=[], ) def test_parses_positional(self): result = self.cli_parser.parse( 'aws logs tail mygroup', ) self.assert_parsed_results_equal( result, current_command='tail', current_param='group_name', parsed_params={}, lineage=['aws', 'logs'], current_fragment='mygroup', unparsed_items=[], ) def test_parses_positional_done(self): result = self.cli_parser.parse( 'aws logs tail mygroup ', ) self.assert_parsed_results_equal( result, current_command='tail', current_param=None, parsed_params={'group_name': 'mygroup'}, lineage=['aws', 'logs'], current_fragment='', unparsed_items=[], ) def test_parses_positional_parsed_only_once(self): result = self.cli_parser.parse( 'aws logs tail mygroup random-content ', ) self.assert_parsed_results_equal( result, current_command='tail', current_param=None, parsed_params={'group_name': 'mygroup'}, lineage=['aws', 'logs'], current_fragment='', unparsed_items=['random-content'], ) def test_parse_positional_with_option(self): result = self.cli_parser.parse( 'aws logs tail mygroup --filter-pattern pat', ) self.assert_parsed_results_equal( result, current_command='tail', current_param='filter-pattern', parsed_params={'filter-pattern': None, 'group_name': 'mygroup'}, lineage=['aws', 'logs'], current_fragment='pat', unparsed_items=[], ) def test_parse_positional_with_option_before(self): result = self.cli_parser.parse( 'aws logs tail --filter-pattern pattern mygroup', ) self.assert_parsed_results_equal( result, current_command='tail', current_param='group_name', parsed_params={ 'filter-pattern': 'pattern', }, lineage=['aws', 'logs'], current_fragment='mygroup', unparsed_items=[], ) def test_parse_outfile_option_in_the_beginning(self): result = self.cli_parser.parse( 'aws s3api get-object filename --bucket myBucket --key foo ', ) self.assert_parsed_s3api_result_correct(result) def test_parse_outfile_option_in_the_middle(self): result = self.cli_parser.parse( 'aws s3api get-object --bucket myBucket filename --key foo ', ) self.assert_parsed_s3api_result_correct(result) def test_parse_outfile_option_at_the_end(self): result = self.cli_parser.parse( 'aws s3api get-object --bucket myBucket --key foo filename ', ) self.assert_parsed_s3api_result_correct(result) class TestParseState(unittest.TestCase): def test_can_set_initial_state(self): state = parser.ParseState() self.assertIsNone(state.current_command) self.assertEqual(state.lineage, []) self.assertEqual(state.full_lineage, []) def test_initial_starting_command_has_correct_lineage(self): state = parser.ParseState() state.current_command = 'aws' self.assertEqual(state.current_command, 'aws') self.assertEqual(state.lineage, []) self.assertEqual(state.full_lineage, ['aws']) def test_can_add_new_state(self): state = parser.ParseState() state.current_command = 'aws' state.current_command = 'ec2' self.assertEqual(state.current_command, 'ec2') self.assertEqual(state.lineage, ['aws']) self.assertEqual(state.full_lineage, ['aws', 'ec2']) def test_multiple_lineage_has_dotted_name(self): state = parser.ParseState() state.current_command = 'aws' state.current_command = 'ec2' state.current_command = 'describe-instances' self.assertEqual(state.current_command, 'describe-instances') self.assertEqual(state.lineage, ['aws', 'ec2']) self.assertEqual( state.full_lineage, ['aws', 'ec2', 'describe-instances'] )