mirror of
https://github.com/certbot/certbot.git
synced 2026-07-27 08:09:10 +02:00
full display integration
This commit is contained in:
@@ -985,7 +985,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
|
||||
self.restart()
|
||||
|
||||
|
||||
def get_version(self):
|
||||
def get_version():
|
||||
"""Return version of Apache Server.
|
||||
|
||||
Version is returned as tuple. (ie. 2.4.7 = (2, 4, 7))
|
||||
|
||||
@@ -20,6 +20,7 @@ from letsencrypt.client import revoker
|
||||
|
||||
from letsencrypt.client.apache import configurator
|
||||
from letsencrypt.client.display import ops
|
||||
from letsencrypt.client.display import enhancements
|
||||
|
||||
|
||||
class Client(object):
|
||||
@@ -102,7 +103,9 @@ class Client(object):
|
||||
cert_file, chain_file = self.save_certificate(
|
||||
certificate_dict, cert_path, chain_path)
|
||||
|
||||
revoker.Revoker.store_cert_key(cert_file, False)
|
||||
print cert_file
|
||||
|
||||
revoker.Revoker.store_cert_key(cert_file, self.authkey.file, False)
|
||||
|
||||
return cert_file, chain_file
|
||||
|
||||
@@ -217,8 +220,7 @@ class Client(object):
|
||||
raise errors.LetsEncryptClientError("No installer available")
|
||||
|
||||
if redirect is None:
|
||||
redirect = zope.component.getUtility(
|
||||
interfaces.IDisplay).redirect_by_default()
|
||||
redirect = enhancements.ask("redirect")
|
||||
|
||||
if redirect:
|
||||
self.redirect_to_ssl(domains)
|
||||
@@ -348,6 +350,8 @@ def determine_authenticator():
|
||||
logging.info("Unable to determine a way to authenticate the server")
|
||||
if len(auths) > 1:
|
||||
return ops.choose_authenticator(auths)
|
||||
elif len(auths) == 1:
|
||||
return auths[0]
|
||||
|
||||
|
||||
def determine_installer():
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Let's Encrypt client.display"""
|
||||
@@ -8,61 +8,51 @@ from letsencrypt.client import interfaces
|
||||
from letsencrypt.client.display import display_util
|
||||
|
||||
|
||||
class EnhanceDisplay(object):
|
||||
"""Class used to display various enhancements.
|
||||
def ask(enhancement):
|
||||
"""Display the enhancement to the user.
|
||||
|
||||
.. note::This is not a subclass of Display. It merely uses Display as a
|
||||
component.
|
||||
:param str enhancement: One of the
|
||||
:class:`letsencrypt.client.CONFIG.ENHANCEMENTS` enhancements
|
||||
|
||||
:ivar displayer: Display singleton
|
||||
:type displayer: :class:`letsencrypt.client.interfaces.IDisplay
|
||||
:returns: True if feature is desired, False otherwise
|
||||
:rtype: bool
|
||||
|
||||
:ivar dict dispatch: Dict mapping enhancements to functions
|
||||
:raises :class:`letsencrypt.client.errors.LetsEncryptClientError`: If
|
||||
the enhancement provided is not supported.
|
||||
|
||||
"""
|
||||
def __init__(self):
|
||||
self.displayer = zope.component.getUtility(interfaces.IDisplay)
|
||||
try:
|
||||
return _dispatch[enhancement]()
|
||||
except KeyError:
|
||||
logging.error("Unsupported enhancement given to ask()")
|
||||
raise errors.LetsEncryptClientError("Unsupported Enhancement")
|
||||
|
||||
self.dispatch = {
|
||||
"redirect": self.redirect_by_default,
|
||||
}
|
||||
|
||||
def ask(self, enhancement):
|
||||
"""Display the enhancement to the user.
|
||||
def redirect_by_default():
|
||||
"""Determines whether the user would like to redirect to HTTPS.
|
||||
|
||||
:param str enhancement: One of the
|
||||
:class:`letsencrypt.client.CONFIG.ENHANCEMENTS` enhancements
|
||||
:returns: True if redirect is desired, False otherwise
|
||||
:rtype: bool
|
||||
|
||||
:returns: True if feature is desired, False otherwise
|
||||
:rtype: bool
|
||||
"""
|
||||
choices = [
|
||||
("Easy", "Allow both HTTP and HTTPS access to these sites"),
|
||||
("Secure", "Make all requests redirect to secure HTTPS access"),
|
||||
]
|
||||
|
||||
:raises :class:`letsencrypt.client.errors.LetsEncryptClientError`: If
|
||||
the enhancement provided is not supported.
|
||||
result = _util(interfaces.IDisplay).menu(
|
||||
"Please choose whether HTTPS access is required or optional.",
|
||||
choices)
|
||||
|
||||
"""
|
||||
try:
|
||||
return self.dispatch[enhancement]
|
||||
except KeyError:
|
||||
logging.error("Unsupported enhancement given to ask()")
|
||||
raise errors.LetsEncryptClientError("Unsupported Enhancement")
|
||||
if result[0] != display_util.OK:
|
||||
return False
|
||||
|
||||
def redirect_by_default(self):
|
||||
"""Determines whether the user would like to redirect to HTTPS.
|
||||
return result[1] == 1
|
||||
|
||||
:returns: True if redirect is desired, False otherwise
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
choices = [
|
||||
("Easy", "Allow both HTTP and HTTPS access to these sites"),
|
||||
("Secure", "Make all requests redirect to secure HTTPS access")]
|
||||
_util = zope.component.getUtility
|
||||
|
||||
result = self.displayer.menu(
|
||||
"Please choose whether HTTPS access is required or optional.",
|
||||
choices, "Please enter the appropriate number")
|
||||
|
||||
if result[0] != display_util.OK:
|
||||
return False
|
||||
|
||||
# different answer for each type of display
|
||||
return str(result[1]) == "Secure" or result[1] == 1
|
||||
_dispatch = {
|
||||
"redirect": redirect_by_default
|
||||
}
|
||||
@@ -73,10 +73,9 @@ def _filter_names(names):
|
||||
:rtype: tuple
|
||||
|
||||
"""
|
||||
choices = [(n, "", 0) for n in names]
|
||||
code, names = util(interfaces.IDisplay).checklist(
|
||||
"Which names would you like to activate HTTPS for?",
|
||||
choices=choices)
|
||||
tags=names)
|
||||
return code, [str(s) for s in names]
|
||||
|
||||
|
||||
|
||||
@@ -17,9 +17,9 @@ class Reverter(object):
|
||||
"""Reverter Class - save and revert configuration checkpoints"""
|
||||
def __init__(self, direc=None):
|
||||
if not direc:
|
||||
direc = {'backup': CONFIG.BACKUP_DIR,
|
||||
'temp': CONFIG.TEMP_CHECKPOINT_DIR,
|
||||
'progress': CONFIG.IN_PROGRESS_DIR}
|
||||
direc = {"backup": CONFIG.BACKUP_DIR,
|
||||
"temp": CONFIG.TEMP_CHECKPOINT_DIR,
|
||||
"progress": CONFIG.IN_PROGRESS_DIR}
|
||||
self.direc = direc
|
||||
|
||||
def revert_temporary_config(self):
|
||||
@@ -32,13 +32,13 @@ class Reverter(object):
|
||||
Unable to revert config
|
||||
|
||||
"""
|
||||
if os.path.isdir(self.direc['temp']):
|
||||
if os.path.isdir(self.direc["temp"]):
|
||||
try:
|
||||
self._recover_checkpoint(self.direc['temp'])
|
||||
self._recover_checkpoint(self.direc["temp"])
|
||||
except errors.LetsEncryptReverterError:
|
||||
# We have a partial or incomplete recovery
|
||||
logging.fatal("Incomplete or failed recovery for %s",
|
||||
self.direc['temp'])
|
||||
self.direc["temp"])
|
||||
raise errors.LetsEncryptReverterError(
|
||||
"Unable to revert temporary config")
|
||||
|
||||
@@ -46,7 +46,7 @@ class Reverter(object):
|
||||
"""Revert 'rollback' number of configuration checkpoints.
|
||||
|
||||
:param int rollback: Number of checkpoints to reverse. A str num will be
|
||||
cast to an integer. So '2' is also acceptable.
|
||||
cast to an integer. So "2" is also acceptable.
|
||||
|
||||
:raises :class:`letsencrypt.client.errors.LetsEncryptReverterError`: If
|
||||
there is a problem with the input or if the function is unable to
|
||||
@@ -63,7 +63,7 @@ class Reverter(object):
|
||||
logging.error("Rollback argument must be a positive integer")
|
||||
raise errors.LetsEncryptReverterError("Invalid Input")
|
||||
|
||||
backups = os.listdir(self.direc['backup'])
|
||||
backups = os.listdir(self.direc["backup"])
|
||||
backups.sort()
|
||||
|
||||
if len(backups) < rollback:
|
||||
@@ -71,7 +71,7 @@ class Reverter(object):
|
||||
rollback, len(backups))
|
||||
|
||||
while rollback > 0 and backups:
|
||||
cp_dir = os.path.join(self.direc['backup'], backups.pop())
|
||||
cp_dir = os.path.join(self.direc["backup"], backups.pop())
|
||||
try:
|
||||
self._recover_checkpoint(cp_dir)
|
||||
except errors.LetsEncryptReverterError:
|
||||
@@ -88,7 +88,7 @@ class Reverter(object):
|
||||
.. todo:: Decide on a policy for error handling, OSError IOError...
|
||||
|
||||
"""
|
||||
backups = os.listdir(self.direc['backup'])
|
||||
backups = os.listdir(self.direc["backup"])
|
||||
backups.sort(reverse=True)
|
||||
|
||||
if not backups:
|
||||
@@ -102,12 +102,12 @@ class Reverter(object):
|
||||
float(bkup)
|
||||
except ValueError:
|
||||
raise errors.LetsEncryptReverterError(
|
||||
"Invalid directories in {0}".format(self.direc['backup']))
|
||||
"Invalid directories in {0}".format(self.direc["backup"]))
|
||||
|
||||
output = []
|
||||
for bkup in backups:
|
||||
output.append(time.ctime(float(bkup)))
|
||||
cur_dir = os.path.join(self.direc['backup'], bkup)
|
||||
cur_dir = os.path.join(self.direc["backup"], bkup)
|
||||
with open(os.path.join(cur_dir, "CHANGES_SINCE")) as changes_fd:
|
||||
output.append(changes_fd.read())
|
||||
|
||||
@@ -136,7 +136,7 @@ class Reverter(object):
|
||||
param str save_notes: notes about changes during the save
|
||||
|
||||
"""
|
||||
self._add_to_checkpoint_dir(self.direc['temp'], save_files, save_notes)
|
||||
self._add_to_checkpoint_dir(self.direc["temp"], save_files, save_notes)
|
||||
|
||||
def add_to_checkpoint(self, save_files, save_notes):
|
||||
"""Add files to a permanent checkpoint
|
||||
@@ -148,7 +148,7 @@ class Reverter(object):
|
||||
# Check to make sure we are not overwriting a temp file
|
||||
self._check_tempfile_saves(save_files)
|
||||
self._add_to_checkpoint_dir(
|
||||
self.direc['progress'], save_files, save_notes)
|
||||
self.direc["progress"], save_files, save_notes)
|
||||
|
||||
def _add_to_checkpoint_dir(self, cp_dir, save_files, save_notes):
|
||||
"""Add save files to checkpoint directory.
|
||||
@@ -179,7 +179,7 @@ class Reverter(object):
|
||||
try:
|
||||
shutil.copy2(filename, os.path.join(
|
||||
cp_dir, os.path.basename(filename) + "_" + str(idx)))
|
||||
op_fd.write(filename + '\n')
|
||||
op_fd.write(filename + os.linesep)
|
||||
# http://stackoverflow.com/questions/4726260/effective-use-of-python-shutil-copy2
|
||||
except IOError:
|
||||
op_fd.close()
|
||||
@@ -192,7 +192,7 @@ class Reverter(object):
|
||||
idx += 1
|
||||
op_fd.close()
|
||||
|
||||
with open(os.path.join(cp_dir, "CHANGES_SINCE"), 'a') as notes_fd:
|
||||
with open(os.path.join(cp_dir, "CHANGES_SINCE"), "a") as notes_fd:
|
||||
notes_fd.write(save_notes)
|
||||
|
||||
def _read_and_append(self, filepath): # pylint: disable=no-self-use
|
||||
@@ -203,11 +203,11 @@ class Reverter(object):
|
||||
"""
|
||||
# Open up filepath differently depending on if it already exists
|
||||
if os.path.isfile(filepath):
|
||||
op_fd = open(filepath, 'r+')
|
||||
op_fd = open(filepath, "r+")
|
||||
lines = op_fd.read().splitlines()
|
||||
else:
|
||||
lines = []
|
||||
op_fd = open(filepath, 'w')
|
||||
op_fd = open(filepath, "w")
|
||||
|
||||
return op_fd, lines
|
||||
|
||||
@@ -229,7 +229,7 @@ class Reverter(object):
|
||||
for idx, path in enumerate(filepaths):
|
||||
shutil.copy2(os.path.join(
|
||||
cp_dir,
|
||||
os.path.basename(path) + '_' + str(idx)), path)
|
||||
os.path.basename(path) + "_" + str(idx)), path)
|
||||
except (IOError, OSError):
|
||||
# This file is required in all checkpoints.
|
||||
logging.error("Unable to recover files from %s", cp_dir)
|
||||
@@ -258,15 +258,15 @@ class Reverter(object):
|
||||
protected_files = []
|
||||
|
||||
# Get temp modified files
|
||||
temp_path = os.path.join(self.direc['temp'], "FILEPATHS")
|
||||
temp_path = os.path.join(self.direc["temp"], "FILEPATHS")
|
||||
if os.path.isfile(temp_path):
|
||||
with open(temp_path, 'r') as protected_fd:
|
||||
with open(temp_path, "r") as protected_fd:
|
||||
protected_files.extend(protected_fd.read().splitlines())
|
||||
|
||||
# Get temp new files
|
||||
new_path = os.path.join(self.direc['temp'], "NEW_FILES")
|
||||
new_path = os.path.join(self.direc["temp"], "NEW_FILES")
|
||||
if os.path.isfile(new_path):
|
||||
with open(new_path, 'r') as protected_fd:
|
||||
with open(new_path, "r") as protected_fd:
|
||||
protected_files.extend(protected_fd.read().splitlines())
|
||||
|
||||
# Verify no save_file is in protected_files
|
||||
@@ -299,9 +299,9 @@ class Reverter(object):
|
||||
"Forgot to provide files to registration call")
|
||||
|
||||
if temporary:
|
||||
cp_dir = self.direc['temp']
|
||||
cp_dir = self.direc["temp"]
|
||||
else:
|
||||
cp_dir = self.direc['progress']
|
||||
cp_dir = self.direc["progress"]
|
||||
|
||||
le_util.make_or_verify_dir(cp_dir, 0o755, os.geteuid())
|
||||
|
||||
@@ -325,7 +325,7 @@ class Reverter(object):
|
||||
def recovery_routine(self):
|
||||
"""Revert all previously modified files.
|
||||
|
||||
First, any changes found in self.direc['temp'] are removed,
|
||||
First, any changes found in self.direc["temp"] are removed,
|
||||
then IN_PROGRESS changes are removed The order is important.
|
||||
IN_PROGRESS is unable to add files that are already added by a TEMP
|
||||
change. Thus TEMP must be rolled back first because that will be the
|
||||
@@ -333,17 +333,17 @@ class Reverter(object):
|
||||
|
||||
"""
|
||||
self.revert_temporary_config()
|
||||
if os.path.isdir(self.direc['progress']):
|
||||
if os.path.isdir(self.direc["progress"]):
|
||||
try:
|
||||
self._recover_checkpoint(self.direc['progress'])
|
||||
self._recover_checkpoint(self.direc["progress"])
|
||||
except errors.LetsEncryptReverterError:
|
||||
# We have a partial or incomplete recovery
|
||||
logging.fatal("Incomplete or failed recovery for IN_PROGRESS "
|
||||
"checkpoint - %s",
|
||||
self.direc['progress'])
|
||||
self.direc["progress"])
|
||||
raise errors.LetsEncryptReverterError(
|
||||
"Incomplete or failed recovery for IN_PROGRESS checkpoint "
|
||||
"- %s" % self.direc['progress'])
|
||||
"- %s" % self.direc["progress"])
|
||||
|
||||
def _remove_contained_files(self, file_list): # pylint: disable=no-self-use
|
||||
"""Erase all files contained within file_list.
|
||||
@@ -362,7 +362,7 @@ class Reverter(object):
|
||||
if not os.path.isfile(file_list):
|
||||
return False
|
||||
try:
|
||||
with open(file_list, 'r') as list_fd:
|
||||
with open(file_list, "r") as list_fd:
|
||||
filepaths = list_fd.read().splitlines()
|
||||
for path in filepaths:
|
||||
# Files are registered before they are added... so
|
||||
@@ -386,8 +386,8 @@ class Reverter(object):
|
||||
def finalize_checkpoint(self, title):
|
||||
"""Move IN_PROGRESS checkpoint to timestamped checkpoint.
|
||||
|
||||
Adds title to self.direc['progress'] CHANGES_SINCE
|
||||
Move self.direc['progress'] to Backups directory and
|
||||
Adds title to self.direc["progress"] CHANGES_SINCE
|
||||
Move self.direc["progress"] to Backups directory and
|
||||
rename the directory as a timestamp
|
||||
|
||||
:param str title: Title describing checkpoint
|
||||
@@ -396,19 +396,18 @@ class Reverter(object):
|
||||
|
||||
"""
|
||||
# Check to make sure an "in progress" directory exists
|
||||
if not os.path.isdir(self.direc['progress']):
|
||||
logging.warning("No IN_PROGRESS checkpoint to finalize")
|
||||
if not os.path.isdir(self.direc["progress"]):
|
||||
return
|
||||
|
||||
changes_since_path = os.path.join(
|
||||
self.direc['progress'], 'CHANGES_SINCE')
|
||||
self.direc["progress"], "CHANGES_SINCE")
|
||||
changes_since_tmp_path = os.path.join(
|
||||
self.direc['progress'], 'CHANGES_SINCE.tmp')
|
||||
self.direc["progress"], "CHANGES_SINCE.tmp")
|
||||
|
||||
try:
|
||||
with open(changes_since_tmp_path, 'w') as changes_tmp:
|
||||
with open(changes_since_tmp_path, "w") as changes_tmp:
|
||||
changes_tmp.write("-- %s --\n" % title)
|
||||
with open(changes_since_path, 'r') as changes_orig:
|
||||
with open(changes_since_path, "r") as changes_orig:
|
||||
changes_tmp.write(changes_orig.read())
|
||||
|
||||
shutil.move(changes_since_tmp_path, changes_since_path)
|
||||
@@ -424,9 +423,9 @@ class Reverter(object):
|
||||
# collisions in the naming convention.
|
||||
cur_time = time.time()
|
||||
for _ in range(10):
|
||||
final_dir = os.path.join(self.direc['backup'], str(cur_time))
|
||||
final_dir = os.path.join(self.direc["backup"], str(cur_time))
|
||||
try:
|
||||
os.rename(self.direc['progress'], final_dir)
|
||||
os.rename(self.direc["progress"], final_dir)
|
||||
return
|
||||
except OSError:
|
||||
# It is possible if the checkpoints are made extremely quickly
|
||||
@@ -437,6 +436,6 @@ class Reverter(object):
|
||||
# After 10 attempts... something is probably wrong here...
|
||||
logging.error(
|
||||
"Unable to finalize checkpoint, %s -> %s",
|
||||
self.direc['progress'], final_dir)
|
||||
self.direc["progress"], final_dir)
|
||||
raise errors.LetsEncryptReverterError(
|
||||
"Unable to finalize checkpoint renaming")
|
||||
|
||||
@@ -61,8 +61,7 @@ class Revoker(object):
|
||||
certs = self._populate_saved_certs(csha1_vhlist)
|
||||
|
||||
if certs:
|
||||
self._insert_installed_status(certs)
|
||||
cert = self.choose_certs(certs)
|
||||
cert = revocation.choose_certs(certs)
|
||||
self.acme_revocation(cert)
|
||||
else:
|
||||
logging.info(
|
||||
|
||||
@@ -132,31 +132,6 @@ class TwoVhost80Test(util.ApacheTest):
|
||||
|
||||
self.assertEqual(len(self.config.vhosts), 5)
|
||||
|
||||
@mock.patch("letsencrypt.client.apache.configurator."
|
||||
"subprocess.Popen")
|
||||
def test_get_version(self, mock_popen):
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache/2.4.2 (Debian)", "")
|
||||
self.assertEqual(self.config.get_version(), (2, 4, 2))
|
||||
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache/2 (Linux)", "")
|
||||
self.assertEqual(self.config.get_version(), (2,))
|
||||
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache (Debian)", "")
|
||||
self.assertRaises(
|
||||
errors.LetsEncryptConfiguratorError, self.config.get_version)
|
||||
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache/2.3\n Apache/2.4.7", "")
|
||||
self.assertRaises(
|
||||
errors.LetsEncryptConfiguratorError, self.config.get_version)
|
||||
|
||||
mock_popen.side_effect = OSError("Can't find program")
|
||||
self.assertRaises(
|
||||
errors.LetsEncryptConfiguratorError, self.config.get_version)
|
||||
|
||||
@mock.patch("letsencrypt.client.apache.configurator."
|
||||
"dvsni.ApacheDvsni.perform")
|
||||
@mock.patch("letsencrypt.client.apache.configurator."
|
||||
@@ -189,5 +164,37 @@ class TwoVhost80Test(util.ApacheTest):
|
||||
|
||||
self.assertEqual(mock_restart.call_count, 1)
|
||||
|
||||
|
||||
class GetVersionTest(unittest.TestCase):
|
||||
@classmethod
|
||||
def _call(cls):
|
||||
from letsencrypt.client.apache.configurator import get_version
|
||||
return get_version()
|
||||
|
||||
@mock.patch("letsencrypt.client.apache.configurator."
|
||||
"subprocess.Popen")
|
||||
def test_get_version(self, mock_popen):
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache/2.4.2 (Debian)", "")
|
||||
self.assertEqual(self._call(), (2, 4, 2))
|
||||
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache/2 (Linux)", "")
|
||||
self.assertEqual(self._call(), (2,))
|
||||
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache (Debian)", "")
|
||||
self.assertRaises(
|
||||
errors.LetsEncryptConfiguratorError, self._call)
|
||||
|
||||
mock_popen().communicate.return_value = (
|
||||
"Server Version: Apache/2.3\n Apache/2.4.7", "")
|
||||
self.assertRaises(
|
||||
errors.LetsEncryptConfiguratorError, self._call)
|
||||
|
||||
mock_popen.side_effect = OSError("Can't find program")
|
||||
self.assertRaises(
|
||||
errors.LetsEncryptConfiguratorError, self._call)
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
|
||||
@@ -64,3 +64,7 @@ class VirtualHostTest(unittest.TestCase):
|
||||
self.assertEqual(vhost1b, self.vhost1)
|
||||
self.assertEqual(str(vhost1b), str(self.vhost1))
|
||||
self.assertNotEqual(vhost1b, 1234)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -8,9 +8,9 @@ import augeas
|
||||
import mock
|
||||
import zope.component
|
||||
|
||||
from letsencrypt.client.display import display_util
|
||||
from letsencrypt.client import errors
|
||||
from letsencrypt.client.apache import parser
|
||||
from letsencrypt.client.display import display_util
|
||||
|
||||
from letsencrypt.client.tests.apache import util
|
||||
|
||||
|
||||
@@ -22,7 +22,7 @@ class DvsniGenCertTest(unittest.TestCase):
|
||||
r_b64 = le_util.jose_b64encode(dvsni_r)
|
||||
pem = pkg_resources.resource_string(
|
||||
__name__, os.path.join("testdata", "rsa256_key.pem"))
|
||||
key = le_util.Client.Key("path", pem)
|
||||
key = le_util.Key("path", pem)
|
||||
nonce = "12345ABCDE"
|
||||
cert_pem, s_b64 = self._call(domain, r_b64, nonce, key)
|
||||
|
||||
@@ -49,3 +49,7 @@ class DvsniGenCertTest(unittest.TestCase):
|
||||
def _call(cls, name, r_b64, nonce, key):
|
||||
from letsencrypt.client.challenge_util import dvsni_gen_cert
|
||||
return dvsni_gen_cert(name, r_b64, nonce, key)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Let's Encrypt Display Tests"""
|
||||
@@ -14,18 +14,16 @@ class DisplayT(unittest.TestCase):
|
||||
self.tags_choices = [("1", "tag1"), ("2", "tag2"), ("3", "tag3")]
|
||||
|
||||
|
||||
def test_visual(displayer, choices):
|
||||
def visual(displayer, choices):
|
||||
"""Visually test all of the display functions."""
|
||||
displayer.notification("Random notification!")
|
||||
displayer.menu("Question?", choices,
|
||||
ok_label="O", cancel_label="Can", help_label="??")
|
||||
ok_label="O", cancel_label="Can", help_label="??")
|
||||
displayer.menu("Question?", [choice[1] for choice in choices],
|
||||
ok_label="O", cancel_label="Can", help_label="??")
|
||||
ok_label="O", cancel_label="Can", help_label="??")
|
||||
displayer.input("Input Message")
|
||||
displayer.yesno(
|
||||
"YesNo Message", yes_label="Yessir", no_label="Nosir")
|
||||
displayer.checklist(
|
||||
"Checklist Message", [choice[0] for choice in choices])
|
||||
displayer.yesno("YesNo Message", yes_label="Yessir", no_label="Nosir")
|
||||
displayer.checklist("Checklist Message", [choice[0] for choice in choices])
|
||||
|
||||
|
||||
class NcursesDisplayTest(DisplayT):
|
||||
@@ -122,7 +120,7 @@ class NcursesDisplayTest(DisplayT):
|
||||
choices=choices)
|
||||
|
||||
# def test_visual(self):
|
||||
# test_visual(self.displayer, self.choices)
|
||||
# visual(self.displayer, self.choices)
|
||||
|
||||
|
||||
class FileOutputDisplayTest(DisplayT):
|
||||
@@ -144,8 +142,7 @@ class FileOutputDisplayTest(DisplayT):
|
||||
self.assertTrue("message" in string)
|
||||
|
||||
def test_notification_pause(self):
|
||||
# Attempt to mock raw_input
|
||||
with mock_raw_input(["enter"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="enter"):
|
||||
self.displayer.notification("message")
|
||||
|
||||
self.assertTrue("message" in self.mock_stdout.write.call_args[0][0])
|
||||
@@ -158,28 +155,26 @@ class FileOutputDisplayTest(DisplayT):
|
||||
self.assertEqual(ret, (display_util.OK, 0))
|
||||
|
||||
def test_input_cancel(self):
|
||||
# Attempt to mock raw_input
|
||||
with mock_raw_input(["c"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="c"):
|
||||
code, _ = self.displayer.input("message")
|
||||
|
||||
self.assertTrue(code, display_util.CANCEL)
|
||||
|
||||
def test_input_normal(self):
|
||||
with mock_raw_input(["domain.com"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="domain.com"):
|
||||
code, input_ = self.displayer.input("message")
|
||||
|
||||
self.assertEqual(code, display_util.OK)
|
||||
self.assertEqual(input_, "domain.com")
|
||||
|
||||
def test_yesno(self):
|
||||
with mock_raw_input(["Yes"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="Yes"):
|
||||
self.assertTrue(self.displayer.yesno("message"))
|
||||
with mock_raw_input(["y"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="y"):
|
||||
self.assertTrue(self.displayer.yesno("message"))
|
||||
with mock_raw_input(["cancel"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="cancel"):
|
||||
self.assertFalse(self.displayer.yesno("message"))
|
||||
|
||||
with mock_raw_input(["a"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="a"):
|
||||
self.assertTrue(self.displayer.yesno("msg", yes_label="Agree"))
|
||||
|
||||
@mock.patch("letsencrypt.client.display.display_util.FileDisplay.input")
|
||||
@@ -252,11 +247,11 @@ class FileOutputDisplayTest(DisplayT):
|
||||
self.assertEqual(text.count(os.linesep), 3)
|
||||
|
||||
def test_get_valid_int_ans_valid(self):
|
||||
with mock_raw_input(["1"]):
|
||||
with mock.patch("__builtin__.raw_input", return_value="1"):
|
||||
self.assertEqual(
|
||||
self.displayer._get_valid_int_ans(1), (display_util.OK, 1))
|
||||
ans = "2"
|
||||
with mock_raw_input([ans]):
|
||||
with mock.patch("__builtin__.raw_input", return_value=ans):
|
||||
self.assertEqual(
|
||||
self.displayer._get_valid_int_ans(3),
|
||||
(display_util.OK, int(ans)))
|
||||
@@ -268,14 +263,14 @@ class FileOutputDisplayTest(DisplayT):
|
||||
["c"],
|
||||
]
|
||||
for ans in answers:
|
||||
with mock_raw_input(ans):
|
||||
with mock.patch("__builtin__.raw_input", side_effect=ans):
|
||||
self.assertEqual(
|
||||
self.displayer._get_valid_int_ans(3),
|
||||
(display_util.CANCEL, -1))
|
||||
|
||||
# def test_visual(self):
|
||||
# self.displayer = display_util.FileDisplay(sys.stdout)
|
||||
# test_visual(self.displayer, self.choices)
|
||||
# visual(self.displayer, self.choices)
|
||||
|
||||
|
||||
class SeparateListInputTest(unittest.TestCase):
|
||||
@@ -323,15 +318,5 @@ class PlaceParensTest(unittest.TestCase):
|
||||
self.assertEqual("(L)abel", ret)
|
||||
|
||||
|
||||
# https://stackoverflow.com/a/25275926
|
||||
@contextlib.contextmanager
|
||||
def mock_raw_input(values):
|
||||
func = mock.MagicMock(side_effect=values)
|
||||
original_raw_input = __builtins__.raw_input
|
||||
__builtins__.raw_input = func
|
||||
yield
|
||||
__builtins__.raw_input = original_raw_input
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -101,18 +101,18 @@ def main(): # pylint: disable=too-many-statements,too-many-branches
|
||||
# Make sure we actually get an installer that is functioning properly
|
||||
# before we begin to try to use it.
|
||||
try:
|
||||
installer = client.determine_authenticator()
|
||||
auth = client.determine_authenticator()
|
||||
except errors.LetsEncryptMisconfigurationError as err:
|
||||
logging.fatal("Please fix your configuration before proceeding.{0}"
|
||||
logging.fatal("Please fix your configuration before proceeding.%s"
|
||||
"The Authenticator exited with the following message: "
|
||||
"{1}".format(os.linesep, err))
|
||||
"%s", (os.linesep, err))
|
||||
sys.exit(1)
|
||||
|
||||
# Use the same object if possible
|
||||
if interfaces.IAuthenticator.providedBy(installer): # pylint: disable=no-member
|
||||
auth = installer
|
||||
if interfaces.IInstaller.providedBy(auth): # pylint: disable=no-member
|
||||
installer = auth
|
||||
else:
|
||||
auth = client.determine_authenticator()
|
||||
installer = client.determine_installer()
|
||||
|
||||
domains = ops.choose_names(installer) if args.domains is None else args.domains
|
||||
|
||||
@@ -131,6 +131,7 @@ def main(): # pylint: disable=too-many-statements,too-many-branches
|
||||
# It should be possible for reconfig only, install-only, no-install
|
||||
# I am not sure the best way to handle all of the unimplemented abilities,
|
||||
# but this code should be safe on all environments.
|
||||
cert_file = None
|
||||
if auth is not None:
|
||||
cert_file, chain_file = acme.obtain_certificate(domains)
|
||||
if installer is not None and cert_file is not None:
|
||||
|
||||
Reference in New Issue
Block a user