mirror of
https://github.com/certbot/certbot.git
synced 2026-08-03 08:03:10 +02:00
Fix circular import (#8967)
* add internal display util * Move display constants internal. * move other utilities internal * fix OK and CANCEL documentation
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
"""This modules define the actual display implementations used in Certbot"""
|
||||
import logging
|
||||
import sys
|
||||
import textwrap
|
||||
from typing import Any
|
||||
from typing import Optional
|
||||
from typing import Union
|
||||
@@ -13,11 +12,22 @@ from certbot import errors
|
||||
from certbot import interfaces
|
||||
from certbot._internal import constants
|
||||
from certbot._internal.display import completer
|
||||
from certbot._internal.display import util
|
||||
from certbot.compat import os
|
||||
from certbot.display import util
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Display exit codes
|
||||
OK = "ok"
|
||||
"""Display exit code indicating user acceptance."""
|
||||
|
||||
CANCEL = "cancel"
|
||||
"""Display exit code for a user canceling the display."""
|
||||
|
||||
# Display constants
|
||||
SIDE_FRAME = ("- " * 39) + "-"
|
||||
"""Display boundary (alternates spaces, so when copy-pasted, markdown doesn't interpret
|
||||
it as a heading)"""
|
||||
|
||||
# This class holds the global state of the display service. Using this class
|
||||
# eliminates potential gotchas that exist if self.display was just a global
|
||||
@@ -62,7 +72,7 @@ class FileDisplay:
|
||||
|
||||
"""
|
||||
if wrap:
|
||||
message = _wrap_lines(message)
|
||||
message = util.wrap_lines(message)
|
||||
|
||||
logger.debug("Notifying user: %s", message)
|
||||
|
||||
@@ -70,7 +80,7 @@ class FileDisplay:
|
||||
(("{line}{frame}{line}" if decorate else "") +
|
||||
"{msg}{line}" +
|
||||
("{frame}{line}" if decorate else ""))
|
||||
.format(line=os.linesep, frame=util.SIDE_FRAME, msg=message)
|
||||
.format(line=os.linesep, frame=SIDE_FRAME, msg=message)
|
||||
)
|
||||
self.outfile.flush()
|
||||
|
||||
@@ -105,7 +115,7 @@ class FileDisplay:
|
||||
|
||||
"""
|
||||
if self._return_default(message, default, cli_flag, force_interactive):
|
||||
return util.OK, default
|
||||
return OK, default
|
||||
|
||||
self._print_menu(message, choices)
|
||||
|
||||
@@ -130,15 +140,16 @@ class FileDisplay:
|
||||
|
||||
"""
|
||||
if self._return_default(message, default, cli_flag, force_interactive):
|
||||
return util.OK, default
|
||||
return OK, default
|
||||
|
||||
# Trailing space must be added outside of _wrap_lines to be preserved
|
||||
message = _wrap_lines("%s (Enter 'c' to cancel):" % message) + " "
|
||||
# Trailing space must be added outside of util.wrap_lines to
|
||||
# be preserved
|
||||
message = util.wrap_lines("%s (Enter 'c' to cancel):" % message) + " "
|
||||
ans = util.input_with_timeout(message)
|
||||
|
||||
if ans in ("c", "C"):
|
||||
return util.CANCEL, "-1"
|
||||
return util.OK, ans
|
||||
return CANCEL, "-1"
|
||||
return OK, ans
|
||||
|
||||
def yesno(self, message, yes_label="Yes", no_label="No", default=None,
|
||||
cli_flag=None, force_interactive=False, **unused_kwargs):
|
||||
@@ -162,16 +173,16 @@ class FileDisplay:
|
||||
if self._return_default(message, default, cli_flag, force_interactive):
|
||||
return default
|
||||
|
||||
message = _wrap_lines(message)
|
||||
message = util.wrap_lines(message)
|
||||
|
||||
self.outfile.write("{0}{frame}{msg}{0}{frame}".format(
|
||||
os.linesep, frame=util.SIDE_FRAME + os.linesep, msg=message))
|
||||
os.linesep, frame=SIDE_FRAME + os.linesep, msg=message))
|
||||
self.outfile.flush()
|
||||
|
||||
while True:
|
||||
ans = util.input_with_timeout("{yes}/{no}: ".format(
|
||||
yes=_parens_around_char(yes_label),
|
||||
no=_parens_around_char(no_label)))
|
||||
yes=util.parens_around_char(yes_label),
|
||||
no=util.parens_around_char(no_label)))
|
||||
|
||||
# Couldn't get pylint indentation right with elif
|
||||
# elif doesn't matter in this situation
|
||||
@@ -200,7 +211,7 @@ class FileDisplay:
|
||||
|
||||
"""
|
||||
if self._return_default(message, default, cli_flag, force_interactive):
|
||||
return util.OK, default
|
||||
return OK, default
|
||||
|
||||
while True:
|
||||
self._print_menu(message, tags)
|
||||
@@ -210,7 +221,7 @@ class FileDisplay:
|
||||
"blank to select all options shown",
|
||||
force_interactive=True)
|
||||
|
||||
if code == util.OK:
|
||||
if code == OK:
|
||||
if not ans.strip():
|
||||
ans = " ".join(str(x) for x in range(1, len(tags)+1))
|
||||
indices = util.separate_list_input(ans)
|
||||
@@ -334,17 +345,17 @@ class FileDisplay:
|
||||
# Write out the message to the user
|
||||
self.outfile.write(
|
||||
"{new}{msg}{new}".format(new=os.linesep, msg=message))
|
||||
self.outfile.write(util.SIDE_FRAME + os.linesep)
|
||||
self.outfile.write(SIDE_FRAME + os.linesep)
|
||||
|
||||
# Write out the menu choices
|
||||
for i, desc in enumerate(choices, 1):
|
||||
msg = "{num}: {desc}".format(num=i, desc=desc)
|
||||
self.outfile.write(_wrap_lines(msg))
|
||||
self.outfile.write(util.wrap_lines(msg))
|
||||
|
||||
# Keep this outside of the textwrap
|
||||
self.outfile.write(os.linesep)
|
||||
|
||||
self.outfile.write(util.SIDE_FRAME + os.linesep)
|
||||
self.outfile.write(SIDE_FRAME + os.linesep)
|
||||
self.outfile.flush()
|
||||
|
||||
def _get_valid_int_ans(self, max_):
|
||||
@@ -369,7 +380,7 @@ class FileDisplay:
|
||||
while selection < 1:
|
||||
ans = util.input_with_timeout(input_msg)
|
||||
if ans.startswith("c") or ans.startswith("C"):
|
||||
return util.CANCEL, -1
|
||||
return CANCEL, -1
|
||||
try:
|
||||
selection = int(ans)
|
||||
if selection < 1 or selection > max_:
|
||||
@@ -381,7 +392,7 @@ class FileDisplay:
|
||||
"{0}** Invalid input **{0}".format(os.linesep))
|
||||
self.outfile.flush()
|
||||
|
||||
return util.OK, selection
|
||||
return OK, selection
|
||||
|
||||
|
||||
# This use of IDisplay can be removed when this class is no longer accessible
|
||||
@@ -414,7 +425,7 @@ class NoninteractiveDisplay:
|
||||
|
||||
"""
|
||||
if wrap:
|
||||
message = _wrap_lines(message)
|
||||
message = util.wrap_lines(message)
|
||||
|
||||
logger.debug("Notifying user: %s", message)
|
||||
|
||||
@@ -422,7 +433,7 @@ class NoninteractiveDisplay:
|
||||
(("{line}{frame}{line}" if decorate else "") +
|
||||
"{msg}{line}" +
|
||||
("{frame}{line}" if decorate else ""))
|
||||
.format(line=os.linesep, frame=util.SIDE_FRAME, msg=message)
|
||||
.format(line=os.linesep, frame=SIDE_FRAME, msg=message)
|
||||
)
|
||||
self.outfile.flush()
|
||||
|
||||
@@ -448,7 +459,7 @@ class NoninteractiveDisplay:
|
||||
if default is None:
|
||||
self._interaction_fail(message, cli_flag, "Choices: " + repr(choices))
|
||||
|
||||
return util.OK, default
|
||||
return OK, default
|
||||
|
||||
def input(self, message, default=None, cli_flag=None, **unused_kwargs):
|
||||
"""Accept input from the user.
|
||||
@@ -464,7 +475,7 @@ class NoninteractiveDisplay:
|
||||
"""
|
||||
if default is None:
|
||||
self._interaction_fail(message, cli_flag)
|
||||
return util.OK, default
|
||||
return OK, default
|
||||
|
||||
def yesno(self, message, yes_label=None, no_label=None, # pylint: disable=unused-argument
|
||||
default=None, cli_flag=None, **unused_kwargs):
|
||||
@@ -498,7 +509,7 @@ class NoninteractiveDisplay:
|
||||
"""
|
||||
if default is None:
|
||||
self._interaction_fail(message, cli_flag, "? ".join(tags))
|
||||
return util.OK, default
|
||||
return OK, default
|
||||
|
||||
def directory_select(self, message, default=None,
|
||||
cli_flag=None, **unused_kwargs):
|
||||
@@ -551,34 +562,3 @@ def set_display(display: Any) -> None:
|
||||
zope.component.provideUtility(display, interfaces.IDisplay)
|
||||
|
||||
_SERVICE.display = display
|
||||
|
||||
|
||||
def _wrap_lines(msg):
|
||||
"""Format lines nicely to 80 chars.
|
||||
|
||||
:param str msg: Original message
|
||||
|
||||
:returns: Formatted message respecting newlines in message
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
lines = msg.splitlines()
|
||||
fixed_l = []
|
||||
|
||||
for line in lines:
|
||||
fixed_l.append(textwrap.fill(
|
||||
line,
|
||||
80,
|
||||
break_long_words=False,
|
||||
break_on_hyphens=False))
|
||||
|
||||
return '\n'.join(fixed_l)
|
||||
|
||||
|
||||
def _parens_around_char(label):
|
||||
"""Place parens around first character of label.
|
||||
|
||||
:param str label: Must contain at least one character
|
||||
|
||||
"""
|
||||
return "({first}){rest}".format(first=label[0], rest=label[1:])
|
||||
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Internal Certbot display utilities."""
|
||||
from typing import List
|
||||
import textwrap
|
||||
import sys
|
||||
|
||||
from certbot.compat import misc
|
||||
|
||||
|
||||
def wrap_lines(msg):
|
||||
"""Format lines nicely to 80 chars.
|
||||
|
||||
:param str msg: Original message
|
||||
|
||||
:returns: Formatted message respecting newlines in message
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
lines = msg.splitlines()
|
||||
fixed_l = []
|
||||
|
||||
for line in lines:
|
||||
fixed_l.append(textwrap.fill(
|
||||
line,
|
||||
80,
|
||||
break_long_words=False,
|
||||
break_on_hyphens=False))
|
||||
|
||||
return '\n'.join(fixed_l)
|
||||
|
||||
|
||||
def parens_around_char(label):
|
||||
"""Place parens around first character of label.
|
||||
|
||||
:param str label: Must contain at least one character
|
||||
|
||||
"""
|
||||
return "({first}){rest}".format(first=label[0], rest=label[1:])
|
||||
|
||||
|
||||
def input_with_timeout(prompt=None, timeout=36000.0):
|
||||
"""Get user input with a timeout.
|
||||
|
||||
Behaves the same as the builtin input, however, an error is raised if
|
||||
a user doesn't answer after timeout seconds. The default timeout
|
||||
value was chosen to place it just under 12 hours for users following
|
||||
our advice and running Certbot twice a day.
|
||||
|
||||
:param str prompt: prompt to provide for input
|
||||
:param float timeout: maximum number of seconds to wait for input
|
||||
|
||||
:returns: user response
|
||||
:rtype: str
|
||||
|
||||
:raises errors.Error if no answer is given before the timeout
|
||||
|
||||
"""
|
||||
# use of sys.stdin and sys.stdout to mimic the builtin input based on
|
||||
# https://github.com/python/cpython/blob/baf7bb30a02aabde260143136bdf5b3738a1d409/Lib/getpass.py#L129
|
||||
if prompt:
|
||||
sys.stdout.write(prompt)
|
||||
sys.stdout.flush()
|
||||
|
||||
line = misc.readline_with_timeout(timeout, prompt)
|
||||
|
||||
if not line:
|
||||
raise EOFError
|
||||
return line.rstrip('\n')
|
||||
|
||||
|
||||
def separate_list_input(input_):
|
||||
"""Separate a comma or space separated list.
|
||||
|
||||
:param str input_: input from the user
|
||||
|
||||
:returns: strings
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
no_commas = input_.replace(",", " ")
|
||||
# Each string is naturally unicode, this causes problems with M2Crypto SANs
|
||||
# TODO: check if above is still true when M2Crypto is gone ^
|
||||
return [str(string) for string in no_commas.split()]
|
||||
|
||||
|
||||
def summarize_domain_list(domains: List[str]) -> str:
|
||||
"""Summarizes a list of domains in the format of:
|
||||
example.com.com and N more domains
|
||||
or if there is are only two domains:
|
||||
example.com and www.example.com
|
||||
or if there is only one domain:
|
||||
example.com
|
||||
|
||||
:param list domains: `str` list of domains
|
||||
:returns: the domain list summary
|
||||
:rtype: str
|
||||
"""
|
||||
if not domains:
|
||||
return ""
|
||||
|
||||
l = len(domains)
|
||||
if l == 1:
|
||||
return domains[0]
|
||||
elif l == 2:
|
||||
return " and ".join(domains)
|
||||
else:
|
||||
return "{0} and {1} more domains".format(domains[0], l-1)
|
||||
@@ -38,6 +38,7 @@ from certbot._internal import snap_config
|
||||
from certbot._internal import storage
|
||||
from certbot._internal import updater
|
||||
from certbot._internal.display import obj as display_obj
|
||||
from certbot._internal.display import util as internal_display_util
|
||||
from certbot._internal.plugins import disco as plugins_disco
|
||||
from certbot._internal.plugins import selection as plug_sel
|
||||
from certbot.compat import filesystem
|
||||
@@ -111,7 +112,7 @@ def _get_and_save_cert(le_client, config, domains=None, certname=None, lineage=N
|
||||
"{action} for {domains}".format(
|
||||
action="Simulating renewal of an existing certificate"
|
||||
if config.dry_run else "Renewing an existing certificate",
|
||||
domains=display_util.summarize_domain_list(domains or lineage.names())
|
||||
domains=internal_display_util.summarize_domain_list(domains or lineage.names())
|
||||
)
|
||||
)
|
||||
renewal.renew_cert(config, domains, le_client, lineage)
|
||||
@@ -122,7 +123,7 @@ def _get_and_save_cert(le_client, config, domains=None, certname=None, lineage=N
|
||||
"{action} for {domains}".format(
|
||||
action="Simulating a certificate request" if config.dry_run else
|
||||
"Requesting a certificate",
|
||||
domains=display_util.summarize_domain_list(domains)
|
||||
domains=internal_display_util.summarize_domain_list(domains)
|
||||
)
|
||||
)
|
||||
lineage = le_client.obtain_and_enroll_certificate(domains, certname)
|
||||
@@ -1339,7 +1340,7 @@ def _csr_get_and_save_cert(config, le_client):
|
||||
"{action} for {domains}".format(
|
||||
action="Simulating a certificate request" if config.dry_run else
|
||||
"Requesting a certificate",
|
||||
domains=display_util.summarize_domain_list(csr_names)
|
||||
domains=internal_display_util.summarize_domain_list(csr_names)
|
||||
)
|
||||
)
|
||||
cert, chain = le_client.obtain_certificate_from_csr(csr)
|
||||
|
||||
@@ -26,6 +26,7 @@ from certbot._internal import constants
|
||||
from certbot._internal import hooks
|
||||
from certbot._internal import storage
|
||||
from certbot._internal import updater
|
||||
from certbot._internal.display import obj as display_obj
|
||||
from certbot._internal.plugins import disco as plugins_disco
|
||||
from certbot.compat import os
|
||||
from certbot.display import util as display_util
|
||||
@@ -363,7 +364,7 @@ def _renew_describe_results(config: interfaces.IConfig, renew_successes: List[st
|
||||
notify = display_util.notify
|
||||
notify_error = logger.error
|
||||
|
||||
notify('\n{}'.format(display_util.SIDE_FRAME))
|
||||
notify('\n{}'.format(display_obj.SIDE_FRAME))
|
||||
|
||||
renewal_noun = "simulated renewal" if config.dry_run else "renewal"
|
||||
|
||||
@@ -393,7 +394,7 @@ def _renew_describe_results(config: interfaces.IConfig, renew_successes: List[st
|
||||
"were invalid: ")
|
||||
notify(report(parse_failures, "parsefail"))
|
||||
|
||||
notify(display_util.SIDE_FRAME)
|
||||
notify(display_obj.SIDE_FRAME)
|
||||
|
||||
|
||||
def handle_renewal_request(config):
|
||||
|
||||
@@ -4,6 +4,7 @@ from textwrap import indent
|
||||
|
||||
from certbot import errors
|
||||
from certbot import util
|
||||
from certbot._internal.display import util as internal_display_util
|
||||
from certbot.compat import os
|
||||
from certbot.display import util as display_util
|
||||
|
||||
@@ -193,7 +194,7 @@ def _choose_names_manually(prompt_prefix=""):
|
||||
invalid_domains = {}
|
||||
retry_message = ""
|
||||
try:
|
||||
domain_list = display_util.separate_list_input(input_)
|
||||
domain_list = internal_display_util.separate_list_input(input_)
|
||||
except UnicodeEncodeError:
|
||||
domain_list = []
|
||||
retry_message = (
|
||||
|
||||
@@ -9,42 +9,42 @@ should be used whenever:
|
||||
Other messages can use the `logging` module. See `log.py`.
|
||||
|
||||
"""
|
||||
import logging
|
||||
import sys
|
||||
from typing import List
|
||||
from typing import Optional
|
||||
from typing import Tuple
|
||||
from typing import Union
|
||||
|
||||
|
||||
from certbot.compat import misc
|
||||
# These imports are done to not break the public API of the module.
|
||||
# These specific imports from certbot._internal.display.obj and
|
||||
# certbot._internal.display.util are done to not break the public API of this
|
||||
# module.
|
||||
from certbot._internal.display.obj import FileDisplay # pylint: disable=unused-import
|
||||
from certbot._internal.display.obj import NoninteractiveDisplay # pylint: disable=unused-import
|
||||
from certbot._internal.display.obj import SIDE_FRAME # pylint: disable=unused-import
|
||||
from certbot._internal.display.util import input_with_timeout # pylint: disable=unused-import
|
||||
from certbot._internal.display.util import separate_list_input # pylint: disable=unused-import
|
||||
from certbot._internal.display.util import summarize_domain_list # pylint: disable=unused-import
|
||||
from certbot._internal.display import obj
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
WIDTH = 72
|
||||
|
||||
# Display exit codes
|
||||
OK = "ok"
|
||||
# These constants are defined this way to make them easier to document with
|
||||
# Sphinx and to not couple our public docstrings to our internal ones.
|
||||
OK = obj.OK
|
||||
"""Display exit code indicating user acceptance."""
|
||||
|
||||
CANCEL = "cancel"
|
||||
CANCEL = obj.CANCEL
|
||||
"""Display exit code for a user canceling the display."""
|
||||
|
||||
# These constants are unused and should be removed in a major release of
|
||||
# Certbot.
|
||||
WIDTH = 72
|
||||
|
||||
HELP = "help"
|
||||
"""Display exit code when for when the user requests more help. (UNUSED)"""
|
||||
|
||||
ESC = "esc"
|
||||
"""Display exit code when the user hits Escape (UNUSED)"""
|
||||
|
||||
# Display constants
|
||||
SIDE_FRAME = ("- " * 39) + "-"
|
||||
"""Display boundary (alternates spaces, so when copy-pasted, markdown doesn't interpret
|
||||
it as a heading)"""
|
||||
|
||||
|
||||
def notify(msg: str) -> None:
|
||||
"""Display a basic status message.
|
||||
@@ -186,36 +186,6 @@ def directory_select(message: str, default: Optional[str] = None, cli_flag: Opti
|
||||
force_interactive=force_interactive)
|
||||
|
||||
|
||||
def input_with_timeout(prompt=None, timeout=36000.0):
|
||||
"""Get user input with a timeout.
|
||||
|
||||
Behaves the same as the builtin input, however, an error is raised if
|
||||
a user doesn't answer after timeout seconds. The default timeout
|
||||
value was chosen to place it just under 12 hours for users following
|
||||
our advice and running Certbot twice a day.
|
||||
|
||||
:param str prompt: prompt to provide for input
|
||||
:param float timeout: maximum number of seconds to wait for input
|
||||
|
||||
:returns: user response
|
||||
:rtype: str
|
||||
|
||||
:raises errors.Error if no answer is given before the timeout
|
||||
|
||||
"""
|
||||
# use of sys.stdin and sys.stdout to mimic the builtin input based on
|
||||
# https://github.com/python/cpython/blob/baf7bb30a02aabde260143136bdf5b3738a1d409/Lib/getpass.py#L129
|
||||
if prompt:
|
||||
sys.stdout.write(prompt)
|
||||
sys.stdout.flush()
|
||||
|
||||
line = misc.readline_with_timeout(timeout, prompt)
|
||||
|
||||
if not line:
|
||||
raise EOFError
|
||||
return line.rstrip('\n')
|
||||
|
||||
|
||||
def assert_valid_call(prompt, default, cli_flag, force_interactive):
|
||||
"""Verify that provided arguments is a valid IDisplay call.
|
||||
|
||||
@@ -232,42 +202,3 @@ def assert_valid_call(prompt, default, cli_flag, force_interactive):
|
||||
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
|
||||
|
||||
|
||||
def separate_list_input(input_):
|
||||
"""Separate a comma or space separated list.
|
||||
|
||||
:param str input_: input from the user
|
||||
|
||||
:returns: strings
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
no_commas = input_.replace(",", " ")
|
||||
# Each string is naturally unicode, this causes problems with M2Crypto SANs
|
||||
# TODO: check if above is still true when M2Crypto is gone ^
|
||||
return [str(string) for string in no_commas.split()]
|
||||
|
||||
|
||||
def summarize_domain_list(domains: List[str]) -> str:
|
||||
"""Summarizes a list of domains in the format of:
|
||||
example.com.com and N more domains
|
||||
or if there is are only two domains:
|
||||
example.com and www.example.com
|
||||
or if there is only one domain:
|
||||
example.com
|
||||
|
||||
:param list domains: `str` list of domains
|
||||
:returns: the domain list summary
|
||||
:rtype: str
|
||||
"""
|
||||
if not domains:
|
||||
return ""
|
||||
|
||||
l = len(domains)
|
||||
if l == 1:
|
||||
return domains[0]
|
||||
elif l == 2:
|
||||
return " and ".join(domains)
|
||||
else:
|
||||
return "{0} and {1} more domains".format(domains[0], l-1)
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
"""Test :mod:`certbot._internal.display.util`."""
|
||||
import io
|
||||
import socket
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from certbot import errors
|
||||
|
||||
try:
|
||||
import mock
|
||||
except ImportError: # pragma: no cover
|
||||
from unittest import mock
|
||||
|
||||
|
||||
class WrapLinesTest(unittest.TestCase):
|
||||
def test_wrap_lines(self):
|
||||
from certbot._internal.display.util import wrap_lines
|
||||
msg = ("This is just a weak test{0}"
|
||||
"This function is only meant to be for easy viewing{0}"
|
||||
"Test a really really really really really really really really "
|
||||
"really really really really long line...".format('\n'))
|
||||
text = wrap_lines(msg)
|
||||
|
||||
self.assertEqual(text.count('\n'), 3)
|
||||
|
||||
|
||||
class PlaceParensTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def _call(cls, label):
|
||||
from certbot._internal.display.util import parens_around_char
|
||||
return parens_around_char(label)
|
||||
|
||||
def test_single_letter(self):
|
||||
self.assertEqual("(a)", self._call("a"))
|
||||
|
||||
def test_multiple(self):
|
||||
self.assertEqual("(L)abel", self._call("Label"))
|
||||
self.assertEqual("(y)es please", self._call("yes please"))
|
||||
|
||||
|
||||
class InputWithTimeoutTest(unittest.TestCase):
|
||||
"""Tests for certbot._internal.display.util.input_with_timeout."""
|
||||
@classmethod
|
||||
def _call(cls, *args, **kwargs):
|
||||
from certbot._internal.display.util import input_with_timeout
|
||||
return input_with_timeout(*args, **kwargs)
|
||||
|
||||
def test_eof(self):
|
||||
with tempfile.TemporaryFile("r+") as f:
|
||||
with mock.patch("certbot._internal.display.util.sys.stdin", new=f):
|
||||
self.assertRaises(EOFError, self._call)
|
||||
|
||||
def test_input(self, prompt=None):
|
||||
expected = "foo bar"
|
||||
stdin = io.StringIO(expected + "\n")
|
||||
with mock.patch("certbot.compat.misc.select.select") as mock_select:
|
||||
mock_select.return_value = ([stdin], [], [],)
|
||||
self.assertEqual(self._call(prompt), expected)
|
||||
|
||||
@mock.patch("certbot._internal.display.util.sys.stdout")
|
||||
def test_input_with_prompt(self, mock_stdout):
|
||||
prompt = "test prompt: "
|
||||
self.test_input(prompt)
|
||||
mock_stdout.write.assert_called_once_with(prompt)
|
||||
mock_stdout.flush.assert_called_once_with()
|
||||
|
||||
def test_timeout(self):
|
||||
stdin = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
stdin.bind(('', 0))
|
||||
stdin.listen(1)
|
||||
with mock.patch("certbot._internal.display.util.sys.stdin", stdin):
|
||||
self.assertRaises(errors.Error, self._call, timeout=0.001)
|
||||
stdin.close()
|
||||
|
||||
|
||||
class SeparateListInputTest(unittest.TestCase):
|
||||
"""Test Module functions."""
|
||||
def setUp(self):
|
||||
self.exp = ["a", "b", "c", "test"]
|
||||
|
||||
@classmethod
|
||||
def _call(cls, input_):
|
||||
from certbot._internal.display.util import separate_list_input
|
||||
return separate_list_input(input_)
|
||||
|
||||
def test_commas(self):
|
||||
self.assertEqual(self._call("a,b,c,test"), self.exp)
|
||||
|
||||
def test_spaces(self):
|
||||
self.assertEqual(self._call("a b c test"), self.exp)
|
||||
|
||||
def test_both(self):
|
||||
self.assertEqual(self._call("a, b, c, test"), self.exp)
|
||||
|
||||
def test_mess(self):
|
||||
actual = [
|
||||
self._call(" a , b c \t test"),
|
||||
self._call(",a, ,, , b c test "),
|
||||
self._call(",,,,, , a b,,, , c,test"),
|
||||
]
|
||||
|
||||
for act in actual:
|
||||
self.assertEqual(act, self.exp)
|
||||
|
||||
|
||||
class SummarizeDomainListTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def _call(cls, domains):
|
||||
from certbot._internal.display.util import summarize_domain_list
|
||||
return summarize_domain_list(domains)
|
||||
|
||||
def test_single_domain(self):
|
||||
self.assertEqual("example.com", self._call(["example.com"]))
|
||||
|
||||
def test_two_domains(self):
|
||||
self.assertEqual("example.com and example.org",
|
||||
self._call(["example.com", "example.org"]))
|
||||
|
||||
def test_many_domains(self):
|
||||
self.assertEqual("example.com and 2 more domains",
|
||||
self._call(["example.com", "example.org", "a.example.com"]))
|
||||
|
||||
def test_empty_domains(self):
|
||||
self.assertEqual("", self._call([]))
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main() # pragma: no cover
|
||||
@@ -32,7 +32,7 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
mock_logger.debug.assert_called_with("Notifying user: %s", "message")
|
||||
|
||||
def test_notification_pause(self):
|
||||
input_with_timeout = "certbot.display.util.input_with_timeout"
|
||||
input_with_timeout = "certbot._internal.display.util.input_with_timeout"
|
||||
with mock.patch(input_with_timeout, return_value="enter"):
|
||||
self.displayer.notification("message", force_interactive=True)
|
||||
|
||||
@@ -67,7 +67,7 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
self.assertIn("- - - ", string)
|
||||
self.assertIn("message2" + os.linesep, string)
|
||||
|
||||
@mock.patch("certbot.display.util."
|
||||
@mock.patch("certbot._internal.display.obj."
|
||||
"FileDisplay._get_valid_int_ans")
|
||||
def test_menu(self, mock_ans):
|
||||
mock_ans.return_value = (display_util.OK, 1)
|
||||
@@ -81,14 +81,14 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
self.assertEqual(result, (display_util.OK, default))
|
||||
|
||||
def test_input_cancel(self):
|
||||
input_with_timeout = "certbot.display.util.input_with_timeout"
|
||||
input_with_timeout = "certbot._internal.display.util.input_with_timeout"
|
||||
with mock.patch(input_with_timeout, return_value="c"):
|
||||
code, _ = self.displayer.input("message", force_interactive=True)
|
||||
|
||||
self.assertTrue(code, display_util.CANCEL)
|
||||
|
||||
def test_input_normal(self):
|
||||
input_with_timeout = "certbot.display.util.input_with_timeout"
|
||||
input_with_timeout = "certbot._internal.display.util.input_with_timeout"
|
||||
with mock.patch(input_with_timeout, return_value="domain.com"):
|
||||
code, input_ = self.displayer.input("message", force_interactive=True)
|
||||
|
||||
@@ -115,7 +115,7 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
self.displayer.input, "msg", cli_flag="--flag")
|
||||
|
||||
def test_yesno(self):
|
||||
input_with_timeout = "certbot.display.util.input_with_timeout"
|
||||
input_with_timeout = "certbot._internal.display.util.input_with_timeout"
|
||||
with mock.patch(input_with_timeout, return_value="Yes"):
|
||||
self.assertTrue(self.displayer.yesno(
|
||||
"message", force_interactive=True))
|
||||
@@ -140,7 +140,7 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
self.assertTrue(self._force_noninteractive(
|
||||
self.displayer.yesno, "message", default=True))
|
||||
|
||||
@mock.patch("certbot.display.util.input_with_timeout")
|
||||
@mock.patch("certbot._internal.display.util.input_with_timeout")
|
||||
def test_checklist_valid(self, mock_input):
|
||||
mock_input.return_value = "2 1"
|
||||
code, tag_list = self.displayer.checklist(
|
||||
@@ -148,21 +148,21 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
self.assertEqual(
|
||||
(code, set(tag_list)), (display_util.OK, {"tag1", "tag2"}))
|
||||
|
||||
@mock.patch("certbot.display.util.input_with_timeout")
|
||||
@mock.patch("certbot._internal.display.util.input_with_timeout")
|
||||
def test_checklist_empty(self, mock_input):
|
||||
mock_input.return_value = ""
|
||||
code, tag_list = self.displayer.checklist("msg", TAGS, force_interactive=True)
|
||||
self.assertEqual(
|
||||
(code, set(tag_list)), (display_util.OK, {"tag1", "tag2", "tag3"}))
|
||||
|
||||
@mock.patch("certbot.display.util.input_with_timeout")
|
||||
@mock.patch("certbot._internal.display.util.input_with_timeout")
|
||||
def test_checklist_miss_valid(self, mock_input):
|
||||
mock_input.side_effect = ["10", "tag1 please", "1"]
|
||||
|
||||
ret = self.displayer.checklist("msg", TAGS, force_interactive=True)
|
||||
self.assertEqual(ret, (display_util.OK, ["tag1"]))
|
||||
|
||||
@mock.patch("certbot.display.util.input_with_timeout")
|
||||
@mock.patch("certbot._internal.display.util.input_with_timeout")
|
||||
def test_checklist_miss_quit(self, mock_input):
|
||||
mock_input.side_effect = ["10", "c"]
|
||||
|
||||
@@ -194,7 +194,7 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
self.displayer._scrub_checklist_input(list_, TAGS))
|
||||
self.assertEqual(set_tags, exp[i])
|
||||
|
||||
@mock.patch("certbot.display.util.input_with_timeout")
|
||||
@mock.patch("certbot._internal.display.util.input_with_timeout")
|
||||
def test_directory_select(self, mock_input):
|
||||
args = ["msg", "/var/www/html", "--flag", True]
|
||||
user_input = "/var/www/html"
|
||||
@@ -214,7 +214,7 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
def _force_noninteractive(self, func, *args, **kwargs):
|
||||
skipped_interaction = self.displayer.skipped_interaction
|
||||
|
||||
with mock.patch("certbot.display.util.sys.stdin") as mock_stdin:
|
||||
with mock.patch("certbot._internal.display.obj.sys.stdin") as mock_stdin:
|
||||
mock_stdin.isatty.return_value = False
|
||||
with mock.patch("certbot._internal.display.obj.logger") as mock_logger:
|
||||
result = func(*args, **kwargs)
|
||||
@@ -245,19 +245,9 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
self.displayer._print_menu("msg", CHOICES)
|
||||
self.displayer._print_menu("msg", TAGS)
|
||||
|
||||
def test_wrap_lines(self):
|
||||
# pylint: disable=protected-access
|
||||
msg = ("This is just a weak test{0}"
|
||||
"This function is only meant to be for easy viewing{0}"
|
||||
"Test a really really really really really really really really "
|
||||
"really really really really long line...".format('\n'))
|
||||
text = display_obj._wrap_lines(msg)
|
||||
|
||||
self.assertEqual(text.count('\n'), 3)
|
||||
|
||||
def test_get_valid_int_ans_valid(self):
|
||||
# pylint: disable=protected-access
|
||||
input_with_timeout = "certbot.display.util.input_with_timeout"
|
||||
input_with_timeout = "certbot._internal.display.util.input_with_timeout"
|
||||
with mock.patch(input_with_timeout, return_value="1"):
|
||||
self.assertEqual(
|
||||
self.displayer._get_valid_int_ans(1), (display_util.OK, 1))
|
||||
@@ -274,7 +264,7 @@ class FileOutputDisplayTest(unittest.TestCase):
|
||||
["4", "one", "C"],
|
||||
["c"],
|
||||
]
|
||||
input_with_timeout = "certbot.display.util.input_with_timeout"
|
||||
input_with_timeout = "certbot._internal.display.util.input_with_timeout"
|
||||
for ans in answers:
|
||||
with mock.patch(input_with_timeout, side_effect=ans):
|
||||
self.assertEqual(
|
||||
@@ -359,15 +349,5 @@ class NoninteractiveDisplayTest(unittest.TestCase):
|
||||
self.assertIsNotNone(result)
|
||||
|
||||
|
||||
class PlaceParensTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def _call(cls, label): # pylint: disable=protected-access
|
||||
from certbot._internal.display.obj import _parens_around_char
|
||||
return _parens_around_char(label)
|
||||
|
||||
def test_single_letter(self):
|
||||
self.assertEqual("(a)", self._call("a"))
|
||||
|
||||
def test_multiple(self):
|
||||
self.assertEqual("(L)abel", self._call("Label"))
|
||||
self.assertEqual("(y)es please", self._call("yes please"))
|
||||
if __name__ == "__main__":
|
||||
unittest.main() # pragma: no cover
|
||||
|
||||
@@ -304,7 +304,7 @@ class ChooseNamesTest(unittest.TestCase):
|
||||
self.assertEqual(_choose_names_manually(), [])
|
||||
# IDN exception with previous mocks
|
||||
with mock.patch(
|
||||
"certbot.display.ops.display_util.separate_list_input"
|
||||
"certbot.display.ops.internal_display_util.separate_list_input"
|
||||
) as mock_sli:
|
||||
unicode_error = UnicodeEncodeError('mock', u'', 0, 1, 'mock')
|
||||
mock_sli.side_effect = unicode_error
|
||||
|
||||
@@ -98,91 +98,5 @@ class DirectorySelectTest(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
class InputWithTimeoutTest(unittest.TestCase):
|
||||
"""Tests for certbot.display.util.input_with_timeout."""
|
||||
@classmethod
|
||||
def _call(cls, *args, **kwargs):
|
||||
from certbot.display.util import input_with_timeout
|
||||
return input_with_timeout(*args, **kwargs)
|
||||
|
||||
def test_eof(self):
|
||||
with tempfile.TemporaryFile("r+") as f:
|
||||
with mock.patch("certbot.display.util.sys.stdin", new=f):
|
||||
self.assertRaises(EOFError, self._call)
|
||||
|
||||
def test_input(self, prompt=None):
|
||||
expected = "foo bar"
|
||||
stdin = io.StringIO(expected + "\n")
|
||||
with mock.patch("certbot.compat.misc.select.select") as mock_select:
|
||||
mock_select.return_value = ([stdin], [], [],)
|
||||
self.assertEqual(self._call(prompt), expected)
|
||||
|
||||
@mock.patch("certbot.display.util.sys.stdout")
|
||||
def test_input_with_prompt(self, mock_stdout):
|
||||
prompt = "test prompt: "
|
||||
self.test_input(prompt)
|
||||
mock_stdout.write.assert_called_once_with(prompt)
|
||||
mock_stdout.flush.assert_called_once_with()
|
||||
|
||||
def test_timeout(self):
|
||||
stdin = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
stdin.bind(('', 0))
|
||||
stdin.listen(1)
|
||||
with mock.patch("certbot.display.util.sys.stdin", stdin):
|
||||
self.assertRaises(errors.Error, self._call, timeout=0.001)
|
||||
stdin.close()
|
||||
|
||||
|
||||
class SeparateListInputTest(unittest.TestCase):
|
||||
"""Test Module functions."""
|
||||
def setUp(self):
|
||||
self.exp = ["a", "b", "c", "test"]
|
||||
|
||||
@classmethod
|
||||
def _call(cls, input_):
|
||||
from certbot.display.util import separate_list_input
|
||||
return separate_list_input(input_)
|
||||
|
||||
def test_commas(self):
|
||||
self.assertEqual(self._call("a,b,c,test"), self.exp)
|
||||
|
||||
def test_spaces(self):
|
||||
self.assertEqual(self._call("a b c test"), self.exp)
|
||||
|
||||
def test_both(self):
|
||||
self.assertEqual(self._call("a, b, c, test"), self.exp)
|
||||
|
||||
def test_mess(self):
|
||||
actual = [
|
||||
self._call(" a , b c \t test"),
|
||||
self._call(",a, ,, , b c test "),
|
||||
self._call(",,,,, , a b,,, , c,test"),
|
||||
]
|
||||
|
||||
for act in actual:
|
||||
self.assertEqual(act, self.exp)
|
||||
|
||||
|
||||
class SummarizeDomainListTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def _call(cls, domains):
|
||||
from certbot.display.util import summarize_domain_list
|
||||
return summarize_domain_list(domains)
|
||||
|
||||
def test_single_domain(self):
|
||||
self.assertEqual("example.com", self._call(["example.com"]))
|
||||
|
||||
def test_two_domains(self):
|
||||
self.assertEqual("example.com and example.org",
|
||||
self._call(["example.com", "example.org"]))
|
||||
|
||||
def test_many_domains(self):
|
||||
self.assertEqual("example.com and 2 more domains",
|
||||
self._call(["example.com", "example.org", "a.example.com"]))
|
||||
|
||||
def test_empty_domains(self):
|
||||
self.assertEqual("", self._call([]))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main() # pragma: no cover
|
||||
|
||||
Reference in New Issue
Block a user