summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/unittests/test_handler_ca_certs.py178
-rw-r--r--tests/unittests/test_util.py199
2 files changed, 376 insertions, 1 deletions
diff --git a/tests/unittests/test_handler_ca_certs.py b/tests/unittests/test_handler_ca_certs.py
new file mode 100644
index 00000000..d6513b5b
--- /dev/null
+++ b/tests/unittests/test_handler_ca_certs.py
@@ -0,0 +1,178 @@
+from mocker import MockerTestCase
+
+from cloudinit.util import write_file, delete_dir_contents
+from cloudinit.CloudConfig.cc_ca_certs import (
+ handle, update_ca_certs, add_ca_certs, remove_default_ca_certs)
+from logging import getLogger
+
+
+class TestNoConfig(MockerTestCase):
+ def setUp(self):
+ super(TestNoConfig, self).setUp()
+ self.name = "ca-certs"
+ self.cloud_init = None
+ self.log = getLogger("TestNoConfig")
+ self.args = []
+
+ def test_no_config(self):
+ """
+ Test that nothing is done if no ca-certs configuration is provided.
+ """
+ config = {"unknown-key": "value"}
+
+ self.mocker.replace(write_file, passthrough=False)
+ self.mocker.replace(update_ca_certs, passthrough=False)
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+
+class TestConfig(MockerTestCase):
+ def setUp(self):
+ super(TestConfig, self).setUp()
+ self.name = "ca-certs"
+ self.cloud_init = None
+ self.log = getLogger("TestNoConfig")
+ self.args = []
+
+ # Mock out the functions that actually modify the system
+ self.mock_add = self.mocker.replace(add_ca_certs, passthrough=False)
+ self.mock_update = self.mocker.replace(update_ca_certs,
+ passthrough=False)
+ self.mock_remove = self.mocker.replace(remove_default_ca_certs,
+ passthrough=False)
+ # Order must be correct
+ self.mocker.order()
+
+ def test_no_trusted_list(self):
+ """
+ Test that no certificates are written if the 'trusted' key is not
+ present.
+ """
+ config = {"ca-certs": {}}
+
+ # No functions should be called
+ self.mock_update()
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+ def test_empty_trusted_list(self):
+ """Test that no certificate are written if 'trusted' list is empty"""
+ config = {"ca-certs": {"trusted": []}}
+
+ # No functions should be called
+ self.mock_update()
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+ def test_single_trusted(self):
+ """Test that a single cert gets passed to add_ca_certs"""
+ config = {"ca-certs": {"trusted": ["CERT1"]}}
+
+ self.mock_add(["CERT1"])
+ self.mock_update()
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+ def test_multiple_trusted(self):
+ """Test that multiple certs get passed to add_ca_certs"""
+ config = {"ca-certs": {"trusted": ["CERT1", "CERT2"]}}
+
+ self.mock_add(["CERT1", "CERT2"])
+ self.mock_update()
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+ def test_remove_default_ca_certs(self):
+ """Test remove_defaults works as expected"""
+ config = {"ca-certs": {"remove-defaults": True}}
+
+ self.mock_remove()
+ self.mock_update()
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+ def test_no_remove_defaults_if_false(self):
+ """Test remove_defaults is not called when config value is False"""
+ config = {"ca-certs": {"remove-defaults": False}}
+
+ self.mock_update()
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+ def test_correct_order_for_remove_then_add(self):
+ """Test remove_defaults is not called when config value is False"""
+ config = {"ca-certs": {"remove-defaults": True, "trusted": ["CERT1"]}}
+
+ self.mock_remove()
+ self.mock_add(["CERT1"])
+ self.mock_update()
+ self.mocker.replay()
+
+ handle(self.name, config, self.cloud_init, self.log, self.args)
+
+
+class TestAddCaCerts(MockerTestCase):
+ def test_no_certs_in_list(self):
+ """Test that no certificate are written if not provided."""
+ self.mocker.replace(write_file, passthrough=False)
+ self.mocker.replay()
+
+ add_ca_certs([])
+
+ def test_single_cert(self):
+ """Test adding a single certificate to the trusted CAs"""
+ cert = "CERT1\nLINE2\nLINE3"
+
+ mock_write = self.mocker.replace(write_file, passthrough=False)
+ mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
+ cert, mode=0644)
+ mock_write("/etc/ca-certificates.conf",
+ "\ncloud-init-ca-certs.crt", omode="a")
+ self.mocker.replay()
+
+ add_ca_certs([cert])
+
+ def test_multiple_certs(self):
+ """Test adding multiple certificates to the trusted CAs"""
+ certs = ["CERT1\nLINE2\nLINE3", "CERT2\nLINE2\nLINE3"]
+ expected_cert_file = "\n".join(certs)
+
+ mock_write = self.mocker.replace(write_file, passthrough=False)
+ mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
+ expected_cert_file, mode=0644)
+ mock_write("/etc/ca-certificates.conf",
+ "\ncloud-init-ca-certs.crt", omode="a")
+ self.mocker.replay()
+
+ add_ca_certs(certs)
+
+
+class TestUpdateCaCerts(MockerTestCase):
+ def test_commands(self):
+ mock_check_call = self.mocker.replace("subprocess.check_call",
+ passthrough=False)
+ mock_check_call(["update-ca-certificates"])
+ self.mocker.replay()
+
+ update_ca_certs()
+
+
+class TestRemoveDefaultCaCerts(MockerTestCase):
+ def test_commands(self):
+ mock_delete_dir_contents = self.mocker.replace(delete_dir_contents,
+ passthrough=False)
+ mock_write = self.mocker.replace(write_file, passthrough=False)
+
+ mock_delete_dir_contents("/usr/share/ca-certificates/")
+ mock_delete_dir_contents("/etc/ssl/certs/")
+ mock_write("/etc/ca-certificates.conf", "", mode=0644)
+ self.mocker.replay()
+
+ remove_default_ca_certs()
diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 34a073d9..d8da8bc9 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -1,15 +1,24 @@
from unittest import TestCase
+from mocker import MockerTestCase
+from tempfile import mkdtemp
+from shutil import rmtree
+import os
+import stat
+
+from cloudinit.util import (mergedict, get_cfg_option_list_or_str, write_file,
+ delete_dir_contents)
-from cloudinit.util import mergedict
class TestMergeDict(TestCase):
def test_simple_merge(self):
+ """Test simple non-conflict merge."""
source = {"key1": "value1"}
candidate = {"key2": "value2"}
result = mergedict(source, candidate)
self.assertEqual({"key1": "value1", "key2": "value2"}, result)
def test_nested_merge(self):
+ """Test nested merge."""
source = {"key1": {"key1.1": "value1.1"}}
candidate = {"key1": {"key1.2": "value1.2"}}
result = mergedict(source, candidate)
@@ -17,37 +26,225 @@ class TestMergeDict(TestCase):
{"key1": {"key1.1": "value1.1", "key1.2": "value1.2"}}, result)
def test_merge_does_not_override(self):
+ """Test that candidate doesn't override source."""
source = {"key1": "value1", "key2": "value2"}
candidate = {"key2": "value2", "key2": "NEW VALUE"}
result = mergedict(source, candidate)
self.assertEqual(source, result)
def test_empty_candidate(self):
+ """Test empty candidate doesn't change source."""
source = {"key": "value"}
candidate = {}
result = mergedict(source, candidate)
self.assertEqual(source, result)
def test_empty_source(self):
+ """Test empty source is replaced by candidate."""
source = {}
candidate = {"key": "value"}
result = mergedict(source, candidate)
self.assertEqual(candidate, result)
def test_non_dict_candidate(self):
+ """Test non-dict candidate is discarded."""
source = {"key": "value"}
candidate = "not a dict"
result = mergedict(source, candidate)
self.assertEqual(source, result)
def test_non_dict_source(self):
+ """Test non-dict source is not modified with a dict candidate."""
source = "not a dict"
candidate = {"key": "value"}
result = mergedict(source, candidate)
self.assertEqual(source, result)
def test_neither_dict(self):
+ """Test if neither candidate or source is dict source wins."""
source = "source"
candidate = "candidate"
result = mergedict(source, candidate)
self.assertEqual(source, result)
+
+
+class TestGetCfgOptionListOrStr(TestCase):
+ def test_not_found_no_default(self):
+ """None is returned if key is not found and no default given."""
+ config = {}
+ result = get_cfg_option_list_or_str(config, "key")
+ self.assertIsNone(result)
+
+ def test_not_found_with_default(self):
+ """Default is returned if key is not found."""
+ config = {}
+ result = get_cfg_option_list_or_str(config, "key", default=["DEFAULT"])
+ self.assertEqual(["DEFAULT"], result)
+
+ def test_found_with_default(self):
+ """Default is not returned if key is found."""
+ config = {"key": ["value1"]}
+ result = get_cfg_option_list_or_str(config, "key", default=["DEFAULT"])
+ self.assertEqual(["value1"], result)
+
+ def test_found_convert_to_list(self):
+ """Single string is converted to one element list."""
+ config = {"key": "value1"}
+ result = get_cfg_option_list_or_str(config, "key")
+ self.assertEqual(["value1"], result)
+
+ def test_value_is_none(self):
+ """If value is None empty list is returned."""
+ config = {"key": None}
+ result = get_cfg_option_list_or_str(config, "key")
+ self.assertEqual([], result)
+
+
+class TestWriteFile(MockerTestCase):
+ def setUp(self):
+ super(TestWriteFile, self).setUp()
+ # Make a temp directoy for tests to use.
+ self.tmp = mkdtemp(prefix="unittest_")
+
+ def tearDown(self):
+ super(TestWriteFile, self).tearDown()
+ # Clean up temp directory
+ rmtree(self.tmp)
+
+ def test_basic_usage(self):
+ """Verify basic usage with default args."""
+ path = os.path.join(self.tmp, "NewFile.txt")
+ contents = "Hey there"
+
+ write_file(path, contents)
+
+ self.assertTrue(os.path.exists(path))
+ self.assertTrue(os.path.isfile(path))
+ with open(path) as f:
+ create_contents = f.read()
+ self.assertEqual(contents, create_contents)
+ file_stat = os.stat(path)
+ self.assertEqual(0644, stat.S_IMODE(file_stat.st_mode))
+
+ def test_dir_is_created_if_required(self):
+ """Verifiy that directories are created is required."""
+ dirname = os.path.join(self.tmp, "subdir")
+ path = os.path.join(dirname, "NewFile.txt")
+ contents = "Hey there"
+
+ write_file(path, contents)
+
+ self.assertTrue(os.path.isdir(dirname))
+ self.assertTrue(os.path.isfile(path))
+
+ def test_custom_mode(self):
+ """Verify custom mode works properly."""
+ path = os.path.join(self.tmp, "NewFile.txt")
+ contents = "Hey there"
+
+ write_file(path, contents, mode=0666)
+
+ self.assertTrue(os.path.exists(path))
+ self.assertTrue(os.path.isfile(path))
+ file_stat = os.stat(path)
+ self.assertEqual(0666, stat.S_IMODE(file_stat.st_mode))
+
+ def test_custom_omode(self):
+ """Verify custom omode works properly."""
+ path = os.path.join(self.tmp, "NewFile.txt")
+ contents = "Hey there"
+
+ # Create file first with basic content
+ with open(path, "wb") as f:
+ f.write("LINE1\n")
+ write_file(path, contents, omode="a")
+
+ self.assertTrue(os.path.exists(path))
+ self.assertTrue(os.path.isfile(path))
+ with open(path) as f:
+ create_contents = f.read()
+ self.assertEqual("LINE1\nHey there", create_contents)
+
+ def test_restorecon_if_possible_is_called(self):
+ """Make sure the restorecon_if_possible is called correctly."""
+ path = os.path.join(self.tmp, "NewFile.txt")
+ contents = "Hey there"
+
+ # Mock out the restorecon_if_possible call to test if it's called.
+ mock_restorecon = self.mocker.replace(
+ "cloudinit.util.restorecon_if_possible", passthrough=False)
+ mock_restorecon(path)
+ self.mocker.replay()
+
+ write_file(path, contents)
+
+
+class TestDeleteDirContents(TestCase):
+ def setUp(self):
+ super(TestDeleteDirContents, self).setUp()
+ # Make a temp directoy for tests to use.
+ self.tmp = mkdtemp(prefix="unittest_")
+
+ def tearDown(self):
+ super(TestDeleteDirContents, self).tearDown()
+ # Clean up temp directory
+ rmtree(self.tmp)
+
+ def assertDirEmpty(self, dirname):
+ self.assertEqual([], os.listdir(dirname))
+
+ def test_does_not_delete_dir(self):
+ """Ensure directory itself is not deleted."""
+ delete_dir_contents(self.tmp)
+
+ self.assertTrue(os.path.isdir(self.tmp))
+ self.assertDirEmpty(self.tmp)
+
+ def test_deletes_files(self):
+ """Single file should be deleted."""
+ with open(os.path.join(self.tmp, "new_file.txt"), "wb") as f:
+ f.write("DELETE ME")
+
+ delete_dir_contents(self.tmp)
+
+ self.assertDirEmpty(self.tmp)
+
+ def test_deletes_empty_dirs(self):
+ """Empty directories should be deleted."""
+ os.mkdir(os.path.join(self.tmp, "new_dir"))
+
+ delete_dir_contents(self.tmp)
+
+ self.assertDirEmpty(self.tmp)
+
+ def test_deletes_nested_dirs(self):
+ """Nested directories should be deleted."""
+ os.mkdir(os.path.join(self.tmp, "new_dir"))
+ os.mkdir(os.path.join(self.tmp, "new_dir", "new_subdir"))
+
+ delete_dir_contents(self.tmp)
+
+ self.assertDirEmpty(self.tmp)
+
+ def test_deletes_non_empty_dirs(self):
+ """Non-empty directories should be deleted."""
+ os.mkdir(os.path.join(self.tmp, "new_dir"))
+ f_name = os.path.join(self.tmp, "new_dir", "new_file.txt")
+ with open(f_name, "wb") as f:
+ f.write("DELETE ME")
+
+ delete_dir_contents(self.tmp)
+
+ self.assertDirEmpty(self.tmp)
+
+ def test_deletes_symlinks(self):
+ """Symlinks should be deleted."""
+ file_name = os.path.join(self.tmp, "new_file.txt")
+ link_name = os.path.join(self.tmp, "new_file_link.txt")
+ with open(file_name, "wb") as f:
+ f.write("DELETE ME")
+ os.symlink(file_name, link_name)
+
+ delete_dir_contents(self.tmp)
+
+ self.assertDirEmpty(self.tmp)
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-14 12:43:38 +0000