diff --git a/cloudinit/sources/helpers/openstack.py b/cloudinit/sources/helpers/openstack.py index 031ac8c9360..094c889caef 100644 --- a/cloudinit/sources/helpers/openstack.py +++ b/cloudinit/sources/helpers/openstack.py @@ -763,7 +763,7 @@ def convert_net_json(network_json=None, known_macs=None): cfg["type"] = "infiniband" for service in services: - cfg = service + cfg = copy.deepcopy(service) cfg.update({"type": "nameserver"}) config.append(cfg) diff --git a/tests/unittests/cmd/devel/test_net_convert.py b/tests/unittests/cmd/devel/test_net_convert.py index fb72963f842..be2fcdd6543 100644 --- a/tests/unittests/cmd/devel/test_net_convert.py +++ b/tests/unittests/cmd/devel/test_net_convert.py @@ -90,6 +90,18 @@ """ +@pytest.fixture +def mock_setup_logging(): + """Mock setup_basic_logging to avoid changing log level. + + net_convert.handle_args() can call setup_basic_logging() with a + WARNING level, which would be a side-effect for future tests. + It's behavior isn't checked in these tests, so mock it out. + """ + with mock.patch(f"{M_PATH}log.setup_basic_logging"): + yield + + class TestNetConvert: missing_required_args = itertools.combinations( @@ -155,7 +167,13 @@ def test_argparse_error_on_missing_args(self, cmdargs, capsys, tmpdir): ), ) def test_convert_output_kind_artifacts( - self, output_kind, outfile_content, debug, capsys, tmpdir + self, + output_kind, + outfile_content, + debug, + capsys, + tmpdir, + mock_setup_logging, ): """Assert proper output-kind artifacts are written.""" network_data = tmpdir.join("network_data") @@ -186,7 +204,9 @@ def test_convert_output_kind_artifacts( ] == chown.call_args_list @pytest.mark.parametrize("debug", (False, True)) - def test_convert_netplan_passthrough(self, debug, tmpdir): + def test_convert_netplan_passthrough( + self, debug, tmpdir, mock_setup_logging + ): """Assert that if the network config's version is 2 and the renderer is Netplan, then the config is passed through as-is. """ diff --git a/tests/unittests/cmd/test_main.py b/tests/unittests/cmd/test_main.py index 2a9e063fe4d..7f580203e82 100644 --- a/tests/unittests/cmd/test_main.py +++ b/tests/unittests/cmd/test_main.py @@ -54,6 +54,18 @@ def setUp(self): self.patchUtils(self.new_root) self.stderr = StringIO() self.patchStdoutAndStderr(stderr=self.stderr) + # Every cc_ module calls get_meta_doc on import. + # This call will fail if filesystem redirection mocks are in place + # and the module hasn't already been imported which can depend + # on test ordering. + self.m_doc = mock.patch( + "cloudinit.config.schema.get_meta_doc", return_value={} + ) + self.m_doc.start() + + def tearDown(self): + self.m_doc.stop() + super().tearDown() def test_main_init_run_net_runs_modules(self): """Modules like write_files are run in 'net' mode.""" diff --git a/tests/unittests/config/test_cc_apt_configure.py b/tests/unittests/config/test_cc_apt_configure.py index a75acd3cdde..0b0a4e8515a 100644 --- a/tests/unittests/config/test_cc_apt_configure.py +++ b/tests/unittests/config/test_cc_apt_configure.py @@ -298,6 +298,7 @@ class TestAptConfigure: ), ) @mock.patch(M_PATH + "get_apt_cfg") + @mock.patch.object(features, "APT_DEB822_SOURCE_LIST_FILE", True) def test_remove_source( self, m_get_apt_cfg, @@ -312,7 +313,6 @@ def test_remove_source( "sourceparts": f"{tmpdir}/etc/apt/sources.list.d/", } cloud = get_cloud(distro_name) - features.APT_DEB822_SOURCE_LIST_FILE = True sources_file = tmpdir.join("/etc/apt/sources.list") deb822_sources_file = tmpdir.join( f"/etc/apt/sources.list.d/{distro_name}.sources" diff --git a/tests/unittests/config/test_cc_ca_certs.py b/tests/unittests/config/test_cc_ca_certs.py index 905bbeb12cb..7013a95dbe8 100644 --- a/tests/unittests/config/test_cc_ca_certs.py +++ b/tests/unittests/config/test_cc_ca_certs.py @@ -380,6 +380,7 @@ def test_non_existent_cert_cfg(self): cc_ca_certs.disable_default_ca_certs(distro_name, conf) +@pytest.mark.usefixtures("clear_deprecation_log") class TestCACertsSchema: """Directly test schema rather than through handle.""" diff --git a/tests/unittests/config/test_cc_set_passwords.py b/tests/unittests/config/test_cc_set_passwords.py index ef34a8c6052..d37faedd4c6 100644 --- a/tests/unittests/config/test_cc_set_passwords.py +++ b/tests/unittests/config/test_cc_set_passwords.py @@ -1,5 +1,6 @@ # This file is part of cloud-init. See LICENSE file for license information. +import copy import logging from unittest import mock @@ -508,6 +509,7 @@ def test_chpasswd_parity(self, list_def, users_def): class TestExpire: @pytest.mark.parametrize("cfg", expire_cases) def test_expire(self, cfg, mocker, caplog): + cfg = copy.deepcopy(cfg) cloud = get_cloud() mocker.patch(f"{MODPATH}subp.subp") mocker.patch.object(cloud.distro, "chpasswd") @@ -533,7 +535,9 @@ def test_expire(self, cfg, mocker, caplog): def test_expire_old_behavior(self, cfg, mocker, caplog): # Previously expire didn't apply to hashed passwords. # Ensure we can preserve that case on older releases - features.EXPIRE_APPLIES_TO_HASHED_USERS = False + mocker.patch.object(features, "EXPIRE_APPLIES_TO_HASHED_USERS", False) + + cfg = copy.deepcopy(cfg) cloud = get_cloud() mocker.patch(f"{MODPATH}subp.subp") mocker.patch.object(cloud.distro, "chpasswd") diff --git a/tests/unittests/config/test_cc_ssh.py b/tests/unittests/config/test_cc_ssh.py index 544e0b67b8a..102519ebf58 100644 --- a/tests/unittests/config/test_cc_ssh.py +++ b/tests/unittests/config/test_cc_ssh.py @@ -38,8 +38,10 @@ def publish_hostkey_test_setup(tmpdir): with open(filepath, "w") as f: f.write(" ".join(test_hostkeys[key_type])) - cc_ssh.KEY_FILE_TPL = os.path.join(hostkey_tmpdir, "ssh_host_%s_key") - yield test_hostkeys, test_hostkey_files + with mock.patch.object( + cc_ssh, "KEY_FILE_TPL", os.path.join(hostkey_tmpdir, "ssh_host_%s_key") + ): + yield test_hostkeys, test_hostkey_files def _replace_options(user: Optional[str] = None) -> str: @@ -255,6 +257,7 @@ def test_handle_default_root( @mock.patch(MODPATH + "ug_util.normalize_users_groups") @mock.patch(MODPATH + "os.path.exists") @mock.patch(MODPATH + "util.fips_enabled", return_value=False) + @mock.patch.object(cc_ssh, "PUBLISH_HOST_KEYS", True) def test_handle_publish_hostkeys( self, m_fips, @@ -268,7 +271,6 @@ def test_handle_publish_hostkeys( ): """Test handle with various configs for ssh_publish_hostkeys.""" test_hostkeys, test_hostkey_files = publish_hostkey_test_setup - cc_ssh.PUBLISH_HOST_KEYS = True keys = ["key1"] user = "clouduser" # Return no matching keys for first glob, test keys for second. @@ -282,7 +284,6 @@ def test_handle_publish_hostkeys( m_path_exists.return_value = True m_nug.return_value = ({user: {"default": user}}, {}) cloud = get_cloud(distro="ubuntu", metadata={"public-keys": keys}) - cloud.datasource.publish_host_keys = mock.Mock() expected_calls = [] if expected_key_types is not None: @@ -294,10 +295,15 @@ def test_handle_publish_hostkeys( ] ) ] - cc_ssh.handle("name", cfg, cloud, None) - assert ( - expected_calls == cloud.datasource.publish_host_keys.call_args_list - ) + + with mock.patch.object( + cloud.datasource, "publish_host_keys", mock.Mock() + ): + cc_ssh.handle("name", cfg, cloud, None) + assert ( + expected_calls + == cloud.datasource.publish_host_keys.call_args_list + ) @pytest.mark.parametrize( "ssh_keys_group_exists,sshd_version,expected_private_permissions", diff --git a/tests/unittests/conftest.py b/tests/unittests/conftest.py index 5a46646f241..26d5bc784af 100644 --- a/tests/unittests/conftest.py +++ b/tests/unittests/conftest.py @@ -139,12 +139,21 @@ def dhclient_exists(): log.configure_root_logger() -@pytest.fixture(autouse=True) -def disable_root_logger_setup(request): - with mock.patch("cloudinit.cmd.main.configure_root_logger", autospec=True): +@pytest.fixture(autouse=True, scope="session") +def disable_root_logger_setup(): + with mock.patch("cloudinit.log.configure_root_logger", autospec=True): yield +@pytest.fixture +def clear_deprecation_log(): + """Clear any deprecation warnings before and after running tests.""" + # Since deprecations are de-duped, the existance (or non-existance) of + # a deprecation warning in a previous test can cause the next test to + # fail. + util.deprecate._log = set() + + PYTEST_VERSION_TUPLE = tuple(map(int, pytest.__version__.split("."))) if PYTEST_VERSION_TUPLE < (3, 9, 0): diff --git a/tests/unittests/distros/test_netconfig.py b/tests/unittests/distros/test_netconfig.py index e3e2c8fe159..b8a22a89aff 100644 --- a/tests/unittests/distros/test_netconfig.py +++ b/tests/unittests/distros/test_netconfig.py @@ -303,7 +303,7 @@ def setUp(self): def _get_distro(self, dname, renderers=None, activators=None): cls = distros.fetch(dname) - cfg = settings.CFG_BUILTIN + cfg = copy.deepcopy(settings.CFG_BUILTIN) cfg["system_info"]["distro"] = dname system_info_network_cfg = {} if renderers: diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py index 8c185e707bd..70ca9dff912 100644 --- a/tests/unittests/helpers.py +++ b/tests/unittests/helpers.py @@ -162,6 +162,7 @@ def setUp(self): self.old_handlers = self.logger.handlers self.logger.handlers = [handler] self.old_level = logging.root.level + self.logger.level = logging.DEBUG if self.allowed_subp is True: subp.subp = _real_subp else: diff --git a/tests/unittests/runs/test_simple_run.py b/tests/unittests/runs/test_simple_run.py index eec2db00bdf..7cb5a28e71f 100644 --- a/tests/unittests/runs/test_simple_run.py +++ b/tests/unittests/runs/test_simple_run.py @@ -2,6 +2,7 @@ import copy import os +from unittest import mock from cloudinit import atomic_helper, safeyaml, stages, util from cloudinit.config.modules import Modules @@ -45,6 +46,15 @@ def setUp(self): self.patchOS(self.new_root) self.patchUtils(self.new_root) + self.m_doc = mock.patch( + "cloudinit.config.schema.get_meta_doc", return_value={} + ) + self.m_doc.start() + + def tearDown(self): + self.m_doc.stop() + super().tearDown() + def test_none_ds_populates_var_lib_cloud(self): """Init and run_section default behavior creates appropriate dirs.""" # Now start verifying whats created diff --git a/tests/unittests/sources/test_lxd.py b/tests/unittests/sources/test_lxd.py index efc24883893..7ec67b4577d 100644 --- a/tests/unittests/sources/test_lxd.py +++ b/tests/unittests/sources/test_lxd.py @@ -333,13 +333,13 @@ def test_network_config_when_unset(self, lxd_ds): assert NETWORK_V1 == lxd_ds.network_config assert LXD_V1_METADATA == lxd_ds._crawled_metadata + @mock.patch.object(lxd, "generate_network_config", return_value=NETWORK_V1) def test_network_config_crawled_metadata_no_network_config( - self, lxd_ds_no_network_config + self, m_generate, lxd_ds_no_network_config ): """network_config is correctly computed when _network_config is unset and _crawled_metadata does not contain network_config. """ - lxd.generate_network_config = mock.Mock(return_value=NETWORK_V1) assert UNSET == lxd_ds_no_network_config._crawled_metadata assert UNSET == lxd_ds_no_network_config._network_config assert None is lxd_ds_no_network_config.userdata_raw @@ -349,7 +349,7 @@ def test_network_config_crawled_metadata_no_network_config( LXD_V1_METADATA_NO_NETWORK_CONFIG == lxd_ds_no_network_config._crawled_metadata ) - assert 1 == lxd.generate_network_config.call_count + assert 1 == m_generate.call_count class TestIsPlatformViable: diff --git a/tests/unittests/sources/test_vultr.py b/tests/unittests/sources/test_vultr.py index 7fa02b1c9bb..117fdab0f60 100644 --- a/tests/unittests/sources/test_vultr.py +++ b/tests/unittests/sources/test_vultr.py @@ -5,7 +5,7 @@ # Vultr Metadata API: # https://www.vultr.com/metadata/ -import json +import copy from cloudinit import helpers, settings from cloudinit.net.dhcp import NoDHCPLeaseError @@ -13,6 +13,20 @@ from cloudinit.sources.helpers import vultr from tests.unittests.helpers import CiTestCase, mock +VENDOR_DATA = """\ +#cloud-config +package_upgrade: true +disable_root: 0 +ssh_pwauth: 1 +chpasswd: + expire: false + list: + - root:$6$SxXx...k2mJNIzZB5vMCDBlYT1 +system_info: + default_user: + name: root +""" + # Vultr metadata test data VULTR_V1_1 = { "bgp": { @@ -58,18 +72,7 @@ "startup-script": "echo No configured startup script", "raid1-script": "", "user-data": [], - "vendor-data": [ - { - "package_upgrade": "true", - "disable_root": 0, - "ssh_pwauth": 1, - "chpasswd": { - "expire": False, - "list": ["root:$6$S2Smuj.../VqxmIR9Urw0jPZ88i4yvB/"], - }, - "system_info": {"default_user": {"name": "root"}}, - } - ], + "vendor-data": VENDOR_DATA, } VULTR_V1_2 = { @@ -130,22 +133,9 @@ "user-defined": [], "startup-script": "echo No configured startup script", "user-data": [], - "vendor-data": [ - { - "package_upgrade": "true", - "disable_root": 0, - "ssh_pwauth": 1, - "chpasswd": { - "expire": False, - "list": ["root:$6$SxXx...k2mJNIzZB5vMCDBlYT1"], - }, - "system_info": {"default_user": {"name": "root"}}, - } - ], + "vendor-data": VENDOR_DATA, } -VULTR_V1_3 = None - SSH_KEYS_1 = ["ssh-rsa AAAAB3NzaC1y...IQQhv5PAOKaIl+mM3c= test3@key"] CLOUD_INTERFACES = { @@ -190,20 +180,6 @@ FILTERED_INTERFACES = ["eth1", "eth2", "eth0"] -# Expected generated objects - -# Expected config -EXPECTED_VULTR_CONFIG = { - "package_upgrade": "true", - "disable_root": 0, - "ssh_pwauth": 1, - "chpasswd": { - "expire": False, - "list": ["root:$6$SxXx...k2mJNIzZB5vMCDBlYT1"], - }, - "system_info": {"default_user": {"name": "root"}}, -} - # Expected network config object from generator EXPECTED_VULTR_NETWORK_1 = { "version": 1, @@ -271,28 +247,9 @@ } -FINAL_INTERFACE_USED = "" - - class TestDataSourceVultr(CiTestCase): def setUp(self): - global VULTR_V1_3 super(TestDataSourceVultr, self).setUp() - - # Create v3 - VULTR_V1_3 = VULTR_V1_2.copy() - VULTR_V1_3["cloud_interfaces"] = CLOUD_INTERFACES.copy() - VULTR_V1_3["interfaces"] = [] - - # Stored as a dict to make it easier to maintain - raw1 = json.dumps(VULTR_V1_1["vendor-data"][0]) - raw2 = json.dumps(VULTR_V1_2["vendor-data"][0]) - - # Make expected format - VULTR_V1_1["vendor-data"] = [raw1] - VULTR_V1_2["vendor-data"] = [raw2] - VULTR_V1_3["vendor-data"] = [raw2] - self.tmp = self.tmp_dir() # Test the datasource itself @@ -330,8 +287,8 @@ def test_datasource(self, mock_getmeta, mock_isvultr, mock_netmap): # Test vendor config self.assertEqual( - EXPECTED_VULTR_CONFIG, - json.loads(vendordata[0].replace("#cloud-config", "")), + VENDOR_DATA, + vendordata, ) self.maxDiff = orig_val @@ -339,6 +296,15 @@ def test_datasource(self, mock_getmeta, mock_isvultr, mock_netmap): # Test network config generation self.assertEqual(EXPECTED_VULTR_NETWORK_2, source.network_config) + def _get_metadata(self): + # Create v1_3 + vultr_v1_3 = VULTR_V1_2.copy() + vultr_v1_3["cloud_interfaces"] = CLOUD_INTERFACES.copy() + vultr_v1_3["interfaces"] = [] + vultr_v1_3["vendor-data"] = copy.deepcopy(VULTR_V1_2["vendor-data"]) + + return vultr_v1_3 + # Test the datasource with new network config type @mock.patch("cloudinit.net.get_interfaces_by_mac") @mock.patch("cloudinit.sources.helpers.vultr.is_vultr") @@ -346,7 +312,7 @@ def test_datasource(self, mock_getmeta, mock_isvultr, mock_netmap): def test_datasource_cloud_interfaces( self, mock_getmeta, mock_isvultr, mock_netmap ): - mock_getmeta.return_value = VULTR_V1_3 + mock_getmeta.return_value = self._get_metadata() mock_isvultr.return_value = True mock_netmap.return_value = INTERFACE_MAP @@ -375,7 +341,7 @@ def test_network_config(self, mock_netmap): @mock.patch("cloudinit.net.get_interfaces_by_mac") def test_private_network_config(self, mock_netmap): mock_netmap.return_value = INTERFACE_MAP - interf = VULTR_V1_2["interfaces"].copy() + interf = copy.deepcopy(VULTR_V1_2["interfaces"]) # Test configuring self.assertEqual( @@ -384,27 +350,10 @@ def test_private_network_config(self, mock_netmap): # Test unconfigured interf[1]["unconfigured"] = True - expected = EXPECTED_VULTR_NETWORK_2.copy() + expected = copy.deepcopy(EXPECTED_VULTR_NETWORK_2) expected["config"].pop(2) self.assertEqual(expected, vultr.generate_network_config(interf)) - # Override ephemeral for proper unit testing - def ephemeral_init( - self, distro, iface="", connectivity_url_data=None, tmp_dir=None - ): - global FINAL_INTERFACE_USED - FINAL_INTERFACE_USED = iface - if iface == "eth0": - return - raise NoDHCPLeaseError("Generic for testing") - - # Override ephemeral for proper unit testing - def ephemeral_init_always( - self, iface="", connectivity_url_data=None, tmp_dir=None - ): - global FINAL_INTERFACE_USED - FINAL_INTERFACE_USED = iface - # Override ephemeral for proper unit testing def override_enter(self): return @@ -415,7 +364,8 @@ def override_exit(self, excp_type, excp_value, excp_traceback): # Test interface seeking to ensure we are able to find the correct one @mock.patch( - "cloudinit.net.ephemeral.EphemeralDHCPv4.__init__", ephemeral_init + "cloudinit.net.ephemeral.EphemeralDHCPv4.__init__", + side_effect=(NoDHCPLeaseError("Generic for testing"), None), ) @mock.patch( "cloudinit.net.ephemeral.EphemeralDHCPv4.__enter__", override_enter @@ -431,6 +381,7 @@ def test_interface_seek( mock_interface_list, mock_read_metadata, mock_isvultr, + mock_eph_init, ): mock_read_metadata.return_value = {} mock_isvultr.return_value = True @@ -447,36 +398,4 @@ def test_interface_seek( except Exception: pass - self.assertEqual(FINAL_INTERFACE_USED, INTERFACES[3]) - - # Test route checking sucessful DHCPs - @mock.patch( - "cloudinit.net.ephemeral.EphemeralDHCPv4.__init__", - ephemeral_init_always, - ) - @mock.patch( - "cloudinit.net.ephemeral.EphemeralDHCPv4.__enter__", override_enter - ) - @mock.patch( - "cloudinit.net.ephemeral.EphemeralDHCPv4.__exit__", override_exit - ) - @mock.patch("cloudinit.sources.helpers.vultr.get_interface_list") - @mock.patch("cloudinit.sources.helpers.vultr.is_vultr") - @mock.patch("cloudinit.sources.helpers.vultr.read_metadata") - def test_interface_seek_route_check( - self, mock_read_metadata, mock_isvultr, mock_interface_list - ): - mock_read_metadata.return_value = {} - mock_interface_list.return_value = FILTERED_INTERFACES - mock_isvultr.return_value = True - - source = DataSourceVultr.DataSourceVultr( - settings.CFG_BUILTIN, None, helpers.Paths({"run_dir": self.tmp}) - ) - - try: - source._get_data() - except Exception: - pass - - self.assertEqual(FINAL_INTERFACE_USED, INTERFACES[3]) + assert mock_eph_init.call_args[1]["iface"] == FILTERED_INTERFACES[1] diff --git a/tests/unittests/test_cli.py b/tests/unittests/test_cli.py index 575b107c758..2681cddde7e 100644 --- a/tests/unittests/test_cli.py +++ b/tests/unittests/test_cli.py @@ -25,6 +25,14 @@ def mock_get_user_data_file(mocker, tmpdir): ) +@pytest.fixture(autouse=True, scope="module") +def disable_setup_logging(): + # setup_basic_logging can change the logging level to WARNING, so + # ensure it is always mocked + with mock.patch(f"{M_PATH}log.setup_basic_logging", autospec=True): + yield + + class TestCLI: def _call_main(self, sysv_args=None): if not sysv_args: @@ -193,7 +201,7 @@ def test_all_subcommands_represented_in_help(self, capsys): ), ), ) - @mock.patch("cloudinit.cmd.main.setup_basic_logging") + @mock.patch("cloudinit.cmd.main.log.setup_basic_logging") def test_subcommands_log_to_stderr_via_setup_basic_logging( self, setup_basic_logging, subcommand, log_to_stderr, mocks ):