diff --git a/cloudinit/net/network_state.py b/cloudinit/net/network_state.py index 4e49a41a01d..3c7ee5a3f7a 100644 --- a/cloudinit/net/network_state.py +++ b/cloudinit/net/network_state.py @@ -1059,7 +1059,8 @@ def parse_net_config_data(net_config, skip_broken=True) -> NetworkState: if not state: raise RuntimeError( "No valid network_state object created from network config. " - "Did you specify the correct version?" + "Did you specify the correct version? Network config:\n" + f"{net_config}" ) return state diff --git a/cloudinit/sources/DataSourceLXD.py b/cloudinit/sources/DataSourceLXD.py index 04d393456d0..640348f4984 100644 --- a/cloudinit/sources/DataSourceLXD.py +++ b/cloudinit/sources/DataSourceLXD.py @@ -109,34 +109,6 @@ def get_connection(self, url, proxies=None): return SocketConnectionPool(LXD_SOCKET_PATH) -def _maybe_remove_top_network(cfg): - """If network-config contains top level 'network' key, then remove it. - - Some providers of network configuration may provide a top level - 'network' key (LP: #1798117) even though it is not necessary. - - Be friendly and remove it if it really seems so. - - Return the original value if no change or the updated value if changed.""" - if "network" not in cfg: - return cfg - network_val = cfg["network"] - bmsg = "Top level network key in network-config %s: %s" - if not isinstance(network_val, dict): - LOG.debug(bmsg, "was not a dict", cfg) - return cfg - if len(list(cfg.keys())) != 1: - LOG.debug(bmsg, "had multiple top level keys", cfg) - return cfg - if network_val.get("config") == "disabled": - LOG.debug(bmsg, "was config/disabled", cfg) - elif not all(("config" in network_val, "version" in network_val)): - LOG.debug(bmsg, "but missing 'config' or 'version'", cfg) - return cfg - LOG.debug(bmsg, "fixed by removing shifting network.", cfg) - return network_val - - def _raw_instance_data_to_dict(metadata_type: str, metadata_value) -> dict: """Convert raw instance data from str, bytes, YAML to dict @@ -211,10 +183,8 @@ def _get_data(self) -> bool: if "user-data" in self._crawled_metadata: self.userdata_raw = self._crawled_metadata["user-data"] if "network-config" in self._crawled_metadata: - self._network_config = _maybe_remove_top_network( - _raw_instance_data_to_dict( - "network-config", self._crawled_metadata["network-config"] - ) + self._network_config = _raw_instance_data_to_dict( + "network-config", self._crawled_metadata["network-config"] ) if "vendor-data" in self._crawled_metadata: self.vendordata_raw = self._crawled_metadata["vendor-data"] diff --git a/cloudinit/sources/DataSourceNoCloud.py b/cloudinit/sources/DataSourceNoCloud.py index 56559630bcb..fba6aaae8c5 100644 --- a/cloudinit/sources/DataSourceNoCloud.py +++ b/cloudinit/sources/DataSourceNoCloud.py @@ -331,35 +331,6 @@ def parse_cmdline_data(ds_id, fill, cmdline=None): return True -def _maybe_remove_top_network(cfg): - """If network-config contains top level 'network' key, then remove it. - - Some providers of network configuration may provide a top level - 'network' key (LP: #1798117) even though it is not necessary. - - Be friendly and remove it if it really seems so. - - Return the original value if no change or the updated value if changed.""" - nullval = object() - network_val = cfg.get("network", nullval) - if network_val is nullval: - return cfg - bmsg = "Top level network key in network-config %s: %s" - if not isinstance(network_val, dict): - LOG.debug(bmsg, "was not a dict", cfg) - return cfg - if len(list(cfg.keys())) != 1: - LOG.debug(bmsg, "had multiple top level keys", cfg) - return cfg - if network_val.get("config") == "disabled": - LOG.debug(bmsg, "was config/disabled", cfg) - elif not all(("config" in network_val, "version" in network_val)): - LOG.debug(bmsg, "but missing 'config' or 'version'", cfg) - return cfg - LOG.debug(bmsg, "fixed by removing shifting network.", cfg) - return network_val - - def _merge_new_seed(cur, seeded): ret = cur.copy() @@ -369,9 +340,7 @@ def _merge_new_seed(cur, seeded): ret["meta-data"] = util.mergemanydict([cur["meta-data"], newmd]) if seeded.get("network-config"): - ret["network-config"] = _maybe_remove_top_network( - util.load_yaml(seeded.get("network-config")) - ) + ret["network-config"] = util.load_yaml(seeded.get("network-config")) if "user-data" in seeded: ret["user-data"] = seeded["user-data"] diff --git a/cloudinit/stages.py b/cloudinit/stages.py index ce43fd5d901..878aa3e6dd9 100644 --- a/cloudinit/stages.py +++ b/cloudinit/stages.py @@ -799,6 +799,16 @@ def _consume_userdata(self, frequency=PER_INSTANCE): # Run the handlers self._do_handlers(user_data_msg, c_handlers_list, frequency) + def _remove_top_level_network_key(self, cfg): + """If network-config contains top level 'network' key, then remove it. + + Some providers of network configuration skip the top-level network + key, so ensure both methods works. + """ + if cfg and "network" in cfg: + return cfg["network"] + return cfg + def _find_networking_config(self): disable_file = os.path.join( self.paths.get_cpath("data"), "upgraded-network" @@ -836,7 +846,9 @@ def _find_networking_config(self): cfg_source, ) continue - ncfg = available_cfgs[cfg_source] + ncfg = self._remove_top_level_network_key( + available_cfgs[cfg_source] + ) if net.is_disabled_cfg(ncfg): LOG.debug("network config disabled by %s", cfg_source) return (None, cfg_source) diff --git a/tests/unittests/net/test_network_state.py b/tests/unittests/net/test_network_state.py index 471d969ada3..ec21d007fa1 100644 --- a/tests/unittests/net/test_network_state.py +++ b/tests/unittests/net/test_network_state.py @@ -217,6 +217,3 @@ def test_mask_to_net_prefix_ipv6_object(self): expected = 48 prefix_value = network_state.ipv6_mask_to_net_prefix(netmask_value) assert prefix_value == expected - - -# vi: ts=4 expandtab diff --git a/tests/unittests/sources/test_nocloud.py b/tests/unittests/sources/test_nocloud.py index 1f6b722d04c..15b25196db7 100644 --- a/tests/unittests/sources/test_nocloud.py +++ b/tests/unittests/sources/test_nocloud.py @@ -7,10 +7,7 @@ from cloudinit import dmi, helpers, util from cloudinit.sources.DataSourceNoCloud import DataSourceNoCloud as dsNoCloud -from cloudinit.sources.DataSourceNoCloud import ( - _maybe_remove_top_network, - parse_cmdline_data, -) +from cloudinit.sources.DataSourceNoCloud import parse_cmdline_data from tests.unittests.helpers import CiTestCase, ExitStack, mock, populate_dir @@ -253,25 +250,6 @@ def test_metadata_network_config(self, m_is_lxd): self.assertTrue(ret) self.assertEqual(netconf, dsrc.network_config) - def test_metadata_network_config_with_toplevel_network(self, m_is_lxd): - """network-config may have 'network' top level key.""" - netconf = {"config": "disabled"} - populate_dir( - os.path.join(self.paths.seed_dir, "nocloud"), - { - "user-data": b"ud", - "meta-data": "instance-id: IID\n", - "network-config": yaml.dump({"network": netconf}) + "\n", - }, - ) - - sys_cfg = {"datasource": {"NoCloud": {"fs_label": None}}} - - dsrc = dsNoCloud(sys_cfg=sys_cfg, distro=None, paths=self.paths) - ret = dsrc.get_data() - self.assertTrue(ret) - self.assertEqual(netconf, dsrc.network_config) - def test_metadata_network_config_over_interfaces(self, m_is_lxd): # network-config should override meta-data/network-interfaces gateway = "103.225.10.1" @@ -406,48 +384,3 @@ def test_parse_cmdline_data_none(self): ret = parse_cmdline_data(ds_id=ds_id, fill=fill, cmdline=cmdline) self.assertEqual(fill, {}) self.assertFalse(ret) - - -class TestMaybeRemoveToplevelNetwork(CiTestCase): - """test _maybe_remove_top_network function.""" - - basecfg = [ - { - "type": "physical", - "name": "interface0", - "subnets": [{"type": "dhcp"}], - } - ] - - def test_should_remove_safely(self): - mcfg = {"config": self.basecfg, "version": 1} - self.assertEqual(mcfg, _maybe_remove_top_network({"network": mcfg})) - - def test_no_remove_if_other_keys(self): - """should not shift if other keys at top level.""" - mcfg = { - "network": {"config": self.basecfg, "version": 1}, - "unknown_keyname": "keyval", - } - self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) - - def test_no_remove_if_non_dict(self): - """should not shift if not a dict.""" - mcfg = {"network": '"content here'} - self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) - - def test_no_remove_if_missing_config_or_version(self): - """should not shift unless network entry has config and version.""" - mcfg = {"network": {"config": self.basecfg}} - self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) - - mcfg = {"network": {"version": 1}} - self.assertEqual(mcfg, _maybe_remove_top_network(mcfg)) - - def test_remove_with_config_disabled(self): - """network/config=disabled should be shifted.""" - mcfg = {"config": "disabled"} - self.assertEqual(mcfg, _maybe_remove_top_network({"network": mcfg})) - - -# vi: ts=4 expandtab diff --git a/tests/unittests/test_stages.py b/tests/unittests/test_stages.py index 2f85914255d..fdf0e490550 100644 --- a/tests/unittests/test_stages.py +++ b/tests/unittests/test_stages.py @@ -10,7 +10,7 @@ from cloudinit.event import EventScope, EventType from cloudinit.sources import NetworkConfigSource from cloudinit.util import write_file -from tests.unittests.helpers import CiTestCase, mock +from tests.unittests.helpers import mock TEST_INSTANCE_ID = "i-testing" @@ -35,15 +35,11 @@ def _get_data(self): return True -class TestInit(CiTestCase): - with_logs = True - allowed_subp = False - - def setUp(self): - super(TestInit, self).setUp() - self.tmpdir = self.tmp_dir() +class TestInit: + @pytest.fixture(autouse=True) + def setup(self, tmpdir): + self.tmpdir = tmpdir self.init = stages.Init() - # Setup fake Paths for Init to reference self.init._cfg = { "system_info": { "distro": "ubuntu", @@ -60,47 +56,63 @@ def test_wb__find_networking_config_disabled(self): self.init.paths.get_cpath("data"), "upgraded-network" ) write_file(disable_file, "") - self.assertEqual( - (None, disable_file), self.init._find_networking_config() - ) + assert (None, disable_file) == self.init._find_networking_config() @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "net_config", + [ + {"config": "disabled"}, + {"network": {"config": "disabled"}}, + ], + ) def test_wb__find_networking_config_disabled_by_kernel( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, net_config, caplog ): """find_networking_config returns when disabled by kernel cmdline.""" - m_cmdline.return_value = {"config": "disabled"} + m_cmdline.return_value = net_config m_initramfs.return_value = {"config": ["fake_initrd"]} - self.assertEqual( - (None, NetworkConfigSource.CMD_LINE), - self.init._find_networking_config(), - ) - self.assertEqual( - "DEBUG: network config disabled by cmdline\n", self.logs.getvalue() - ) + assert ( + None, + NetworkConfigSource.CMD_LINE, + ) == self.init._find_networking_config() + assert caplog.records[0].levelname == "DEBUG" + assert "network config disabled by cmdline" in caplog.text @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "net_config", + [ + {"config": "disabled"}, + {"network": {"config": "disabled"}}, + ], + ) def test_wb__find_networking_config_disabled_by_initrd( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, net_config, caplog ): """find_networking_config returns when disabled by kernel cmdline.""" m_cmdline.return_value = {} - m_initramfs.return_value = {"config": "disabled"} - self.assertEqual( - (None, NetworkConfigSource.INITRAMFS), - self.init._find_networking_config(), - ) - self.assertEqual( - "DEBUG: network config disabled by initramfs\n", - self.logs.getvalue(), - ) + m_initramfs.return_value = net_config + assert ( + None, + NetworkConfigSource.INITRAMFS, + ) == self.init._find_networking_config() + assert caplog.records[0].levelname == "DEBUG" + assert "network config disabled by initramfs" in caplog.text @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "net_config", + [ + {"config": "disabled"}, + {"network": {"config": "disabled"}}, + ], + ) def test_wb__find_networking_config_disabled_by_datasrc( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, net_config, caplog ): """find_networking_config returns when disabled by datasource cfg.""" m_cmdline.return_value = {} # Kernel doesn't disable networking @@ -110,41 +122,51 @@ def test_wb__find_networking_config_disabled_by_datasrc( "network": {}, } # system config doesn't disable - self.init.datasource = FakeDataSource( - network_config={"config": "disabled"} - ) - self.assertEqual( - (None, NetworkConfigSource.DS), self.init._find_networking_config() - ) - self.assertEqual( - "DEBUG: network config disabled by ds\n", self.logs.getvalue() - ) + self.init.datasource = FakeDataSource(network_config=net_config) + assert ( + None, + NetworkConfigSource.DS, + ) == self.init._find_networking_config() + assert caplog.records[0].levelname == "DEBUG" + assert "network config disabled by ds" in caplog.text @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "net_config", + [ + {"config": "disabled"}, + {"network": {"config": "disabled"}}, + ], + ) def test_wb__find_networking_config_disabled_by_sysconfig( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, net_config, caplog ): """find_networking_config returns when disabled by system config.""" m_cmdline.return_value = {} # Kernel doesn't disable networking m_initramfs.return_value = {} # initramfs doesn't disable networking self.init._cfg = { "system_info": {"paths": {"cloud_dir": self.tmpdir}}, - "network": {"config": "disabled"}, + "network": net_config, } - self.assertEqual( - (None, NetworkConfigSource.SYSTEM_CFG), - self.init._find_networking_config(), - ) - self.assertEqual( - "DEBUG: network config disabled by system_cfg\n", - self.logs.getvalue(), - ) + assert ( + None, + NetworkConfigSource.SYSTEM_CFG, + ) == self.init._find_networking_config() + assert caplog.records[0].levelname == "DEBUG" + assert "network config disabled by system_cfg" in caplog.text @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "in_config,out_config", + [ + ({"config": {"a": True}}, {"config": {"a": True}}), + ({"network": {"config": {"a": True}}}, {"config": {"a": True}}), + ], + ) def test__find_networking_config_uses_datasrc_order( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, in_config, out_config ): """find_networking_config should check sources in DS defined order""" # cmdline and initramfs, which would normally be preferred over other @@ -153,8 +175,7 @@ def test__find_networking_config_uses_datasrc_order( m_cmdline.return_value = {"config": "disabled"} m_initramfs.return_value = {"config": "disabled"} - ds_net_cfg = {"config": {"needle": True}} - self.init.datasource = FakeDataSource(network_config=ds_net_cfg) + self.init.datasource = FakeDataSource(network_config=in_config) self.init.datasource.network_config_sources = [ NetworkConfigSource.DS, NetworkConfigSource.SYSTEM_CFG, @@ -162,65 +183,83 @@ def test__find_networking_config_uses_datasrc_order( NetworkConfigSource.INITRAMFS, ] - self.assertEqual( - (ds_net_cfg, NetworkConfigSource.DS), - self.init._find_networking_config(), - ) + assert ( + out_config, + NetworkConfigSource.DS, + ) == self.init._find_networking_config() @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "in_config,out_config", + [ + ({"config": {"a": True}}, {"config": {"a": True}}), + ({"network": {"config": {"a": True}}}, {"config": {"a": True}}), + ], + ) def test__find_networking_config_warns_if_datasrc_uses_invalid_src( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, in_config, out_config, caplog ): """find_networking_config should check sources in DS defined order""" - ds_net_cfg = {"config": {"needle": True}} - self.init.datasource = FakeDataSource(network_config=ds_net_cfg) + self.init.datasource = FakeDataSource(network_config=in_config) self.init.datasource.network_config_sources = [ "invalid_src", NetworkConfigSource.DS, ] - self.assertEqual( - (ds_net_cfg, NetworkConfigSource.DS), - self.init._find_networking_config(), - ) - self.assertIn( - "WARNING: data source specifies an invalid network" - " cfg_source: invalid_src", - self.logs.getvalue(), + assert ( + out_config, + NetworkConfigSource.DS, + ) == self.init._find_networking_config() + assert caplog.records[0].levelname == "WARNING" + assert ( + "data source specifies an invalid network cfg_source: invalid_src" + in caplog.text ) @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "in_config,out_config", + [ + ({"config": {"a": True}}, {"config": {"a": True}}), + ({"network": {"config": {"a": True}}}, {"config": {"a": True}}), + ], + ) def test__find_networking_config_warns_if_datasrc_uses_unavailable_src( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, in_config, out_config, caplog ): """find_networking_config should check sources in DS defined order""" - ds_net_cfg = {"config": {"needle": True}} - self.init.datasource = FakeDataSource(network_config=ds_net_cfg) + self.init.datasource = FakeDataSource(network_config=in_config) self.init.datasource.network_config_sources = [ NetworkConfigSource.FALLBACK, NetworkConfigSource.DS, ] - self.assertEqual( - (ds_net_cfg, NetworkConfigSource.DS), - self.init._find_networking_config(), - ) - self.assertIn( - "WARNING: data source specifies an unavailable network" - " cfg_source: fallback", - self.logs.getvalue(), + assert ( + out_config, + NetworkConfigSource.DS, + ) == self.init._find_networking_config() + assert caplog.records[0].levelname == "WARNING" + assert ( + "data source specifies an unavailable network cfg_source: fallback" + in caplog.text ) @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "in_config,out_config", + [ + ({"config": {"a": True}}, {"config": {"a": True}}), + ({"network": {"config": {"a": True}}}, {"config": {"a": True}}), + ], + ) def test_wb__find_networking_config_returns_kernel( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, in_config, out_config ): """find_networking_config returns kernel cmdline config if present.""" - expected_cfg = {"config": ["fakekernel"]} - m_cmdline.return_value = expected_cfg + m_cmdline.return_value = in_config m_initramfs.return_value = {"config": ["fake_initrd"]} self.init._cfg = { "system_info": {"paths": {"cloud_dir": self.tmpdir}}, @@ -229,20 +268,26 @@ def test_wb__find_networking_config_returns_kernel( self.init.datasource = FakeDataSource( network_config={"config": ["fakedatasource"]} ) - self.assertEqual( - (expected_cfg, NetworkConfigSource.CMD_LINE), - self.init._find_networking_config(), - ) + assert ( + out_config, + NetworkConfigSource.CMD_LINE, + ) == self.init._find_networking_config() @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "in_config,out_config", + [ + ({"config": {"a": True}}, {"config": {"a": True}}), + ({"network": {"config": {"a": True}}}, {"config": {"a": True}}), + ], + ) def test_wb__find_networking_config_returns_initramfs( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, in_config, out_config ): - """find_networking_config returns kernel cmdline config if present.""" - expected_cfg = {"config": ["fake_initrd"]} + """find_networking_config returns initramfs config if present.""" m_cmdline.return_value = {} - m_initramfs.return_value = expected_cfg + m_initramfs.return_value = in_config self.init._cfg = { "system_info": {"paths": {"cloud_dir": self.tmpdir}}, "network": {"config": ["fakesys_config"]}, @@ -250,52 +295,63 @@ def test_wb__find_networking_config_returns_initramfs( self.init.datasource = FakeDataSource( network_config={"config": ["fakedatasource"]} ) - self.assertEqual( - (expected_cfg, NetworkConfigSource.INITRAMFS), - self.init._find_networking_config(), - ) + assert ( + out_config, + NetworkConfigSource.INITRAMFS, + ) == self.init._find_networking_config() @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "in_config,out_config", + [ + ({"config": {"a": True}}, {"config": {"a": True}}), + ({"network": {"config": {"a": True}}}, {"config": {"a": True}}), + ], + ) def test_wb__find_networking_config_returns_system_cfg( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, in_config, out_config ): """find_networking_config returns system config when present.""" m_cmdline.return_value = {} # No kernel network config m_initramfs.return_value = {} # no initramfs network config - expected_cfg = {"config": ["fakesys_config"]} self.init._cfg = { "system_info": {"paths": {"cloud_dir": self.tmpdir}}, - "network": expected_cfg, + "network": in_config, } self.init.datasource = FakeDataSource( network_config={"config": ["fakedatasource"]} ) - self.assertEqual( - (expected_cfg, NetworkConfigSource.SYSTEM_CFG), - self.init._find_networking_config(), - ) + assert ( + out_config, + NetworkConfigSource.SYSTEM_CFG, + ) == self.init._find_networking_config() @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") + @pytest.mark.parametrize( + "in_config,out_config", + [ + ({"config": {"a": True}}, {"config": {"a": True}}), + ({"network": {"config": {"a": True}}}, {"config": {"a": True}}), + ], + ) def test_wb__find_networking_config_returns_datasrc_cfg( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, in_config, out_config ): """find_networking_config returns datasource net config if present.""" m_cmdline.return_value = {} # No kernel network config m_initramfs.return_value = {} # no initramfs network config - # No system config for network in setUp - expected_cfg = {"config": ["fakedatasource"]} - self.init.datasource = FakeDataSource(network_config=expected_cfg) - self.assertEqual( - (expected_cfg, NetworkConfigSource.DS), - self.init._find_networking_config(), - ) + self.init.datasource = FakeDataSource(network_config=in_config) + assert ( + out_config, + NetworkConfigSource.DS, + ) == self.init._find_networking_config() @mock.patch("cloudinit.stages.cmdline.read_initramfs_config") @mock.patch("cloudinit.stages.cmdline.read_kernel_cmdline_config") def test_wb__find_networking_config_returns_fallback( - self, m_cmdline, m_initramfs + self, m_cmdline, m_initramfs, caplog ): """find_networking_config returns fallback config if not defined.""" m_cmdline.return_value = {} # Kernel doesn't disable networking @@ -313,13 +369,13 @@ def fake_generate_fallback(): # Monkey patch distro which gets cached on self.init distro = self.init.distro distro.generate_fallback_config = fake_generate_fallback - self.assertEqual( - (fake_cfg, NetworkConfigSource.FALLBACK), - self.init._find_networking_config(), - ) - self.assertNotIn("network config disabled", self.logs.getvalue()) + assert ( + fake_cfg, + NetworkConfigSource.FALLBACK, + ) == self.init._find_networking_config() + assert "network config disabled" not in caplog.text - def test_apply_network_config_disabled(self): + def test_apply_network_config_disabled(self, caplog): """Log when network is disabled by upgraded-network.""" disable_file = os.path.join( self.init.paths.get_cpath("data"), "upgraded-network" @@ -331,10 +387,8 @@ def fake_network_config(): self.init._find_networking_config = fake_network_config self.init.apply_network_config(True) - self.assertIn( - "INFO: network config is disabled by %s" % disable_file, - self.logs.getvalue(), - ) + assert caplog.records[0].levelname == "INFO" + assert f"network config is disabled by {disable_file}" in caplog.text @mock.patch("cloudinit.net.get_interfaces_by_mac") @mock.patch("cloudinit.distros.ubuntu.Distro") @@ -367,7 +421,7 @@ def fake_network_config(): ) @mock.patch("cloudinit.distros.ubuntu.Distro") - def test_apply_network_on_same_instance_id(self, m_ubuntu): + def test_apply_network_on_same_instance_id(self, m_ubuntu, caplog): """Only call distro.networking.apply_network_config_names on same instance id.""" self.init.is_new_instance = self._real_is_new_instance @@ -398,12 +452,9 @@ def fake_network_config(): self.init.distro.apply_network_config.assert_not_called() assert ( "No network config applied. Neither a new instance nor datasource " - "network update allowed" in self.logs.getvalue() + "network update allowed" in caplog.text ) - # CiTestCase doesn't work with pytest.mark.parametrize, and moving this - # functionality to a separate class is more cumbersome than it'd be worth - # at the moment, so use this as a simple setup def _apply_network_setup(self, m_macs): old_instance_id = os.path.join( self.init.paths.get_cpath("data"), "instance-id" @@ -459,7 +510,7 @@ def test_apply_network_allowed_when_default_boot(self, m_ubuntu, m_macs): {EventScope.NETWORK: {EventType.BOOT_NEW_INSTANCE}}, ) def test_apply_network_disabled_when_no_default_boot( - self, m_ubuntu, m_macs + self, m_ubuntu, m_macs, caplog ): """Don't apply network if datasource has no BOOT event.""" self._apply_network_setup(m_macs) @@ -467,7 +518,7 @@ def test_apply_network_disabled_when_no_default_boot( self.init.distro.apply_network_config.assert_not_called() assert ( "No network config applied. Neither a new instance nor datasource " - "network update allowed" in self.logs.getvalue() + "network update allowed" in caplog.text ) @mock.patch("cloudinit.net.get_interfaces_by_mac") @@ -495,7 +546,9 @@ def test_apply_network_allowed_with_userdata_overrides( sources.DataSource.supported_update_events, {EventScope.NETWORK: {EventType.BOOT_NEW_INSTANCE}}, ) - def test_apply_network_disabled_when_unsupported(self, m_ubuntu, m_macs): + def test_apply_network_disabled_when_unsupported( + self, m_ubuntu, m_macs, caplog + ): """Don't apply network config if unsupported. Shouldn't work even when specified as userdata @@ -507,7 +560,7 @@ def test_apply_network_disabled_when_unsupported(self, m_ubuntu, m_macs): self.init.distro.apply_network_config.assert_not_called() assert ( "No network config applied. Neither a new instance nor datasource " - "network update allowed" in self.logs.getvalue() + "network update allowed" in caplog.text ) @@ -568,6 +621,3 @@ def test_existing_file_permissions_are_not_modified(self, init, tmpdir): init._initialize_filesystem() assert mode == stat.S_IMODE(log_file.stat().mode) - - -# vi: ts=4 expandtab