From 57db2b5fc8773da57cfb3f136cf77742a7a86578 Mon Sep 17 00:00:00 2001 From: Alberto Contreras Date: Wed, 20 Aug 2025 14:55:48 +0200 Subject: [PATCH] test: drop citestcase from net/test_init.py --- tests/unittests/net/test_init.py | 257 ++++++++++++++----------------- 1 file changed, 117 insertions(+), 140 deletions(-) diff --git a/tests/unittests/net/test_init.py b/tests/unittests/net/test_init.py index af43a87c765..7bbaae1e392 100644 --- a/tests/unittests/net/test_init.py +++ b/tests/unittests/net/test_init.py @@ -17,7 +17,11 @@ from cloudinit.net.ephemeral import EphemeralIPv4Network, EphemeralIPv6Network from cloudinit.subp import ProcessExecutionError from cloudinit.util import ensure_file, write_file -from tests.unittests.helpers import CiTestCase, example_netdev, random_string +from tests.unittests.helpers import ( + assert_count_equal, + example_netdev, + random_string, +) from tests.unittests.util import MockDistro @@ -223,26 +227,20 @@ def test_is_vlan(self): assert net.is_vlan("eth0") -class TestGenerateFallbackConfig(CiTestCase): - def setUp(self): - super(TestGenerateFallbackConfig, self).setUp() - sys_mock = mock.patch("cloudinit.net.get_sys_class_path") - self.m_sys_path = sys_mock.start() - self.sysdir = self.tmp_dir() + "/" - self.m_sys_path.return_value = self.sysdir - self.addCleanup(sys_mock.stop) - self.add_patch( - "cloudinit.net.util.is_container", - "m_is_container", - return_value=False, +class TestGenerateFallbackConfig: + @pytest.fixture(autouse=True) + def fixtures(self, mocker, tmp_path): + self.sysdir = str(tmp_path) + "/" + mocker.patch( + "cloudinit.net.get_sys_class_path", return_value=self.sysdir ) - self.add_patch("cloudinit.net.util.udevadm_settle", "m_settle") - self.add_patch( - "cloudinit.net.is_netfailover", "m_netfail", return_value=False + mocker.patch("cloudinit.net.util.is_container", return_value=False) + mocker.patch("cloudinit.net.util.udevadm_settle") + self.m_is_netfail = mocker.patch( + "cloudinit.net.is_netfailover", return_value=False ) - self.add_patch( + mocker.patch( "cloudinit.net.is_netfail_master", - "m_netfail_master", return_value=False, ) @@ -263,7 +261,7 @@ def test_generate_fallback_finds_connected_eth_with_mac(self): }, "version": 2, } - self.assertEqual(expected, net.generate_fallback_config()) + assert expected == net.generate_fallback_config() def test_generate_fallback_finds_dormant_eth_with_mac(self): """generate_fallback_config finds any dormant device with a mac.""" @@ -281,7 +279,7 @@ def test_generate_fallback_finds_dormant_eth_with_mac(self): }, "version": 2, } - self.assertEqual(expected, net.generate_fallback_config()) + assert expected == net.generate_fallback_config() def test_generate_fallback_finds_eth_by_operstate(self): """generate_fallback_config finds any dormant device with a mac.""" @@ -301,15 +299,15 @@ def test_generate_fallback_finds_eth_by_operstate(self): valid_operstates = ["dormant", "down", "lowerlayerdown", "unknown"] for state in valid_operstates: write_file(os.path.join(self.sysdir, "eth0", "operstate"), state) - self.assertEqual(expected, net.generate_fallback_config()) + assert expected == net.generate_fallback_config() write_file(os.path.join(self.sysdir, "eth0", "operstate"), "noworky") - self.assertIsNone(net.generate_fallback_config()) + assert net.generate_fallback_config() is None def test_generate_fallback_config_skips_veth(self): """generate_fallback_config will skip any veth interfaces.""" # A connected veth which gets ignored write_file(os.path.join(self.sysdir, "veth0", "carrier"), "1") - self.assertIsNone(net.generate_fallback_config()) + assert net.generate_fallback_config() is None def test_generate_fallback_config_skips_bridges(self): """generate_fallback_config will skip any bridges interfaces.""" @@ -318,7 +316,7 @@ def test_generate_fallback_config_skips_bridges(self): mac = "aa:bb:cc:aa:bb:cc" write_file(os.path.join(self.sysdir, "eth0", "address"), mac) ensure_file(os.path.join(self.sysdir, "eth0", "bridge")) - self.assertIsNone(net.generate_fallback_config()) + assert net.generate_fallback_config() is None def test_generate_fallback_config_skips_bonds(self): """generate_fallback_config will skip any bonded interfaces.""" @@ -327,9 +325,12 @@ def test_generate_fallback_config_skips_bonds(self): mac = "aa:bb:cc:aa:bb:cc" write_file(os.path.join(self.sysdir, "eth0", "address"), mac) ensure_file(os.path.join(self.sysdir, "eth0", "bonding")) - self.assertIsNone(net.generate_fallback_config()) + assert net.generate_fallback_config() is None - def test_generate_fallback_config_skips_netfail_devs(self): + @mock.patch("cloudinit.net.is_netfail_master") + def test_generate_fallback_config_skips_netfail_devs( + self, m_is_netfail_master + ): """gen_fallback_config ignores netfail primary,sby no mac on master.""" mac = "aa:bb:cc:aa:bb:cc" # netfailover devs share the same mac for iface in ["ens3", "ens3sby", "enP0s1f3"]: @@ -345,7 +346,7 @@ def is_netfail(iface, _driver=None): return False return True - self.m_netfail.side_effect = is_netfail + self.m_is_netfail.side_effect = is_netfail def is_netfail_master(iface, _driver=None): # ens3 is the master @@ -353,7 +354,7 @@ def is_netfail_master(iface, _driver=None): return True return False - self.m_netfail_master.side_effect = is_netfail_master + m_is_netfail_master.side_effect = is_netfail_master expected = { "ethernets": { "ens3": { @@ -366,23 +367,21 @@ def is_netfail_master(iface, _driver=None): "version": 2, } result = net.generate_fallback_config() - self.assertEqual(expected, result) - - -class TestNetFindFallBackNic(CiTestCase): - def setUp(self): - super(TestNetFindFallBackNic, self).setUp() - sys_mock = mock.patch("cloudinit.net.get_sys_class_path") - self.m_sys_path = sys_mock.start() - self.sysdir = self.tmp_dir() + "/" - self.m_sys_path.return_value = self.sysdir - self.addCleanup(sys_mock.stop) - self.add_patch( + assert expected == result + + +class TestNetFindFallBackNic: + @pytest.fixture(autouse=True) + def fixtures(self, mocker, tmp_path): + self.sysdir = str(tmp_path) + "/" + mocker.patch( + "cloudinit.net.get_sys_class_path", return_value=self.sysdir + ) + mocker.patch( "cloudinit.net.util.is_container", - "m_is_container", return_value=False, ) - self.add_patch("cloudinit.net.util.udevadm_settle", "m_settle") + mocker.patch("cloudinit.net.util.udevadm_settle") def test_generate_fallback_finds_first_connected_eth_with_mac(self): """find_fallback_nic finds any connected device with a mac.""" @@ -390,7 +389,7 @@ def test_generate_fallback_finds_first_connected_eth_with_mac(self): write_file(os.path.join(self.sysdir, "eth1", "carrier"), "1") mac = "aa:bb:cc:aa:bb:cc" write_file(os.path.join(self.sysdir, "eth1", "address"), mac) - self.assertEqual("eth1", net.find_fallback_nic()) + assert "eth1" == net.find_fallback_nic() class TestNetFindCandidateNics: @@ -617,66 +616,62 @@ def test_udevadm_settle_failure_handled_gracefully(self, caplog): self.m_settle.assert_called_once() -class TestGetDeviceList(CiTestCase): - def setUp(self): - super(TestGetDeviceList, self).setUp() - sys_mock = mock.patch("cloudinit.net.get_sys_class_path") - self.m_sys_path = sys_mock.start() - self.sysdir = self.tmp_dir() + "/" - self.m_sys_path.return_value = self.sysdir - self.addCleanup(sys_mock.stop) +class TestGetDeviceList: + @pytest.fixture(autouse=True) + def fixtures(self, mocker, tmp_path): + self.sysdir = str(tmp_path) + "/" + self.m_sys_path = mocker.patch( + "cloudinit.net.get_sys_class_path", return_value=self.sysdir + ) def test_get_devicelist_raise_oserror(self): """get_devicelist raise any non-ENOENT OSerror.""" error = OSError("Can not do it") error.errno = errno.EPERM # Set non-ENOENT self.m_sys_path.side_effect = error - with self.assertRaises(OSError) as context_manager: + with pytest.raises(OSError, match="Can not do it"): net.get_devicelist() - exception = context_manager.exception - self.assertEqual("Can not do it", str(exception)) def test_get_devicelist_empty_without_sys_net(self): """get_devicelist returns empty list when missing SYS_CLASS_NET.""" self.m_sys_path.return_value = "idontexist" - self.assertEqual([], net.get_devicelist()) + assert [] == net.get_devicelist() def test_get_devicelist_empty_with_no_devices_in_sys_net(self): """get_devicelist returns empty directoty listing for SYS_CLASS_NET.""" - self.assertEqual([], net.get_devicelist()) + assert [] == net.get_devicelist() def test_get_devicelist_lists_any_subdirectories_in_sys_net(self): """get_devicelist returns a directory listing for SYS_CLASS_NET.""" write_file(os.path.join(self.sysdir, "eth0", "operstate"), "up") write_file(os.path.join(self.sysdir, "eth1", "operstate"), "up") - self.assertCountEqual(["eth0", "eth1"], net.get_devicelist()) + assert_count_equal(["eth0", "eth1"], net.get_devicelist()) @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", mock.Mock(return_value=False), ) -class TestGetInterfaceMAC(CiTestCase): - def setUp(self): - super(TestGetInterfaceMAC, self).setUp() - sys_mock = mock.patch("cloudinit.net.get_sys_class_path") - self.m_sys_path = sys_mock.start() - self.sysdir = self.tmp_dir() + "/" - self.m_sys_path.return_value = self.sysdir - self.addCleanup(sys_mock.stop) +class TestGetInterfaceMAC: + @pytest.fixture(autouse=True) + def fixtures(self, mocker, tmp_path): + self.sysdir = str(tmp_path) + "/" + self.m_sys_path = mocker.patch( + "cloudinit.net.get_sys_class_path", return_value=self.sysdir + ) def test_get_interface_mac_false_with_no_mac(self): """get_device_list returns False when no mac is reported.""" ensure_file(os.path.join(self.sysdir, "eth0", "bonding")) mac_path = os.path.join(self.sysdir, "eth0", "address") - self.assertFalse(os.path.exists(mac_path)) - self.assertFalse(net.get_interface_mac("eth0")) + assert not os.path.exists(mac_path) + assert net.get_interface_mac("eth0") is False def test_get_interface_mac(self): """get_interfaces returns the mac from SYS_CLASS_NET/dev/address.""" mac = "aa:bb:cc:aa:bb:cc" write_file(os.path.join(self.sysdir, "eth1", "address"), mac) - self.assertEqual(mac, net.get_interface_mac("eth1")) + assert mac == net.get_interface_mac("eth1") def test_get_interface_mac_grabs_bonding_address(self): """get_interfaces returns the source device mac for bonded devices.""" @@ -687,12 +682,12 @@ def test_get_interface_mac_grabs_bonding_address(self): os.path.join(self.sysdir, "eth1", "bonding_slave", "perm_hwaddr"), source_dev_mac, ) - self.assertEqual(source_dev_mac, net.get_interface_mac("eth1")) + assert source_dev_mac == net.get_interface_mac("eth1") def test_get_interfaces_empty_list_without_sys_net(self): """get_interfaces returns an empty list when missing SYS_CLASS_NET.""" self.m_sys_path.return_value = "idontexist" - self.assertEqual([], net.get_interfaces()) + assert [] == net.get_interfaces() def test_get_interfaces_by_mac_skips_empty_mac(self): """Ignore 00:00:00:00:00:00 addresses from get_interfaces_by_mac.""" @@ -703,18 +698,18 @@ def test_get_interfaces_by_mac_skips_empty_mac(self): write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0") write_file(os.path.join(self.sysdir, "eth2", "address"), mac) expected = [("eth2", "aa:bb:cc:aa:bb:cc", None, None)] - self.assertEqual(expected, net.get_interfaces()) + assert expected == net.get_interfaces() def test_get_interfaces_by_mac_skips_missing_mac(self): """Ignore interfaces without an address from get_interfaces_by_mac.""" write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "0") address_path = os.path.join(self.sysdir, "eth1", "address") - self.assertFalse(os.path.exists(address_path)) + assert not os.path.exists(address_path) mac = "aa:bb:cc:aa:bb:cc" write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0") write_file(os.path.join(self.sysdir, "eth2", "address"), mac) expected = [("eth2", "aa:bb:cc:aa:bb:cc", None, None)] - self.assertEqual(expected, net.get_interfaces()) + assert expected == net.get_interfaces() def test_get_interfaces_by_mac_skips_master_devs(self): """Ignore interfaces with a master device which would have dup mac.""" @@ -725,7 +720,7 @@ def test_get_interfaces_by_mac_skips_master_devs(self): write_file(os.path.join(self.sysdir, "eth2", "addr_assign_type"), "0") write_file(os.path.join(self.sysdir, "eth2", "address"), mac2) expected = [("eth2", mac2, None, None)] - self.assertEqual(expected, net.get_interfaces()) + assert expected == net.get_interfaces() @mock.patch("cloudinit.net.is_netfailover") def test_get_interfaces_by_mac_skips_netfailvoer(self, m_netfail): @@ -746,7 +741,7 @@ def is_netfail(iface, _driver=None): m_netfail.side_effect = is_netfail expected = [("ens3", mac, None, None)] - self.assertEqual(expected, net.get_interfaces()) + assert expected == net.get_interfaces() def test_get_interfaces_does_not_skip_phys_members_of_bridges_and_bonds( self, @@ -779,24 +774,23 @@ def test_get_interfaces_does_not_skip_phys_members_of_bridges_and_bonds( ) interface_names = [interface[0] for interface in net.get_interfaces()] - self.assertEqual( - ["eth1", "eth2", "eth3", "ovs-system"], sorted(interface_names) + assert ["eth1", "eth2", "eth3", "ovs-system"] == sorted( + interface_names ) -class TestInterfaceHasOwnMAC(CiTestCase): - def setUp(self): - super(TestInterfaceHasOwnMAC, self).setUp() - sys_mock = mock.patch("cloudinit.net.get_sys_class_path") - self.m_sys_path = sys_mock.start() - self.sysdir = self.tmp_dir() + "/" - self.m_sys_path.return_value = self.sysdir - self.addCleanup(sys_mock.stop) +class TestInterfaceHasOwnMAC: + @pytest.fixture(autouse=True) + def fixtures(self, mocker, tmp_path): + self.sysdir = str(tmp_path) + "/" + mocker.patch( + "cloudinit.net.get_sys_class_path", return_value=self.sysdir + ) def test_interface_has_own_mac_false_when_stolen(self): """Return False from interface_has_own_mac when address is stolen.""" write_file(os.path.join(self.sysdir, "eth1", "addr_assign_type"), "2") - self.assertFalse(net.interface_has_own_mac("eth1")) + assert net.interface_has_own_mac("eth1") is False def test_interface_has_own_mac_true_when_not_stolen(self): """Return False from interface_has_own_mac when mac isn't stolen.""" @@ -804,27 +798,23 @@ def test_interface_has_own_mac_true_when_not_stolen(self): assign_path = os.path.join(self.sysdir, "eth1", "addr_assign_type") for _type in valid_assign_types: write_file(assign_path, _type) - self.assertTrue(net.interface_has_own_mac("eth1")) + assert net.interface_has_own_mac("eth1") is True def test_interface_has_own_mac_strict_errors_on_absent_assign_type(self): """When addr_assign_type is absent, interface_has_own_mac errors.""" - with self.assertRaises(ValueError): + with pytest.raises(ValueError): net.interface_has_own_mac("eth1", strict=True) @mock.patch("cloudinit.net.subp.subp") @pytest.mark.usefixtures("disable_netdev_info") -class TestEphemeralIPV4Network(CiTestCase): - - with_logs = True - - def setUp(self): - super(TestEphemeralIPV4Network, self).setUp() - sys_mock = mock.patch("cloudinit.net.get_sys_class_path") - self.m_sys_path = sys_mock.start() - self.sysdir = self.tmp_dir() + "/" - self.m_sys_path.return_value = self.sysdir - self.addCleanup(sys_mock.stop) +class TestEphemeralIPV4Network: + @pytest.fixture(autouse=True) + def fixtures(self, mocker, tmp_path): + self.sysdir = str(tmp_path) + "/" + mocker.patch( + "cloudinit.net.get_sys_class_path", return_value=self.sysdir + ) def test_ephemeral_ipv4_network_errors_on_missing_params(self, m_subp): """No required params for EphemeralIPv4Network can be None.""" @@ -837,15 +827,13 @@ def test_ephemeral_ipv4_network_errors_on_missing_params(self, m_subp): for key in required_params.keys(): params = copy.deepcopy(required_params) params[key] = None - with self.assertRaises(ValueError) as context_manager: + with pytest.raises(ValueError, match="Cannot init network on"): EphemeralIPv4Network( MockDistro(), interface_addrs_before_dhcp=example_netdev, **params, ) - error = context_manager.exception - self.assertIn("Cannot init network on", str(error)) - self.assertEqual(0, m_subp.call_count) + assert 0 == m_subp.call_count def test_ephemeral_ipv4_network_errors_invalid_mask_prefix(self, m_subp): """Raise an error when prefix_or_mask is not a netmask or prefix.""" @@ -858,14 +846,13 @@ def test_ephemeral_ipv4_network_errors_invalid_mask_prefix(self, m_subp): invalid_masks = ("invalid", "invalid.", "123.123.123") for error_val in invalid_masks: params["prefix_or_mask"] = error_val - with self.assertRaises(ValueError) as context_manager: + with pytest.raises( + ValueError, + match="Cannot setup network, invalid prefix or netmask: ", + ): with EphemeralIPv4Network(MockDistro(), **params): pass - error = context_manager.exception - self.assertIn( - "Cannot setup network, invalid prefix or netmask: ", str(error) - ) - self.assertEqual(0, m_subp.call_count) + assert 0 == m_subp.call_count def test_ephemeral_ipv4_network_performs_teardown(self, m_subp): """EphemeralIPv4Network performs teardown on the device if setup.""" @@ -907,7 +894,7 @@ def test_ephemeral_ipv4_network_performs_teardown(self, m_subp): "interface_addrs_before_dhcp": example_netdev, } with EphemeralIPv4Network(MockDistro(), **params): - self.assertEqual(expected_setup_calls, m_subp.call_args_list) + assert expected_setup_calls == m_subp.call_args_list m_subp.assert_has_calls(expected_teardown_calls) def test_teardown_on_enter_exception(self, m_subp): @@ -969,7 +956,7 @@ def side_effect(args, **kwargs): for teardown in expected_teardown_calls: assert teardown in m_subp.call_args_list - def test_ephemeral_ipv4_network_noop_when_configured(self, m_subp): + def test_ephemeral_ipv4_network_noop_when_configured(self, m_subp, caplog): """EphemeralIPv4Network handles exception when address is setup. It performs no cleanup as the interface was already setup. @@ -988,8 +975,8 @@ def test_ephemeral_ipv4_network_noop_when_configured(self, m_subp): with EphemeralIPv4Network(MockDistro(), **params): pass assert expected_calls == m_subp.call_args_list - assert "Skip bringing up network link" in self.logs.getvalue() - assert "Skip adding ip address" in self.logs.getvalue() + assert "Skip bringing up network link" in caplog.text + assert "Skip adding ip address" in caplog.text def test_ephemeral_ipv4_network_with_prefix(self, m_subp): """EphemeralIPv4Network takes a valid prefix to setup the network.""" @@ -1116,7 +1103,7 @@ def test_ephemeral_ipv4_network_with_new_default_route(self, m_subp): ] with EphemeralIPv4Network(MockDistro(), **params): - self.assertEqual(expected_setup_calls, m_subp.call_args_list) + assert expected_setup_calls == m_subp.call_args_list m_subp.assert_has_calls(expected_teardown_calls) def test_ephemeral_ipv4_network_with_rfc3442_static_routes(self, m_subp): @@ -1230,7 +1217,7 @@ def test_ephemeral_ipv4_network_with_rfc3442_static_routes(self, m_subp): ), ] with EphemeralIPv4Network(MockDistro(), **params): - self.assertEqual(expected_setup_calls, m_subp.call_args_list) + assert expected_setup_calls == m_subp.call_args_list m_subp.assert_has_calls(expected_setup_calls + expected_teardown_calls) @@ -1272,11 +1259,13 @@ def _mk_v2_phys(mac, name, driver=None, device_id=None): return v2_cfg -class TestExtractPhysdevs(CiTestCase): - def setUp(self): - super(TestExtractPhysdevs, self).setUp() - self.add_patch("cloudinit.net.device_driver", "m_driver") - self.add_patch("cloudinit.net.device_devid", "m_devid") +class TestExtractPhysdevs: + @pytest.fixture(autouse=True) + def fixtures(self, mocker): + self.m_driver = mocker.patch("cloudinit.net.device_driver") + self.m_devid = mocker.patch( + "cloudinit.net.device_devid", + ) def test_extract_physdevs_looks_up_driver_v1(self): driver = "virtio" @@ -1290,9 +1279,7 @@ def test_extract_physdevs_looks_up_driver_v1(self): } # insert the driver value for verification physdevs[0][2] = driver - self.assertEqual( - sorted(physdevs), sorted(net.extract_physdevs(netcfg)) - ) + assert sorted(physdevs) == sorted(net.extract_physdevs(netcfg)) self.m_driver.assert_called_with("eth0") def test_extract_physdevs_looks_up_driver_v2(self): @@ -1307,9 +1294,7 @@ def test_extract_physdevs_looks_up_driver_v2(self): } # insert the driver value for verification physdevs[0][2] = driver - self.assertEqual( - sorted(physdevs), sorted(net.extract_physdevs(netcfg)) - ) + assert sorted(physdevs) == sorted(net.extract_physdevs(netcfg)) self.m_driver.assert_called_with("eth0") def test_extract_physdevs_looks_up_devid_v1(self): @@ -1324,9 +1309,7 @@ def test_extract_physdevs_looks_up_devid_v1(self): } # insert the driver value for verification physdevs[0][3] = devid - self.assertEqual( - sorted(physdevs), sorted(net.extract_physdevs(netcfg)) - ) + assert sorted(physdevs) == sorted(net.extract_physdevs(netcfg)) self.m_devid.assert_called_with("eth0") def test_extract_physdevs_looks_up_devid_v2(self): @@ -1341,9 +1324,7 @@ def test_extract_physdevs_looks_up_devid_v2(self): } # insert the driver value for verification physdevs[0][3] = devid - self.assertEqual( - sorted(physdevs), sorted(net.extract_physdevs(netcfg)) - ) + assert sorted(physdevs) == sorted(net.extract_physdevs(netcfg)) self.m_devid.assert_called_with("eth0") def test_get_v1_type_physical(self): @@ -1356,9 +1337,7 @@ def test_get_v1_type_physical(self): "version": 1, "config": [_mk_v1_phys(*args) for args in physdevs], } - self.assertEqual( - sorted(physdevs), sorted(net.extract_physdevs(netcfg)) - ) + assert sorted(physdevs) == sorted(net.extract_physdevs(netcfg)) def test_get_v2_type_physical(self): physdevs = [ @@ -1370,9 +1349,7 @@ def test_get_v2_type_physical(self): "version": 2, "ethernets": {args[1]: _mk_v2_phys(*args) for args in physdevs}, } - self.assertEqual( - sorted(physdevs), sorted(net.extract_physdevs(netcfg)) - ) + assert sorted(physdevs) == sorted(net.extract_physdevs(netcfg)) def test_get_v2_type_physical_skips_if_no_set_name(self): netcfg = { @@ -1383,10 +1360,10 @@ def test_get_v2_type_physical_skips_if_no_set_name(self): } }, } - self.assertEqual([], net.extract_physdevs(netcfg)) + assert [] == net.extract_physdevs(netcfg) def test_runtime_error_on_unknown_netcfg_version(self): - with self.assertRaises(RuntimeError): + with pytest.raises(RuntimeError): net.extract_physdevs({"version": 3, "awesome_config": []})