Viewing file:  account_test.py (13.79 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
"""Tests for certbot.account."""
import datetime
import json
import unittest
import josepy as jose
import mock
import pytz
from acme import messages
import certbot.tests.util as test_util
from certbot import errors
from certbot.compat import filesystem
from certbot.compat import misc
from certbot.compat import os
KEY = jose.JWKRSA.load(test_util.load_vector("rsa512_key.pem"))
class AccountTest(unittest.TestCase):
"""Tests for certbot.account.Account."""
def setUp(self):
from certbot.account import Account
self.regr = mock.MagicMock()
self.meta = Account.Meta(
creation_host="test.certbot.org",
creation_dt=datetime.datetime(
2015, 7, 4, 14, 4, 10, tzinfo=pytz.UTC))
self.acc = Account(self.regr, KEY, self.meta)
self.regr.__repr__ = mock.MagicMock(return_value="i_am_a_regr")
with mock.patch("certbot.account.socket") as mock_socket:
mock_socket.getfqdn.return_value = "test.certbot.org"
with mock.patch("certbot.account.datetime") as mock_dt:
mock_dt.datetime.now.return_value = self.meta.creation_dt
self.acc_no_meta = Account(self.regr, KEY)
def test_init(self):
self.assertEqual(self.regr, self.acc.regr)
self.assertEqual(KEY, self.acc.key)
self.assertEqual(self.meta, self.acc_no_meta.meta)
def test_id(self):
self.assertEqual(
self.acc.id, "7adac10320f585ddf118429c0c4af2cd")
def test_slug(self):
self.assertEqual(
self.acc.slug, "test.certbot.org@2015-07-04T14:04:10Z (7ada)")
def test_repr(self):
self.assertTrue(repr(self.acc).startswith(
"<Account(i_am_a_regr, 7adac10320f585ddf118429c0c4af2cd, Meta("))
class ReportNewAccountTest(test_util.ConfigTestCase):
"""Tests for certbot.account.report_new_account."""
def _call(self):
from certbot.account import report_new_account
report_new_account(self.config)
@mock.patch("certbot.account.zope.component.queryUtility")
def test_no_reporter(self, mock_zope):
mock_zope.return_value = None
self._call()
@mock.patch("certbot.account.zope.component.queryUtility")
def test_it(self, mock_zope):
self._call()
call_list = mock_zope().add_message.call_args_list
self.assertTrue(self.config.config_dir in call_list[0][0][0])
class AccountMemoryStorageTest(unittest.TestCase):
"""Tests for certbot.account.AccountMemoryStorage."""
def setUp(self):
from certbot.account import AccountMemoryStorage
self.storage = AccountMemoryStorage()
def test_it(self):
account = mock.Mock(id="x")
self.assertEqual([], self.storage.find_all())
self.assertRaises(errors.AccountNotFound, self.storage.load, "x")
self.storage.save(account, None)
self.assertEqual([account], self.storage.find_all())
self.assertEqual(account, self.storage.load("x"))
self.storage.save(account, None)
self.assertEqual([account], self.storage.find_all())
class AccountFileStorageTest(test_util.ConfigTestCase):
"""Tests for certbot.account.AccountFileStorage."""
#pylint: disable=too-many-public-methods
def setUp(self):
super(AccountFileStorageTest, self).setUp()
from certbot.account import AccountFileStorage
self.storage = AccountFileStorage(self.config)
from certbot.account import Account
new_authzr_uri = "hi"
self.acc = Account(
regr=messages.RegistrationResource(
uri=None, body=messages.Registration(),
new_authzr_uri=new_authzr_uri),
key=KEY)
self.mock_client = mock.MagicMock()
self.mock_client.directory.new_authz = new_authzr_uri
def test_init_creates_dir(self):
self.assertTrue(os.path.isdir(
misc.underscores_for_unsupported_characters_in_path(self.config.accounts_dir)))
def test_save_and_restore(self):
self.storage.save(self.acc, self.mock_client)
account_path = os.path.join(self.config.accounts_dir, self.acc.id)
self.assertTrue(os.path.exists(account_path))
for file_name in "regr.json", "meta.json", "private_key.json":
self.assertTrue(os.path.exists(
os.path.join(account_path, file_name)))
self.assertTrue(
filesystem.check_mode(os.path.join(account_path, "private_key.json"), 0o400))
# restore
loaded = self.storage.load(self.acc.id)
self.assertEqual(self.acc, loaded)
def test_save_and_restore_old_version(self):
"""Saved regr should include a new_authzr_uri for older Certbots"""
self.storage.save(self.acc, self.mock_client)
path = os.path.join(self.config.accounts_dir, self.acc.id, "regr.json")
with open(path, "r") as f:
regr = json.load(f)
self.assertTrue("new_authzr_uri" in regr)
def test_save_regr(self):
self.storage.save_regr(self.acc, self.mock_client)
account_path = os.path.join(self.config.accounts_dir, self.acc.id)
self.assertTrue(os.path.exists(account_path))
self.assertTrue(os.path.exists(os.path.join(
account_path, "regr.json")))
for file_name in "meta.json", "private_key.json":
self.assertFalse(os.path.exists(
os.path.join(account_path, file_name)))
def test_find_all(self):
self.storage.save(self.acc, self.mock_client)
self.assertEqual([self.acc], self.storage.find_all())
def test_find_all_none_empty_list(self):
self.assertEqual([], self.storage.find_all())
def test_find_all_accounts_dir_absent(self):
os.rmdir(self.config.accounts_dir)
self.assertEqual([], self.storage.find_all())
def test_find_all_load_skips(self):
# pylint: disable=protected-access
self.storage._load_for_server_path = mock.MagicMock(
side_effect=["x", errors.AccountStorageError, "z"])
with mock.patch("certbot.account.os.listdir") as mock_listdir:
mock_listdir.return_value = ["x", "y", "z"]
self.assertEqual(["x", "z"], self.storage.find_all())
def test_load_non_existent_raises_error(self):
self.assertRaises(errors.AccountNotFound, self.storage.load, "missing")
def _set_server(self, server):
self.config.server = server
from certbot.account import AccountFileStorage
self.storage = AccountFileStorage(self.config)
def test_find_all_neither_exists(self):
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.assertEqual([], self.storage.find_all())
self.assertEqual([], self.storage.find_all())
self.assertFalse(os.path.islink(self.config.accounts_dir))
def test_find_all_find_before_save(self):
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.assertEqual([], self.storage.find_all())
self.storage.save(self.acc, self.mock_client)
self.assertEqual([self.acc], self.storage.find_all())
self.assertEqual([self.acc], self.storage.find_all())
self.assertFalse(os.path.islink(self.config.accounts_dir))
# we shouldn't have created a v1 account
prev_server_path = 'https://acme-staging.api.letsencrypt.org/directory'
self.assertFalse(os.path.isdir(self.config.accounts_dir_for_server_path(prev_server_path)))
def test_find_all_save_before_find(self):
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
self.assertEqual([self.acc], self.storage.find_all())
self.assertEqual([self.acc], self.storage.find_all())
self.assertFalse(os.path.islink(self.config.accounts_dir))
self.assertTrue(os.path.isdir(self.config.accounts_dir))
prev_server_path = 'https://acme-staging.api.letsencrypt.org/directory'
self.assertFalse(os.path.isdir(self.config.accounts_dir_for_server_path(prev_server_path)))
def test_find_all_server_downgrade(self):
# don't use v2 accounts with a v1 url
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.assertEqual([], self.storage.find_all())
self.storage.save(self.acc, self.mock_client)
self.assertEqual([self.acc], self.storage.find_all())
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.assertEqual([], self.storage.find_all())
def test_upgrade_version_staging(self):
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.assertEqual([self.acc], self.storage.find_all())
def test_upgrade_version_production(self):
self._set_server('https://acme-v01.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
self._set_server('https://acme-v02.api.letsencrypt.org/directory')
self.assertEqual([self.acc], self.storage.find_all())
@mock.patch('certbot.compat.os.rmdir')
def test_corrupted_account(self, mock_rmdir):
# pylint: disable=protected-access
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
mock_rmdir.side_effect = OSError
self.storage._load_for_server_path = mock.MagicMock(
side_effect=errors.AccountStorageError)
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.assertEqual([], self.storage.find_all())
def test_upgrade_load(self):
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
prev_account = self.storage.load(self.acc.id)
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
account = self.storage.load(self.acc.id)
self.assertEqual(prev_account, account)
def test_upgrade_load_single_account(self):
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
prev_account = self.storage.load(self.acc.id)
self._set_server_and_stop_symlink('https://acme-staging-v02.api.letsencrypt.org/directory')
account = self.storage.load(self.acc.id)
self.assertEqual(prev_account, account)
def test_load_ioerror(self):
self.storage.save(self.acc, self.mock_client)
mock_open = mock.mock_open()
mock_open.side_effect = IOError
with mock.patch("six.moves.builtins.open", mock_open):
self.assertRaises(
errors.AccountStorageError, self.storage.load, self.acc.id)
def test_save_ioerrors(self):
mock_open = mock.mock_open()
mock_open.side_effect = IOError # TODO: [None, None, IOError]
with mock.patch("six.moves.builtins.open", mock_open):
self.assertRaises(
errors.AccountStorageError, self.storage.save,
self.acc, self.mock_client)
def test_delete(self):
self.storage.save(self.acc, self.mock_client)
self.storage.delete(self.acc.id)
self.assertRaises(errors.AccountNotFound, self.storage.load, self.acc.id)
def test_delete_no_account(self):
self.assertRaises(errors.AccountNotFound, self.storage.delete, self.acc.id)
def _assert_symlinked_account_removed(self):
# create v1 account
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
# ensure v2 isn't already linked to it
with mock.patch('certbot.constants.LE_REUSE_SERVERS', {}):
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.assertRaises(errors.AccountNotFound, self.storage.load, self.acc.id)
def _test_delete_folders(self, server_url):
# create symlinked servers
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.storage.save(self.acc, self.mock_client)
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.storage.load(self.acc.id)
# delete starting at given server_url
self._set_server(server_url)
self.storage.delete(self.acc.id)
# make sure we're gone from both urls
self._set_server('https://acme-staging.api.letsencrypt.org/directory')
self.assertRaises(errors.AccountNotFound, self.storage.load, self.acc.id)
self._set_server('https://acme-staging-v02.api.letsencrypt.org/directory')
self.assertRaises(errors.AccountNotFound, self.storage.load, self.acc.id)
def test_delete_folders_up(self):
self._test_delete_folders('https://acme-staging.api.letsencrypt.org/directory')
self._assert_symlinked_account_removed()
def test_delete_folders_down(self):
self._test_delete_folders('https://acme-staging-v02.api.letsencrypt.org/directory')
self._assert_symlinked_account_removed()
def _set_server_and_stop_symlink(self, server_path):
self._set_server(server_path)
with open(os.path.join(self.config.accounts_dir, 'foo'), 'w') as f:
f.write('bar')
def test_delete_shared_account_up(self):
self._set_server_and_stop_symlink('https://acme-staging-v02.api.letsencrypt.org/directory')
self._test_delete_folders('https://acme-staging.api.letsencrypt.org/directory')
def test_delete_shared_account_down(self):
self._set_server_and_stop_symlink('https://acme-staging-v02.api.letsencrypt.org/directory')
self._test_delete_folders('https://acme-staging-v02.api.letsencrypt.org/directory')
if __name__ == "__main__":
unittest.main() # pragma: no cover
|