From 5d3b07c0458df3f28b40587d8bd80d7ce3b77ccf Mon Sep 17 00:00:00 2001 From: James Falcon Date: Thu, 28 Apr 2022 21:20:02 -0500 Subject: [PATCH 1/4] Consistently strip top-level network key Our NetworkState parsing code expects no top level network key even though it's part of our v1 and v2 standard. NoCloud and LXD datasources had their own code to strip off the top-level network key, but it failed for v2 configs. This commit simplifies and moves that code into the parser so that it is consistent across all datasources. LP: #1906187 --- cloudinit/net/network_state.py | 2 + cloudinit/sources/DataSourceLXD.py | 34 +---------- cloudinit/sources/DataSourceNoCloud.py | 33 +---------- tests/unittests/net/test_network_state.py | 48 +++++++++++++++- tests/unittests/sources/test_nocloud.py | 69 +---------------------- 5 files changed, 53 insertions(+), 133 deletions(-) diff --git a/cloudinit/net/network_state.py b/cloudinit/net/network_state.py index 4e49a41a01d..cf229d293bd 100644 --- a/cloudinit/net/network_state.py +++ b/cloudinit/net/network_state.py @@ -1044,6 +1044,8 @@ def parse_net_config_data(net_config, skip_broken=True) -> NetworkState: :param net_config: curtin network config dict """ state = None + if "network" in net_config: + net_config = net_config["network"] version = net_config.get("version") config = net_config.get("config") if version == 2: 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/tests/unittests/net/test_network_state.py b/tests/unittests/net/test_network_state.py index 471d969ada3..2ae81ab94be 100644 --- a/tests/unittests/net/test_network_state.py +++ b/tests/unittests/net/test_network_state.py @@ -58,6 +58,40 @@ addresses: [4.4.4.4] """ +V1_CONFIG_WITH_NETWORK = """\ +network: + version: 1 + config: + - type: physical + name: eth0 + mac_address: '00:11:22:33:44:55' +""" + +V1_CONFIG_NO_NETWORK = """\ +version: 1 +config: + - type: physical + name: eth0 + mac_address: '00:11:22:33:44:55' +""" + +V2_CONFIG_WITH_NETWORK = """\ +network: + version: 2 + ethernets: + eth0: + match: + macaddress: '00:11:22:33:44:55' +""" + +V2_CONFIG_NO_NETWORK = """\ +version: 2 +ethernets: + eth0: + match: + macaddress: '00:11:22:33:44:55' +""" + class TestNetworkStateParseConfig(CiTestCase): def setUp(self): @@ -219,4 +253,16 @@ def test_mask_to_net_prefix_ipv6_object(self): assert prefix_value == expected -# vi: ts=4 expandtab +class TestNetworkKeyOptional: + def _load_config(self, config): + return network_state.parse_net_config_data(safeyaml.load(config)) + + def test_v1(self): + no_net = self._load_config(V1_CONFIG_NO_NETWORK) + with_net = self._load_config(V1_CONFIG_WITH_NETWORK) + assert no_net.config == with_net.config + + def test_v2(self): + no_net = self._load_config(V2_CONFIG_NO_NETWORK) + with_net = self._load_config(V2_CONFIG_WITH_NETWORK) + assert no_net.config == with_net.config 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 From 1b45c09a6c07e0ff50b069d1c5132d4a0d85ef98 Mon Sep 17 00:00:00 2001 From: James Falcon Date: Fri, 29 Apr 2022 16:41:24 -0500 Subject: [PATCH 2/4] round 2 --- cloudinit/net/network_state.py | 5 ++--- cloudinit/stages.py | 14 +++++++++++++- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/cloudinit/net/network_state.py b/cloudinit/net/network_state.py index cf229d293bd..3c7ee5a3f7a 100644 --- a/cloudinit/net/network_state.py +++ b/cloudinit/net/network_state.py @@ -1044,8 +1044,6 @@ def parse_net_config_data(net_config, skip_broken=True) -> NetworkState: :param net_config: curtin network config dict """ state = None - if "network" in net_config: - net_config = net_config["network"] version = net_config.get("version") config = net_config.get("config") if version == 2: @@ -1061,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/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) From b01edd772df9a3611ad18b8ace920e6e1dc4a9c3 Mon Sep 17 00:00:00 2001 From: James Falcon Date: Tue, 3 May 2022 08:51:52 -0500 Subject: [PATCH 3/4] Update testing. I updated test_stages.py: * Stop using the CiTestCase base class * Use a common autouse fixture as a replacement of the setup method * Replace unittest assert calls to raw assert calls * Replace custom tmpdir implementation with the pytest fixture * parametrize the tests that test _find_networking_config to include a network config with a top-level network key and one without --- tests/unittests/net/test_network_state.py | 49 ---- tests/unittests/test_stages.py | 316 +++++++++++++--------- 2 files changed, 183 insertions(+), 182 deletions(-) diff --git a/tests/unittests/net/test_network_state.py b/tests/unittests/net/test_network_state.py index 2ae81ab94be..ec21d007fa1 100644 --- a/tests/unittests/net/test_network_state.py +++ b/tests/unittests/net/test_network_state.py @@ -58,40 +58,6 @@ addresses: [4.4.4.4] """ -V1_CONFIG_WITH_NETWORK = """\ -network: - version: 1 - config: - - type: physical - name: eth0 - mac_address: '00:11:22:33:44:55' -""" - -V1_CONFIG_NO_NETWORK = """\ -version: 1 -config: - - type: physical - name: eth0 - mac_address: '00:11:22:33:44:55' -""" - -V2_CONFIG_WITH_NETWORK = """\ -network: - version: 2 - ethernets: - eth0: - match: - macaddress: '00:11:22:33:44:55' -""" - -V2_CONFIG_NO_NETWORK = """\ -version: 2 -ethernets: - eth0: - match: - macaddress: '00:11:22:33:44:55' -""" - class TestNetworkStateParseConfig(CiTestCase): def setUp(self): @@ -251,18 +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 - - -class TestNetworkKeyOptional: - def _load_config(self, config): - return network_state.parse_net_config_data(safeyaml.load(config)) - - def test_v1(self): - no_net = self._load_config(V1_CONFIG_NO_NETWORK) - with_net = self._load_config(V1_CONFIG_WITH_NETWORK) - assert no_net.config == with_net.config - - def test_v2(self): - no_net = self._load_config(V2_CONFIG_NO_NETWORK) - with_net = self._load_config(V2_CONFIG_WITH_NETWORK) - assert no_net.config == with_net.config diff --git a/tests/unittests/test_stages.py b/tests/unittests/test_stages.py index 2f85914255d..4ad210faddf 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, tmp_path): + self.tmpdir = tmp_path 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 From 1f56948cc2e833ac479f7b81bacd6ea2732eacdd Mon Sep 17 00:00:00 2001 From: James Falcon Date: Wed, 4 May 2022 10:51:14 -0500 Subject: [PATCH 4/4] tmp_path not available in 3.6 --- tests/unittests/test_stages.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unittests/test_stages.py b/tests/unittests/test_stages.py index 4ad210faddf..fdf0e490550 100644 --- a/tests/unittests/test_stages.py +++ b/tests/unittests/test_stages.py @@ -37,8 +37,8 @@ def _get_data(self): class TestInit: @pytest.fixture(autouse=True) - def setup(self, tmp_path): - self.tmpdir = tmp_path + def setup(self, tmpdir): + self.tmpdir = tmpdir self.init = stages.Init() self.init._cfg = { "system_info": {