Refactor main (#4127)

* Refactor main to simplify logic

* Update tests and comments

* Correct main test

* increase timeout limit

* reset timeout limit

* call renew_cert in appropriate main test

* Update docstrings and revert signatures of _report_new_cert and _suggest_donation_of_appropriate

* replace renew_cert logic

* update tests

* rename _csr_obtain_cert and add a check to _report_new_cert
This commit is contained in:
Erica Portnoy
2017-02-22 13:08:56 -08:00
committed by Brad Warren
parent ebf5170d12
commit 5bab6b512f
5 changed files with 187 additions and 130 deletions
+2 -2
View File
@@ -412,8 +412,8 @@ class HelpfulArgumentParser(object):
def __init__(self, args, plugins, detect_defaults=False):
from certbot import main
self.VERBS = {
"auth": main.obtain_cert,
"certonly": main.obtain_cert,
"auth": main.certonly,
"certonly": main.certonly,
"config_changes": main.config_changes,
"run": main.run,
"install": main.install,
+108 -94
View File
@@ -49,61 +49,45 @@ USER_CANCELLED = ("User chose to cancel the operation and may "
logger = logging.getLogger(__name__)
def _suggest_donation_if_appropriate(config, action):
def _suggest_donation_if_appropriate(config):
"""Potentially suggest a donation to support Certbot."""
if config.staging or config.verb == "renew":
assert config.verb != "renew"
if config.staging:
# --dry-run implies --staging
return
if action not in ["renew", "newcert"]:
return
reporter_util = zope.component.getUtility(interfaces.IReporter)
msg = ("If you like Certbot, please consider supporting our work by:\n\n"
"Donating to ISRG / Let's Encrypt: https://letsencrypt.org/donate\n"
"Donating to EFF: https://eff.org/donate-le\n\n")
reporter_util.add_message(msg, reporter_util.LOW_PRIORITY)
def _report_successful_dry_run(config):
reporter_util = zope.component.getUtility(interfaces.IReporter)
if config.verb != "renew":
reporter_util.add_message("The dry run was successful.",
reporter_util.HIGH_PRIORITY, on_crash=False)
assert config.verb != "renew"
reporter_util.add_message("The dry run was successful.",
reporter_util.HIGH_PRIORITY, on_crash=False)
def _auth_from_available(le_client, config, domains=None, certname=None, lineage=None):
def _get_and_save_cert(le_client, config, domains=None, certname=None, lineage=None):
"""Authenticate and enroll certificate.
This method finds the relevant lineage, figures out what to do with it,
then performs that action. Includes calls to hooks, various reports,
checks, and requests for user input.
:returns: Tuple of (str action, cert_or_None) as per _find_lineage_for_domains_and_certname
action can be: "newcert" | "renew" | "reinstall"
:returns: the issued certificate or `None` if doing a dry run
:rtype: `storage.RenewableCert` or `None`
"""
# If lineage is specified, use that one instead of looking around for
# a matching one.
if lineage is None:
# This will find a relevant matching lineage that exists
action, lineage = _find_lineage_for_domains_and_certname(config, domains, certname)
else:
# Renewal, where we already know the specific lineage we're
# interested in
action = "renew"
if action == "reinstall":
# The lineage already exists; allow the caller to try installing
# it without getting a new certificate at all.
logger.info("Keeping the existing certificate")
return "reinstall", lineage
hooks.pre_hook(config)
try:
if action == "renew":
if lineage is not None:
# Renewal, where we already know the specific lineage we're
# interested in
logger.info("Renewing an existing certificate")
renewal.renew_cert(config, domains, le_client, lineage)
elif action == "newcert":
else:
# TREAT AS NEW REQUEST
assert domains is not None
logger.info("Obtaining a new certificate")
lineage = le_client.obtain_and_enroll_certificate(domains, certname)
if lineage is False:
@@ -111,10 +95,7 @@ def _auth_from_available(le_client, config, domains=None, certname=None, lineage
finally:
hooks.post_hook(config)
if not config.dry_run and not config.verb == "renew":
_report_new_cert(config, lineage.cert, lineage.fullchain)
return action, lineage
return lineage
def _handle_subset_cert_request(config, domains, cert):
@@ -236,6 +217,18 @@ def _find_lineage_for_domains(config, domains):
elif subset_names_cert is not None:
return _handle_subset_cert_request(config, domains, subset_names_cert)
def _find_cert(config, domains, certname):
"""Finds an existing certificate object given domains and/or a certificate name.
:returns: Two-element tuple of a boolean that indicates if this function should be
followed by a call to fetch a certificate from the server, and either a
RenewableCert instance or None.
"""
action, lineage = _find_lineage_for_domains_and_certname(config, domains, certname)
if action == "reinstall":
logger.info("Keeping the existing certificate")
return (action != "reinstall"), lineage
def _find_lineage_for_domains_and_certname(config, domains, certname):
"""Find appropriate lineage based on given domains and/or certname.
@@ -314,26 +307,25 @@ def _report_new_cert(config, cert_path, fullchain_path):
:param str fullchain_path: path to full chain
"""
if config.dry_run:
_report_successful_dry_run(config)
return
assert cert_path and fullchain_path, "No certificates saved to report."
expiry = crypto_util.notAfter(cert_path).date()
reporter_util = zope.component.getUtility(interfaces.IReporter)
if fullchain_path:
# Print the path to fullchain.pem because that's what modern webservers
# (Nginx and Apache2.4) will want.
and_chain = "and chain have"
path = fullchain_path
else:
# Unless we're in .csr mode and there really isn't one
and_chain = "has "
path = cert_path
# Print the path to fullchain.pem because that's what modern webservers
# (Nginx and Apache2.4) will want.
verbswitch = ' with the "certonly" option' if config.verb == "run" else ""
# XXX Perhaps one day we could detect the presence of known old webservers
# and say something more informative here.
msg = ('Congratulations! Your certificate {0} been saved at {1}.'
' Your cert will expire on {2}. To obtain a new or tweaked version of this '
'certificate in the future, simply run {3} again{4}. '
'To non-interactively renew *all* of your certificates, run "{3} renew"'
.format(and_chain, path, expiry, cli.cli_command, verbswitch))
msg = ('Congratulations! Your certificate and chain have been saved at {0}.'
' Your cert will expire on {1}. To obtain a new or tweaked version of this '
'certificate in the future, simply run {2} again{3}. '
'To non-interactively renew *all* of your certificates, run "{2} renew"'
.format(fullchain_path, expiry, cli.cli_command, verbswitch))
reporter_util.add_message(msg, reporter_util.MEDIUM_PRIORITY)
@@ -478,6 +470,13 @@ def register(config, unused_plugins):
eff.handle_subscription(config)
add_msg("Your e-mail address was updated to {0}.".format(config.email))
def _install_cert(config, le_client, domains, lineage=None):
path_provider = lineage if lineage else config
assert path_provider.cert_path is not None
le_client.deploy_certificate(domains, path_provider.key_path,
path_provider.cert_path, path_provider.chain_path, path_provider.fullchain_path)
le_client.enhance_config(domains, path_provider.chain_path)
def install(config, plugins):
"""Install a previously obtained cert in a server."""
@@ -492,11 +491,7 @@ def install(config, plugins):
domains, _ = _find_domains_or_certname(config, installer)
le_client = _init_le_client(config, authenticator=None, installer=installer)
assert config.cert_path is not None # required=True in the subparser
le_client.deploy_certificate(
domains, config.key_path, config.cert_path, config.chain_path,
config.fullchain_path)
le_client.enhance_config(domains, config.chain_path)
_install_cert(config, le_client, domains)
def plugins_cmd(config, plugins): # TODO: Use IDisplay rather than print
@@ -600,28 +595,32 @@ def run(config, plugins): # pylint: disable=too-many-branches,too-many-locals
except errors.PluginSelectionError as e:
return e.message
domains, certname = _find_domains_or_certname(config, installer)
# TODO: Handle errors from _init_le_client?
le_client = _init_le_client(config, authenticator, installer)
action, lineage = _auth_from_available(le_client, config, domains, certname)
domains, certname = _find_domains_or_certname(config, installer)
should_get_cert, lineage = _find_cert(config, domains, certname)
le_client.deploy_certificate(
domains, lineage.privkey, lineage.cert,
lineage.chain, lineage.fullchain)
new_lineage = lineage
if should_get_cert:
new_lineage = _get_and_save_cert(le_client, config, domains,
certname, lineage)
le_client.enhance_config(domains, lineage.chain)
cert_path = new_lineage.cert_path if new_lineage else None
fullchain_path = new_lineage.fullchain_path if new_lineage else None
_report_new_cert(config, cert_path, fullchain_path)
if action in ("newcert", "reinstall",):
_install_cert(config, le_client, domains, new_lineage)
if lineage is None or not should_get_cert:
display_ops.success_installation(domains)
else:
display_ops.success_renewal(domains)
_suggest_donation_if_appropriate(config, action)
_suggest_donation_if_appropriate(config)
def _csr_obtain_cert(config, le_client):
def _csr_get_and_save_cert(config, le_client):
"""Obtain a cert using a user-supplied CSR
This works differently in the CSR case (for now) because we don't
@@ -633,16 +632,39 @@ def _csr_obtain_cert(config, le_client):
if config.dry_run:
logger.debug(
"Dry run: skipping saving certificate to %s", config.cert_path)
else:
cert_path, _, cert_fullchain = le_client.save_certificate(
return None, None
cert_path, _, fullchain_path = le_client.save_certificate(
certr, chain, config.cert_path, config.chain_path, config.fullchain_path)
_report_new_cert(config, cert_path, cert_fullchain)
return cert_path, fullchain_path
def obtain_cert(config, plugins, lineage=None):
def renew_cert(config, plugins, lineage):
"""Renew & save an existing cert. Do not install it."""
try:
# installers are used in auth mode to determine domain names
installer, auth = plug_sel.choose_configurator_plugins(config, plugins, "certonly")
except errors.PluginSelectionError as e:
logger.info("Could not choose appropriate plugin: %s", e)
raise
le_client = _init_le_client(config, auth, installer)
_get_and_save_cert(le_client, config, lineage=lineage)
notify = zope.component.getUtility(interfaces.IDisplay).notification
if installer is None:
notify("new certificate deployed without reload, fullchain is {0}".format(
lineage.fullchain), pause=False)
else:
# In case of a renewal, reload server to pick up new certificate.
# In principle we could have a configuration option to inhibit this
# from happening.
installer.restart()
notify("new certificate deployed with reload of {0} server; fullchain is {1}".format(
config.installer, lineage.fullchain), pause=False)
def certonly(config, plugins):
"""Authenticate & obtain cert, but do not install it.
This implements the 'certonly' subcommand, and is also called from within the
'renew' command."""
This implements the 'certonly' subcommand."""
# SETUP: Select plugins and construct a client instance
try:
@@ -653,34 +675,26 @@ def obtain_cert(config, plugins, lineage=None):
raise
le_client = _init_le_client(config, auth, installer)
# SHOWTIME: Possibly obtain/renew a cert, and set action to renew | newcert | reinstall
if config.csr is None: # the common case
domains, certname = _find_domains_or_certname(config, installer)
action, _ = _auth_from_available(le_client, config, domains, certname, lineage)
else:
assert lineage is None, "Did not expect a CSR with a RenewableCert"
_csr_obtain_cert(config, le_client)
action = "newcert"
if config.csr:
cert_path, fullchain_path = _csr_get_and_save_cert(config, le_client)
_report_new_cert(config, cert_path, fullchain_path)
_suggest_donation_if_appropriate(config)
return
# POSTPRODUCTION: Cleanup, deployment & reporting
notify = zope.component.getUtility(interfaces.IDisplay).notification
if config.dry_run:
_report_successful_dry_run(config)
elif config.verb == "renew":
if installer is None:
notify("new certificate deployed without reload, fullchain is {0}".format(
lineage.fullchain), pause=False)
else:
# In case of a renewal, reload server to pick up new certificate.
# In principle we could have a configuration option to inhibit this
# from happening.
installer.restart()
notify("new certificate deployed with reload of {0} server; fullchain is {1}".format(
config.installer, lineage.fullchain), pause=False)
elif action == "reinstall" and config.verb == "certonly":
domains, certname = _find_domains_or_certname(config, installer)
should_get_cert, lineage = _find_cert(config, domains, certname)
if not should_get_cert:
notify = zope.component.getUtility(interfaces.IDisplay).notification
notify("Certificate not yet due for renewal; no action taken.", pause=False)
_suggest_donation_if_appropriate(config, action)
return
lineage = _get_and_save_cert(le_client, config, domains, certname, lineage)
cert_path = lineage.cert_path if lineage else None
fullchain_path = lineage.fullchain_path if lineage else None
_report_new_cert(config, cert_path, fullchain_path)
_suggest_donation_if_appropriate(config)
def renew(config, unused_plugins):
"""Renew previously-obtained certificates."""
+6 -1
View File
@@ -410,7 +410,12 @@ def handle_renewal_request(config):
if should_renew(lineage_config, renewal_candidate):
plugins = plugins_disco.PluginsRegistry.find_all()
from certbot import main
main.obtain_cert(lineage_config, plugins, renewal_candidate)
# domains have been restored into lineage_config by reconstitute
# but they're unnecessary anyway because renew_cert here
# will just grab them from the certificate
# we already know it's time to renew based on should_renew
# and we have a lineage in renewal_candidate
main.renew_cert(lineage_config, plugins, renewal_candidate)
renew_successes.append(renewal_candidate.fullchain)
else:
renew_skipped.append(renewal_candidate.fullchain)
+20
View File
@@ -391,6 +391,26 @@ class RenewableCert(object):
self._update_symlinks()
self._check_symlinks()
@property
def key_path(self):
"""Duck type for self.privkey"""
return self.privkey
@property
def cert_path(self):
"""Duck type for self.cert"""
return self.cert
@property
def chain_path(self):
"""Duck type for self.chain"""
return self.chain
@property
def fullchain_path(self):
"""Duck type for self.fullchain"""
return self.fullchain
@property
def target_expiry(self):
"""The current target certificate's expiration datetime
+51 -33
View File
@@ -57,17 +57,21 @@ class RunTest(unittest.TestCase):
def setUp(self):
self.domain = 'example.org'
self.patches = [
mock.patch('certbot.main._auth_from_available'),
mock.patch('certbot.main._get_and_save_cert'),
mock.patch('certbot.main.display_ops.success_installation'),
mock.patch('certbot.main.display_ops.success_renewal'),
mock.patch('certbot.main._init_le_client'),
mock.patch('certbot.main._suggest_donation_if_appropriate')]
mock.patch('certbot.main._suggest_donation_if_appropriate'),
mock.patch('certbot.main._report_new_cert'),
mock.patch('certbot.main._find_cert')]
self.mock_auth = self.patches[0].start()
self.mock_success_installation = self.patches[1].start()
self.mock_success_renewal = self.patches[2].start()
self.mock_init = self.patches[3].start()
self.mock_suggest_donation = self.patches[4].start()
self.mock_report_cert = self.patches[5].start()
self.mock_find_cert = self.patches[6].start()
def tearDown(self):
for patch in self.patches:
@@ -83,23 +87,26 @@ class RunTest(unittest.TestCase):
run(config, plugins)
def test_newcert_success(self):
self.mock_auth.return_value = ('newcert', mock.Mock())
self.mock_auth.return_value = mock.Mock()
self.mock_find_cert.return_value = True, None
self._call()
self.mock_success_installation.assert_called_once_with([self.domain])
def test_reinstall_success(self):
self.mock_auth.return_value = ('reinstall', mock.Mock())
self.mock_auth.return_value = mock.Mock()
self.mock_find_cert.return_value = False, mock.Mock()
self._call()
self.mock_success_installation.assert_called_once_with([self.domain])
def test_renewal_success(self):
self.mock_auth.return_value = ('renewal', mock.Mock())
self.mock_auth.return_value = mock.Mock()
self.mock_find_cert.return_value = True, mock.Mock()
self._call()
self.mock_success_renewal.assert_called_once_with([self.domain])
class ObtainCertTest(unittest.TestCase):
"""Tests for certbot.main.obtain_cert."""
class CertonlyTest(unittest.TestCase):
"""Tests for certbot.main.certonly."""
def setUp(self):
self.get_utility_patch = test_util.patch_get_utility()
@@ -114,15 +121,20 @@ class ObtainCertTest(unittest.TestCase):
cli.prepare_and_parse_args(plugins, args))
with mock.patch('certbot.main._init_le_client') as mock_init:
main.obtain_cert(config, plugins)
with mock.patch('certbot.main._suggest_donation_if_appropriate'):
main.certonly(config, plugins)
return mock_init() # returns the client
@mock.patch('certbot.main._auth_from_available')
def test_no_reinstall_text_pause(self, mock_auth):
@mock.patch('certbot.main._find_cert')
@mock.patch('certbot.main._get_and_save_cert')
@mock.patch('certbot.main._report_new_cert')
def test_no_reinstall_text_pause(self, unused_report, mock_auth,
mock_find_cert):
mock_notification = self.mock_get_utility().notification
mock_notification.side_effect = self._assert_no_pause
mock_auth.return_value = ('reinstall', mock.ANY)
mock_auth.return_value = mock.Mock()
mock_find_cert.return_value = False, None
self._call('certonly --webroot -d example.com'.split())
def _assert_no_pause(self, message, pause=True):
@@ -495,22 +507,23 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
self._cli_missing_flag(args, "specify a plugin")
args.extend(['--standalone', '-d', 'eg.is'])
self._cli_missing_flag(args, "register before running")
with mock.patch('certbot.main._auth_from_available'):
with mock.patch('certbot.main._get_and_save_cert'):
with mock.patch('certbot.main.client.acme_from_config_key'):
args.extend(['--email', 'io@io.is'])
self._cli_missing_flag(args, "--agree-tos")
@mock.patch('certbot.main._report_new_cert')
@mock.patch('certbot.main.client.acme_client.Client')
@mock.patch('certbot.main._determine_account')
@mock.patch('certbot.main.client.Client.obtain_and_enroll_certificate')
@mock.patch('certbot.main._auth_from_available')
def test_user_agent(self, afa, _obt, det, _client):
@mock.patch('certbot.main._get_and_save_cert')
def test_user_agent(self, gsc, _obt, det, _client, unused_report):
# Normally the client is totally mocked out, but here we need more
# arguments to automate it...
args = ["--standalone", "certonly", "-m", "none@none.com",
"-d", "example.com", '--agree-tos'] + self.standard_args
det.return_value = mock.MagicMock(), None
afa.return_value = "newcert", mock.MagicMock()
gsc.return_value = mock.MagicMock()
with mock.patch('certbot.main.client.acme_client.ClientNetwork') as acme_net:
self._call_no_clientmock(args)
@@ -535,8 +548,9 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
'--key-path', 'key', '--chain-path', 'chain'])
self.assertEqual(mock_pick_installer.call_count, 1)
@mock.patch('certbot.main._report_new_cert')
@mock.patch('certbot.util.exe_exists')
def test_configurator_selection(self, mock_exe_exists):
def test_configurator_selection(self, mock_exe_exists, unused_report):
mock_exe_exists.return_value = True
real_plugins = disco.PluginsRegistry.find_all()
args = ['--apache', '--authenticator', 'standalone']
@@ -562,13 +576,13 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
self._cli_missing_flag(["--standalone"], "With the standalone plugin, you probably")
with mock.patch("certbot.main._init_le_client") as mock_init:
with mock.patch("certbot.main._auth_from_available") as mock_afa:
mock_afa.return_value = (mock.MagicMock(), mock.MagicMock())
with mock.patch("certbot.main._get_and_save_cert") as mock_gsc:
mock_gsc.return_value = mock.MagicMock()
self._call(["certonly", "--manual", "-d", "foo.bar"])
unused_config, auth, unused_installer = mock_init.call_args[0]
self.assertTrue(isinstance(auth, manual.Authenticator))
with mock.patch('certbot.main.obtain_cert') as mock_certonly:
with mock.patch('certbot.main.certonly') as mock_certonly:
self._call(["auth", "--standalone"])
self.assertEqual(1, mock_certonly.call_count)
@@ -656,12 +670,12 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
chain = 'chain'
fullchain = 'fullchain'
with mock.patch('certbot.main.obtain_cert') as mock_obtaincert:
with mock.patch('certbot.main.certonly') as mock_certonly:
self._call(['certonly', '--cert-path', cert, '--key-path', 'key',
'--chain-path', 'chain',
'--fullchain-path', 'fullchain'])
config, unused_plugins = mock_obtaincert.call_args[0]
config, unused_plugins = mock_certonly.call_args[0]
self.assertEqual(config.cert_path, os.path.abspath(cert))
self.assertEqual(config.key_path, os.path.abspath(key))
self.assertEqual(config.chain_path, os.path.abspath(chain))
@@ -747,7 +761,8 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
date = '1970-01-01'
mock_notAfter().date.return_value = date
mock_lineage = mock.MagicMock(cert=cert_path, fullchain=cert_path)
mock_lineage = mock.MagicMock(cert=cert_path, fullchain=cert_path,
fullchain_path=cert_path)
mock_client = mock.MagicMock()
mock_client.obtain_and_enroll_certificate.return_value = mock_lineage
self._certonly_new_request_common(mock_client)
@@ -770,7 +785,8 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
# pylint: disable=too-many-locals,too-many-arguments
cert_path = test_util.vector_path('cert.pem')
chain_path = '/etc/letsencrypt/live/foo.bar/fullchain.pem'
mock_lineage = mock.MagicMock(cert=cert_path, fullchain=chain_path)
mock_lineage = mock.MagicMock(cert=cert_path, fullchain=chain_path,
cert_path=cert_path, fullchain_path=chain_path)
mock_lineage.should_autorenew.return_value = due_for_renewal
mock_lineage.has_pending_deployment.return_value = False
mock_lineage.names.return_value = ['isnot.org']
@@ -821,7 +837,8 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
return mock_lineage, mock_get_utility, stdout
def test_certonly_renewal(self):
@mock.patch('certbot.crypto_util.notAfter')
def test_certonly_renewal(self, unused_notafter):
lineage, get_utility, _ = self._test_renewal_common(True, [])
self.assertEqual(lineage.save_successor.call_count, 1)
lineage.update_all_links_to.assert_called_once_with(
@@ -830,7 +847,8 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertTrue('fullchain.pem' in cert_msg)
self.assertTrue('donate' in get_utility().add_message.call_args[0][0])
def test_certonly_renewal_triggers(self):
@mock.patch('certbot.crypto_util.notAfter')
def test_certonly_renewal_triggers(self, unused_notafter):
# --dry-run should force renewal
_, get_utility, _ = self._test_renewal_common(False, ['--dry-run', '--keep'],
log_out="simulating renewal")
@@ -932,15 +950,15 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
if names is not None:
mock_lineage.names.return_value = names
mock_rc.return_value = mock_lineage
with mock.patch('certbot.main.obtain_cert') as mock_obtain_cert:
with mock.patch('certbot.main.renew_cert') as mock_renew_cert:
kwargs.setdefault('args', ['renew'])
self._test_renewal_common(True, None, should_renew=False, **kwargs)
if assert_oc_called is not None:
if assert_oc_called:
self.assertTrue(mock_obtain_cert.called)
self.assertTrue(mock_renew_cert.called)
else:
self.assertFalse(mock_obtain_cert.called)
self.assertFalse(mock_renew_cert.called)
def test_renew_no_renewalparams(self):
self._test_renew_common(assert_oc_called=False, error_expected=True)
@@ -1000,8 +1018,8 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_rc.return_value = mock_lineage
mock_lineage.configuration = {
'renewalparams': {'authenticator': 'webroot'}}
with mock.patch('certbot.main.obtain_cert') as mock_obtain_cert:
mock_obtain_cert.side_effect = Exception
with mock.patch('certbot.main.renew_cert') as mock_renew_cert:
mock_renew_cert.side_effect = Exception
self._test_renewal_common(True, None, error_expected=True,
args=['renew'], should_renew=False)
@@ -1035,12 +1053,12 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_client = mock.MagicMock()
mock_client.obtain_certificate_from_csr.return_value = (certr, chain)
cert_path = '/etc/letsencrypt/live/example.com/cert.pem'
mock_client.save_certificate.return_value = cert_path, None, None
full_path = '/etc/letsencrypt/live/example.com/fullchain.pem'
mock_client.save_certificate.return_value = cert_path, None, full_path
with mock.patch('certbot.main._init_le_client') as mock_init:
mock_init.return_value = mock_client
with test_util.patch_get_utility() as mock_get_utility:
chain_path = '/etc/letsencrypt/live/example.com/chain.pem'
full_path = '/etc/letsencrypt/live/example.com/fullchain.pem'
args = ('-a standalone certonly --csr {0} --cert-path {1} '
'--chain-path {2} --fullchain-path {3}').format(
CSR, cert_path, chain_path, full_path).split()
@@ -1060,7 +1078,7 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods
def test_certonly_csr(self):
mock_get_utility = self._test_certonly_csr_common()
cert_msg = mock_get_utility().add_message.call_args_list[0][0][0]
self.assertTrue('cert.pem' in cert_msg)
self.assertTrue('fullchain.pem' in cert_msg)
self.assertTrue(
'donate' in mock_get_utility().add_message.call_args[0][0])