Mitigate problems for people who run without -n (#3916)

* CLI flag for forcing interactivity

* add --force-interactive

* Add force_interactive error checking and tests

* Add force_interactive parameter to FileDisplay

* add _can_interact

* Add _return_default

* Add **unused_kwargs to NoninteractiveDisplay

* improve _return_default assertion

* Change IDisplay calls and write tests

* Document force_interactive in interfaces.py

* Don't force_interactive with a new prompt

* Warn when skipping an interaction for the first time

* add specific logger.debug message
This commit is contained in:
Brad Warren
2016-12-19 12:45:40 -08:00
committed by Peter Eckersley
parent 81fd0cd32c
commit ae379568b1
23 changed files with 320 additions and 81 deletions
@@ -463,7 +463,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
zope.component.getUtility(interfaces.IDisplay).notification( zope.component.getUtility(interfaces.IDisplay).notification(
"Apache mod_macro seems to be in use in file(s):\n{0}" "Apache mod_macro seems to be in use in file(s):\n{0}"
"\n\nUnfortunately mod_macro is not yet supported".format( "\n\nUnfortunately mod_macro is not yet supported".format(
"\n ".join(vhost_macro))) "\n ".join(vhost_macro)), force_interactive=True)
return all_names return all_names
+4 -2
View File
@@ -85,7 +85,8 @@ def _vhost_menu(domain, vhosts):
"or Address of {0}.{1}Which virtual host would you " "or Address of {0}.{1}Which virtual host would you "
"like to choose?\n(note: conf files with multiple " "like to choose?\n(note: conf files with multiple "
"vhosts are not yet supported)".format(domain, os.linesep), "vhosts are not yet supported)".format(domain, os.linesep),
choices, help_label="More Info", ok_label="Select") choices, help_label="More Info",
ok_label="Select", force_interactive=True)
except errors.MissingCommandlineFlag: except errors.MissingCommandlineFlag:
msg = ("Encountered vhost ambiguity but unable to ask for user guidance in " msg = ("Encountered vhost ambiguity but unable to ask for user guidance in "
"non-interactive mode. Currently Certbot needs each vhost to be " "non-interactive mode. Currently Certbot needs each vhost to be "
@@ -100,4 +101,5 @@ def _vhost_menu(domain, vhosts):
def _more_info_vhost(vhost): def _more_info_vhost(vhost):
zope.component.getUtility(interfaces.IDisplay).notification( zope.component.getUtility(interfaces.IDisplay).notification(
"Virtual Host Information:{0}{1}{0}{2}".format( "Virtual Host Information:{0}{1}{0}{2}".format(
os.linesep, "-" * (display_util.WIDTH - 4), str(vhost))) os.linesep, "-" * (display_util.WIDTH - 4), str(vhost)),
force_interactive=True)
@@ -17,7 +17,8 @@ class SelectVhostTest(unittest.TestCase):
"""Tests for certbot_apache.display_ops.select_vhost.""" """Tests for certbot_apache.display_ops.select_vhost."""
def setUp(self): def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout)) zope.component.provideUtility(display_util.FileDisplay(sys.stdout,
False))
self.base_dir = "/example_path" self.base_dir = "/example_path"
self.vhosts = util.get_vh_truth( self.vhosts = util.get_vh_truth(
self.base_dir, "debian_apache_2_4/multiple_vhosts") self.base_dir, "debian_apache_2_4/multiple_vhosts")
+2 -1
View File
@@ -64,7 +64,8 @@ class ParserTest(ApacheTest): # pytlint: disable=too-few-public-methods
vhost_root="debian_apache_2_4/multiple_vhosts/apache2/sites-available"): vhost_root="debian_apache_2_4/multiple_vhosts/apache2/sites-available"):
super(ParserTest, self).setUp(test_dir, config_root, vhost_root) super(ParserTest, self).setUp(test_dir, config_root, vhost_root)
zope.component.provideUtility(display_util.FileDisplay(sys.stdout)) zope.component.provideUtility(display_util.FileDisplay(sys.stdout,
False))
from certbot_apache.parser import ApacheParser from certbot_apache.parser import ApacheParser
self.aug = augeas.Augeas( self.aug = augeas.Augeas(
+3 -2
View File
@@ -47,8 +47,9 @@ def rename_lineage(config):
new_certname = config.new_certname new_certname = config.new_certname
if not new_certname: if not new_certname:
code, new_certname = disp.input("Enter the new name for certificate {0}" code, new_certname = disp.input(
.format(certname), flag="--updated-cert-name") "Enter the new name for certificate {0}".format(certname),
flag="--updated-cert-name", force_interactive=True)
if code != display_util.OK or not new_certname: if code != display_util.OK or not new_certname:
raise errors.Error("User ended interaction.") raise errors.Error("User ended interaction.")
+15
View File
@@ -521,8 +521,17 @@ class HelpfulArgumentParser(object):
# Do any post-parsing homework here # Do any post-parsing homework here
if self.verb == "renew": if self.verb == "renew":
if parsed_args.force_interactive:
raise errors.Error(
"{0} cannot be used with renew".format(
constants.FORCE_INTERACTIVE_FLAG))
parsed_args.noninteractive_mode = True parsed_args.noninteractive_mode = True
if parsed_args.force_interactive and parsed_args.noninteractive_mode:
raise errors.Error(
"Flag for non-interactive mode and {0} conflict".format(
constants.FORCE_INTERACTIVE_FLAG))
if parsed_args.staging or parsed_args.dry_run: if parsed_args.staging or parsed_args.dry_run:
self.set_test_server(parsed_args) self.set_test_server(parsed_args)
@@ -808,6 +817,12 @@ def prepare_and_parse_args(plugins, args, detect_defaults=False): # pylint: dis
help="Run without ever asking for user input. This may require " help="Run without ever asking for user input. This may require "
"additional command line flags; the client will try to explain " "additional command line flags; the client will try to explain "
"which ones are required if it finds one missing") "which ones are required if it finds one missing")
helpful.add(
[None, "register", "run", "certonly"],
constants.FORCE_INTERACTIVE_FLAG, action="store_true",
help="Force Certbot to be interactive even if it detects it's not "
"being run in a terminal. This flag cannot be used with the "
"renew subcommand.")
helpful.add( helpful.add(
[None, "run", "certonly"], [None, "run", "certonly"],
"-d", "--domains", "--domain", dest="domains", "-d", "--domains", "--domain", dest="domains",
+3
View File
@@ -92,3 +92,6 @@ TEMP_CHECKPOINT_DIR = "temp_checkpoint"
RENEWAL_CONFIGS_DIR = "renewal" RENEWAL_CONFIGS_DIR = "renewal"
"""Renewal configs directory, relative to `IConfig.config_dir`.""" """Renewal configs directory, relative to `IConfig.config_dir`."""
FORCE_INTERACTIVE_FLAG = "--force-interactive"
"""Flag to disable TTY checking in IDisplay."""
+2 -1
View File
@@ -48,7 +48,8 @@ def redirect_by_default():
code, selection = util(interfaces.IDisplay).menu( code, selection = util(interfaces.IDisplay).menu(
"Please choose whether HTTPS access is required or optional.", "Please choose whether HTTPS access is required or optional.",
choices, default=0, cli_flag="--redirect / --no-redirect") choices, default=0,
cli_flag="--redirect / --no-redirect", force_interactive=True)
if code != display_util.OK: if code != display_util.OK:
return False return False
+7 -5
View File
@@ -46,7 +46,8 @@ def get_email(invalid=False, optional=True):
while True: while True:
try: try:
code, email = z_util(interfaces.IDisplay).input( code, email = z_util(interfaces.IDisplay).input(
invalid_prefix + msg if invalid else msg) invalid_prefix + msg if invalid else msg,
force_interactive=True)
except errors.MissingCommandlineFlag: except errors.MissingCommandlineFlag:
msg = ("You should register before running non-interactively, " msg = ("You should register before running non-interactively, "
"or provide --agree-tos and --email <email_address> flags.") "or provide --agree-tos and --email <email_address> flags.")
@@ -79,7 +80,7 @@ def choose_account(accounts):
labels = [acc.slug for acc in accounts] labels = [acc.slug for acc in accounts]
code, index = z_util(interfaces.IDisplay).menu( code, index = z_util(interfaces.IDisplay).menu(
"Please choose an account", labels) "Please choose an account", labels, force_interactive=True)
if code == display_util.OK: if code == display_util.OK:
return accounts[index] return accounts[index]
else: else:
@@ -157,7 +158,7 @@ def _filter_names(names):
code, names = z_util(interfaces.IDisplay).checklist( code, names = z_util(interfaces.IDisplay).checklist(
"Which names would you like to activate HTTPS for?", "Which names would you like to activate HTTPS for?",
tags=sorted_names, cli_flag="--domains") tags=sorted_names, cli_flag="--domains", force_interactive=True)
return code, [str(s) for s in names] return code, [str(s) for s in names]
@@ -173,7 +174,7 @@ def _choose_names_manually(prompt_prefix=""):
code, input_ = z_util(interfaces.IDisplay).input( code, input_ = z_util(interfaces.IDisplay).input(
prompt_prefix + prompt_prefix +
"Please enter in your domain name(s) (comma and/or space separated) ", "Please enter in your domain name(s) (comma and/or space separated) ",
cli_flag="--domains") cli_flag="--domains", force_interactive=True)
if code == display_util.OK: if code == display_util.OK:
invalid_domains = dict() invalid_domains = dict()
@@ -211,7 +212,8 @@ def _choose_names_manually(prompt_prefix=""):
if retry_message: if retry_message:
# We had error in input # We had error in input
retry = z_util(interfaces.IDisplay).yesno(retry_message) retry = z_util(interfaces.IDisplay).yesno(retry_message,
force_interactive=True)
if retry: if retry:
return _choose_names_manually() return _choose_names_manually()
else: else:
+120 -22
View File
@@ -1,14 +1,18 @@
"""Certbot display.""" """Certbot display."""
import logging
import os import os
import textwrap import textwrap
import sys
import six import six
import zope.interface import zope.interface
from certbot import constants
from certbot import interfaces from certbot import interfaces
from certbot import errors from certbot import errors
from certbot.display import completer from certbot.display import completer
logger = logging.getLogger(__name__)
WIDTH = 72 WIDTH = 72
@@ -50,19 +54,25 @@ def _wrap_lines(msg):
@zope.interface.implementer(interfaces.IDisplay) @zope.interface.implementer(interfaces.IDisplay)
class FileDisplay(object): class FileDisplay(object):
"""File-based display.""" """File-based display."""
# pylint: disable=too-many-arguments
# see https://github.com/certbot/certbot/issues/3915
def __init__(self, outfile): def __init__(self, outfile, force_interactive):
super(FileDisplay, self).__init__() super(FileDisplay, self).__init__()
self.outfile = outfile self.outfile = outfile
self.force_interactive = force_interactive
self.skipped_interaction = False
def notification(self, message, pause=True, wrap=True): def notification(self, message, pause=True,
# pylint: disable=unused-argument wrap=True, force_interactive=False):
"""Displays a notification and waits for user acceptance. """Displays a notification and waits for user acceptance.
:param str message: Message to display :param str message: Message to display
:param bool pause: Whether or not the program should pause for the :param bool pause: Whether or not the program should pause for the
user's confirmation user's confirmation
:param bool wrap: Whether or not the application should wrap text :param bool wrap: Whether or not the application should wrap text
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
""" """
side_frame = "-" * 79 side_frame = "-" * 79
@@ -72,10 +82,14 @@ class FileDisplay(object):
"{line}{frame}{line}{msg}{line}{frame}{line}".format( "{line}{frame}{line}{msg}{line}{frame}{line}".format(
line=os.linesep, frame=side_frame, msg=message)) line=os.linesep, frame=side_frame, msg=message))
if pause: if pause:
six.moves.input("Press Enter to Continue") if self._can_interact(force_interactive):
six.moves.input("Press Enter to Continue")
else:
logger.debug("Not pausing for user confirmation")
def menu(self, message, choices, ok_label="", cancel_label="", def menu(self, message, choices, ok_label="", cancel_label="",
help_label="", **unused_kwargs): help_label="", default=None,
cli_flag=None, force_interactive=False, **unused_kwargs):
# pylint: disable=unused-argument # pylint: disable=unused-argument
"""Display a menu. """Display a menu.
@@ -86,7 +100,10 @@ class FileDisplay(object):
:param choices: Menu lines, len must be > 0 :param choices: Menu lines, len must be > 0
:type choices: list of tuples (tag, item) or :type choices: list of tuples (tag, item) or
list of descriptions (tags will be enumerated) list of descriptions (tags will be enumerated)
:param dict _kwargs: absorbs default / cli_args :param default: default value to return (if one exists)
:param str cli_flag: option used to set this value with the CLI
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of (`code`, `index`) where :returns: tuple of (`code`, `index`) where
`code` - str display exit code `code` - str display exit code
@@ -95,18 +112,25 @@ class FileDisplay(object):
:rtype: tuple :rtype: tuple
""" """
if self._return_default(message, default, cli_flag, force_interactive):
return OK, default
self._print_menu(message, choices) self._print_menu(message, choices)
code, selection = self._get_valid_int_ans(len(choices)) code, selection = self._get_valid_int_ans(len(choices))
return code, selection - 1 return code, selection - 1
def input(self, message, **unused_kwargs): def input(self, message, default=None,
cli_flag=None, force_interactive=False, **unused_kwargs):
# pylint: disable=no-self-use # pylint: disable=no-self-use
"""Accept input from the user. """Accept input from the user.
:param str message: message to display to the user :param str message: message to display to the user
:param dict _kwargs: absorbs default / cli_args :param default: default value to return (if one exists)
:param str cli_flag: option used to set this value with the CLI
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of (`code`, `input`) where :returns: tuple of (`code`, `input`) where
`code` - str display exit code `code` - str display exit code
@@ -114,6 +138,9 @@ class FileDisplay(object):
:rtype: tuple :rtype: tuple
""" """
if self._return_default(message, default, cli_flag, force_interactive):
return OK, default
ans = six.moves.input( ans = six.moves.input(
textwrap.fill( textwrap.fill(
"%s (Enter 'c' to cancel): " % message, "%s (Enter 'c' to cancel): " % message,
@@ -126,7 +153,8 @@ class FileDisplay(object):
else: else:
return OK, ans return OK, ans
def yesno(self, message, yes_label="Yes", no_label="No", **unused_kwargs): def yesno(self, message, yes_label="Yes", no_label="No", default=None,
cli_flag=None, force_interactive=False, **unused_kwargs):
"""Query the user with a yes/no question. """Query the user with a yes/no question.
Yes and No label must begin with different letters, and must contain at Yes and No label must begin with different letters, and must contain at
@@ -135,12 +163,18 @@ class FileDisplay(object):
:param str message: question for the user :param str message: question for the user
:param str yes_label: Label of the "Yes" parameter :param str yes_label: Label of the "Yes" parameter
:param str no_label: Label of the "No" parameter :param str no_label: Label of the "No" parameter
:param dict _kwargs: absorbs default / cli_args :param default: default value to return (if one exists)
:param str cli_flag: option used to set this value with the CLI
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: True for "Yes", False for "No" :returns: True for "Yes", False for "No"
:rtype: bool :rtype: bool
""" """
if self._return_default(message, default, cli_flag, force_interactive):
return default
side_frame = ("-" * 79) + os.linesep side_frame = ("-" * 79) + os.linesep
message = _wrap_lines(message) message = _wrap_lines(message)
@@ -162,14 +196,18 @@ class FileDisplay(object):
ans.startswith(no_label[0].upper())): ans.startswith(no_label[0].upper())):
return False return False
def checklist(self, message, tags, default_status=True, **unused_kwargs): def checklist(self, message, tags, default_status=True, default=None,
cli_flag=None, force_interactive=False, **unused_kwargs):
# pylint: disable=unused-argument # pylint: disable=unused-argument
"""Display a checklist. """Display a checklist.
:param str message: Message to display to user :param str message: Message to display to user
:param list tags: `str` tags to select, len(tags) > 0 :param list tags: `str` tags to select, len(tags) > 0
:param bool default_status: Not used for FileDisplay :param bool default_status: Not used for FileDisplay
:param dict _kwargs: absorbs default / cli_args :param default: default value to return (if one exists)
:param str cli_flag: option used to set this value with the CLI
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of (`code`, `tags`) where :returns: tuple of (`code`, `tags`) where
`code` - str display exit code `code` - str display exit code
@@ -177,6 +215,9 @@ class FileDisplay(object):
:rtype: tuple :rtype: tuple
""" """
if self._return_default(message, default, cli_flag, force_interactive):
return OK, default
while True: while True:
self._print_menu(message, tags) self._print_menu(message, tags)
@@ -197,10 +238,65 @@ class FileDisplay(object):
else: else:
return code, [] return code, []
def directory_select(self, message, **unused_kwargs): def _return_default(self, prompt, default, cli_flag, force_interactive):
"""Should we return the default instead of prompting the user?
:param str prompt: prompt for the user
:param default: default answer to prompt
:param str cli_flag: command line option for setting an answer
to this question
:param bool force_interactive: if interactivity is forced by the
IDisplay call
:returns: True if we should return the default without prompting
:rtype: bool
"""
msg = "Invalid IDisplay call for this prompt:\n{0}".format(prompt)
if cli_flag:
msg += ("\nYou can set an answer to "
"this prompt with the {0} flag".format(cli_flag))
assert default is not None or force_interactive, msg
if self._can_interact(force_interactive):
return False
else:
logger.debug(
"Falling back to default %s for the prompt:\n%s",
default, prompt)
return True
def _can_interact(self, force_interactive):
"""Can we safely interact with the user?
:param bool force_interactive: if interactivity is forced by the
IDisplay call
:returns: True if the display can interact with the user
:rtype: bool
"""
if (self.force_interactive or force_interactive or
sys.stdin.isatty() and self.outfile.isatty()):
return True
elif not self.skipped_interaction:
logger.warning(
"Skipped user interaction because Certbot doesn't appear to "
"be running in a terminal. You should probably include "
"--non-interactive or %s on the command line.",
constants.FORCE_INTERACTIVE_FLAG)
self.skipped_interaction = True
return False
def directory_select(self, message, default=None, cli_flag=None,
force_interactive=False, **unused_kwargs):
"""Display a directory selection screen. """Display a directory selection screen.
:param str message: prompt to give the user :param str message: prompt to give the user
:param default: default value to return (if one exists)
:param str cli_flag: option used to set this value with the CLI
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of the form (`code`, `string`) where :returns: tuple of the form (`code`, `string`) where
`code` - display exit code `code` - display exit code
@@ -208,7 +304,7 @@ class FileDisplay(object):
""" """
with completer.Completer(): with completer.Completer():
return self.input(message) return self.input(message, default, cli_flag, force_interactive)
def _scrub_checklist_input(self, indices, tags): def _scrub_checklist_input(self, indices, tags):
# pylint: disable=no-self-use # pylint: disable=no-self-use
@@ -310,7 +406,7 @@ class FileDisplay(object):
class NoninteractiveDisplay(object): class NoninteractiveDisplay(object):
"""An iDisplay implementation that never asks for interactive user input""" """An iDisplay implementation that never asks for interactive user input"""
def __init__(self, outfile): def __init__(self, outfile, *unused_args, **unused_kwargs):
super(NoninteractiveDisplay, self).__init__() super(NoninteractiveDisplay, self).__init__()
self.outfile = outfile self.outfile = outfile
@@ -324,7 +420,7 @@ class NoninteractiveDisplay(object):
msg += "\n\n(You can set this with the {0} flag)".format(cli_flag) msg += "\n\n(You can set this with the {0} flag)".format(cli_flag)
raise errors.MissingCommandlineFlag(msg) raise errors.MissingCommandlineFlag(msg)
def notification(self, message, pause=False, wrap=True): def notification(self, message, pause=False, wrap=True, **unused_kwargs):
# pylint: disable=unused-argument # pylint: disable=unused-argument
"""Displays a notification without waiting for user acceptance. """Displays a notification without waiting for user acceptance.
@@ -341,7 +437,7 @@ class NoninteractiveDisplay(object):
line=os.linesep, frame=side_frame, msg=message)) line=os.linesep, frame=side_frame, msg=message))
def menu(self, message, choices, ok_label=None, cancel_label=None, def menu(self, message, choices, ok_label=None, cancel_label=None,
help_label=None, default=None, cli_flag=None): help_label=None, default=None, cli_flag=None, *unused_kwargs):
# pylint: disable=unused-argument,too-many-arguments # pylint: disable=unused-argument,too-many-arguments
"""Avoid displaying a menu. """Avoid displaying a menu.
@@ -364,7 +460,7 @@ class NoninteractiveDisplay(object):
return OK, default return OK, default
def input(self, message, default=None, cli_flag=None): def input(self, message, default=None, cli_flag=None, **unused_kwargs):
"""Accept input from the user. """Accept input from the user.
:param str message: message to display to the user :param str message: message to display to the user
@@ -381,7 +477,8 @@ class NoninteractiveDisplay(object):
else: else:
return OK, default return OK, default
def yesno(self, message, yes_label=None, no_label=None, default=None, cli_flag=None): def yesno(self, message, yes_label=None, no_label=None,
default=None, cli_flag=None, **unused_kwargs):
# pylint: disable=unused-argument # pylint: disable=unused-argument
"""Decide Yes or No, without asking anybody """Decide Yes or No, without asking anybody
@@ -398,8 +495,8 @@ class NoninteractiveDisplay(object):
else: else:
return default return default
def checklist(self, message, tags, default=None, cli_flag=None, **kwargs): def checklist(self, message, tags, default=None,
# pylint: disable=unused-argument cli_flag=None, **unused_kwargs):
"""Display a checklist. """Display a checklist.
:param str message: Message to display to user :param str message: Message to display to user
@@ -417,7 +514,8 @@ class NoninteractiveDisplay(object):
else: else:
return OK, default return OK, default
def directory_select(self, message, default=None, cli_flag=None): def directory_select(self, message, default=None,
cli_flag=None, **unused_kwargs):
"""Simulate prompting the user for a directory. """Simulate prompting the user for a directory.
This function returns default if it is not ``None``, otherwise, This function returns default if it is not ``None``, otherwise,
+40 -7
View File
@@ -361,21 +361,29 @@ class IInstaller(IPlugin):
class IDisplay(zope.interface.Interface): class IDisplay(zope.interface.Interface):
"""Generic display.""" """Generic display."""
# pylint: disable=too-many-arguments
# see https://github.com/certbot/certbot/issues/3915
def notification(message, pause, wrap=True): def notification(message, pause, wrap=True, force_interactive=False):
"""Displays a string message """Displays a string message
:param str message: Message to display :param str message: Message to display
:param bool pause: Whether or not the application should pause for :param bool pause: Whether or not the application should pause for
confirmation (if available) confirmation (if available)
:param bool wrap: Whether or not the application should wrap text :param bool wrap: Whether or not the application should wrap text
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
""" """
def menu(message, choices, ok_label="OK", # pylint: disable=too-many-arguments def menu(message, choices, ok_label="OK",
cancel_label="Cancel", help_label="", default=None, cli_flag=None): cancel_label="Cancel", help_label="",
default=None, cli_flag=None, force_interactive=False):
"""Displays a generic menu. """Displays a generic menu.
When not setting force_interactive=True, you must provide a
default value.
:param str message: message to display :param str message: message to display
:param choices: choices :param choices: choices
@@ -386,6 +394,8 @@ class IDisplay(zope.interface.Interface):
:param str help_label: label for Help button :param str help_label: label for Help button
:param int default: default (non-interactive) choice from the menu :param int default: default (non-interactive) choice from the menu
:param str cli_flag: to automate choice from the menu, eg "--keep" :param str cli_flag: to automate choice from the menu, eg "--keep"
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of (`code`, `index`) where :returns: tuple of (`code`, `index`) where
`code` - str display exit code `code` - str display exit code
@@ -396,10 +406,16 @@ class IDisplay(zope.interface.Interface):
""" """
def input(message, default=None, cli_args=None): def input(message, default=None, cli_args=None, force_interactive=False):
"""Accept input from the user. """Accept input from the user.
When not setting force_interactive=True, you must provide a
default value.
:param str message: message to display to the user :param str message: message to display to the user
:param str default: default (non-interactive) response to prompt
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of (`code`, `input`) where :returns: tuple of (`code`, `input`) where
`code` - str display exit code `code` - str display exit code
@@ -412,14 +428,19 @@ class IDisplay(zope.interface.Interface):
""" """
def yesno(message, yes_label="Yes", no_label="No", default=None, def yesno(message, yes_label="Yes", no_label="No", default=None,
cli_args=None): cli_args=None, force_interactive=False):
"""Query the user with a yes/no question. """Query the user with a yes/no question.
Yes and No label must begin with different letters. Yes and No label must begin with different letters.
When not setting force_interactive=True, you must provide a
default value.
:param str message: question for the user :param str message: question for the user
:param str default: default (non-interactive) choice from the menu :param str default: default (non-interactive) choice from the menu
:param str cli_flag: to automate choice from the menu, eg "--redirect / --no-redirect" :param str cli_flag: to automate choice from the menu, eg "--redirect / --no-redirect"
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: True for "Yes", False for "No" :returns: True for "Yes", False for "No"
:rtype: bool :rtype: bool
@@ -429,14 +450,20 @@ class IDisplay(zope.interface.Interface):
""" """
def checklist(message, tags, default_state, default=None, cli_args=None): def checklist(message, tags, default_state,
default=None, cli_args=None, force_interactive=False):
"""Allow for multiple selections from a menu. """Allow for multiple selections from a menu.
When not setting force_interactive=True, you must provide a
default value.
:param str message: message to display to the user :param str message: message to display to the user
:param list tags: where each is of type :class:`str` len(tags) > 0 :param list tags: where each is of type :class:`str` len(tags) > 0
:param bool default_status: If True, items are in a selected state by default. :param bool default_status: If True, items are in a selected state by default.
:param str default: default (non-interactive) state of the checklist :param str default: default (non-interactive) state of the checklist
:param str cli_flag: to automate choice from the menu, eg "--domains" :param str cli_flag: to automate choice from the menu, eg "--domains"
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of the form (code, list_tags) where :returns: tuple of the form (code, list_tags) where
`code` - int display exit code `code` - int display exit code
@@ -448,15 +475,21 @@ class IDisplay(zope.interface.Interface):
""" """
def directory_select(self, message, default=None, cli_flag=None): def directory_select(self, message, default=None,
cli_flag=None, force_interactive=False):
"""Display a directory selection screen. """Display a directory selection screen.
When not setting force_interactive=True, you must provide a
default value.
:param str message: prompt to give the user :param str message: prompt to give the user
:param default: the default value to return, if one exists, when :param default: the default value to return, if one exists, when
using the NoninteractiveDisplay using the NoninteractiveDisplay
:param str cli_flag: option used to set this value with the CLI, :param str cli_flag: option used to set this value with the CLI,
if one exists, to be included in error messages given by if one exists, to be included in error messages given by
NoninteractiveDisplay NoninteractiveDisplay
:param bool force_interactive: True if it's safe to prompt the user
because it won't cause any workflow regressions
:returns: tuple of the form (`code`, `string`) where :returns: tuple of the form (`code`, `string`) where
`code` - int display exit code `code` - int display exit code
+8 -4
View File
@@ -139,7 +139,8 @@ def _handle_subset_cert_request(config, domains, cert):
br=os.linesep) br=os.linesep)
if config.expand or config.renew_by_default or zope.component.getUtility( if config.expand or config.renew_by_default or zope.component.getUtility(
interfaces.IDisplay).yesno(question, "Expand", "Cancel", interfaces.IDisplay).yesno(question, "Expand", "Cancel",
cli_flag="--expand"): cli_flag="--expand",
force_interactive=True):
return "renew", cert return "renew", cert
else: else:
reporter_util = zope.component.getUtility(interfaces.IReporter) reporter_util = zope.component.getUtility(interfaces.IReporter)
@@ -188,7 +189,8 @@ def _handle_identical_cert_request(config, lineage):
"Renew & replace the cert (limit ~5 per 7 days)"] "Renew & replace the cert (limit ~5 per 7 days)"]
display = zope.component.getUtility(interfaces.IDisplay) display = zope.component.getUtility(interfaces.IDisplay)
response = display.menu(question, choices, "OK", "Cancel", default=0) response = display.menu(question, choices, "OK", "Cancel",
default=0, force_interactive=True)
if response[0] == display_util.CANCEL: if response[0] == display_util.CANCEL:
# TODO: Add notification related to command-line options for # TODO: Add notification related to command-line options for
# skipping the menu for this case. # skipping the menu for this case.
@@ -365,7 +367,8 @@ def _determine_account(config):
"server at {1}".format( "server at {1}".format(
regr.terms_of_service, config.server)) regr.terms_of_service, config.server))
obj = zope.component.getUtility(interfaces.IDisplay) obj = zope.component.getUtility(interfaces.IDisplay)
return obj.yesno(msg, "Agree", "Cancel", cli_flag="--agree-tos") return obj.yesno(msg, "Agree", "Cancel",
cli_flag="--agree-tos", force_interactive=True)
try: try:
acc, acme = client.register( acc, acme = client.register(
@@ -788,7 +791,8 @@ def set_displayer(config):
elif config.noninteractive_mode: elif config.noninteractive_mode:
displayer = display_util.NoninteractiveDisplay(sys.stdout) displayer = display_util.NoninteractiveDisplay(sys.stdout)
else: else:
displayer = display_util.FileDisplay(sys.stdout) displayer = display_util.FileDisplay(sys.stdout,
config.force_interactive)
zope.component.provideUtility(displayer) zope.component.provideUtility(displayer)
+2 -1
View File
@@ -251,7 +251,8 @@ s.serve_forever()" """
if not (self.conf("test-mode") or self.conf("public-ip-logging-ok")): if not (self.conf("test-mode") or self.conf("public-ip-logging-ok")):
if not zope.component.getUtility(interfaces.IDisplay).yesno( if not zope.component.getUtility(interfaces.IDisplay).yesno(
self.IP_DISCLAIMER, "Yes", "No", self.IP_DISCLAIMER, "Yes", "No",
cli_flag="--manual-public-ip-logging-ok"): cli_flag="--manual-public-ip-logging-ok",
force_interactive=True):
raise errors.PluginError("Must agree to IP logging to proceed") raise errors.PluginError("Must agree to IP logging to proceed")
else: else:
self.config.namespace.manual_public_ip_logging_ok = True self.config.namespace.manual_public_ip_logging_ok = True
+4 -2
View File
@@ -111,7 +111,8 @@ def choose_plugin(prepared, question):
while True: while True:
disp = z_util(interfaces.IDisplay) disp = z_util(interfaces.IDisplay)
code, index = disp.menu(question, opts, help_label="More Info") code, index = disp.menu(
question, opts, help_label="More Info", force_interactive=True)
if code == display_util.OK: if code == display_util.OK:
plugin_ep = prepared[index] plugin_ep = prepared[index]
@@ -127,7 +128,8 @@ def choose_plugin(prepared, question):
msg = "Reported Error: %s" % prepared[index].prepare() msg = "Reported Error: %s" % prepared[index].prepare()
else: else:
msg = prepared[index].init().more_info() msg = prepared[index].init().more_info()
z_util(interfaces.IDisplay).notification(msg) z_util(interfaces.IDisplay).notification(msg,
force_interactive=True)
else: else:
return None return None
+2 -1
View File
@@ -110,7 +110,8 @@ class ChoosePluginTest(unittest.TestCase):
"""Tests for certbot.plugins.selection.choose_plugin.""" """Tests for certbot.plugins.selection.choose_plugin."""
def setUp(self): def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout)) zope.component.provideUtility(display_util.FileDisplay(sys.stdout,
False))
self.mock_apache = mock.Mock( self.mock_apache = mock.Mock(
description_with_name="a", misconfigured=True) description_with_name="a", misconfigured=True)
self.mock_stand = mock.Mock( self.mock_stand = mock.Mock(
+2 -2
View File
@@ -243,13 +243,13 @@ class Authenticator(common.Plugin):
"Could not bind TCP port {0} because you don't have " "Could not bind TCP port {0} because you don't have "
"the appropriate permissions (for example, you " "the appropriate permissions (for example, you "
"aren't running this program as " "aren't running this program as "
"root).".format(error.port)) "root).".format(error.port), force_interactive=True)
elif error.socket_error.errno == socket.errno.EADDRINUSE: elif error.socket_error.errno == socket.errno.EADDRINUSE:
display.notification( display.notification(
"Could not bind TCP port {0} because it is already in " "Could not bind TCP port {0} because it is already in "
"use by another process on this system (such as a web " "use by another process on this system (such as a web "
"server). Please stop the program in question and then " "server). Please stop the program in question and then "
"try again.".format(error.port)) "try again.".format(error.port), force_interactive=True)
else: else:
raise # XXX: How to handle unknown errors in binding? raise # XXX: How to handle unknown errors in binding?
+3 -2
View File
@@ -103,7 +103,7 @@ def already_listening_socket(port, renewer=False):
"Port {0} is already in use by another process. This will " "Port {0} is already in use by another process. This will "
"prevent us from binding to that port. Please stop the " "prevent us from binding to that port. Please stop the "
"process that is populating the port in question and try " "process that is populating the port in question and try "
"again. {1}".format(port, extra)) "again. {1}".format(port, extra), force_interactive=True)
return True return True
finally: finally:
testsocket.close() testsocket.close()
@@ -151,7 +151,8 @@ def already_listening_psutil(port, renewer=False):
"The program {0} (process ID {1}) is already listening " "The program {0} (process ID {1}) is already listening "
"on TCP port {2}. This will prevent us from binding to " "on TCP port {2}. This will prevent us from binding to "
"that port. Please stop the {0} program temporarily " "that port. Please stop the {0} program temporarily "
"and then try again.{3}".format(name, pid, port, extra)) "and then try again.{3}".format(name, pid, port, extra),
force_interactive=True)
return True return True
except (psutil.NoSuchProcess, psutil.AccessDenied): except (psutil.NoSuchProcess, psutil.AccessDenied):
# Perhaps the result of a race where the process could have # Perhaps the result of a race where the process could have
+4 -2
View File
@@ -129,7 +129,8 @@ to serve all files under specified web root ({0})."""
"public_html or webroot directory. The webroot " "public_html or webroot directory. The webroot "
"plugin works by temporarily saving necessary " "plugin works by temporarily saving necessary "
"resources in the HTTP server's webroot directory " "resources in the HTTP server's webroot directory "
"to pass domain validation challenges.") "to pass domain validation challenges.",
force_interactive=True)
else: # code == display_util.OK else: # code == display_util.OK
return None if index == 0 else known_webroots[index - 1] return None if index == 0 else known_webroots[index - 1]
@@ -138,7 +139,8 @@ to serve all files under specified web root ({0})."""
while True: while True:
code, webroot = display.directory_select( code, webroot = display.directory_select(
"Input the webroot for {0}:".format(domain)) "Input the webroot for {0}:".format(domain),
force_interactive=True)
if code == display_util.HELP: if code == display_util.HELP:
# Displaying help is not currently implemented # Displaying help is not currently implemented
return None return None
+1 -1
View File
@@ -181,7 +181,7 @@ class Reverter(object):
if for_logging: if for_logging:
return os.linesep.join(output) return os.linesep.join(output)
zope.component.getUtility(interfaces.IDisplay).notification( zope.component.getUtility(interfaces.IDisplay).notification(
os.linesep.join(output)) os.linesep.join(output), force_interactive=True)
def add_to_temp_checkpoint(self, save_files, save_notes): def add_to_temp_checkpoint(self, save_files, save_notes):
"""Add files to temporary checkpoint. """Add files to temporary checkpoint.
+6
View File
@@ -262,6 +262,12 @@ class ParseTest(unittest.TestCase):
self.assertFalse(cli.option_was_set( self.assertFalse(cli.option_was_set(
config_dir_option, cli.flag_default(config_dir_option))) config_dir_option, cli.flag_default(config_dir_option)))
def test_force_interactive(self):
self.assertRaises(
errors.Error, self.parse, "renew --force-interactive".split())
self.assertRaises(
errors.Error, self.parse, "-n --force-interactive".split())
class DefaultTest(unittest.TestCase): class DefaultTest(unittest.TestCase):
"""Tests for certbot.cli._Default.""" """Tests for certbot.cli._Default."""
+8 -4
View File
@@ -84,7 +84,8 @@ class GetEmailTest(unittest.TestCase):
class ChooseAccountTest(unittest.TestCase): class ChooseAccountTest(unittest.TestCase):
"""Tests for certbot.display.ops.choose_account.""" """Tests for certbot.display.ops.choose_account."""
def setUp(self): def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout)) zope.component.provideUtility(display_util.FileDisplay(sys.stdout,
False))
self.accounts_dir = tempfile.mkdtemp("accounts") self.accounts_dir = tempfile.mkdtemp("accounts")
self.account_keys_dir = os.path.join(self.accounts_dir, "keys") self.account_keys_dir = os.path.join(self.accounts_dir, "keys")
@@ -127,7 +128,8 @@ class ChooseAccountTest(unittest.TestCase):
class GenSSLLabURLs(unittest.TestCase): class GenSSLLabURLs(unittest.TestCase):
"""Loose test of _gen_ssl_lab_urls. URL can change easily in the future.""" """Loose test of _gen_ssl_lab_urls. URL can change easily in the future."""
def setUp(self): def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout)) zope.component.provideUtility(display_util.FileDisplay(sys.stdout,
False))
@classmethod @classmethod
def _call(cls, domains): def _call(cls, domains):
@@ -146,7 +148,8 @@ class GenSSLLabURLs(unittest.TestCase):
class GenHttpsNamesTest(unittest.TestCase): class GenHttpsNamesTest(unittest.TestCase):
"""Test _gen_https_names.""" """Test _gen_https_names."""
def setUp(self): def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout)) zope.component.provideUtility(display_util.FileDisplay(sys.stdout,
False))
@classmethod @classmethod
def _call(cls, domains): def _call(cls, domains):
@@ -193,7 +196,8 @@ class GenHttpsNamesTest(unittest.TestCase):
class ChooseNamesTest(unittest.TestCase): class ChooseNamesTest(unittest.TestCase):
"""Test choose names.""" """Test choose names."""
def setUp(self): def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout)) zope.component.provideUtility(display_util.FileDisplay(sys.stdout,
False))
self.mock_install = mock.MagicMock() self.mock_install = mock.MagicMock()
@classmethod @classmethod
+80 -19
View File
@@ -20,10 +20,11 @@ class FileOutputDisplayTest(unittest.TestCase):
functions look to a user, uncomment the test_visual function. functions look to a user, uncomment the test_visual function.
""" """
# pylint:disable=too-many-public-methods
def setUp(self): def setUp(self):
super(FileOutputDisplayTest, self).setUp() super(FileOutputDisplayTest, self).setUp()
self.mock_stdout = mock.MagicMock() self.mock_stdout = mock.MagicMock()
self.displayer = display_util.FileDisplay(self.mock_stdout) self.displayer = display_util.FileDisplay(self.mock_stdout, False)
def test_notification_no_pause(self): def test_notification_no_pause(self):
self.displayer.notification("message", False) self.displayer.notification("message", False)
@@ -33,56 +34,84 @@ class FileOutputDisplayTest(unittest.TestCase):
def test_notification_pause(self): def test_notification_pause(self):
with mock.patch("six.moves.input", return_value="enter"): with mock.patch("six.moves.input", return_value="enter"):
self.displayer.notification("message") self.displayer.notification("message", force_interactive=True)
self.assertTrue("message" in self.mock_stdout.write.call_args[0][0]) self.assertTrue("message" in self.mock_stdout.write.call_args[0][0])
def test_notification_noninteractive(self):
self._force_noninteractive(self.displayer.notification, "message")
string = self.mock_stdout.write.call_args[0][0]
self.assertTrue("message" in string)
@mock.patch("certbot.display.util." @mock.patch("certbot.display.util."
"FileDisplay._get_valid_int_ans") "FileDisplay._get_valid_int_ans")
def test_menu(self, mock_ans): def test_menu(self, mock_ans):
mock_ans.return_value = (display_util.OK, 1) mock_ans.return_value = (display_util.OK, 1)
ret = self.displayer.menu("message", CHOICES) ret = self.displayer.menu("message", CHOICES, force_interactive=True)
self.assertEqual(ret, (display_util.OK, 0)) self.assertEqual(ret, (display_util.OK, 0))
def test_input_cancel(self): def test_input_cancel(self):
with mock.patch("six.moves.input", return_value="c"): with mock.patch("six.moves.input", return_value="c"):
code, _ = self.displayer.input("message") code, _ = self.displayer.input("message", force_interactive=True)
self.assertTrue(code, display_util.CANCEL) self.assertTrue(code, display_util.CANCEL)
def test_input_normal(self): def test_input_normal(self):
with mock.patch("six.moves.input", return_value="domain.com"): with mock.patch("six.moves.input", return_value="domain.com"):
code, input_ = self.displayer.input("message") code, input_ = self.displayer.input("message", force_interactive=True)
self.assertEqual(code, display_util.OK) self.assertEqual(code, display_util.OK)
self.assertEqual(input_, "domain.com") self.assertEqual(input_, "domain.com")
def test_input_noninteractive(self):
default = "foo"
code, input_ = self._force_noninteractive(
self.displayer.input, "message", default=default)
self.assertEqual(code, display_util.OK)
self.assertEqual(input_, default)
def test_input_assertion_fail(self):
self.assertRaises(AssertionError, self._force_noninteractive,
self.displayer.input, "message", cli_flag="--flag")
def test_yesno(self): def test_yesno(self):
with mock.patch("six.moves.input", return_value="Yes"): with mock.patch("six.moves.input", return_value="Yes"):
self.assertTrue(self.displayer.yesno("message")) self.assertTrue(self.displayer.yesno(
"message", force_interactive=True))
with mock.patch("six.moves.input", return_value="y"): with mock.patch("six.moves.input", return_value="y"):
self.assertTrue(self.displayer.yesno("message")) self.assertTrue(self.displayer.yesno(
"message", force_interactive=True))
with mock.patch("six.moves.input", side_effect=["maybe", "y"]): with mock.patch("six.moves.input", side_effect=["maybe", "y"]):
self.assertTrue(self.displayer.yesno("message")) self.assertTrue(self.displayer.yesno(
"message", force_interactive=True))
with mock.patch("six.moves.input", return_value="No"): with mock.patch("six.moves.input", return_value="No"):
self.assertFalse(self.displayer.yesno("message")) self.assertFalse(self.displayer.yesno(
"message", force_interactive=True))
with mock.patch("six.moves.input", side_effect=["cancel", "n"]): with mock.patch("six.moves.input", side_effect=["cancel", "n"]):
self.assertFalse(self.displayer.yesno("message")) self.assertFalse(self.displayer.yesno(
"message", force_interactive=True))
with mock.patch("six.moves.input", return_value="a"): with mock.patch("six.moves.input", return_value="a"):
self.assertTrue(self.displayer.yesno("msg", yes_label="Agree")) self.assertTrue(self.displayer.yesno(
"msg", yes_label="Agree", force_interactive=True))
def test_yesno_noninteractive(self):
self.assertTrue(self._force_noninteractive(
self.displayer.yesno, "message", default=True))
@mock.patch("certbot.display.util.FileDisplay.input") @mock.patch("certbot.display.util.FileDisplay.input")
def test_checklist_valid(self, mock_input): def test_checklist_valid(self, mock_input):
mock_input.return_value = (display_util.OK, "2 1") mock_input.return_value = (display_util.OK, "2 1")
code, tag_list = self.displayer.checklist("msg", TAGS) code, tag_list = self.displayer.checklist(
"msg", TAGS, force_interactive=True)
self.assertEqual( self.assertEqual(
(code, set(tag_list)), (display_util.OK, set(["tag1", "tag2"]))) (code, set(tag_list)), (display_util.OK, set(["tag1", "tag2"])))
@mock.patch("certbot.display.util.FileDisplay.input") @mock.patch("certbot.display.util.FileDisplay.input")
def test_checklist_empty(self, mock_input): def test_checklist_empty(self, mock_input):
mock_input.return_value = (display_util.OK, "") mock_input.return_value = (display_util.OK, "")
code, tag_list = self.displayer.checklist("msg", TAGS) code, tag_list = self.displayer.checklist("msg", TAGS, force_interactive=True)
self.assertEqual( self.assertEqual(
(code, set(tag_list)), (display_util.OK, set(["tag1", "tag2", "tag3"]))) (code, set(tag_list)), (display_util.OK, set(["tag1", "tag2", "tag3"])))
@@ -94,7 +123,7 @@ class FileOutputDisplayTest(unittest.TestCase):
(display_util.OK, "1") (display_util.OK, "1")
] ]
ret = self.displayer.checklist("msg", TAGS) ret = self.displayer.checklist("msg", TAGS, force_interactive=True)
self.assertEqual(ret, (display_util.OK, ["tag1"])) self.assertEqual(ret, (display_util.OK, ["tag1"]))
@mock.patch("certbot.display.util.FileDisplay.input") @mock.patch("certbot.display.util.FileDisplay.input")
@@ -103,9 +132,17 @@ class FileOutputDisplayTest(unittest.TestCase):
(display_util.OK, "10"), (display_util.OK, "10"),
(display_util.CANCEL, "1") (display_util.CANCEL, "1")
] ]
ret = self.displayer.checklist("msg", TAGS) ret = self.displayer.checklist("msg", TAGS, force_interactive=True)
self.assertEqual(ret, (display_util.CANCEL, [])) self.assertEqual(ret, (display_util.CANCEL, []))
def test_checklist_noninteractive(self):
default = TAGS
code, input_ = self._force_noninteractive(
self.displayer.checklist, "msg", TAGS, default=default)
self.assertEqual(code, display_util.OK)
self.assertEqual(input_, default)
def test_scrub_checklist_input_valid(self): def test_scrub_checklist_input_valid(self):
# pylint: disable=protected-access # pylint: disable=protected-access
indices = [ indices = [
@@ -125,12 +162,36 @@ class FileOutputDisplayTest(unittest.TestCase):
@mock.patch("certbot.display.util.FileDisplay.input") @mock.patch("certbot.display.util.FileDisplay.input")
def test_directory_select(self, mock_input): def test_directory_select(self, mock_input):
message = "msg" # pylint: disable=star-args
result = (display_util.OK, "/var/www/html",) args = ["msg", "/var/www/html", "--flag", True]
result = (display_util.OK, "/var/www/html")
mock_input.return_value = result mock_input.return_value = result
self.assertEqual(self.displayer.directory_select(message), result) self.assertEqual(self.displayer.directory_select(*args), result)
mock_input.assert_called_once_with(message) mock_input.assert_called_once_with(*args)
def test_directory_select_noninteractive(self):
default = "/var/www/html"
code, input_ = self._force_noninteractive(
self.displayer.directory_select, "msg", default=default)
self.assertEqual(code, display_util.OK)
self.assertEqual(input_, default)
def _force_noninteractive(self, func, *args, **kwargs):
skipped_interaction = self.displayer.skipped_interaction
with mock.patch("certbot.display.util.sys.stdin") as mock_stdin:
mock_stdin.isatty.return_value = False
with mock.patch("certbot.display.util.logger") as mock_logger:
result = func(*args, **kwargs)
if skipped_interaction:
self.assertFalse(mock_logger.warning.called)
else:
self.assertEqual(mock_logger.warning.call_count, 1)
return result
def test_scrub_checklist_input_invalid(self): def test_scrub_checklist_input_invalid(self):
# pylint: disable=protected-access # pylint: disable=protected-access
+1 -1
View File
@@ -18,5 +18,5 @@ def test_visual(displayer, choices):
if __name__ == "__main__": if __name__ == "__main__":
displayer = util.FileDisplay(sys.stdout): displayer = util.FileDisplay(sys.stdout, False)
test_visual(displayer, util_test.CHOICES) test_visual(displayer, util_test.CHOICES)