Alias 0.3.0

src/alias/azext_alias/__init__.py

@@ -8,32 +8,69 @@
 from knack.log import get_logger
 
 from azure.cli.core import AzCommandsLoader
-from azure.cli.core.commands import CliCommandType
+from azure.cli.core.decorators import Completer
 from azure.cli.core.commands.events import EVENT_INVOKER_PRE_CMD_TBL_TRUNCATE
-from azext_alias.alias import AliasManager
+from azext_alias.alias import GLOBAL_ALIAS_PATH, AliasManager
+from azext_alias.util import get_config_parser, is_alias_create_command, cache_reserved_commands
 from azext_alias._const import DEBUG_MSG_WITH_TIMING
+from azext_alias._validators import process_alias_create_namespace
 from azext_alias import telemetry
+from azext_alias import _help  # pylint: disable=unused-import
 
 logger = get_logger(__name__)
+"""
+We don't have access to load_cmd_tbl_func in custom.py (need the entire command table
+for alias and command validation when the user invokes alias create).
+This cache saves the entire command table globally so custom.py can have access to it.
+Alter this cache through cache_reserved_commands(load_cmd_tbl_func) in util.py
+"""
+cached_reserved_commands = []
 
 
-class AliasExtensionLoader(AzCommandsLoader):
+class AliasExtCommandLoader(AzCommandsLoader):
 
     def __init__(self, cli_ctx=None):
-        super(AliasExtensionLoader, self).__init__(cli_ctx=cli_ctx,
-                                                   custom_command_type=CliCommandType())
-
+        from azure.cli.core.commands import CliCommandType
+        custom_command_type = CliCommandType(operations_tmpl='azext_alias.custom#{}')
+        super(AliasExtCommandLoader, self).__init__(cli_ctx=cli_ctx,
+                                                    custom_command_type=custom_command_type)
         self.cli_ctx.register_event(EVENT_INVOKER_PRE_CMD_TBL_TRUNCATE, alias_event_handler)
 
-    def load_command_table(self, _):  # pylint:disable=no-self-use
-        return {}
+    def load_command_table(self, _):
+        with self.command_group('alias') as g:
+            g.custom_command('create', 'create_alias', validator=process_alias_create_namespace)
+            g.custom_command('list', 'list_alias')
+            g.custom_command('remove', 'remove_alias')
+
+        return self.command_table
 
-    def load_arguments(self, _):  # pylint:disable=no-self-use
-        pass
+    def load_arguments(self, _):
+        with self.argument_context('alias create') as c:
+            c.argument('alias_name', options_list=['--name', '-n'], help='The name of the alias.')
+            c.argument('alias_command', options_list=['--command', '-c'], help='The command that the alias points to.')
+
+        with self.argument_context('alias remove') as c:
+            c.argument('alias_name', options_list=['--name', '-n'], help='The name of the alias.',
+                       completer=get_alias_completer)
+
+
+@Completer
+def get_alias_completer(cmd, prefix, namespace, **kwargs):  # pylint: disable=unused-argument
+    """
+    An argument completer for alias name.
+    """
+    try:
+        alias_table = get_config_parser()
+        alias_table.read(GLOBAL_ALIAS_PATH)
+        return alias_table.sections()
+    except Exception:  # pylint: disable=broad-except
+        return []
 
 
 def alias_event_handler(_, **kwargs):
-    """ An event handler for alias transformation when EVENT_INVOKER_PRE_TRUNCATE_CMD_TBL event is invoked """
+    """
+    An event handler for alias transformation when EVENT_INVOKER_PRE_TRUNCATE_CMD_TBL event is invoked
+    """
     try:
         telemetry.start()
 
@@ -44,6 +81,10 @@ def alias_event_handler(_, **kwargs):
         # [:] will keep the reference of the original args
         args[:] = alias_manager.transform(args)
 
+        if is_alias_create_command(args):
+            load_cmd_tbl_func = kwargs.get('load_cmd_tbl_func', lambda _: {})
+            cache_reserved_commands(load_cmd_tbl_func)
+
         elapsed_time = (timeit.default_timer() - start_time) * 1000
         logger.debug(DEBUG_MSG_WITH_TIMING, args, elapsed_time)
 
@@ -55,4 +96,4 @@ def alias_event_handler(_, **kwargs):
         telemetry.conclude()
 
 
