diff options
Diffstat (limited to 'tests/unittests/config/test_cc_ca_certs.py')
| -rw-r--r-- | tests/unittests/config/test_cc_ca_certs.py | 220 |
1 files changed, 134 insertions, 86 deletions
diff --git a/tests/unittests/config/test_cc_ca_certs.py b/tests/unittests/config/test_cc_ca_certs.py index 91b005d0..c49922e6 100644 --- a/tests/unittests/config/test_cc_ca_certs.py +++ b/tests/unittests/config/test_cc_ca_certs.py @@ -6,13 +6,9 @@ import unittest from contextlib import ExitStack from unittest import mock -from cloudinit import distros +from cloudinit import distros, helpers, subp, util from cloudinit.config import cc_ca_certs -from cloudinit import helpers -from cloudinit import subp -from cloudinit import util from tests.unittests.helpers import TestCase - from tests.unittests.util import get_cloud @@ -31,12 +27,15 @@ class TestNoConfig(unittest.TestCase): config = util.get_builtin_cfg() with ExitStack() as mocks: util_mock = mocks.enter_context( - mock.patch.object(util, 'write_file')) + mock.patch.object(util, "write_file") + ) certs_mock = mocks.enter_context( - mock.patch.object(cc_ca_certs, 'update_ca_certs')) + mock.patch.object(cc_ca_certs, "update_ca_certs") + ) - cc_ca_certs.handle(self.name, config, self.cloud_init, self.log, - self.args) + cc_ca_certs.handle( + self.name, config, self.cloud_init, self.log, self.args + ) self.assertEqual(util_mock.call_count, 0) self.assertEqual(certs_mock.call_count, 0) @@ -61,11 +60,14 @@ class TestConfig(TestCase): # Mock out the functions that actually modify the system self.mock_add = self.mocks.enter_context( - mock.patch.object(cc_ca_certs, 'add_ca_certs')) + mock.patch.object(cc_ca_certs, "add_ca_certs") + ) self.mock_update = self.mocks.enter_context( - mock.patch.object(cc_ca_certs, 'update_ca_certs')) + mock.patch.object(cc_ca_certs, "update_ca_certs") + ) self.mock_remove = self.mocks.enter_context( - mock.patch.object(cc_ca_certs, 'remove_default_ca_certs')) + mock.patch.object(cc_ca_certs, "remove_default_ca_certs") + ) def test_no_trusted_list(self): """ @@ -106,7 +108,7 @@ class TestConfig(TestCase): conf = cc_ca_certs._distro_ca_certs_configs(distro_name) cc_ca_certs.handle(self.name, config, cloud, self.log, self.args) - self.mock_add.assert_called_once_with(conf, ['CERT1']) + self.mock_add.assert_called_once_with(conf, ["CERT1"]) self.assertEqual(self.mock_update.call_count, 1) self.assertEqual(self.mock_remove.call_count, 0) @@ -120,7 +122,7 @@ class TestConfig(TestCase): conf = cc_ca_certs._distro_ca_certs_configs(distro_name) cc_ca_certs.handle(self.name, config, cloud, self.log, self.args) - self.mock_add.assert_called_once_with(conf, ['CERT1', 'CERT2']) + self.mock_add.assert_called_once_with(conf, ["CERT1", "CERT2"]) self.assertEqual(self.mock_update.call_count, 1) self.assertEqual(self.mock_remove.call_count, 0) @@ -160,20 +162,21 @@ class TestConfig(TestCase): conf = cc_ca_certs._distro_ca_certs_configs(distro_name) cc_ca_certs.handle(self.name, config, cloud, self.log, self.args) - self.mock_add.assert_called_once_with(conf, ['CERT1']) + self.mock_add.assert_called_once_with(conf, ["CERT1"]) self.assertEqual(self.mock_update.call_count, 1) self.assertEqual(self.mock_remove.call_count, 1) class TestAddCaCerts(TestCase): - def setUp(self): super(TestAddCaCerts, self).setUp() tmpdir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, tmpdir) - self.paths = helpers.Paths({ - 'cloud_dir': tmpdir, - }) + self.paths = helpers.Paths( + { + "cloud_dir": tmpdir, + } + ) self.add_patch("cloudinit.config.cc_ca_certs.os.stat", "m_stat") def _fetch_distro(self, kind): @@ -185,7 +188,7 @@ class TestAddCaCerts(TestCase): """Test that no certificate are written if not provided.""" for distro_name in cc_ca_certs.distros: conf = cc_ca_certs._distro_ca_certs_configs(distro_name) - with mock.patch.object(util, 'write_file') as mockobj: + with mock.patch.object(util, "write_file") as mockobj: cc_ca_certs.add_ca_certs(conf, []) self.assertEqual(mockobj.call_count, 0) @@ -204,21 +207,28 @@ class TestAddCaCerts(TestCase): with ExitStack() as mocks: mock_write = mocks.enter_context( - mock.patch.object(util, 'write_file')) + mock.patch.object(util, "write_file") + ) mock_load = mocks.enter_context( - mock.patch.object(util, 'load_file', - return_value=ca_certs_content)) + mock.patch.object( + util, "load_file", return_value=ca_certs_content + ) + ) cc_ca_certs.add_ca_certs(conf, [cert]) - mock_write.assert_has_calls([ - mock.call(conf['ca_cert_full_path'], - cert, mode=0o644)]) - if conf['ca_cert_config'] is not None: - mock_write.assert_has_calls([ - mock.call(conf['ca_cert_config'], - expected, omode="wb")]) - mock_load.assert_called_once_with(conf['ca_cert_config']) + mock_write.assert_has_calls( + [mock.call(conf["ca_cert_full_path"], cert, mode=0o644)] + ) + if conf["ca_cert_config"] is not None: + mock_write.assert_has_calls( + [ + mock.call( + conf["ca_cert_config"], expected, omode="wb" + ) + ] + ) + mock_load.assert_called_once_with(conf["ca_cert_config"]) def test_single_cert_no_trailing_cr(self): """Test adding a single certificate to the trusted CAs @@ -234,24 +244,32 @@ class TestAddCaCerts(TestCase): with ExitStack() as mocks: mock_write = mocks.enter_context( - mock.patch.object(util, 'write_file')) + mock.patch.object(util, "write_file") + ) mock_load = mocks.enter_context( - mock.patch.object(util, 'load_file', - return_value=ca_certs_content)) + mock.patch.object( + util, "load_file", return_value=ca_certs_content + ) + ) cc_ca_certs.add_ca_certs(conf, [cert]) - mock_write.assert_has_calls([ - mock.call(conf['ca_cert_full_path'], - cert, mode=0o644)]) - if conf['ca_cert_config'] is not None: - mock_write.assert_has_calls([ - mock.call(conf['ca_cert_config'], - "%s\n%s\n" % (ca_certs_content, - conf['ca_cert_filename']), - omode="wb")]) - - mock_load.assert_called_once_with(conf['ca_cert_config']) + mock_write.assert_has_calls( + [mock.call(conf["ca_cert_full_path"], cert, mode=0o644)] + ) + if conf["ca_cert_config"] is not None: + mock_write.assert_has_calls( + [ + mock.call( + conf["ca_cert_config"], + "%s\n%s\n" + % (ca_certs_content, conf["ca_cert_filename"]), + omode="wb", + ) + ] + ) + + mock_load.assert_called_once_with(conf["ca_cert_config"]) def test_single_cert_to_empty_existing_ca_file(self): """Test adding a single certificate to the trusted CAs @@ -264,18 +282,23 @@ class TestAddCaCerts(TestCase): for distro_name in cc_ca_certs.distros: conf = cc_ca_certs._distro_ca_certs_configs(distro_name) - with mock.patch.object(util, 'write_file', - autospec=True) as m_write: + with mock.patch.object( + util, "write_file", autospec=True + ) as m_write: cc_ca_certs.add_ca_certs(conf, [cert]) - m_write.assert_has_calls([ - mock.call(conf['ca_cert_full_path'], - cert, mode=0o644)]) - if conf['ca_cert_config'] is not None: - m_write.assert_has_calls([ - mock.call(conf['ca_cert_config'], - expected, omode="wb")]) + m_write.assert_has_calls( + [mock.call(conf["ca_cert_full_path"], cert, mode=0o644)] + ) + if conf["ca_cert_config"] is not None: + m_write.assert_has_calls( + [ + mock.call( + conf["ca_cert_config"], expected, omode="wb" + ) + ] + ) def test_multiple_certs(self): """Test adding multiple certificates to the trusted CAs.""" @@ -290,45 +313,61 @@ class TestAddCaCerts(TestCase): with ExitStack() as mocks: mock_write = mocks.enter_context( - mock.patch.object(util, 'write_file')) + mock.patch.object(util, "write_file") + ) mock_load = mocks.enter_context( - mock.patch.object(util, 'load_file', - return_value=ca_certs_content)) + mock.patch.object( + util, "load_file", return_value=ca_certs_content + ) + ) cc_ca_certs.add_ca_certs(conf, certs) - mock_write.assert_has_calls([ - mock.call(conf['ca_cert_full_path'], - expected_cert_file, mode=0o644)]) - if conf['ca_cert_config'] is not None: - mock_write.assert_has_calls([ - mock.call(conf['ca_cert_config'], - "%s\n%s\n" % (ca_certs_content, - conf['ca_cert_filename']), - omode='wb')]) - - mock_load.assert_called_once_with(conf['ca_cert_config']) + mock_write.assert_has_calls( + [ + mock.call( + conf["ca_cert_full_path"], + expected_cert_file, + mode=0o644, + ) + ] + ) + if conf["ca_cert_config"] is not None: + mock_write.assert_has_calls( + [ + mock.call( + conf["ca_cert_config"], + "%s\n%s\n" + % (ca_certs_content, conf["ca_cert_filename"]), + omode="wb", + ) + ] + ) + + mock_load.assert_called_once_with(conf["ca_cert_config"]) class TestUpdateCaCerts(unittest.TestCase): def test_commands(self): for distro_name in cc_ca_certs.distros: conf = cc_ca_certs._distro_ca_certs_configs(distro_name) - with mock.patch.object(subp, 'subp') as mockobj: + with mock.patch.object(subp, "subp") as mockobj: cc_ca_certs.update_ca_certs(conf) mockobj.assert_called_once_with( - conf['ca_cert_update_cmd'], capture=False) + conf["ca_cert_update_cmd"], capture=False + ) class TestRemoveDefaultCaCerts(TestCase): - def setUp(self): super(TestRemoveDefaultCaCerts, self).setUp() tmpdir = tempfile.mkdtemp() self.addCleanup(shutil.rmtree, tmpdir) - self.paths = helpers.Paths({ - 'cloud_dir': tmpdir, - }) + self.paths = helpers.Paths( + { + "cloud_dir": tmpdir, + } + ) def test_commands(self): for distro_name in cc_ca_certs.distros: @@ -336,26 +375,35 @@ class TestRemoveDefaultCaCerts(TestCase): with ExitStack() as mocks: mock_delete = mocks.enter_context( - mock.patch.object(util, 'delete_dir_contents')) + mock.patch.object(util, "delete_dir_contents") + ) mock_write = mocks.enter_context( - mock.patch.object(util, 'write_file')) + mock.patch.object(util, "write_file") + ) mock_subp = mocks.enter_context( - mock.patch.object(subp, 'subp')) + mock.patch.object(subp, "subp") + ) cc_ca_certs.remove_default_ca_certs(distro_name, conf) - mock_delete.assert_has_calls([ - mock.call(conf['ca_cert_path']), - mock.call(conf['ca_cert_system_path'])]) + mock_delete.assert_has_calls( + [ + mock.call(conf["ca_cert_path"]), + mock.call(conf["ca_cert_system_path"]), + ] + ) - if conf['ca_cert_config'] is not None: + if conf["ca_cert_config"] is not None: mock_write.assert_called_once_with( - conf['ca_cert_config'], "", mode=0o644) + conf["ca_cert_config"], "", mode=0o644 + ) - if distro_name in ['debian', 'ubuntu']: + if distro_name in ["debian", "ubuntu"]: mock_subp.assert_called_once_with( - ('debconf-set-selections', '-'), - "ca-certificates \ -ca-certificates/trust_new_crts select no") + ("debconf-set-selections", "-"), + "ca-certificates ca-certificates/trust_new_crts" + " select no", + ) + # vi: ts=4 expandtab |