diff options
| -rw-r--r-- | .gitignore | 7 | ||||
| -rw-r--r-- | CHANGELOG.md | 18 | ||||
| -rw-r--r-- | CODEOWNERS | 40 | ||||
| -rw-r--r-- | cmd2/__init__.py | 1 | ||||
| -rw-r--r-- | cmd2/argcomplete_bridge.py | 44 | ||||
| -rwxr-xr-x | cmd2/argparse_completer.py | 81 | ||||
| -rw-r--r-- | cmd2/cmd2.py | 611 | ||||
| -rw-r--r-- | cmd2/parsing.py | 24 | ||||
| -rw-r--r-- | cmd2/pyscript_bridge.py | 9 | ||||
| -rw-r--r-- | cmd2/transcript.py | 18 | ||||
| -rw-r--r-- | cmd2/utils.py | 17 | ||||
| -rwxr-xr-x | examples/paged_output.py | 43 | ||||
| -rwxr-xr-x | examples/tab_autocompletion.py | 17 | ||||
| -rwxr-xr-x | mtime.sh | 14 | ||||
| -rw-r--r-- | speedup_import.md | 99 | ||||
| -rw-r--r-- | tasks.py | 15 | ||||
| -rw-r--r-- | tests/test_bashcompletion.py | 24 | ||||
| -rw-r--r-- | tests/test_cmd2.py | 42 | ||||
| -rw-r--r-- | tests/test_transcript.py | 26 |
19 files changed, 665 insertions, 485 deletions
@@ -17,3 +17,10 @@ htmlcov # Visual Studio Code .vscode + +# mypy optional static type checker +.mypy_cache + +# mypy plugin for PyCharm +dmypy.json +dmypy.sock diff --git a/CHANGELOG.md b/CHANGELOG.md index 08c4e15e..078af8d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,21 @@ +## 0.9.2 (TBD, 2018) +* Bug Fixes + * Fixed issue where piping and redirecting did not work correctly with paths that had spaces +* Enhancements + * Added ability to print a header above tab-completion suggestions using `completion_header` member + * Added ``pager`` and ``pager_chop`` attributes to the ``cmd2.Cmd`` class + * ``pager`` defaults to **less -RXF** on POSIX and **more** on Windows + * ``pager_chop`` defaults to **less -SRXF** on POSIX and **more** on Windows + * Added ``chop`` argument to ``cmd2.Cmd.ppaged()`` method for displaying output using a pager + * If ``chop`` is ``False``, then ``self.pager`` is used as the pager + * Otherwise ``self.pager_chop`` is used as the pager + +## 0.8.8 (TBD, 2018) +* Bug Fixes + * Prevent crashes that could occur attempting to open a file in non-existent directory or with very long filename +* Enhancements + * `display_matches` is no longer restricted to delimited strings + ## 0.9.1 (May 28, 2018) * Bug Fixes * fix packaging error for 0.8.x versions (yes we had to deploy a new version @@ -12,3 +12,43 @@ # You can also use email addresses if you prefer. #docs/* docs@example.com + +# cmd2 code +cmd2/__init__.py @tleonhardt @kotfu +cmd2/arg*.py @anselor +cmd2/cmd2.py @tleonhardt @kmvanbrunt @kotfu +cmd2/constants.py @kotfu +cmd2/parsing.py @kotfu @kmvanbrunt +cmd2/pyscript*.py @anselor +cmd2/rl_utils.py @kmvanbrunt +cmd2/transcript.py @kotfu +cmd2/utils.py @tleonhardt @kotfu @kmvanbrunt + +# Sphinx documentation +docs/* @tleonhardt @kotfu + +# Examples +examples/env*.py @kotfu +examples/help*.py @anselor +examples/tab_au*.py @anselor +examples/tab_co*.py @kmvanbrunt + +# Unit Tests +tests/pyscript/* @anselor +tests/transcripts/* @kotfu +tests/__init__.py @kotfu +tests/conftest.py @kotfu @tleonhardt +tests/test_acar*.py @anselor +tests/test_argp*.py @kotfu +tests/test_auto*.py @anselor +tests/test_bash*.py @anselor @tleonhardt +tests/test_comp*.py @kmvanbrunt +tests/test_pars*.py @kotfu +tests/test_pysc*.py @anselor +tests/test_tran*.py @kotfu + +# Top-level project stuff +CONTRIBUTING.md @tleonhardt @kotfu +setup.py @tleonhardt @kotfu +tasks.py @kotfu + diff --git a/cmd2/__init__.py b/cmd2/__init__.py index 617d643b..e9a82acb 100644 --- a/cmd2/__init__.py +++ b/cmd2/__init__.py @@ -1,4 +1,5 @@ # # -*- coding: utf-8 -*- +"""This simply imports certain things for backwards compatibility.""" from .cmd2 import __version__, Cmd, CmdResult, Statement, EmptyStatement, categorize from .cmd2 import with_argument_list, with_argparser, with_argparser_and_unknown_args, with_category diff --git a/cmd2/argcomplete_bridge.py b/cmd2/argcomplete_bridge.py index 824710b0..7bdb816f 100644 --- a/cmd2/argcomplete_bridge.py +++ b/cmd2/argcomplete_bridge.py @@ -106,18 +106,18 @@ else: class CompletionFinder(argcomplete.CompletionFinder): """Hijack the functor from argcomplete to call AutoCompleter""" - def __call__(self, argument_parser, completer=None, always_complete_options=True, exit_method=os._exit, output_stream=None, - exclude=None, validator=None, print_suppressed=False, append_space=None, + def __call__(self, argument_parser, completer=None, always_complete_options=True, exit_method=os._exit, + output_stream=None, exclude=None, validator=None, print_suppressed=False, append_space=None, default_completer=DEFAULT_COMPLETER): """ :param argument_parser: The argument parser to autocomplete on :type argument_parser: :class:`argparse.ArgumentParser` :param always_complete_options: - Controls the autocompletion of option strings if an option string opening character (normally ``-``) has not - been entered. If ``True`` (default), both short (``-x``) and long (``--x``) option strings will be - suggested. If ``False``, no option strings will be suggested. If ``long``, long options and short options - with no long variant will be suggested. If ``short``, short options and long options with no short variant - will be suggested. + Controls the autocompletion of option strings if an option string opening character (normally ``-``) has + not been entered. If ``True`` (default), both short (``-x``) and long (``--x``) option strings will be + suggested. If ``False``, no option strings will be suggested. If ``long``, long options and short + options with no long variant will be suggested. If ``short``, short options and long options with no + short variant will be suggested. :type always_complete_options: boolean or string :param exit_method: Method used to stop the program after printing completions. Defaults to :meth:`os._exit`. If you want to @@ -126,8 +126,8 @@ else: :param exclude: List of strings representing options to be omitted from autocompletion :type exclude: iterable :param validator: - Function to filter all completions through before returning (called with two string arguments, completion - and prefix; return value is evaluated as a boolean) + Function to filter all completions through before returning (called with two string arguments, + completion and prefix; return value is evaluated as a boolean) :type validator: callable :param print_suppressed: Whether or not to autocomplete options that have the ``help=argparse.SUPPRESS`` keyword argument set. @@ -142,18 +142,18 @@ else: Produces tab completions for ``argument_parser``. See module docs for more info. - Argcomplete only executes actions if their class is known not to have side effects. Custom action classes can be - added to argcomplete.safe_actions, if their values are wanted in the ``parsed_args`` completer argument, or - their execution is otherwise desirable. + Argcomplete only executes actions if their class is known not to have side effects. Custom action classes + can be added to argcomplete.safe_actions, if their values are wanted in the ``parsed_args`` completer + argument, or their execution is otherwise desirable. """ # Older versions of argcomplete have fewer keyword arguments if sys.version_info >= (3, 5): self.__init__(argument_parser, always_complete_options=always_complete_options, exclude=exclude, - validator=validator, print_suppressed=print_suppressed, append_space=append_space, - default_completer=default_completer) + validator=validator, print_suppressed=print_suppressed, append_space=append_space, + default_completer=default_completer) else: self.__init__(argument_parser, always_complete_options=always_complete_options, exclude=exclude, - validator=validator, print_suppressed=print_suppressed) + validator=validator, print_suppressed=print_suppressed) if "_ARGCOMPLETE" not in os.environ: # not an argument completion invocation @@ -171,10 +171,6 @@ else: argcomplete.debug("Unable to open fd 8 for writing, quitting") exit_method(1) - # print("", stream=debug_stream) - # for v in "COMP_CWORD COMP_LINE COMP_POINT COMP_TYPE COMP_KEY _ARGCOMPLETE_COMP_WORDBREAKS COMP_WORDS".split(): - # print(v, os.environ[v], stream=debug_stream) - ifs = os.environ.get("_ARGCOMPLETE_IFS", "\013") if len(ifs) != 1: argcomplete.debug("Invalid value for IFS, quitting [{v}]".format(v=ifs)) @@ -190,8 +186,6 @@ else: # # Replaced with our own tokenizer function ############################## - - # cword_prequote, cword_prefix, cword_suffix, comp_words, last_wordbreak_pos = split_line(comp_line, comp_point) tokens, _, begidx, endidx = tokens_for_completion(comp_line, comp_point) # _ARGCOMPLETE is set by the shell script to tell us where comp_words @@ -259,9 +253,13 @@ else: exit_method(0) - def bash_complete(action, show_hint: bool=True): - """Helper function to configure an argparse action to fall back to bash completion""" + def bash_complete(action, show_hint: bool = True): + """Helper function to configure an argparse action to fall back to bash completion. + + This function tags a parameter for bash completion, bypassing the autocompleter (for file input). + """ def complete_none(*args, **kwargs): return None + setattr(action, ACTION_SUPPRESS_HINT, not show_hint) setattr(action, ACTION_ARG_CHOICES, (complete_none,)) diff --git a/cmd2/argparse_completer.py b/cmd2/argparse_completer.py index d98a6eac..995aeb48 100755 --- a/cmd2/argparse_completer.py +++ b/cmd2/argparse_completer.py @@ -59,6 +59,7 @@ Released under MIT license, see LICENSE file import argparse from colorama import Fore +import os import sys from typing import List, Dict, Tuple, Callable, Union @@ -76,10 +77,20 @@ from .rl_utils import rl_force_redisplay # define the completion choices for the argument. You may provide a Collection or a Function. ACTION_ARG_CHOICES = 'arg_choices' ACTION_SUPPRESS_HINT = 'suppress_hint' +ACTION_DESCRIPTIVE_COMPLETION_HEADER = 'desc_header' + + +class CompletionItem(str): + def __new__(cls, o, desc='', *args, **kwargs) -> str: + return str.__new__(cls, o, *args, **kwargs) + + # noinspection PyMissingConstructor,PyUnusedLocal + def __init__(self, o, desc='', *args, **kwargs) -> None: + self.description = desc class _RangeAction(object): - def __init__(self, nargs: Union[int, str, Tuple[int, int], None]): + def __init__(self, nargs: Union[int, str, Tuple[int, int], None]) -> None: self.nargs_min = None self.nargs_max = None @@ -118,7 +129,7 @@ class _StoreRangeAction(argparse._StoreAction, _RangeAction): choices=None, required=False, help=None, - metavar=None): + metavar=None) -> None: _RangeAction.__init__(self, nargs) @@ -147,7 +158,7 @@ class _AppendRangeAction(argparse._AppendAction, _RangeAction): choices=None, required=False, help=None, - metavar=None): + metavar=None) -> None: _RangeAction.__init__(self, nargs) @@ -164,7 +175,7 @@ class _AppendRangeAction(argparse._AppendAction, _RangeAction): metavar=metavar) -def register_custom_actions(parser: argparse.ArgumentParser): +def register_custom_actions(parser: argparse.ArgumentParser) -> None: """Register custom argument action types""" parser.register('action', None, _StoreRangeAction) parser.register('action', 'store', _StoreRangeAction) @@ -175,14 +186,14 @@ class AutoCompleter(object): """Automatically command line tab completion based on argparse parameters""" class _ArgumentState(object): - def __init__(self): + def __init__(self) -> None: self.min = None self.max = None self.count = 0 self.needed = False self.variable = False - def reset(self): + def reset(self) -> None: """reset tracking values""" self.min = None self.max = None @@ -196,7 +207,7 @@ class AutoCompleter(object): arg_choices: Dict[str, Union[List, Tuple, Callable]] = None, subcmd_args_lookup: dict = None, tab_for_arg_help: bool = True, - cmd2_app=None): + cmd2_app=None) -> None: """ Create an AutoCompleter @@ -413,6 +424,8 @@ class AutoCompleter(object): completion_results = self._complete_for_arg(flag_action, text, line, begidx, endidx, consumed) if not completion_results: self._print_action_help(flag_action) + elif len(completion_results) > 1: + completion_results = self._format_completions(flag_action, completion_results) # ok, we're not a flag, see if there's a positional argument to complete else: @@ -422,9 +435,39 @@ class AutoCompleter(object): completion_results = self._complete_for_arg(pos_action, text, line, begidx, endidx, consumed) if not completion_results: self._print_action_help(pos_action) + elif len(completion_results) > 1: + completion_results = self._format_completions(pos_action, completion_results) return completion_results + def _format_completions(self, action, completions: List[Union[str, CompletionItem]]) -> List[str]: + if completions and len(completions) > 1 and isinstance(completions[0], CompletionItem): + token_width = len(action.dest) + completions_with_desc = [] + + for item in completions: + if len(item) > token_width: + token_width = len(item) + + term_size = os.get_terminal_size() + fill_width = int(term_size.columns * .6) - (token_width + 2) + for item in completions: + entry = '{: <{token_width}}{: <{fill_width}}'.format(item, item.description, + token_width=token_width+2, + fill_width=fill_width) + completions_with_desc.append(entry) + + try: + desc_header = action.desc_header + except AttributeError: + desc_header = 'Description' + header = '\n{: <{token_width}}{}'.format(action.dest.upper(), desc_header, token_width=token_width+2) + + self._cmd2_app.completion_header = header + self._cmd2_app.display_matches = completions_with_desc + + return completions + def complete_command_help(self, tokens: List[str], text: str, line: str, begidx: int, endidx: int) -> List[str]: """Supports the completion of sub-commands for commands through the cmd2 help command.""" for idx, token in enumerate(tokens): @@ -627,7 +670,7 @@ class AutoCompleter(object): class ACHelpFormatter(argparse.HelpFormatter): """Custom help formatter to configure ordering of help text""" - def _format_usage(self, usage, actions, groups, prefix): + def _format_usage(self, usage, actions, groups, prefix) -> str: if prefix is None: prefix = _('Usage: ') @@ -740,7 +783,7 @@ class ACHelpFormatter(argparse.HelpFormatter): # prefix with 'usage:' return '%s%s\n\n' % (prefix, usage) - def _format_action_invocation(self, action): + def _format_action_invocation(self, action) -> str: if not action.option_strings: default = self._get_default_metavar_for_positional(action) metavar, = self._metavar_formatter(action, default)(1) @@ -765,7 +808,7 @@ class ACHelpFormatter(argparse.HelpFormatter): return ', '.join(action.option_strings) + ' ' + args_string # End cmd2 customization - def _metavar_formatter(self, action, default_metavar): + def _metavar_formatter(self, action, default_metavar) -> Callable: if action.metavar is not None: result = action.metavar elif action.choices is not None: @@ -784,7 +827,7 @@ class ACHelpFormatter(argparse.HelpFormatter): return (result, ) * tuple_size return format - def _format_args(self, action, default_metavar): + def _format_args(self, action, default_metavar) -> str: get_metavar = self._metavar_formatter(action, default_metavar) # Begin cmd2 customization (less verbose) if isinstance(action, _RangeAction) and \ @@ -799,7 +842,7 @@ class ACHelpFormatter(argparse.HelpFormatter): result = super()._format_args(action, default_metavar) return result - def _split_lines(self, text, width): + def _split_lines(self, text: str, width) -> List[str]: return text.splitlines() @@ -807,7 +850,7 @@ class ACHelpFormatter(argparse.HelpFormatter): class ACArgumentParser(argparse.ArgumentParser): """Custom argparse class to override error method to change default help text.""" - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs) -> None: if 'formatter_class' not in kwargs: kwargs['formatter_class'] = ACHelpFormatter @@ -817,7 +860,7 @@ class ACArgumentParser(argparse.ArgumentParser): self._custom_error_message = '' # Begin cmd2 customization - def set_custom_message(self, custom_message=''): + def set_custom_message(self, custom_message: str='') -> None: """ Allows an error message override to the error() function, useful when forcing a re-parse of arguments with newly required parameters @@ -825,7 +868,7 @@ class ACArgumentParser(argparse.ArgumentParser): self._custom_error_message = custom_message # End cmd2 customization - def error(self, message): + def error(self, message: str) -> None: """Custom error override. Allows application to control the error being displayed by argparse""" if len(self._custom_error_message) > 0: message = self._custom_error_message @@ -846,7 +889,7 @@ class ACArgumentParser(argparse.ArgumentParser): self.print_help() sys.exit(1) - def format_help(self): + def format_help(self) -> str: """Copy of format_help() from argparse.ArgumentParser with tweaks to separately display required parameters""" formatter = self._get_formatter() @@ -896,7 +939,7 @@ class ACArgumentParser(argparse.ArgumentParser): # determine help from format above return formatter.format_help() - def _get_nargs_pattern(self, action): + def _get_nargs_pattern(self, action) -> str: # Override _get_nargs_pattern behavior to use the nargs ranges provided by AutoCompleter if isinstance(action, _RangeAction) and \ action.nargs_min is not None and action.nargs_max is not None: @@ -909,7 +952,7 @@ class ACArgumentParser(argparse.ArgumentParser): return nargs_pattern return super(ACArgumentParser, self)._get_nargs_pattern(action) - def _match_argument(self, action, arg_strings_pattern): + def _match_argument(self, action, arg_strings_pattern) -> int: # match the pattern for this action to the arg strings nargs_pattern = self._get_nargs_pattern(action) match = _re.match(nargs_pattern, arg_strings_pattern) @@ -925,7 +968,7 @@ class ACArgumentParser(argparse.ArgumentParser): # This is the official python implementation with a 5 year old patch applied # See the comment below describing the patch - def _parse_known_args(self, arg_strings, namespace): # pragma: no cover + def _parse_known_args(self, arg_strings, namespace) -> Tuple[argparse.Namespace, List[str]]: # pragma: no cover # replace arg strings that are file references if self.fromfile_prefix_chars is not None: arg_strings = self._read_args_from_files(arg_strings) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index 0179de2b..4b1e4afc 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -40,7 +40,7 @@ import platform import re import shlex import sys -from typing import Callable, List, Union, Tuple +from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple, Union import pyperclip @@ -51,7 +51,7 @@ from .parsing import StatementParser, Statement # Set up readline from .rl_utils import rl_type, RlType -if rl_type == RlType.NONE: +if rl_type == RlType.NONE: # pragma: no cover rl_warning = "Readline features including tab completion have been disabled since no \n" \ "supported version of readline was found. To resolve this, install \n" \ "pyreadline on Windows or gnureadline on Mac.\n\n" @@ -85,7 +85,7 @@ from .argparse_completer import AutoCompleter, ACArgumentParser # Newer versions of pyperclip are released as a single file, but older versions had a more complicated structure try: from pyperclip.exceptions import PyperclipException -except ImportError: +except ImportError: # pragma: no cover # noinspection PyUnresolvedReferences from pyperclip import PyperclipException @@ -122,7 +122,7 @@ ipython_available = True try: # noinspection PyUnresolvedReferences,PyPackageRequirements from IPython import embed -except ImportError: +except ImportError: # pragma: no cover ipython_available = False __version__ = '0.9.2a' @@ -147,6 +147,7 @@ def categorize(func: Union[Callable, Iterable], category: str) -> None: else: setattr(func, HELP_CATEGORY, category) + def parse_quoted_string(cmdline: str) -> List[str]: """Parse a quoted string into a list of arguments.""" if isinstance(cmdline, list): @@ -323,6 +324,30 @@ class EmptyStatement(Exception): pass +class HistoryItem(str): + """Class used to represent an item in the History list. + + Thin wrapper around str class which adds a custom format for printing. It + also keeps track of its index in the list as well as a lowercase + representation of itself for convenience/efficiency. + + """ + listformat = '-------------------------[{}]\n{}\n' + + # noinspection PyUnusedLocal + def __init__(self, instr: str) -> None: + str.__init__(self) + self.lowercase = self.lower() + self.idx = None + + def pr(self) -> str: + """Represent a HistoryItem in a pretty fashion suitable for printing. + + :return: pretty print string version of a HistoryItem + """ + return self.listformat.format(self.idx, str(self).rstrip()) + + class Cmd(cmd.Cmd): """An easy but powerful framework for writing line-oriented command interpreters. @@ -377,18 +402,19 @@ class Cmd(cmd.Cmd): 'quiet': "Don't print nonessential feedback", 'timing': 'Report execution times'} - def __init__(self, completekey='tab', stdin=None, stdout=None, persistent_history_file='', - persistent_history_length=1000, startup_script=None, use_ipython=False, transcript_files=None): + def __init__(self, completekey: str='tab', stdin=None, stdout=None, persistent_history_file: str='', + persistent_history_length: int=1000, startup_script: Optional[str]=None, use_ipython: bool=False, + transcript_files: Optional[List[str]]=None) -> None: """An easy but powerful framework for writing line-oriented command interpreters, extends Python's cmd package. - :param completekey: str - (optional) readline name of a completion key, default to Tab + :param completekey: (optional) readline name of a completion key, default to Tab :param stdin: (optional) alternate input file object, if not specified, sys.stdin is used :param stdout: (optional) alternate output file object, if not specified, sys.stdout is used - :param persistent_history_file: str - (optional) file path to load a persistent readline history from - :param persistent_history_length: int - (optional) max number of lines which will be written to the history file - :param startup_script: str - (optional) file path to a a script to load and execute at startup + :param persistent_history_file: (optional) file path to load a persistent readline history from + :param persistent_history_length: (optional) max number of lines which will be written to the history file + :param startup_script: (optional) file path to a a script to load and execute at startup :param use_ipython: (optional) should the "ipy" command be included for an embedded IPython shell - :param transcript_files: str - (optional) allows running transcript tests when allow_cli_args is False + :param transcript_files: (optional) allows running transcript tests when allow_cli_args is False """ # If use_ipython is False, make sure the do_ipy() method doesn't exit if not use_ipython: @@ -499,6 +525,9 @@ class Cmd(cmd.Cmd): # will be added if there is an unmatched opening quote self.allow_closing_quote = True + # An optional header that prints above the tab-completion suggestions + self.completion_header = '' + # Use this list if you are completing strings that contain a common delimiter and you only want to # display the final portion of the matches as the tab-completion suggestions. The full matches # still must be returned from your completer function. For an example, look at path_complete() @@ -506,34 +535,51 @@ class Cmd(cmd.Cmd): # populates this list. self.display_matches = [] + # Used by functions like path_complete() and delimiter_complete() to properly + # quote matches that are completed in a delimited fashion + self.matches_delimited = False + + # Set the pager(s) for use with the ppaged() method for displaying output using a pager + if sys.platform.startswith('win'): + self.pager = self.pager_chop = 'more' + else: + # Here is the meaning of the various flags we are using with the less command: + # -S causes lines longer than the screen width to be chopped (truncated) rather than wrapped + # -R causes ANSI "color" escape sequences to be output in raw form (i.e. colors are displayed) + # -X disables sending the termcap initialization and deinitialization strings to the terminal + # -F causes less to automatically exit if the entire file can be displayed on the first screen + self.pager = 'less -RXF' + self.pager_chop = 'less -SRXF' + # ----- Methods related to presenting output to the user ----- @property - def visible_prompt(self): + def visible_prompt(self) -> str: """Read-only property to get the visible prompt with any ANSI escape codes stripped. Used by transcript testing to make it easier and more reliable when users are doing things like coloring the prompt using ANSI color codes. - :return: str - prompt stripped of any ANSI escape codes + :return: prompt stripped of any ANSI escape codes """ return utils.strip_ansi(self.prompt) - def _finalize_app_parameters(self): + def _finalize_app_parameters(self) -> None: + """Finalize the shortcuts and settable parameters.""" # noinspection PyUnresolvedReferences self.shortcuts = sorted(self.shortcuts.items(), reverse=True) # Make sure settable parameters are sorted alphabetically by key self.settable = collections.OrderedDict(sorted(self.settable.items(), key=lambda t: t[0])) - def poutput(self, msg, end='\n'): + def poutput(self, msg: str, end: str='\n') -> None: """Convenient shortcut for self.stdout.write(); by default adds newline to end if not already present. Also handles BrokenPipeError exceptions for when a commands's output has been piped to another process and that process terminates before the cmd2 command is finished executing. - :param msg: str - message to print to current stdout - anything convertible to a str with '{}'.format() is OK - :param end: str - string appended after the end of the message if not already present, default a newline + :param msg: message to print to current stdout - anything convertible to a str with '{}'.format() is OK + :param end: string appended after the end of the message if not already present, default a newline """ if msg is not None and msg != '': try: @@ -548,30 +594,29 @@ class Cmd(cmd.Cmd): if self.broken_pipe_warning: sys.stderr.write(self.broken_pipe_warning) - def perror(self, errmsg, exception_type=None, traceback_war=True): + def perror(self, err: Union[str, Exception], traceback_war: bool=True) -> None: """ Print error message to sys.stderr and if debug is true, print an exception Traceback if one exists. - :param errmsg: str - error message to print out - :param exception_type: str - (optional) type of exception which precipitated this error message - :param traceback_war: bool - (optional) if True, print a message to let user know they can enable debug + :param err: an Exception or error message to print out + :param traceback_war: (optional) if True, print a message to let user know they can enable debug :return: """ if self.debug: import traceback traceback.print_exc() - if exception_type is None: - err = self.colorize("ERROR: {}\n".format(errmsg), 'red') - sys.stderr.write(err) + if isinstance(err, Exception): + err_msg = "EXCEPTION of type '{}' occurred with message: '{}'\n".format(type(err).__name__, err) + sys.stderr.write(self.colorize(err_msg, 'red')) else: - err = "EXCEPTION of type '{}' occurred with message: '{}'\n".format(exception_type, errmsg) - sys.stderr.write(self.colorize(err, 'red')) + err_msg = self.colorize("ERROR: {}\n".format(err), 'red') + sys.stderr.write(err_msg) if traceback_war: war = "To enable full traceback, run the following command: 'set debug true'\n" sys.stderr.write(self.colorize(war, 'yellow')) - def pfeedback(self, msg): + def pfeedback(self, msg: str) -> None: """For printing nonessential feedback. Can be silenced with `quiet`. Inclusion in redirected output is controlled by `feedback_to_output`.""" if not self.quiet: @@ -580,14 +625,20 @@ class Cmd(cmd.Cmd): else: sys.stderr.write("{}\n".format(msg)) - def ppaged(self, msg, end='\n'): + def ppaged(self, msg: str, end: str='\n', chop: bool=False) -> None: """Print output using a pager if it would go off screen and stdout isn't currently being redirected. Never uses a pager inside of a script (Python or text) or when output is being redirected or piped or when stdout or stdin are not a fully functional terminal. - :param msg: str - message to print to current stdout - anything convertible to a str with '{}'.format() is OK - :param end: str - string appended after the end of the message if not already present, default a newline + :param msg: message to print to current stdout - anything convertible to a str with '{}'.format() is OK + :param end: string appended after the end of the message if not already present, default a newline + :param chop: True -> causes lines longer than the screen width to be chopped (truncated) rather than wrapped + - truncated text is still accessible by scrolling with the right & left arrow keys + - chopping is ideal for displaying wide tabular data as is done in utilities like pgcli + False -> causes lines longer than the screen width to wrap to the next line + - wrapping is ideal when you want to avoid users having to use horizontal scrolling + WARNING: On Windows, the text always wraps regardless of what the chop argument is set to """ import subprocess if msg is not None and msg != '': @@ -607,17 +658,10 @@ class Cmd(cmd.Cmd): # Don't attempt to use a pager that can block if redirecting or running a script (either text or Python) # Also only attempt to use a pager if actually running in a real fully functional terminal if functional_terminal and not self.redirecting and not self._in_py and not self._script_dir: - - if sys.platform.startswith('win'): - pager_cmd = 'more' - else: - # Here is the meaning of the various flags we are using with the less command: - # -S causes lines longer than the screen width to be chopped (truncated) rather than wrapped - # -R causes ANSI "color" escape sequences to be output in raw form (i.e. colors are displayed) - # -X disables sending the termcap initialization and deinitialization strings to the terminal - # -F causes less to automatically exit if the entire file can be displayed on the first screen - pager_cmd = 'less -SRXF' - self.pipe_proc = subprocess.Popen(pager_cmd, shell=True, stdin=subprocess.PIPE) + pager = self.pager + if chop: + pager = self.pager_chop + self.pipe_proc = subprocess.Popen(pager, shell=True, stdin=subprocess.PIPE) try: self.pipe_proc.stdin.write(msg_str.encode('utf-8', 'replace')) self.pipe_proc.stdin.close() @@ -642,7 +686,7 @@ class Cmd(cmd.Cmd): if self.broken_pipe_warning: sys.stderr.write(self.broken_pipe_warning) - def colorize(self, val, color): + def colorize(self, val: str, color: str) -> str: """Given a string (``val``), returns that string wrapped in UNIX-style special characters that turn on (and then off) text color and style. If the ``colors`` environment parameter is ``False``, or the application @@ -655,26 +699,29 @@ class Cmd(cmd.Cmd): # ----- Methods related to tab completion ----- - def reset_completion_defaults(self): + def reset_completion_defaults(self) -> None: """ Resets tab completion settings Needs to be called each time readline runs tab completion """ self.allow_appended_space = True self.allow_closing_quote = True + self.completion_header = '' self.display_matches = [] + self.matches_delimited = False if rl_type == RlType.GNU: readline.set_completion_display_matches_hook(self._display_matches_gnu_readline) elif rl_type == RlType.PYREADLINE: readline.rl.mode._display_completions = self._display_matches_pyreadline - def tokens_for_completion(self, line, begidx, endidx): + def tokens_for_completion(self, line: str, begidx: int, endidx: int) -> Tuple[Optional[List[str]], + Optional[List[str]]]: """ Used by tab completion functions to get all tokens through the one being completed - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text :return: A 2 item tuple where the items are On Success tokens: list of unquoted tokens @@ -791,20 +838,21 @@ class Cmd(cmd.Cmd): # noinspection PyUnusedLocal @staticmethod - def basic_complete(text, line, begidx, endidx, match_against): + def basic_complete(text: str, line: str, begidx: int, endidx: int, match_against: Iterable) -> List[str]: """ Performs tab completion against a list - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text - :param match_against: Collection - the list being matched against - :return: List[str] - a list of possible tab completions + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text + :param match_against: the list being matched against + :return: a list of possible tab completions """ return [cur_match for cur_match in match_against if cur_match.startswith(text)] - def delimiter_complete(self, text, line, begidx, endidx, match_against, delimiter): + def delimiter_complete(self, text: str, line: str, begidx: int, endidx: int, match_against: Iterable, + delimiter: str) -> List[str]: """ Performs tab completion against a list but each match is split on a delimiter and only the portion of the match being tab completed is shown as the completion suggestions. @@ -829,18 +877,20 @@ class Cmd(cmd.Cmd): In this case the delimiter would be :: and the user could easily narrow down what they are looking for if they were only shown suggestions in the category they are at in the string. - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text - :param match_against: Collection - the list being matched against - :param delimiter: str - what delimits each portion of the matches (ex: paths are delimited by a slash) - :return: List[str] - a list of possible tab completions + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text + :param match_against: the list being matched against + :param delimiter: what delimits each portion of the matches (ex: paths are delimited by a slash) + :return: a list of possible tab completions """ matches = self.basic_complete(text, line, begidx, endidx, match_against) # Display only the portion of the match that's being completed based on delimiter if matches: + # Set this to True for proper quoting of matches with spaces + self.matches_delimited = True # Get the common beginning for the matches common_prefix = os.path.commonprefix(matches) @@ -862,13 +912,15 @@ class Cmd(cmd.Cmd): return matches - def flag_based_complete(self, text, line, begidx, endidx, flag_dict, all_else=None): + def flag_based_complete(self, text: str, line: str, begidx: int, endidx: int, + flag_dict: Dict[str, Union[Iterable, Callable]], + all_else: Union[None, Iterable, Callable]=None) -> List[str]: """ Tab completes based on a particular flag preceding the token being completed - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text :param flag_dict: dict - dictionary whose structure is the following: keys - flags (ex: -c, --create) that result in tab completion for the next argument in the command line @@ -877,7 +929,7 @@ class Cmd(cmd.Cmd): 2. function that performs tab completion (ex: path_complete) :param all_else: Collection or function - an optional parameter for tab completing any token that isn't preceded by a flag in flag_dict - :return: List[str] - a list of possible tab completions + :return: a list of possible tab completions """ # Get all tokens through the one being completed tokens, _ = self.tokens_for_completion(line, begidx, endidx) @@ -903,13 +955,15 @@ class Cmd(cmd.Cmd): return completions_matches - def index_based_complete(self, text, line, begidx, endidx, index_dict, all_else=None): + def index_based_complete(self, text: str, line: str, begidx: int, endidx: int, + index_dict: Mapping[int, Union[Iterable, Callable]], + all_else: Union[None, Iterable, Callable] = None) -> List[str]: """ Tab completes based on a fixed position in the input string - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text :param index_dict: dict - dictionary whose structure is the following: keys - 0-based token indexes into command line that determine which tokens perform tab completion @@ -918,7 +972,7 @@ class Cmd(cmd.Cmd): 2. function that performs tab completion (ex: path_complete) :param all_else: Collection or function - an optional parameter for tab completing any token that isn't at an index in index_dict - :return: List[str] - a list of possible tab completions + :return: a list of possible tab completions """ # Get all tokens through the one being completed tokens, _ = self.tokens_for_completion(line, begidx, endidx) @@ -947,16 +1001,17 @@ class Cmd(cmd.Cmd): return matches # noinspection PyUnusedLocal - def path_complete(self, text, line, begidx, endidx, dir_exe_only=False, dir_only=False): + def path_complete(self, text: str, line: str, begidx: int, endidx: int, dir_exe_only: bool=False, + dir_only: bool=False) -> List[str]: """Performs completion of local file system paths - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text - :param dir_exe_only: bool - only return directories and executables, not non-executable files - :param dir_only: bool - only return directories - :return: List[str] - a list of possible tab completions + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text + :param dir_exe_only: only return directories and executables, not non-executable files + :param dir_only: only return directories + :return: a list of possible tab completions """ # Used to complete ~ and ~user strings @@ -1042,6 +1097,9 @@ class Cmd(cmd.Cmd): search_str = os.path.join(os.getcwd(), search_str) cwd_added = True + # Set this to True for proper quoting of paths with spaces + self.matches_delimited = True + # Find all matching path completions matches = glob.glob(search_str) @@ -1078,11 +1136,11 @@ class Cmd(cmd.Cmd): return matches @staticmethod - def get_exes_in_path(starts_with): - """ - Returns names of executables in a user's path - :param starts_with: str - what the exes should start with. leave blank for all exes in path. - :return: List[str] - a list of matching exe names + def get_exes_in_path(starts_with: str) -> List[str]: + """Returns names of executables in a user's path + + :param starts_with: what the exes should start with. leave blank for all exes in path. + :return: a list of matching exe names """ # Purposely don't match any executable containing wildcards wildcards = ['*', '?'] @@ -1106,16 +1164,17 @@ class Cmd(cmd.Cmd): return list(exes_set) - def shell_cmd_complete(self, text, line, begidx, endidx, complete_blank=False): + def shell_cmd_complete(self, text: str, line: str, begidx: int, endidx: int, + complete_blank: bool=False) -> List[str]: """Performs completion of executables either in a user's path or a given path - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text - :param complete_blank: bool - If True, then a blank will complete all shell commands in a user's path - If False, then no completion is performed - Defaults to False to match Bash shell behavior - :return: List[str] - a list of possible tab completions + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text + :param complete_blank: If True, then a blank will complete all shell commands in a user's path + If False, then no completion is performed + Defaults to False to match Bash shell behavior + :return: a list of possible tab completions """ # Don't tab complete anything if no shell command has been started if not complete_blank and not text: @@ -1129,19 +1188,18 @@ class Cmd(cmd.Cmd): else: return self.path_complete(text, line, begidx, endidx, dir_exe_only=True) - def _redirect_complete(self, text, line, begidx, endidx, compfunc): - """ - Called by complete() as the first tab completion function for all commands + def _redirect_complete(self, text: str, line: str, begidx: int, endidx: int, compfunc: Callable) -> List[str]: + """Called by complete() as the first tab completion function for all commands It determines if it should tab complete for redirection (|, <, >, >>) or use the completer function for the current command - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text - :param compfunc: Callable - the completer function for the current command - this will be called if we aren't completing for redirection - :return: List[str] - a list of possible tab completions + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text + :param compfunc: the completer function for the current command + this will be called if we aren't completing for redirection + :return: a list of possible tab completions """ if self.allow_redirection: @@ -1184,9 +1242,8 @@ class Cmd(cmd.Cmd): return compfunc(text, line, begidx, endidx) @staticmethod - def _pad_matches_to_display(matches_to_display): # pragma: no cover - """ - Adds padding to the matches being displayed as tab completion suggestions. + def _pad_matches_to_display(matches_to_display: List[str]) -> Tuple[List[str], int]: # pragma: no cover + """Adds padding to the matches being displayed as tab completion suggestions. The default padding of readline/pyreadine is small and not visually appealing especially if matches have spaces. It appears very squished together. @@ -1206,14 +1263,14 @@ class Cmd(cmd.Cmd): return [cur_match + padding for cur_match in matches_to_display], len(padding) - def _display_matches_gnu_readline(self, substitution, matches, longest_match_length): # pragma: no cover - """ - Prints a match list using GNU readline's rl_display_match_list() + def _display_matches_gnu_readline(self, substitution: str, matches: List[str], + longest_match_length: int) -> None: # pragma: no cover + """Prints a match list using GNU readline's rl_display_match_list() This exists to print self.display_matches if it has data. Otherwise matches prints. - :param substitution: str - the substitution written to the command line - :param matches: list[str] - the tab completion matches to display - :param longest_match_length: int - longest printed length of the matches + :param substitution: the substitution written to the command line + :param matches: the tab completion matches to display + :param longest_match_length: longest printed length of the matches """ if rl_type == RlType.GNU: @@ -1250,6 +1307,10 @@ class Cmd(cmd.Cmd): strings_array[1:-1] = encoded_matches strings_array[-1] = None + # Print the header if one exists + if self.completion_header: + sys.stdout.write('\n' + self.completion_header) + # Call readline's display function # rl_display_match_list(strings_array, number of completion matches, longest match length) readline_lib.rl_display_match_list(strings_array, len(encoded_matches), longest_match_length) @@ -1257,12 +1318,11 @@ class Cmd(cmd.Cmd): # Redraw prompt and input line rl_force_redisplay() - def _display_matches_pyreadline(self, matches): # pragma: no cover - """ - Prints a match list using pyreadline's _display_completions() + def _display_matches_pyreadline(self, matches: List[str]) -> None: # pragma: no cover + """Prints a match list using pyreadline's _display_completions() This exists to print self.display_matches if it has data. Otherwise matches prints. - :param matches: list[str] - the tab completion matches to display + :param matches: the tab completion matches to display """ if rl_type == RlType.PYREADLINE: @@ -1275,12 +1335,16 @@ class Cmd(cmd.Cmd): # Add padding for visual appeal matches_to_display, _ = self._pad_matches_to_display(matches_to_display) + # Print the header if one exists + if self.completion_header: + readline.rl.mode.console.write('\n' + self.completion_header) + # Display matches using actual display function. This also redraws the prompt and line. orig_pyreadline_display(matches_to_display) # ----- Methods which override stuff in cmd ----- - def complete(self, text, state): + def complete(self, text: str, state: int) -> Optional[str]: """Override of command method which returns the next possible completion for 'text'. If a command has not been entered, then complete against command list. @@ -1291,8 +1355,8 @@ class Cmd(cmd.Cmd): This completer function is called as complete(text, state), for state in 0, 1, 2, …, until it returns a non-string value. It should return the next possible completion starting with text. - :param text: str - the current word that user is typing - :param state: int - non-negative integer + :param text: the current word that user is typing + :param state: non-negative integer """ import functools if state == 0 and rl_type != RlType.NONE: @@ -1314,7 +1378,7 @@ class Cmd(cmd.Cmd): # from text and update the indexes. This only applies if we are at the the beginning of the line. shortcut_to_restore = '' if begidx == 0: - for (shortcut, expansion) in self.shortcuts: + for (shortcut, _) in self.shortcuts: if text.startswith(shortcut): # Save the shortcut to restore later shortcut_to_restore = shortcut @@ -1419,13 +1483,7 @@ class Cmd(cmd.Cmd): display_matches_set = set(self.display_matches) self.display_matches = list(display_matches_set) - # Check if display_matches has been used. If so, then matches - # on delimited strings like paths was done. - if self.display_matches: - matches_delimited = True - else: - matches_delimited = False - + if not self.display_matches: # Since self.display_matches is empty, set it to self.completion_matches # before we alter them. That way the suggestions will reflect how we parsed # the token being completed and not how readline did. @@ -1440,7 +1498,7 @@ class Cmd(cmd.Cmd): # This is the tab completion text that will appear on the command line. common_prefix = os.path.commonprefix(self.completion_matches) - if matches_delimited: + if self.matches_delimited: # Check if any portion of the display matches appears in the tab completion display_prefix = os.path.commonprefix(self.display_matches) @@ -1514,16 +1572,12 @@ class Cmd(cmd.Cmd): return results - def get_all_commands(self): - """ - Returns a list of all commands - """ + def get_all_commands(self) -> List[str]: + """Returns a list of all commands.""" return [cur_name[3:] for cur_name in self.get_names() if cur_name.startswith('do_')] - def get_visible_commands(self): - """ - Returns a list of commands that have not been hidden - """ + def get_visible_commands(self) -> List[str]: + """Returns a list of commands that have not been hidden.""" commands = self.get_all_commands() # Remove the hidden commands @@ -1533,11 +1587,11 @@ class Cmd(cmd.Cmd): return commands - def get_help_topics(self): + def get_help_topics(self) -> List[str]: """ Returns a list of help topics """ return [name[5:] for name in self.get_names() if name.startswith('help_')] - def complete_help(self, text, line, begidx, endidx): + def complete_help(self, text: str, line: str, begidx: int, endidx: int) -> List[str]: """ Override of parent class method to handle tab completing subcommands and not showing hidden commands Returns a list of possible tab completions @@ -1581,12 +1635,12 @@ class Cmd(cmd.Cmd): return matches # noinspection PyUnusedLocal - def sigint_handler(self, signum, frame): + def sigint_handler(self, signum: int, frame) -> None: """Signal handler for SIGINTs which typically come from Ctrl-C events. If you need custom SIGINT behavior, then override this function. - :param signum: int - signal number + :param signum: signal number :param frame """ @@ -1599,8 +1653,8 @@ class Cmd(cmd.Cmd): # Re-raise a KeyboardInterrupt so other parts of the code can catch it raise KeyboardInterrupt("Got a keyboard interrupt") - def preloop(self): - """"Hook method executed once when the cmdloop() method is called.""" + def preloop(self) -> None: + """Hook method executed once when the cmdloop() method is called.""" import signal # Register a default SIGINT signal handler for Ctrl+C signal.signal(signal.SIGINT, self.sigint_handler) @@ -1608,8 +1662,8 @@ class Cmd(cmd.Cmd): def precmd(self, statement: Statement) -> Statement: """Hook method executed just before the command is processed by ``onecmd()`` and after adding it to the history. - :param statement: Statement - subclass of str which also contains the parsed input - :return: Statement - a potentially modified version of the input Statement object + :param statement: subclass of str which also contains the parsed input + :return: a potentially modified version of the input Statement object """ return statement @@ -1619,8 +1673,8 @@ class Cmd(cmd.Cmd): def preparse(self, raw: str) -> str: """Hook method executed just before the command line is interpreted, but after the input prompt is generated. - :param raw: str - raw command line input - :return: str - potentially modified raw command line input + :param raw: raw command line input + :return: potentially modified raw command line input """ return raw @@ -1661,25 +1715,24 @@ class Cmd(cmd.Cmd): proc.communicate() return stop - def parseline(self, line): + def parseline(self, line: str) -> Tuple[str, str, str]: """Parse the line into a command name and a string containing the arguments. NOTE: This is an override of a parent class method. It is only used by other parent class methods. Different from the parent class method, this ignores self.identchars. - :param line: str - line read by readline - :return: (str, str, str) - tuple containing (command, args, line) + :param line: line read by readline + :return: tuple containing (command, args, line) """ - statement = self.statement_parser.parse_command_only(line) return statement.command, statement.args, statement.command_and_args def onecmd_plus_hooks(self, line: str) -> bool: """Top-level function called by cmdloop() to handle parsing a line and running the command and all of its hooks. - :param line: str - line of text read from input - :return: bool - True if cmdloop() should exit, False otherwise + :param line: line of text read from input + :return: True if cmdloop() should exit, False otherwise """ import datetime stop = False @@ -1723,7 +1776,7 @@ class Cmd(cmd.Cmd): if self.timing: self.pfeedback('Elapsed: %s' % str(datetime.datetime.now() - timestart)) finally: - if self.allow_redirection: + if self.allow_redirection and self.redirecting: self._restore_output(statement) except EmptyStatement: pass @@ -1731,11 +1784,11 @@ class Cmd(cmd.Cmd): # If shlex.split failed on syntax, let user know whats going on self.perror("Invalid syntax: {}".format(ex), traceback_war=False) except Exception as ex: - self.perror(ex, type(ex).__name__) + self.perror(ex) finally: return self.postparsing_postcmd(stop) - def runcmds_plus_hooks(self, cmds): + def runcmds_plus_hooks(self, cmds: List[str]) -> bool: """Convenience method to run multiple commands by onecmd_plus_hooks. This method adds the given cmds to the command queue and processes the @@ -1753,8 +1806,8 @@ class Cmd(cmd.Cmd): Example: cmd_obj.runcmds_plus_hooks(['load myscript.txt']) - :param cmds: list - Command strings suitable for onecmd_plus_hooks. - :return: bool - True implies the entire application should exit. + :param cmds: command strings suitable for onecmd_plus_hooks. + :return: True implies the entire application should exit. """ stop = False @@ -1777,7 +1830,7 @@ class Cmd(cmd.Cmd): # necessary/desired here. return stop - def _complete_statement(self, line): + def _complete_statement(self, line: str) -> Statement: """Keep accepting lines of input until the command is complete. There is some pretty hacky code here to handle some quirks of @@ -1818,10 +1871,10 @@ class Cmd(cmd.Cmd): raise EmptyStatement() return statement - def _redirect_output(self, statement): + def _redirect_output(self, statement: Statement) -> None: """Handles output redirection for >, >>, and |. - :param statement: Statement - a parsed statement from the user + :param statement: a parsed statement from the user """ import io import subprocess @@ -1867,19 +1920,22 @@ class Cmd(cmd.Cmd): # REDIRECTION_APPEND or REDIRECTION_OUTPUT if statement.output == constants.REDIRECTION_APPEND: mode = 'a' - sys.stdout = self.stdout = open(os.path.expanduser(statement.output_to), mode) + try: + sys.stdout = self.stdout = open(statement.output_to, mode) + except OSError as ex: + self.perror('Not Redirecting because - {}'.format(ex), traceback_war=False) + self.redirecting = False else: # going to a paste buffer sys.stdout = self.stdout = tempfile.TemporaryFile(mode="w+") if statement.output == constants.REDIRECTION_APPEND: self.poutput(get_paste_buffer()) - def _restore_output(self, statement): + def _restore_output(self, statement: Statement) -> None: """Handles restoring state after output redirection as well as the actual pipe operation if present. - :param statement: Statement object which contains the parsed - input from the user + :param statement: Statement object which contains the parsed input from the user """ # If we have redirected output to a file or the clipboard or piped it to a shell command, then restore state if self.kept_state is not None: @@ -1910,11 +1966,11 @@ class Cmd(cmd.Cmd): self.redirecting = False - def _func_named(self, arg): + def _func_named(self, arg: str) -> str: """Gets the method name associated with a given command. - :param arg: str - command to look up method name which implements it - :return: str - method name which implements the given command + :param arg: command to look up method name which implements it + :return: method name which implements the given command """ result = None target = 'do_' + arg @@ -1922,17 +1978,18 @@ class Cmd(cmd.Cmd): result = target return result - def onecmd(self, statement): + def onecmd(self, statement: Statement) -> Optional[bool]: """ This executes the actual do_* method for a command. If the command provided doesn't exist, then it executes _default() instead. :param statement: Command - a parsed command from the input stream - :return: bool - a flag indicating whether the interpretation of commands should stop + :return: a flag indicating whether the interpretation of commands should stop """ funcname = self._func_named(statement.command) if not funcname: - return self.default(statement) + self.default(statement) + return # Since we have a valid command store it in the history if statement.command not in self.exclude_from_history: @@ -1941,16 +1998,16 @@ class Cmd(cmd.Cmd): try: func = getattr(self, funcname) except AttributeError: - return self.default(statement) + self.default(statement) + return stop = func(statement) return stop - def default(self, statement): + def default(self, statement: Statement) -> None: """Executed when the command given isn't a recognized command implemented by a do_* method. :param statement: Statement object with parsed input - :return: """ arg = statement.raw if self.default_to_shell: @@ -1963,13 +2020,13 @@ class Cmd(cmd.Cmd): self.poutput('*** Unknown syntax: {}\n'.format(arg)) @staticmethod - def _surround_ansi_escapes(prompt, start="\x01", end="\x02"): + def _surround_ansi_escapes(prompt: str, start: str="\x01", end: str="\x02") -> str: """Overcome bug in GNU Readline in relation to calculation of prompt length in presence of ANSI escape codes. - :param prompt: str - original prompt - :param start: str - start code to tell GNU Readline about beginning of invisible characters - :param end: str - end code to tell GNU Readline about end of invisible characters - :return: str - prompt safe to pass to GNU Readline + :param prompt: original prompt + :param start: start code to tell GNU Readline about beginning of invisible characters + :param end: end code to tell GNU Readline about end of invisible characters + :return: prompt safe to pass to GNU Readline """ # Windows terminals don't use ANSI escape codes and Windows readline isn't based on GNU Readline if sys.platform == "win32": @@ -1990,9 +2047,8 @@ class Cmd(cmd.Cmd): return result - def pseudo_raw_input(self, prompt): - """ - began life as a copy of cmd's cmdloop; like raw_input but + def pseudo_raw_input(self, prompt: str) -> str: + """Began life as a copy of cmd's cmdloop; like raw_input but - accounts for changed stdin, stdout - if input is a pipe (instead of a tty), look at self.echo @@ -2033,14 +2089,14 @@ class Cmd(cmd.Cmd): line = 'eof' return line.strip() - def _cmdloop(self): + def _cmdloop(self) -> bool: """Repeatedly issue a prompt, accept input, parse an initial prefix off the received input, and dispatch to action methods, passing them the remainder of the line as argument. This serves the same role as cmd.cmdloop(). - :return: bool - True implies the entire application should exit. + :return: True implies the entire application should exit. """ # An almost perfect copy from Cmd; however, the pseudo_raw_input portion # has been split out so that it can be called separately @@ -2070,7 +2126,7 @@ class Cmd(cmd.Cmd): # Enable tab completion readline.parse_and_bind(self.completekey + ": complete") - stop = None + stop = False try: while not stop: if self.cmdqueue: @@ -2111,7 +2167,7 @@ class Cmd(cmd.Cmd): return stop @with_argument_list - def do_alias(self, arglist): + def do_alias(self, arglist: List[str]) -> None: """Define or display aliases Usage: Usage: alias [name] | [<name> <value>] @@ -2167,7 +2223,7 @@ Usage: Usage: alias [name] | [<name> <value>] errmsg = "Aliases can not contain: {}".format(invalidchars) self.perror(errmsg, traceback_war=False) - def complete_alias(self, text, line, begidx, endidx): + def complete_alias(self, text: str, line: str, begidx: int, endidx: int) -> List[str]: """ Tab completion for alias """ alias_names = set(self.aliases.keys()) visible_commands = set(self.get_visible_commands()) @@ -2180,7 +2236,7 @@ Usage: Usage: alias [name] | [<name> <value>] return self.index_based_complete(text, line, begidx, endidx, index_dict, self.path_complete) @with_argument_list - def do_unalias(self, arglist): + def do_unalias(self, arglist: List[str]) -> None: """Unsets aliases Usage: Usage: unalias [-a] name [name ...] @@ -2191,7 +2247,7 @@ Usage: Usage: unalias [-a] name [name ...] -a remove all alias definitions """ if not arglist: - self.do_help('unalias') + self.do_help(['unalias']) if '-a' in arglist: self.aliases.clear() @@ -2208,12 +2264,12 @@ Usage: Usage: unalias [-a] name [name ...] else: self.perror("Alias {!r} does not exist".format(cur_arg), traceback_war=False) - def complete_unalias(self, text, line, begidx, endidx): + def complete_unalias(self, text: str, line: str, begidx: int, endidx: int) -> List[str]: """ Tab completion for unalias """ return self.basic_complete(text, line, begidx, endidx, self.aliases) @with_argument_list - def do_help(self, arglist): + def do_help(self, arglist: List[str]) -> None: """List available commands with "help" or detailed help with "help cmd".""" if not arglist or (len(arglist) == 1 and arglist[0] in ('--verbose', '-v')): verbose = len(arglist) == 1 and arglist[0] in ('--verbose', '-v') @@ -2240,7 +2296,7 @@ Usage: Usage: unalias [-a] name [name ...] # This could be a help topic cmd.Cmd.do_help(self, arglist[0]) - def _help_menu(self, verbose=False): + def _help_menu(self, verbose: bool=False) -> None: """Show a list of commands which help can be displayed for. """ # Get a sorted list of help topics @@ -2283,7 +2339,7 @@ Usage: Usage: unalias [-a] name [name ...] self.print_topics(self.misc_header, help_topics, 15, 80) self.print_topics(self.undoc_header, cmds_undoc, 15, 80) - def _print_topics(self, header, cmds, verbose): + def _print_topics(self, header: str, cmds: List[str], verbose: bool) -> None: """Customized version of print_topics that can switch between verbose or traditional output""" import io @@ -2354,23 +2410,23 @@ Usage: Usage: unalias [-a] name [name ...] command = '' self.stdout.write("\n") - def do_shortcuts(self, _): + def do_shortcuts(self, _: str) -> None: """Lists shortcuts (aliases) available.""" result = "\n".join('%s: %s' % (sc[0], sc[1]) for sc in sorted(self.shortcuts)) self.poutput("Shortcuts for other commands:\n{}\n".format(result)) - def do_eof(self, _): + def do_eof(self, _: str) -> bool: """Called when <Ctrl>-D is pressed.""" # End of script should not exit app, but <Ctrl>-D should. print('') # Required for clearing line when exiting submenu return self._STOP_AND_EXIT - def do_quit(self, _): + def do_quit(self, _: str) -> bool: """Exits this application.""" self._should_quit = True return self._STOP_AND_EXIT - def select(self, opts, prompt='Your choice? '): + def select(self, opts: Union[str, List[str], List[Tuple[str, Optional[str]]]], prompt: str='Your choice? ') -> str: """Presents a numbered menu to the user. Modelled after the bash shell's SELECT. Returns the item chosen. @@ -2393,7 +2449,7 @@ Usage: Usage: unalias [-a] name [name ...] fulloptions.append((opt[0], opt[1])) except IndexError: fulloptions.append((opt[0], opt[0])) - for (idx, (value, text)) in enumerate(fulloptions): + for (idx, (_, text)) in enumerate(fulloptions): self.poutput(' %2d. %s\n' % (idx + 1, text)) while True: response = input(prompt) @@ -2404,18 +2460,18 @@ Usage: Usage: unalias [-a] name [name ...] readline.remove_history_item(hlen - 1) try: - response = int(response) - result = fulloptions[response - 1][0] + choice = int(response) + result = fulloptions[choice - 1][0] break except (ValueError, IndexError): self.poutput("{!r} isn't a valid choice. Pick a number between 1 and {}:\n".format(response, len(fulloptions))) return result - def cmdenvironment(self): + def cmdenvironment(self) -> str: """Get a summary report of read-only settings which the user cannot modify at runtime. - :return: str - summary report of read-only settings which the user cannot modify at runtime + :return: summary report of read-only settings which the user cannot modify at runtime """ read_only_settings = """ Commands may be terminated with: {} @@ -2423,7 +2479,13 @@ Usage: Usage: unalias [-a] name [name ...] Output redirection and pipes allowed: {}""" return read_only_settings.format(str(self.terminators), self.allow_cli_args, self.allow_redirection) - def show(self, args, parameter): + def show(self, args: argparse.Namespace, parameter: str) -> None: + """Shows current settings of parameters. + + :param args: argparse parsed arguments from the set command + :param parameter: + :return: + """ param = '' if parameter: param = parameter.strip().lower() @@ -2452,7 +2514,7 @@ Usage: Usage: unalias [-a] name [name ...] set_parser.add_argument('settable', nargs=(0, 2), help='[param_name] [value]') @with_argparser(set_parser) - def do_set(self, args): + def do_set(self, args: argparse.Namespace) -> None: """Sets a settable parameter or shows current settings of parameters. Accepts abbreviated parameter names so long as there is no ambiguity. @@ -2487,7 +2549,7 @@ Usage: Usage: unalias [-a] name [name ...] param = args.settable[0] self.show(args, param) - def do_shell(self, command): + def do_shell(self, command: str) -> None: """Execute a command as if at the OS prompt. Usage: shell <command> [arguments]""" @@ -2519,14 +2581,14 @@ Usage: Usage: unalias [-a] name [name ...] proc = subprocess.Popen(expanded_command, stdout=self.stdout, shell=True) proc.communicate() - def complete_shell(self, text, line, begidx, endidx): + def complete_shell(self, text: str, line: str, begidx: int, endidx: int) -> List[str]: """Handles tab completion of executable commands and local file system paths for the shell command - :param text: str - the string prefix we are attempting to match (all returned matches must begin with it) - :param line: str - the current input line with leading whitespace removed - :param begidx: int - the beginning index of the prefix text - :param endidx: int - the ending index of the prefix text - :return: List[str] - a list of possible tab completions + :param text: the string prefix we are attempting to match (all returned matches must begin with it) + :param line: the current input line with leading whitespace removed + :param begidx: the beginning index of the prefix text + :param endidx: the ending index of the prefix text + :return: a list of possible tab completions """ index_dict = {1: self.shell_cmd_complete} return self.index_based_complete(text, line, begidx, endidx, index_dict, self.path_complete) @@ -2555,7 +2617,7 @@ Usage: Usage: unalias [-a] name [name ...] sys.displayhook = sys.__displayhook__ sys.excepthook = sys.__excepthook__ - def do_py(self, arg): + def do_py(self, arg: str) -> bool: """ Invoke python command, shell, or script @@ -2568,7 +2630,7 @@ Usage: Usage: unalias [-a] name [name ...] from .pyscript_bridge import PyscriptBridge if self._in_py: self.perror("Recursively entering interactive Python consoles is not allowed.", traceback_war=False) - return + return False self._in_py = True # noinspection PyBroadException @@ -2711,7 +2773,7 @@ Usage: Usage: unalias [-a] name [name ...] return self._should_quit @with_argument_list - def do_pyscript(self, arglist): + def do_pyscript(self, arglist: List[str]) -> None: """\nRuns a python script file inside the console Usage: pyscript <script_path> [script_arguments] @@ -2722,7 +2784,7 @@ Paths or arguments that contain spaces must be enclosed in quotes """ if not arglist: self.perror("pyscript command requires at least 1 argument ...", traceback_war=False) - self.do_help('pyscript') + self.do_help(['pyscript']) return # Get the absolute path of the script @@ -2741,15 +2803,15 @@ Paths or arguments that contain spaces must be enclosed in quotes # Restore command line arguments to original state sys.argv = orig_args - # Enable tab-completion for pyscript command - def complete_pyscript(self, text, line, begidx, endidx): + def complete_pyscript(self, text: str, line: str, begidx: int, endidx: int) -> List[str]: + """Enable tab-completion for pyscript command.""" index_dict = {1: self.path_complete} return self.index_based_complete(text, line, begidx, endidx, index_dict) # Only include the do_ipy() method if IPython is available on the system if ipython_available: # noinspection PyMethodMayBeStatic,PyUnusedLocal - def do_ipy(self, arg): + def do_ipy(self, arg: str) -> None: """Enters an interactive IPython shell. Run python code from external files with ``run filename.py`` @@ -2787,7 +2849,7 @@ a..b, a:b, a:, ..b items by indices (inclusive) history_parser.add_argument('arg', nargs='?', help=_history_arg_help) @with_argparser(history_parser) - def do_history(self, args): + def do_history(self, args: argparse.Namespace) -> None: """View, run, edit, and save previously entered commands.""" # If an argument was supplied, then retrieve partial contents of the history cowardly_refuse_to_run = False @@ -2852,7 +2914,7 @@ a..b, a:b, a:, ..b items by indices (inclusive) else: self.poutput(hi.pr()) - def _generate_transcript(self, history, transcript_file): + def _generate_transcript(self, history: List[HistoryItem], transcript_file: str) -> None: """Generate a transcript file from a given history of commands.""" # Save the current echo state, and turn it off. We inject commands into the # output using a different mechanism @@ -2905,19 +2967,22 @@ a..b, a:b, a:, ..b items by indices (inclusive) self.echo = saved_echo # finally, we can write the transcript out to the file - with open(transcript_file, 'w') as fout: - fout.write(transcript) - - # and let the user know what we did - if len(history) > 1: - plural = 'commands and their outputs' + try: + with open(transcript_file, 'w') as fout: + fout.write(transcript) + except OSError as ex: + self.perror('Failed to save transcript: {}'.format(ex), traceback_war=False) else: - plural = 'command and its output' - msg = '{} {} saved to transcript file {!r}' - self.pfeedback(msg.format(len(history), plural, transcript_file)) + # and let the user know what we did + if len(history) > 1: + plural = 'commands and their outputs' + else: + plural = 'command and its output' + msg = '{} {} saved to transcript file {!r}' + self.pfeedback(msg.format(len(history), plural, transcript_file)) @with_argument_list - def do_edit(self, arglist): + def do_edit(self, arglist: List[str]) -> None: """Edit a file in a text editor. Usage: edit [file_path] @@ -2935,13 +3000,13 @@ The editor used is determined by the ``editor`` settable parameter. else: os.system('"{}"'.format(self.editor)) - # Enable tab-completion for edit command - def complete_edit(self, text, line, begidx, endidx): + def complete_edit(self, text: str, line: str, begidx: int, endidx: int) -> List[str]: + """Enable tab-completion for edit command.""" index_dict = {1: self.path_complete} return self.index_based_complete(text, line, begidx, endidx, index_dict) @property - def _current_script_dir(self): + def _current_script_dir(self) -> Optional[str]: """Accessor to get the current script directory from the _script_dir LIFO queue.""" if self._script_dir: return self._script_dir[-1] @@ -2949,7 +3014,7 @@ The editor used is determined by the ``editor`` settable parameter. return None @with_argument_list - def do__relative_load(self, arglist): + def do__relative_load(self, arglist: List[str]) -> None: """Runs commands in script file that is encoded as either ASCII or UTF-8 text. Usage: _relative_load <file_path> @@ -2972,15 +3037,15 @@ NOTE: This command is intended to only be used within text file scripts. file_path = arglist[0].strip() # NOTE: Relative path is an absolute path, it is just relative to the current script directory relative_path = os.path.join(self._current_script_dir or '', file_path) - self.do_load(relative_path) + self.do_load([relative_path]) - def do_eos(self, _): + def do_eos(self, _: str) -> None: """Handles cleanup when a script has finished executing.""" if self._script_dir: self._script_dir.pop() @with_argument_list - def do_load(self, arglist): + def do_load(self, arglist: List[str]) -> None: """Runs commands in script file that is encoded as either ASCII or UTF-8 text. Usage: load <file_path> @@ -3024,21 +3089,22 @@ Script should contain one command per line, just like command would be typed in self._script_dir.append(os.path.dirname(expanded_path)) - # Enable tab-completion for load command - def complete_load(self, text, line, begidx, endidx): + def complete_load(self, text: str, line: str, begidx: int, endidx: int) -> List[str]: + """Enable tab-completion for load command.""" index_dict = {1: self.path_complete} return self.index_based_complete(text, line, begidx, endidx, index_dict) - def run_transcript_tests(self, callargs): + def run_transcript_tests(self, callargs: List[str]) -> None: """Runs transcript tests for provided file(s). This is called when either -t is provided on the command line or the transcript_files argument is provided during construction of the cmd2.Cmd instance. - :param callargs: List[str] - list of transcript test file names + :param callargs: list of transcript test file names """ import unittest from .transcript import Cmd2TestCase + class TestMyAppCase(Cmd2TestCase): cmdapp = self @@ -3048,7 +3114,7 @@ Script should contain one command per line, just like command would be typed in runner = unittest.TextTestRunner() runner.run(testcase) - def cmdloop(self, intro=None): + def cmdloop(self, intro: Optional[str]=None) -> None: """This is an outer wrapper around _cmdloop() which deals with extra features provided by cmd2. _cmdloop() provides the main loop equivalent to cmd.cmdloop(). This is a wrapper around that which deals with @@ -3057,7 +3123,7 @@ Script should contain one command per line, just like command would be typed in - transcript testing - intro banner - :param intro: str - if provided this overrides self.intro and serves as the intro banner printed once at start + :param intro: if provided this overrides self.intro and serves as the intro banner printed once at start """ if self.allow_cli_args: parser = argparse.ArgumentParser() @@ -3183,41 +3249,18 @@ Script should contain one command per line, just like command would be typed in # TODO check signature of registered func and throw error if it's wrong -class HistoryItem(str): - """Class used to represent an item in the History list. - - Thin wrapper around str class which adds a custom format for printing. It - also keeps track of its index in the list as well as a lowercase - representation of itself for convenience/efficiency. - - """ - listformat = '-------------------------[{}]\n{}\n' - - # noinspection PyUnusedLocal - def __init__(self, instr): - str.__init__(self) - self.lowercase = self.lower() - self.idx = None - - def pr(self): - """Represent a HistoryItem in a pretty fashion suitable for printing. - - :return: str - pretty print string version of a HistoryItem - """ - return self.listformat.format(self.idx, str(self).rstrip()) - - class History(list): """ A list of HistoryItems that knows how to respond to user requests. """ # noinspection PyMethodMayBeStatic - def _zero_based_index(self, onebased): + def _zero_based_index(self, onebased: int) -> int: + """Convert a one-based index to a zero-based index.""" result = onebased if result > 0: result -= 1 return result - def _to_index(self, raw): + def _to_index(self, raw: str) -> Optional[int]: if raw: result = self._zero_based_index(int(raw)) else: @@ -3226,11 +3269,11 @@ class History(list): spanpattern = re.compile(r'^\s*(?P<start>-?\d+)?\s*(?P<separator>:|(\.{2,}))?\s*(?P<end>-?\d+)?\s*$') - def span(self, raw): + def span(self, raw: str) -> List[HistoryItem]: """Parses the input string search for a span pattern and if if found, returns a slice from the History list. - :param raw: str - string potentially containing a span of the forms a..b, a:b, a:, ..b - :return: List[HistoryItem] - slice from the History list + :param raw: string potentially containing a span of the forms a..b, a:b, a:, ..b + :return: slice from the History list """ if raw.lower() in ('*', '-', 'all'): raw = ':' @@ -3254,20 +3297,20 @@ class History(list): rangePattern = re.compile(r'^\s*(?P<start>[\d]+)?\s*-\s*(?P<end>[\d]+)?\s*$') - def append(self, new): + def append(self, new: str) -> None: """Append a HistoryItem to end of the History list - :param new: str - command line to convert to HistoryItem and add to the end of the History list + :param new: command line to convert to HistoryItem and add to the end of the History list """ new = HistoryItem(new) list.append(self, new) new.idx = len(self) - def get(self, getme=None): + def get(self, getme: Optional[Union[int, str]]=None) -> List[HistoryItem]: """Get an item or items from the History list using 1-based indexing. - :param getme: int or str - item(s) to get - either an integer index or string to search for - :return: List[str] - list of HistoryItems matching the retrieval criteria + :param getme: item(s) to get - either an integer index or string to search for + :return: list of HistoryItems matching the retrieval criteria """ if not getme: return self @@ -3316,23 +3359,23 @@ class History(list): class Statekeeper(object): """Class used to save and restore state during load and py commands as well as when redirecting output or pipes.""" - def __init__(self, obj, attribs): + def __init__(self, obj: Any, attribs: Iterable) -> None: """Use the instance attributes as a generic key-value store to copy instance attributes from outer object. :param obj: instance of cmd2.Cmd derived class (your application instance) - :param attribs: Tuple[str] - tuple of strings listing attributes of obj to save a copy of + :param attribs: tuple of strings listing attributes of obj to save a copy of """ self.obj = obj self.attribs = attribs if self.obj: self._save() - def _save(self): + def _save(self) -> None: """Create copies of attributes from self.obj inside this Statekeeper instance.""" for attrib in self.attribs: setattr(self, attrib, getattr(self.obj, attrib)) - def restore(self): + def restore(self) -> None: """Overwrite attributes in self.obj with the saved values stored in this Statekeeper instance.""" if self.obj: for attrib in self.attribs: @@ -3356,6 +3399,6 @@ class CmdResult(utils.namedtuple_with_two_defaults('CmdResult', ['out', 'err', ' NOTE: Named tuples are immutable. So the contents are there for access, not for modification. """ - def __bool__(self): + def __bool__(self) -> bool: """If err is an empty string, treat the result as a success; otherwise treat it as a failure.""" return not self.err diff --git a/cmd2/parsing.py b/cmd2/parsing.py index 1db78526..b220f1c4 100644 --- a/cmd2/parsing.py +++ b/cmd2/parsing.py @@ -2,6 +2,7 @@ # -*- coding: utf-8 -*- """Statement parsing classes for cmd2""" +import os import re import shlex from typing import List, Tuple, Dict @@ -42,7 +43,7 @@ class Statement(str): from the elements of the list, and aliases and shortcuts are expanded :type argv: list - :var terminator: the charater which terminated the multiline command, if + :var terminator: the character which terminated the multiline command, if there was one :type terminator: str or None :var suffix: characters appearing after the terminator but before output @@ -53,7 +54,7 @@ class Statement(str): :type pipe_to: list :var output: if output was redirected, the redirection token, i.e. '>>' :type output: str or None - :var output_to: if output was redirected, the destination, usually a filename + :var output_to: if output was redirected, the destination file :type output_to: str or None """ @@ -329,6 +330,11 @@ class StatementParser: pipe_pos = tokens.index(constants.REDIRECTION_PIPE) # save everything after the first pipe as tokens pipe_to = tokens[pipe_pos+1:] + + for pos, cur_token in enumerate(pipe_to): + unquoted_token = utils.strip_quotes(cur_token) + pipe_to[pos] = os.path.expanduser(unquoted_token) + # remove all the tokens after the pipe tokens = tokens[:pipe_pos] except ValueError: @@ -341,7 +347,12 @@ class StatementParser: try: output_pos = tokens.index(constants.REDIRECTION_OUTPUT) output = constants.REDIRECTION_OUTPUT - output_to = ' '.join(tokens[output_pos+1:]) + + # Check if we are redirecting to a file + if len(tokens) > output_pos + 1: + unquoted_path = utils.strip_quotes(tokens[output_pos + 1]) + output_to = os.path.expanduser(unquoted_path) + # remove all the tokens after the output redirect tokens = tokens[:output_pos] except ValueError: @@ -350,7 +361,12 @@ class StatementParser: try: output_pos = tokens.index(constants.REDIRECTION_APPEND) output = constants.REDIRECTION_APPEND - output_to = ' '.join(tokens[output_pos+1:]) + + # Check if we are redirecting to a file + if len(tokens) > output_pos + 1: + unquoted_path = utils.strip_quotes(tokens[output_pos + 1]) + output_to = os.path.expanduser(unquoted_path) + # remove all tokens after the output redirect tokens = tokens[:output_pos] except ValueError: diff --git a/cmd2/pyscript_bridge.py b/cmd2/pyscript_bridge.py index 196be82b..9353e611 100644 --- a/cmd2/pyscript_bridge.py +++ b/cmd2/pyscript_bridge.py @@ -10,7 +10,7 @@ Released under MIT license, see LICENSE file import argparse import functools import sys -from typing import List, Tuple, Callable +from typing import List, Callable # Python 3.4 require contextlib2 for temporarily redirecting stderr and stdout if sys.version_info < (3, 5): @@ -40,7 +40,7 @@ class CommandResult(namedtuple_with_defaults('CmdResult', ['stdout', 'stderr', ' class CopyStream(object): """Copies all data written to a stream""" - def __init__(self, inner_stream, echo: bool = False): + def __init__(self, inner_stream, echo: bool = False) -> None: self.buffer = '' self.inner_stream = inner_stream self.echo = echo @@ -212,7 +212,7 @@ class ArgparseFunctor: def process_flag(action, value): if isinstance(action, argparse._CountAction): if isinstance(value, int): - for c in range(value): + for _ in range(value): cmd_str[0] += '{} '.format(action.option_strings[0]) return else: @@ -298,4 +298,5 @@ class PyscriptBridge(object): return commands def __call__(self, args: str): - return _exec_cmd(self._cmd2_app, functools.partial(self._cmd2_app.onecmd_plus_hooks, args + '\n'), self.cmd_echo) + return _exec_cmd(self._cmd2_app, functools.partial(self._cmd2_app.onecmd_plus_hooks, args + '\n'), + self.cmd_echo) diff --git a/cmd2/transcript.py b/cmd2/transcript.py index 8a9837a6..5ba8d20d 100644 --- a/cmd2/transcript.py +++ b/cmd2/transcript.py @@ -12,9 +12,11 @@ classes are used in cmd2.py::run_transcript_tests() import re import glob import unittest +from typing import Tuple from . import utils + class Cmd2TestCase(unittest.TestCase): """A unittest class used for transcript testing. @@ -50,7 +52,7 @@ class Cmd2TestCase(unittest.TestCase): for (fname, transcript) in its: self._test_transcript(fname, transcript) - def _test_transcript(self, fname, transcript): + def _test_transcript(self, fname: str, transcript): line_num = 0 finished = False line = utils.strip_ansi(next(transcript)) @@ -103,7 +105,7 @@ class Cmd2TestCase(unittest.TestCase): fname, line_num, command, expected, result) self.assertTrue(re.match(expected, result, re.MULTILINE | re.DOTALL), message) - def _transform_transcript_expected(self, s): + def _transform_transcript_expected(self, s: str) -> str: """Parse the string with slashed regexes into a valid regex. Given a string like: @@ -151,7 +153,7 @@ class Cmd2TestCase(unittest.TestCase): return regex @staticmethod - def _escaped_find(regex, s, start, in_regex): + def _escaped_find(regex: str, s: str, start: int, in_regex: bool) -> Tuple[str, int, int]: """Find the next slash in {s} after {start} that is not preceded by a backslash. If we find an escaped slash, add everything up to and including it to regex, @@ -162,7 +164,6 @@ class Cmd2TestCase(unittest.TestCase): {in_regex} specifies whether we are currently searching in a regex, we behave differently if we are or if we aren't. """ - while True: pos = s.find('/', start) if pos == -1: @@ -211,14 +212,11 @@ class OutputTrap(object): def __init__(self): self.contents = '' - def write(self, txt): - """Add text to the internal contents. - - :param txt: str - """ + def write(self, txt: str): + """Add text to the internal contents.""" self.contents += txt - def read(self): + def read(self) -> str: """Read from the internal contents and then clear them out. :return: str - text from the internal contents diff --git a/cmd2/utils.py b/cmd2/utils.py index 07969ff1..11d48b78 100644 --- a/cmd2/utils.py +++ b/cmd2/utils.py @@ -4,10 +4,11 @@ import collections import os -from typing import Optional +from typing import Any, List, Optional, Union from . import constants + def strip_ansi(text: str) -> str: """Strip ANSI escape codes from a string. @@ -30,7 +31,8 @@ def strip_quotes(arg: str) -> str: return arg -def namedtuple_with_defaults(typename, field_names, default_values=()): +def namedtuple_with_defaults(typename: str, field_names: Union[str, List[str]], + default_values: collections.Iterable=()): """ Convenience function for defining a namedtuple with default values @@ -58,7 +60,9 @@ def namedtuple_with_defaults(typename, field_names, default_values=()): T.__new__.__defaults__ = tuple(prototype) return T -def namedtuple_with_two_defaults(typename, field_names, default_values=('', '')): + +def namedtuple_with_two_defaults(typename: str, field_names: Union[str, List[str]], + default_values: collections.Iterable=('', '')): """Wrapper around namedtuple which lets you treat the last value as optional. :param typename: str - type name for the Named tuple @@ -72,7 +76,8 @@ def namedtuple_with_two_defaults(typename, field_names, default_values=('', '')) T.__new__.__defaults__ = default_values return T -def cast(current, new): + +def cast(current: Any, new: str) -> Any: """Tries to force a new value into the same type as the current when trying to set the value for a parameter. :param current: current value for the parameter, type varies @@ -101,6 +106,7 @@ def cast(current, new): print("Problem setting parameter (now %s) to %s; incorrect type?" % (current, new)) return current + def which(editor: str) -> Optional[str]: """Find the full path of a given editor. @@ -118,7 +124,8 @@ def which(editor: str) -> Optional[str]: editor_path = None return editor_path -def is_text_file(file_path): + +def is_text_file(file_path: str) -> bool: """Returns if a file contains only ASCII or UTF-8 encoded text :param file_path: path to the file being checked diff --git a/examples/paged_output.py b/examples/paged_output.py index c56dcb89..d1b1b2c2 100755 --- a/examples/paged_output.py +++ b/examples/paged_output.py @@ -2,28 +2,55 @@ # coding=utf-8 """A simple example demonstrating the using paged output via the ppaged() method. """ +import os +from typing import List import cmd2 class PagedOutput(cmd2.Cmd): - """ Example cmd2 application where we create commands that just print the arguments they are called with.""" + """ Example cmd2 application which shows how to display output using a pager.""" def __init__(self): super().__init__() + def page_file(self, file_path: str, chop: bool=False): + """Helper method to prevent having too much duplicated code.""" + filename = os.path.expanduser(file_path) + try: + with open(filename, 'r') as f: + text = f.read() + self.ppaged(text, chop=chop) + except FileNotFoundError as ex: + self.perror('ERROR: file {!r} not found'.format(filename), traceback_war=False) + @cmd2.with_argument_list - def do_page_file(self, args): - """Read in a text file and display its output in a pager.""" + def do_page_wrap(self, args: List[str]): + """Read in a text file and display its output in a pager, wrapping long lines if they don't fit. + + Usage: page_wrap <file_path> + """ if not args: - self.perror('page_file requires a path to a file as an argument', traceback_war=False) + self.perror('page_wrap requires a path to a file as an argument', traceback_war=False) return + self.page_file(args[0], chop=False) + + complete_page_wrap = cmd2.Cmd.path_complete + + @cmd2.with_argument_list + def do_page_truncate(self, args: List[str]): + """Read in a text file and display its output in a pager, truncating long lines if they don't fit. + + Truncated lines can still be accessed by scrolling to the right using the arrow keys. - with open(args[0], 'r') as f: - text = f.read() - self.ppaged(text) + Usage: page_chop <file_path> + """ + if not args: + self.perror('page_truncate requires a path to a file as an argument', traceback_war=False) + return + self.page_file(args[0], chop=True) - complete_page_file = cmd2.Cmd.path_complete + complete_page_truncate = cmd2.Cmd.path_complete if __name__ == '__main__': diff --git a/examples/tab_autocompletion.py b/examples/tab_autocompletion.py index d1726841..342cfff5 100755 --- a/examples/tab_autocompletion.py +++ b/examples/tab_autocompletion.py @@ -109,6 +109,15 @@ class TabCompleteExample(cmd2.Cmd): """Simulating a function that queries and returns a completion values""" return actors + def instance_query_movie_ids(self) -> List[str]: + """Demonstrates showing tabular hinting of tab completion information""" + completions_with_desc = [] + + for movie_id, movie_entry in self.MOVIE_DATABASE.items(): + completions_with_desc.append(argparse_completer.CompletionItem(movie_id, movie_entry['title'])) + + return completions_with_desc + # This demonstrates a number of customizations of the AutoCompleter version of ArgumentParser # - The help output will separately group required vs optional flags # - The help output for arguments with multiple flags or with append=True is more concise @@ -253,6 +262,9 @@ class TabCompleteExample(cmd2.Cmd): ('path_complete', [False, False])) vid_movies_delete_parser = vid_movies_commands_subparsers.add_parser('delete') + vid_delete_movie_id = vid_movies_delete_parser.add_argument('movie_id', help='Movie ID') + setattr(vid_delete_movie_id, argparse_completer.ACTION_ARG_CHOICES, instance_query_movie_ids) + setattr(vid_delete_movie_id, argparse_completer.ACTION_DESCRIPTIVE_COMPLETION_HEADER, 'Title') vid_shows_parser = video_types_subparsers.add_parser('shows') vid_shows_parser.set_defaults(func=_do_vid_media_shows) @@ -328,6 +340,9 @@ class TabCompleteExample(cmd2.Cmd): movies_add_parser.add_argument('actor', help='Actors', nargs='*') movies_delete_parser = movies_commands_subparsers.add_parser('delete') + movies_delete_movie_id = movies_delete_parser.add_argument('movie_id', help='Movie ID') + setattr(movies_delete_movie_id, argparse_completer.ACTION_ARG_CHOICES, 'instance_query_movie_ids') + setattr(movies_delete_movie_id, argparse_completer.ACTION_DESCRIPTIVE_COMPLETION_HEADER, 'Title') movies_load_parser = movies_commands_subparsers.add_parser('load') movie_file_action = movies_load_parser.add_argument('movie_file', help='Movie database') @@ -362,7 +377,7 @@ class TabCompleteExample(cmd2.Cmd): 'director': TabCompleteExample.static_list_directors, # static list 'movie_file': (self.path_complete, [False, False]) } - completer = argparse_completer.AutoCompleter(TabCompleteExample.media_parser, arg_choices=choices) + completer = argparse_completer.AutoCompleter(TabCompleteExample.media_parser, arg_choices=choices, cmd2_app=self) tokens, _ = self.tokens_for_completion(line, begidx, endidx) results = completer.complete_command(tokens, text, line, begidx, endidx) diff --git a/mtime.sh b/mtime.sh deleted file mode 100755 index 1cb5f8dc..00000000 --- a/mtime.sh +++ /dev/null @@ -1,14 +0,0 @@ -#!/bin/bash - -TMPFILE=`mktemp /tmp/mtime.XXXXXX` || exit 1 - -for x in {1..100} -do - gtime -f "real %e user %U sys %S" -a -o $TMPFILE "$@" - #tail -1 $TMPFILE -done - -awk '{ et += $2; ut += $4; st += $6; count++ } END { printf "%d iterations\n", count ; printf "average: real %.3f user %.3f sys %.3f\n", et/count, ut/count, st/count }' $TMPFILE - -rm $TMPFILE - diff --git a/speedup_import.md b/speedup_import.md deleted file mode 100644 index c49f1e86..00000000 --- a/speedup_import.md +++ /dev/null @@ -1,99 +0,0 @@ -# Speedup Import - -## Assumptions - -I created a simple script to run a command 20 times and calculate -the average clock time for each run of the command. This script requires -some unix tools, including the gnu flavor of the `time` command. This script -can is called `mtime.sh` and is included in this branch. - -These tests were all run on my 2015 MacBook Pro with a 3.1 GHz Intel Core i7 -and 16GB of memory. - - -## Baseline measurement - -First let's see how long it takes to start up python. The longish path here -ensures we aren't measuring the time it takes the pyenv shims to run: -``` -$./mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "" -100 iterations -average: real 0.028 user 0.020 sys 0.000 -``` - - -## Initial measurement - -From commit fbbfe256, which has `__init.py__` importing `cmd2.cmd2.Cmd` -and a bunch of other stuff, we get: -``` -$ ./mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "import cmd2" -100 iterations -average: real 0.140 user 0.100 sys 0.030 -``` - -From the baseline and this initial measurement, we infer it takes ~110 ms -to import the `cmd2` module. - - -## Defer unittest - -In commit 8bc2c37a we defer the import of `unittest` until we need it to -test a transcript. -``` -$./mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "import cmd2" -100 iterations -average: real 0.131 user 0.091 sys 0.030 -``` - - -## Defer InteractiveConsole from code - -In commit 6e49661f we defer the import of `InteractiveConsole` until the user -wants to run the `py` command. -``` -$ ./mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "import cmd2" -100 iterations -average: real 0.131 user 0.090 sys 0.030 -``` - -## Defer atexit, codes, signal, tempfile, copy - -In commit a479fa94 we defer 5 imports: atexit, codecs, signal, tempfile, and copy. -``` -$ ./mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "import cmd2"100 iterations -average: real 0.120 user 0.081 sys 0.021 -``` - -## Defer datetime, functools, io, subprocess, traceback - -In commit d9ca07a9 we defer 5 more imports: datetime, functools, io, subprocess, traceback. -``` -$ ./mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "import cmd2" -100 iterations -average: real 0.115 user 0.080 sys 0.020 -``` - -## extract AddSubmenu to its own file - -In commit ccfdf0f9 we extract AddSubmenu() to it's own file, so it is not -imported or processed by default. -``` -$ ./mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "import cmd2" -100 iterations -average: real 0.117 user 0.081 sys 0.021 -``` - -## Progress Update - -Python takes ~30ms to start up and do nothing. When we began we estimated it took -~110ms to import cmd2. We are now down to about ~90ms, which is approximately a -20% improvement. - -## Move more functions into utils - -Commit fc495a42 moves a few functions from `cmd2.py` into `utils.py`. -``` -$ ~/bin/mtime.sh ~/.pyenv/versions/cmd2-3.6/bin/python -c "import cmd2" -100 iterations -average: real 0.119 user 0.081 sys 0.021 @@ -54,6 +54,21 @@ def pytest_clean(context): namespace_clean.add_task(pytest_clean, 'pytest') @invoke.task +def mypy(context): + "Run mypy optional static type checker" + context.run("mypy main.py") + namespace.add_task(mypy) +namespace.add_task(mypy) + +@invoke.task +def mypy_clean(context): + "Remove mypy cache directory" + #pylint: disable=unused-argument + dirs = ['.mypy_cache'] + rmrf(dirs) +namespace_clean.add_task(mypy_clean, 'mypy') + +@invoke.task def tox(context): "Run unit and integration tests on multiple python versions using tox" context.run("tox") diff --git a/tests/test_bashcompletion.py b/tests/test_bashcompletion.py index e2c28fce..47fcdb34 100644 --- a/tests/test_bashcompletion.py +++ b/tests/test_bashcompletion.py @@ -16,22 +16,22 @@ from cmd2.argparse_completer import ACArgumentParser, AutoCompleter try: from cmd2.argcomplete_bridge import CompletionFinder, tokens_for_completion - skip_reason1 = False + skip_no_argcomplete = False skip_reason = '' except ImportError: # Don't test if argcomplete isn't present (likely on Windows) - skip_reason1 = True + skip_no_argcomplete = True skip_reason = "argcomplete isn't installed\n" -skip_reason2 = "TRAVIS" in os.environ and os.environ["TRAVIS"] == "true" -if skip_reason2: +skip_travis = "TRAVIS" in os.environ and os.environ["TRAVIS"] == "true" +if skip_travis: skip_reason += 'These tests cannot run on TRAVIS\n' -skip_reason3 = sys.platform.startswith('win') -if skip_reason3: +skip_windows = sys.platform.startswith('win') +if skip_windows: skip_reason = 'argcomplete doesn\'t support Windows' -skip = skip_reason1 or skip_reason2 or skip_reason3 +skip = skip_no_argcomplete or skip_travis or skip_windows skip_mac = sys.platform.startswith('dar') @@ -103,7 +103,7 @@ def parser1(): # noinspection PyShadowingNames -@pytest.mark.skipif(skip, reason=skip_reason) +@pytest.mark.skipif(skip_no_argcomplete or skip_windows, reason=skip_reason) def test_bash_nocomplete(parser1): completer = CompletionFinder() result = completer(parser1, AutoCompleter(parser1)) @@ -122,7 +122,7 @@ def my_fdopen(fd, mode, *args): # noinspection PyShadowingNames -@pytest.mark.skipif(skip, reason=skip_reason) +@pytest.mark.skipif(skip_no_argcomplete or skip_windows, reason=skip_reason) def test_invalid_ifs(parser1, mock): completer = CompletionFinder() @@ -176,7 +176,7 @@ def fdopen_fail_8(fd, mode, *args): # noinspection PyShadowingNames -@pytest.mark.skipif(skip, reason=skip_reason) +@pytest.mark.skipif(skip_no_argcomplete or skip_windows, reason=skip_reason) def test_fail_alt_stdout(parser1, mock): completer = CompletionFinder() @@ -232,7 +232,7 @@ Hint: assert out == exp_out assert err == exp_err -@pytest.mark.skipif(skip_reason1, reason=skip_reason) +@pytest.mark.skipif(skip_no_argcomplete, reason=skip_reason) def test_argcomplete_tokens_for_completion_simple(): line = 'this is "a test"' endidx = len(line) @@ -243,7 +243,7 @@ def test_argcomplete_tokens_for_completion_simple(): assert begin_idx == line.rfind("is ") + len("is ") assert end_idx == end_idx -@pytest.mark.skipif(skip_reason1, reason=skip_reason) +@pytest.mark.skipif(skip_no_argcomplete, reason=skip_reason) def test_argcomplete_tokens_for_completion_unclosed_quotee_exception(): line = 'this is "a test' endidx = len(line) diff --git a/tests/test_cmd2.py b/tests/test_cmd2.py index 24a14ea2..f167793e 100644 --- a/tests/test_cmd2.py +++ b/tests/test_cmd2.py @@ -230,7 +230,7 @@ def test_pyscript_with_nonexist_file(base_app, capsys): python_script = 'does_not_exist.py' run_cmd(base_app, "pyscript {}".format(python_script)) out, err = capsys.readouterr() - assert err.startswith('ERROR: [Errno 2] No such file or directory:') + assert err.startswith("EXCEPTION of type 'FileNotFoundError' occurred with message:") def test_pyscript_with_exception(base_app, capsys, request): test_dir = os.path.dirname(request.module.__file__) @@ -615,6 +615,44 @@ def test_output_redirection(base_app): finally: os.remove(filename) +def test_output_redirection_to_nonexistent_directory(base_app): + filename = '~/fakedir/this_does_not_exist.txt' + + # Verify that writing to a file in a non-existent directory doesn't work + run_cmd(base_app, 'help > {}'.format(filename)) + expected = normalize(BASE_HELP) + with pytest.raises(FileNotFoundError): + with open(filename) as f: + content = normalize(f.read()) + assert content == expected + + # Verify that appending to a file also works + run_cmd(base_app, 'help history >> {}'.format(filename)) + expected = normalize(BASE_HELP + '\n' + HELP_HISTORY) + with pytest.raises(FileNotFoundError): + with open(filename) as f: + content = normalize(f.read()) + assert content == expected + +def test_output_redirection_to_too_long_filename(base_app): + filename = '~/sdkfhksdjfhkjdshfkjsdhfkjsdhfkjdshfkjdshfkjshdfkhdsfkjhewfuihewiufhweiufhiweufhiuewhiuewhfiuwehfiuewhfiuewhfiuewhfiuewhiuewhfiuewhfiuewfhiuwehewiufhewiuhfiweuhfiuwehfiuewfhiuwehiuewfhiuewhiewuhfiuewhfiuwefhewiuhewiufhewiufhewiufhewiufhewiufhewiufhewiufhewiuhewiufhewiufhewiuheiufhiuewheiwufhewiufheiufheiufhieuwhfewiuhfeiufhiuewfhiuewheiwuhfiuewhfiuewhfeiuwfhewiufhiuewhiuewhfeiuwhfiuwehfuiwehfiuehiuewhfieuwfhieufhiuewhfeiuwfhiuefhueiwhfw' + + # Verify that writing to a file in a non-existent directory doesn't work + run_cmd(base_app, 'help > {}'.format(filename)) + expected = normalize(BASE_HELP) + with pytest.raises(OSError): + with open(filename) as f: + content = normalize(f.read()) + assert content == expected + + # Verify that appending to a file also works + run_cmd(base_app, 'help history >> {}'.format(filename)) + expected = normalize(BASE_HELP + '\n' + HELP_HISTORY) + with pytest.raises(OSError): + with open(filename) as f: + content = normalize(f.read()) + assert content == expected + def test_feedback_to_output_true(base_app): base_app.feedback_to_output = True @@ -1271,7 +1309,7 @@ def test_select_invalid_option(select_app): expected = normalize(""" 1. sweet 2. salty -3 isn't a valid choice. Pick a number between 1 and 2: +'3' isn't a valid choice. Pick a number between 1 and 2: {} with sweet sauce, yum! """.format(food)) diff --git a/tests/test_transcript.py b/tests/test_transcript.py index 302d80c8..3caf6a37 100644 --- a/tests/test_transcript.py +++ b/tests/test_transcript.py @@ -154,6 +154,32 @@ this is a \/multiline\/ command assert transcript == expected +def test_history_transcript_bad_filename(request, capsys): + app = CmdLineApp() + app.stdout = StdOut() + run_cmd(app, 'orate this is\na /multiline/\ncommand;\n') + run_cmd(app, 'speak /tmp/file.txt is not a regex') + + expected = r"""(Cmd) orate this is +> a /multiline/ +> command; +this is a \/multiline\/ command +(Cmd) speak /tmp/file.txt is not a regex +\/tmp\/file.txt is not a regex +""" + + # make a tmp file + history_fname = '~/fakedir/this_does_not_exist.txt' + + # tell the history command to create a transcript + run_cmd(app, 'history -t "{}"'.format(history_fname)) + + # read in the transcript created by the history command + with pytest.raises(FileNotFoundError): + with open(history_fname) as f: + transcript = f.read() + assert transcript == expected + @pytest.mark.parametrize('expected, transformed', [ # strings with zero or one slash or with escaped slashes means no regular # expression present, so the result should just be what re.escape returns. |