-COMMAND_LOADER_CLS = AliasExtensionLoader
+COMMAND_LOADER_CLS = AliasExtCommandLoader

src/alias/azext_alias/_const.py

@@ -9,14 +9,20 @@
 ALIAS_FILE_NAME = 'alias'
 ALIAS_HASH_FILE_NAME = 'alias.sha1'
 COLLIDED_ALIAS_FILE_NAME = 'collided_alias'
-COLLISION_CHECK_LEVEL_DEPTH = 4
+COLLISION_CHECK_LEVEL_DEPTH = 5
 
 INSUFFICIENT_POS_ARG_ERROR = 'alias: "{}" takes exactly {} positional argument{} ({} given)'
-CONFIG_PARSING_ERROR = 'alias: Error parsing the configuration file - {}. Please fix the problem manually.'
+CONFIG_PARSING_ERROR = 'alias: Error parsing the configuration file - %s. Please fix the problem manually.'
 DEBUG_MSG = 'Alias Manager: Transforming "%s" to "%s"'
 DEBUG_MSG_WITH_TIMING = 'Alias Manager: Transformed args to %s in %.3fms'
 POS_ARG_DEBUG_MSG = 'Alias Manager: Transforming "%s" to "%s", with the following positional arguments: %s'
 DUPLICATED_PLACEHOLDER_ERROR = 'alias: Duplicated placeholders found when transforming "{}"'
 RENDER_TEMPLATE_ERROR = 'alias: Encounted the following error when injecting positional arguments to "{}" - {}'
 PLACEHOLDER_EVAL_ERROR = 'alias: Encounted the following error when evaluating "{}" - {}'
 PLACEHOLDER_BRACKETS_ERROR = 'alias: Brackets in "{}" are not enclosed properly'
+ALIAS_NOT_FOUND_ERROR = 'alias: "{}" alias not found'
+INVALID_ALIAS_COMMAND_ERROR = 'alias: Invalid Azure CLI command "{}"'
+EMPTY_ALIAS_ERROR = 'alias: Empty alias name or command is invalid'
+INVALID_STARTING_CHAR_ERROR = 'alias: Alias name should not start with "{}"'
+INCONSISTENT_ARG_ERROR = 'alias: Positional argument{} {} {} not in both alias name and alias command'
+COMMAND_LVL_ERROR = 'alias: "{}" is a reserved command and cannot be used to represent "{}"'

src/alias/azext_alias/_help.py

@@ -0,0 +1,48 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from knack.help_files import helps  # pylint: disable=unused-import
+
+
+helps['alias'] = """
+    type: group
+    short-summary: Manage Azure CLI Aliases.
+"""
+
+
+helps['alias create'] = """
+    type: command
+    short-summary: Create an alias.
+    examples:
+        - name: Create a simple alias.
+          text: >
+            az alias create --name rg --command group\n
+            az alias create --name ls --command list
+        - name: Create a complex alias.
+          text: >
+            az alias create --name list-vm --command 'vm list --resource-group myResourceGroup'
+
+        - name: Create an alias with positional arguments.
+          text: >
+            az alias create --name 'list-vm {{ resource_group }}' --command 'vm list --resource-group {{ resource_group }}'
+
+        - name: Create an alias with positional arguments and additional string processing.
+          text: >
+            az alias create --name 'storage-ls {{ url }}' --command 'storage blob list \n
+            --account-name {{ url.replace("https://", "").split(".")[0] }}\n
+            --container-name {{ url.replace("https://", "").split("/")[1] }}'
+"""
+
+
+helps['alias list'] = """
+    type: command
+    short-summary: List the registered aliases.
+"""
+
+
+helps['alias remove'] = """
+    type: command
+    short-summary: Remove an alias.
+"""

src/alias/azext_alias/_validators.py

