# Third party modules
from fb_tools.common import to_bool
+from fb_tools.argparse_actions import TimeoutOptionAction
from fb_tools.cfg_app import FbConfigApplication
-from fb_tools.errors import FbAppError, IoTimeoutError
+from fb_tools.errors import FbAppError, IoTimeoutError, TimeoutOnPromptError
from fb_tools.multi_config import BaseMultiConfig
# Own modules
_ = XLATOR.gettext
ngettext = XLATOR.ngettext
-__version__ = '0.3.4'
+__version__ = '0.4.1'
# =============================================================================
pass
-# =============================================================================
-class TimeoutOnPromptError(AbortAppError, IoTimeoutError):
- """Special exception class on timout on a prompt."""
-
- # -------------------------------------------------------------------------
- def __init__(self, timeout):
-
- strerror = _("Timeout on answering on the console.")
- super(TimeoutOnPromptError, self).__init__(strerror, timeout)
-
-
-# =============================================================================
-class TimeoutOptionAction(argparse.Action):
-
- # -------------------------------------------------------------------------
- def __init__(self, option_strings, *args, **kwargs):
-
- super(TimeoutOptionAction, self).__init__(
- option_strings=option_strings, *args, **kwargs)
-
- # -------------------------------------------------------------------------
- def __call__(self, parser, namespace, given_timeout, option_string=None):
-
- try:
- timeout = int(given_timeout)
- if timeout <= 0 or timeout > MAX_TIMEOUT:
- msg = _(
- "a timeout must be greater than zero and less "
- "or equal to {}.").format(MAX_TIMEOUT)
- raise ValueError(msg)
- except (ValueError, TypeError) as e:
- msg = _("Wrong timeout {!r}:").format(given_timeout)
- msg += ' ' + str(e)
- raise argparse.ArgumentError(self, msg)
-
- setattr(namespace, self.dest, timeout)
-
# =============================================================================
class BaseDPXApplication(FbConfigApplication):
"""
Base class for all DPX application classes.
"""
- default_prompt_timeout = 30
- max_prompt_timeout = 600
-
- yes_list = ['y', 'yes']
- no_list = ['n', 'no']
-
- pattern_yes_no = r'^\s*(' + '|'.join(yes_list) + '|' + '|'.join(no_list) + r')?\s*$'
- re_yes_no = re.compile(pattern_yes_no, re.IGNORECASE)
+ show_assume_options = True
+ show_console_timeout_option = True
+ # show_force_option = False
+ show_simulate_option = True
# -------------------------------------------------------------------------
def __init__(
env_prefix=env_prefix, config_dir=config_dir
)
- # -------------------------------------------------------------------------
- @classmethod
- def init_yes_no_lists(cls):
-
- yes = _('yes')
- if yes not in cls.yes_list:
- cls.yes_list.append(yes)
- yes_fc = yes[0]
- if yes_fc not in cls.yes_list:
- cls.yes_list.append(yes_fc)
-
- no = _('no')
- if no not in cls.no_list and no not in cls.yes_list:
- cls.no_list.append(no)
- no_fc = no[0]
- if no_fc not in cls.no_list and no_fc not in cls.yes_list:
- cls.no_list.append(no_fc)
-
- cls.pattern_yes_no = (
- r'^\s*(' + '|'.join(cls.yes_list) + '|' + '|'.join(cls.no_list) + r')?\s*$')
- cls.re_yes_no = re.compile(cls.pattern_yes_no, re.IGNORECASE)
-
- # -----------------------------------------------------------
- @property
- def yes(self):
- """Assume 'yes' as an answer to all questions."""
- return self._yes
-
- @yes.setter
- def yes(self, value):
- self._yes = to_bool(value)
-
- # -----------------------------------------------------------
- @property
- def prompt_timeout(self):
- """The timeout in seconds for waiting for an answer on a prompt."""
- return getattr(self, '_prompt_timeout', self.default_prompt_timeout)
-
- @prompt_timeout.setter
- def prompt_timeout(self, value):
- v = int(value)
- if v < 0 or v > self.max_prompt_timeout:
- msg = _(
- "Wrong prompt timeout {v!r}, must be greater or equal to Null "
- "and less or equal to {max}.").format(v=value, max=self.max_prompt_timeout)
- LOG.warning(msg)
- else:
- self._prompt_timeout = v
-
- # -------------------------------------------------------------------------
- def as_dict(self, short=True):
- """
- Transforms the elements of the object into a dict
-
- @param short: don't include local properties in resulting dict.
- @type short: bool
-
- @return: structure as dict
- @rtype: dict
- """
-
- res = super(BaseDPXApplication, self).as_dict(short=short)
-
- res['prompt_timeout'] = self.prompt_timeout
- res['yes'] = self.yes
- res['yes_list'] = self.yes_list
- res['no_list'] = self.no_list
- res['pattern_yes_no'] = self.pattern_yes_no
-
- return res
-
- # -------------------------------------------------------------------------
- def init_arg_parser(self):
- """
- Public available method to initiate the argument parser.
- """
-
- super(BaseDPXApplication, self).init_arg_parser()
-
- self.arg_parser.add_argument(
- '-Y', '--yes', '--assume-yes', action="store_true", dest="yes", help=argparse.SUPPRESS)
-
# -------------------------------------------------------------------------
def post_init(self):
"""
super(BaseDPXApplication, self).post_init()
- self.yes = self.args.yes
-
- # -------------------------------------------------------------------------
- def get_password(self, first_prompt=None, second_prompt=None, may_empty=True, repeat=True):
- """
- Ask the user for a password on the console.
-
- @raise AbortAppError: if the user presses Ctrl-D (EOF)
- @raise TimeoutOnPromptError: if the user does not finishing after a time
-
- @param first_prompt: the prompt for the first password question
- @type first_prompt: str
- @param second_prompt: the prompt for the second password question
- @type second_prompt: str
- @param may_empty: The behaviour, if the user inputs an empty password:
- if True, an empty password will be returned
- if False, the question will be repeated.
- @type may_empty: bool
- @param repeat: Asking for the password a second time, which must be equal
- to the first given password.
- @type repeat: bool
-
- @return: The entered password
- @rtype: str
-
- """
-
- if not first_prompt:
- first_prompt = _('Password:') + ' '
-
- if not second_prompt:
- second_prompt = _('Repeat password:') + ' '
-
- ret_passwd = None
- second_passwd = None
-
- while True:
-
- if not self.quiet:
- print()
- ret_passwd = self._get_password(first_prompt, may_empty)
- if ret_passwd:
- if repeat:
- second_passwd = self._get_password(second_prompt, may_empty=False)
- if ret_passwd != second_passwd:
- msg = _("The entered passwords does not match.")
- LOG.error(msg)
- continue
- break
-
- return ret_passwd
-
- # -------------------------------------------------------------------------
- def _get_password(self, prompt, may_empty=True):
-
- def passwd_alarm_caller(signum, sigframe):
- raise TimeoutOnPromptError(self.prompt_timeout)
-
- msg_intr = _("Interrupted on demand.")
- ret_passwd = ''
-
- try:
- signal.signal(signal.SIGALRM, passwd_alarm_caller)
- signal.alarm(self.prompt_timeout)
-
- while True:
-
- try:
- ret_passwd = getpass.getpass(prompt)
- except EOFError:
- raise AbortAppError(msg_intr)
-
- signal.alarm(self.prompt_timeout)
-
- if ret_passwd == '':
- if may_empty:
- return ''
- else:
- continue
- else:
- break
-
- except (TimeoutOnPromptError, AbortAppError) as e:
- msg = _("Got a {}:").format(e.__class__.__name__) + ' ' + str(e)
- LOG.error(msg)
- self.exit(10)
-
- except KeyboardInterrupt:
- msg = _("Got a {}:").format('KeyboardInterrupt') + ' ' + msg_intr
- LOG.error(msg)
- self.exit(10)
-
- finally:
- signal.alarm(0)
-
- return ret_passwd
-
- # -------------------------------------------------------------------------
- def ask_for_yes_or_no(self, prompt, default_on_empty=None):
- """
- Ask the user for yes or no.
-
- @raise AbortAppError: if the user presses Ctrl-D (EOF)
- @raise TimeoutOnPromptError: if the user does not correct answer after a time
-
- @param prompt: the prompt for the question
- @type prompt: str
- @param default_on_empty: behaviour on an empty reply:
- * if None, repeat the question
- * if True, return True
- * else return False
- @type default_on_empty: bool or None
-
- @return: True, if the user answered Yes, else False
- @rtype: bool
-
- """
-
- if not prompt:
- prompt = _('Yes/No') + ' '
-
- def prompt_alarm_caller(signum, sigframe):
- raise TimeoutOnPromptError(self.prompt_timeout)
-
- msg_intr = _("Interrupted on demand.")
- try:
- signal.signal(signal.SIGALRM, prompt_alarm_caller)
- signal.alarm(self.prompt_timeout)
-
- reply = ''
- while True:
- try:
- reply = input(prompt)
- except EOFError:
- raise AbortAppError(msg_intr)
- signal.alarm(self.prompt_timeout)
- match = self.re_yes_no.match(reply)
- if match:
- if match.group(1) is None:
- if default_on_empty is None:
- continue
- return bool(default_on_empty)
- # There is an answer
- r = match.group(1).lower()
- if r in self.no_list:
- # Continue == no
- return False
- elif r in self.yes_list:
- # Continue == yes
- return True
- else:
- continue
- else:
- continue
- # Repeat the question
-
- except (TimeoutOnPromptError, AbortAppError) as e:
- print()
- msg = _("Got a {}:").format(e.__class__.__name__) + ' ' + str(e)
- LOG.error(msg)
- self.exit(10)
-
- except KeyboardInterrupt:
- msg = _("Got a {}:").format('KeyboardInterrupt') + ' ' + msg_intr
- print()
- LOG.error(msg)
- self.exit(10)
-
- finally:
- signal.alarm(0)
-
# vim: ts=4 et list