@@ -0,0 +1,122 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import re
+import shlex
+
+from knack.util import CLIError
+
+import azext_alias
+from azext_alias.argument import get_placeholders
+from azext_alias._const import (
+    COLLISION_CHECK_LEVEL_DEPTH,
+    INVALID_ALIAS_COMMAND_ERROR,
+    EMPTY_ALIAS_ERROR,
+    INVALID_STARTING_CHAR_ERROR,
+    INCONSISTENT_ARG_ERROR,
+    COMMAND_LVL_ERROR
+)
+from azext_alias.alias import AliasManager
+
+
+def process_alias_create_namespace(namespace):
+    """
+    Validate input arguments when the user invokes 'az alias create'.
+
+    Args:
+        namespace: argparse namespace object.
+    """
+    _validate_alias_name(namespace.alias_name)
+    _validate_alias_command(namespace.alias_command)
+    _validate_alias_command_level(namespace.alias_name, namespace.alias_command)
+    _validate_pos_args_syntax(namespace.alias_name, namespace.alias_command)
+
+
+def _validate_alias_name(alias_name):
+    """
+    Check if the alias name is valid.
+
+    Args:
+        alias_name: The name of the alias to validate.
+    """
+    if not alias_name:
+        raise CLIError(EMPTY_ALIAS_ERROR)
+
+    if not re.match('^[a-zA-Z]', alias_name):
+        raise CLIError(INVALID_STARTING_CHAR_ERROR.format(alias_name[0]))
+
+
+def _validate_alias_command(alias_command):
+    """
+    Check if the alias command is valid.
+
+    Args:
+        alias_command: The command to validate.
+    """
+    if not alias_command:
+        raise CLIError(EMPTY_ALIAS_ERROR)
+
+    # Boundary index is the index at which named argument or positional argument starts
+    split_command = shlex.split(alias_command)
+    boundary_index = len(split_command)
+    for i, subcommand in enumerate(split_command):
+        if not re.match('^[a-z]', subcommand.lower()) or i > COLLISION_CHECK_LEVEL_DEPTH:
+            boundary_index = i
+            break
+
+    # Extract possible CLI commands and validate
+    command_to_validate = ' '.join(split_command[:boundary_index]).lower()
+    for command in azext_alias.cached_reserved_commands:
+        if re.match(r'([a-z\-]*\s)*{}($|\s)'.format(command_to_validate), command):
+            return
+
+    raise CLIError(INVALID_ALIAS_COMMAND_ERROR.format(command_to_validate if command_to_validate else alias_command))
+
+
+def _validate_pos_args_syntax(alias_name, alias_command):
+    """
+    Check if the positional argument syntax is valid in alias name and alias command.
+
+    Args:
+        alias_name: The name of the alias to validate.
+        alias_command: The command to validate.
+    """
+    pos_args_from_alias = get_placeholders(alias_name)
+    # Split by '|' to extract positional argument name from Jinja filter (e.g. {{ arg_name | upper }})
+    # Split by '.' to extract positional argument name from function call (e.g. {{ arg_name.split()[0] }})
+    pos_args_from_command = [x.split('|')[0].split('.')[0].strip() for x in get_placeholders(alias_command)]
+
+    if set(pos_args_from_alias) != set(pos_args_from_command):
+        arg_diff = set(pos_args_from_alias) ^ set(pos_args_from_command)
+        raise CLIError(INCONSISTENT_ARG_ERROR.format('' if len(arg_diff) == 1 else 's',
+                                                     arg_diff,
+                                                     'is' if len(arg_diff) == 1 else 'are'))
+
+
+def _validate_alias_command_level(alias, command):
+    """
+    Make sure that if the alias is a reserved command, the command that the alias points to
+    in the command tree does not conflict in levels.
+
+    e.g. 'dns' -> 'network dns' is valid because dns is a level 2 command and network dns starts at level 1.
+    However, 'list' -> 'show' is not valid because list and show are both reserved commands at level 2.
+
+    Args:
+        alias: The name of the alias.
+        command: The command that the alias points to.
+    """
+    alias_collision_table = AliasManager.build_collision_table([alias], azext_alias.cached_reserved_commands)
+
+    # Alias is not a reserved command, so it can point to any command
+    if not alias_collision_table:
+        return
+
+    command_collision_table = AliasManager.build_collision_table([command], azext_alias.cached_reserved_commands)
+    alias_collision_levels = alias_collision_table.get(alias.split()[0], [])
+    command_collision_levels = command_collision_table.get(command.split()[0], [])
+
+    # Check if there is a command level conflict
+    if set(alias_collision_levels) & set(command_collision_levels):
+        raise CLIError(COMMAND_LVL_ERROR.format(alias, command))