diff --git a/tests/unittests/net/test_dhcp.py b/tests/unittests/net/test_dhcp.py index c6a3518445a..3982472af43 100644 --- a/tests/unittests/net/test_dhcp.py +++ b/tests/unittests/net/test_dhcp.py @@ -374,15 +374,13 @@ def test_redhat_format_with_a_space_too_much_after_comma(self): ) -class TestDHCPDiscoveryClean(CiTestCase): - with_logs = True - +class TestDHCPDiscoveryClean: @mock.patch("cloudinit.distros.net.find_fallback_nic", return_value="eth9") @mock.patch("cloudinit.net.dhcp.os.remove") @mock.patch("cloudinit.net.dhcp.subp.subp") @mock.patch("cloudinit.net.dhcp.subp.which") def test_dhcpcd_exits_with_error( - self, m_which, m_subp, m_remove, m_fallback + self, m_which, m_subp, m_remove, m_fallback, caplog ): """Log and do nothing when nic is absent and no fallback is found.""" m_subp.side_effect = [ @@ -393,16 +391,15 @@ def test_dhcpcd_exits_with_error( with pytest.raises(NoDHCPLeaseError): maybe_perform_dhcp_discovery(Distro("fake but not", {}, None)) - self.assertIn( - "DHCP client selected: dhcpcd", - self.logs.getvalue(), - ) + assert "DHCP client selected: dhcpcd" in caplog.text @mock.patch("cloudinit.distros.net.find_fallback_nic", return_value="eth9") @mock.patch("cloudinit.net.dhcp.os.remove") @mock.patch("cloudinit.net.dhcp.subp.subp") @mock.patch("cloudinit.net.dhcp.subp.which") - def test_dhcp_client_failover(self, m_which, m_subp, m_remove, m_fallback): + def test_dhcp_client_failover( + self, m_which, m_subp, m_remove, m_fallback, caplog + ): """Log and do nothing when nic is absent and no fallback client is found.""" m_subp.side_effect = [ @@ -414,40 +411,22 @@ def test_dhcp_client_failover(self, m_which, m_subp, m_remove, m_fallback): with pytest.raises(NoDHCPLeaseError): maybe_perform_dhcp_discovery(Distro("somename", {}, None)) - self.assertIn( - "DHCP client not found: dhclient", - self.logs.getvalue(), - ) - self.assertIn( - "DHCP client not found: dhcpcd", - self.logs.getvalue(), - ) - self.assertIn( - "DHCP client not found: udhcpc", - self.logs.getvalue(), - ) + assert "DHCP client not found: dhclient" in caplog.text + assert "DHCP client not found: dhcpcd" in caplog.text + assert "DHCP client not found: udhcpc" in caplog.text @mock.patch("cloudinit.net.dhcp.subp.which") @mock.patch("cloudinit.distros.net.find_fallback_nic") - def test_absent_dhclient_command(self, m_fallback, m_which): + def test_absent_dhclient_command(self, m_fallback, m_which, caplog): """When dhclient doesn't exist in the OS, log the issue and no-op.""" m_fallback.return_value = "eth9" m_which.return_value = None # dhclient isn't found with pytest.raises(NoDHCPLeaseMissingDhclientError): maybe_perform_dhcp_discovery(Distro("whoa", {}, None)) - self.assertIn( - "DHCP client not found: dhclient", - self.logs.getvalue(), - ) - self.assertIn( - "DHCP client not found: dhcpcd", - self.logs.getvalue(), - ) - self.assertIn( - "DHCP client not found: udhcpc", - self.logs.getvalue(), - ) + assert "DHCP client not found: dhclient" in caplog.text + assert "DHCP client not found: dhcpcd" in caplog.text + assert "DHCP client not found: udhcpc" in caplog.text @mock.patch("cloudinit.net.dhcp.os.remove") @mock.patch("time.sleep", mock.MagicMock()) @@ -456,7 +435,7 @@ def test_absent_dhclient_command(self, m_fallback, m_which): @mock.patch("cloudinit.net.dhcp.subp.which", return_value="/sbin/dhclient") @mock.patch("cloudinit.net.dhcp.util.wait_for_files", return_value=False) def test_dhcp_discovery_warns_invalid_pid( - self, m_wait, m_which, m_subp, m_kill, m_remove + self, m_wait, m_which, m_subp, m_kill, m_remove, caplog ): """dhcp_discovery logs a warning when pidfile contains invalid content. @@ -478,22 +457,18 @@ def test_dhcp_discovery_warns_invalid_pid( with mock.patch( "cloudinit.util.load_text_file", return_value=lease_content ): - self.assertCountEqual( - { - "interface": "eth9", - "fixed-address": "192.168.2.74", - "subnet-mask": "255.255.255.0", - "routers": "192.168.2.1", - }, - IscDhclient().get_newest_lease("eth0"), - ) - with self.assertRaises(InvalidDHCPLeaseFileError): + assert { + "interface": "eth9", + "fixed-address": "192.168.2.74", + "subnet-mask": "255.255.255.0", + "routers": "192.168.2.1", + } == IscDhclient().get_newest_lease("eth0") + with pytest.raises(InvalidDHCPLeaseFileError): with mock.patch("cloudinit.util.load_text_file", return_value=""): IscDhclient().dhcp_discovery("eth9", distro=MockDistro()) - self.assertIn( - "dhclient(pid=, parentpid=unknown) failed " - "to daemonize after 10.0 seconds", - self.logs.getvalue(), + assert ( + "dhclient(pid=, parentpid=unknown) failed to daemonize after" + " 10.0 seconds" in caplog.text ) m_kill.assert_not_called() @@ -503,23 +478,20 @@ def test_dhcp_discovery_warns_invalid_pid( @mock.patch("cloudinit.net.dhcp.subp.which", return_value="/sbin/dhclient") @mock.patch("cloudinit.net.dhcp.subp.subp") def test_dhcp_discovery_waits_on_lease_and_pid( - self, m_subp, m_which, m_wait, m_kill, m_remove + self, m_subp, m_which, m_wait, m_kill, m_remove, caplog ): """dhcp_discovery waits for the presence of pidfile and dhcp.leases.""" m_subp.return_value = ("", "") # Don't create pid or leases file m_wait.return_value = [PID_F] # Return the missing pidfile wait for - self.assertEqual( - {}, IscDhclient().dhcp_discovery("eth9", distro=MockDistro()) - ) - self.assertEqual( - mock.call([PID_F, LEASE_F], maxwait=5, naplen=0.01), - m_wait.call_args_list[0], + assert {} == IscDhclient().dhcp_discovery("eth9", distro=MockDistro()) + m_wait.assert_called_once_with( + [PID_F, LEASE_F], maxwait=5, naplen=0.01 ) - self.assertIn( - "WARNING: dhclient did not produce expected files: dhclient.pid", - self.logs.getvalue(), + assert ( + "dhclient did not produce expected files: dhclient.pid" + in caplog.text ) m_kill.assert_not_called() @@ -557,15 +529,12 @@ def test_dhcp_discovery( with mock.patch( "cloudinit.util.load_text_file", side_effect=["1", lease_content] ): - self.assertCountEqual( - { - "interface": "eth9", - "fixed-address": "192.168.2.74", - "subnet-mask": "255.255.255.0", - "routers": "192.168.2.1", - }, - IscDhclient().dhcp_discovery("eth9", distro=MockDistro()), - ) + assert { + "interface": "eth9", + "fixed-address": "192.168.2.74", + "subnet-mask": "255.255.255.0", + "routers": "192.168.2.1", + } == IscDhclient().dhcp_discovery("eth9", distro=MockDistro()) # Interface was brought up before dhclient called m_subp.assert_has_calls( [ @@ -633,15 +602,12 @@ def test_dhcp_discovery_ib( with mock.patch( "cloudinit.util.load_text_file", side_effect=["1", lease_content] ): - self.assertCountEqual( - { - "interface": "ib0", - "fixed-address": "192.168.2.74", - "subnet-mask": "255.255.255.0", - "routers": "192.168.2.1", - }, - IscDhclient().dhcp_discovery("ib0", distro=MockDistro()), - ) + assert { + "interface": "ib0", + "fixed-address": "192.168.2.74", + "subnet-mask": "255.255.255.0", + "routers": "192.168.2.1", + } == IscDhclient().dhcp_discovery("ib0", distro=MockDistro()) # Interface was brought up before dhclient called m_subp.assert_has_calls( [ @@ -682,14 +648,13 @@ def test_dhcp_discovery_ib( @mock.patch("cloudinit.net.dhcp.subp.which", return_value="/sbin/dhclient") @mock.patch("cloudinit.util.wait_for_files") def test_dhcp_output_error_stream( - self, m_wait, m_which, m_subp, m_kill, m_remove + self, m_wait, m_which, m_subp, m_kill, m_remove, tmpdir ): """ "dhcp_log_func is called with the output and error streams of dhclient when the callable is passed.""" dhclient_err = "FAKE DHCLIENT ERROR" dhclient_out = "FAKE DHCLIENT OUT" m_subp.return_value = (dhclient_out, dhclient_err) - tmpdir = self.tmp_dir() lease_content = dedent( """ lease { @@ -707,8 +672,8 @@ def test_dhcp_output_error_stream( write_file(pid_file, "%d\n" % my_pid) def dhcp_log_func(out, err): - self.assertEqual(out, dhclient_out) - self.assertEqual(err, dhclient_err) + assert out == dhclient_out + assert err == dhclient_err IscDhclient().dhcp_discovery( "eth9", dhcp_log_func=dhcp_log_func, distro=MockDistro() diff --git a/tests/unittests/net/test_init.py b/tests/unittests/net/test_init.py index 60a44186efa..8ee4e1432c3 100644 --- a/tests/unittests/net/test_init.py +++ b/tests/unittests/net/test_init.py @@ -17,49 +17,55 @@ 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, ResponsesTestCase +from tests.unittests.helpers import ( + CiTestCase, + ResponsesTestCase, + random_string, +) from tests.unittests.util import MockDistro -class TestSysDevPath(CiTestCase): +class TestSysDevPath: def test_sys_dev_path(self): """sys_dev_path returns a path under SYS_CLASS_NET for a device.""" dev = "something" path = "attribute" - expected = net.SYS_CLASS_NET + dev + "/" + path - self.assertEqual(expected, net.sys_dev_path(dev, path)) + expected = net.get_sys_class_path() + dev + "/" + path + assert expected == net.sys_dev_path(dev, path) def test_sys_dev_path_without_path(self): """When path param isn't provided it defaults to empty string.""" dev = "something" - expected = net.SYS_CLASS_NET + dev + "/" - self.assertEqual(expected, net.sys_dev_path(dev)) - + expected = net.get_sys_class_path() + dev + "/" + assert expected == net.sys_dev_path(dev) -class TestReadSysNet(CiTestCase): - with_logs = True - def setUp(self): - super(TestReadSysNet, 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 TestReadSysNet: + @pytest.fixture(autouse=True) + @pytest.mark.parametrize( + "disable_sysfs_net", [False], indirect=["disable_sysfs_net"] + ) + def setup(self, disable_sysfs_net, tmpdir_factory): + # We mock invididual numbered tmpdirs here because these tests write + # to the sysfs directory and stale test artifacts break later tests. + mock_sysfs = f"{tmpdir_factory.mktemp('sysfs', numbered=True)}/" + with mock.patch( + "cloudinit.net.get_sys_class_path", return_value=mock_sysfs + ): + self.sysdir = mock_sysfs + yield def test_read_sys_net_strips_contents_of_sys_path(self): """read_sys_net strips whitespace from the contents of a sys file.""" content = "some stuff with trailing whitespace\t\r\n" write_file(os.path.join(self.sysdir, "dev", "attr"), content) - self.assertEqual(content.strip(), net.read_sys_net("dev", "attr")) + assert content.strip() == net.read_sys_net("dev", "attr") def test_read_sys_net_reraises_oserror(self): """read_sys_net raises OSError/IOError when file doesn't exist.""" # Non-specific Exception because versions of python OSError vs IOError. - with self.assertRaises(Exception) as context_manager: # noqa: H202 + with pytest.raises(Exception, match="No such file or directory"): net.read_sys_net("dev", "attr") - error = context_manager.exception - self.assertIn("No such file or directory", str(error)) def test_read_sys_net_handles_error_with_on_enoent(self): """read_sys_net handles OSError/IOError with on_enoent if provided.""" @@ -70,30 +76,27 @@ def on_enoent(e): net.read_sys_net("dev", "attr", on_enoent=on_enoent) error = handled_errors[0] - self.assertIsInstance(error, Exception) - self.assertIn("No such file or directory", str(error)) + assert isinstance(error, Exception) + assert "No such file or directory" in str(error) def test_read_sys_net_translates_content(self): """read_sys_net translates content when translate dict is provided.""" content = "you're welcome\n" write_file(os.path.join(self.sysdir, "dev", "attr"), content) translate = {"you're welcome": "de nada"} - self.assertEqual( - "de nada", net.read_sys_net("dev", "attr", translate=translate) + assert "de nada" == net.read_sys_net( + "dev", "attr", translate=translate ) - def test_read_sys_net_errors_on_translation_failures(self): + def test_read_sys_net_errors_on_translation_failures(self, caplog): """read_sys_net raises a KeyError and logs details on failure.""" content = "you're welcome\n" write_file(os.path.join(self.sysdir, "dev", "attr"), content) - with self.assertRaises(KeyError) as context_manager: + with pytest.raises(KeyError, match='"you\'re welcome"'): net.read_sys_net("dev", "attr", translate={}) - error = context_manager.exception - self.assertEqual('"you\'re welcome"', str(error)) - self.assertIn( + assert ( "Found unexpected (not translatable) value 'you're welcome' in " - "'{0}dev/attr".format(self.sysdir), - self.logs.getvalue(), + "'{0}dev/attr".format(self.sysdir) in caplog.text ) def test_read_sys_net_handles_handles_with_onkeyerror(self): @@ -107,63 +110,63 @@ def on_keyerror(e): net.read_sys_net("dev", "attr", translate={}, on_keyerror=on_keyerror) error = handled_errors[0] - self.assertIsInstance(error, KeyError) - self.assertEqual('"you\'re welcome"', str(error)) + assert isinstance(error, KeyError) + assert '"you\'re welcome"' == str(error) def test_read_sys_net_safe_false_on_translate_failure(self): """read_sys_net_safe returns False on translation failures.""" content = "you're welcome\n" write_file(os.path.join(self.sysdir, "dev", "attr"), content) - self.assertFalse(net.read_sys_net_safe("dev", "attr", translate={})) + assert not net.read_sys_net_safe("dev", "attr", translate={}) def test_read_sys_net_safe_returns_false_on_noent_failure(self): """read_sys_net_safe returns False on file not found failures.""" - self.assertFalse(net.read_sys_net_safe("dev", "attr")) + assert not net.read_sys_net_safe("dev", "attr") def test_read_sys_net_int_returns_none_on_error(self): """read_sys_net_safe returns None on failures.""" - self.assertFalse(net.read_sys_net_int("dev", "attr")) + assert not net.read_sys_net_int("dev", "attr") def test_read_sys_net_int_returns_none_on_valueerror(self): """read_sys_net_safe returns None when content is not an int.""" write_file(os.path.join(self.sysdir, "dev", "attr"), "NOTINT\n") - self.assertFalse(net.read_sys_net_int("dev", "attr")) + assert not net.read_sys_net_int("dev", "attr") def test_read_sys_net_int_returns_integer_from_content(self): """read_sys_net_safe returns None on failures.""" write_file(os.path.join(self.sysdir, "dev", "attr"), "1\n") - self.assertEqual(1, net.read_sys_net_int("dev", "attr")) + assert 1 == net.read_sys_net_int("dev", "attr") def test_is_up_true(self): """is_up is True if sys/net/devname/operstate is 'up' or 'unknown'.""" for state in ["up", "unknown"]: write_file(os.path.join(self.sysdir, "eth0", "operstate"), state) - self.assertTrue(net.is_up("eth0")) + assert net.is_up("eth0") def test_is_up_false(self): """is_up is False if sys/net/devname/operstate is 'down' or invalid.""" for state in ["down", "incomprehensible"]: write_file(os.path.join(self.sysdir, "eth0", "operstate"), state) - self.assertFalse(net.is_up("eth0")) + assert not net.is_up("eth0") def test_is_bridge(self): """is_bridge is True when /sys/net/devname/bridge exists.""" - self.assertFalse(net.is_bridge("eth0")) + assert not net.is_bridge("eth0") ensure_file(os.path.join(self.sysdir, "eth0", "bridge")) - self.assertTrue(net.is_bridge("eth0")) + assert net.is_bridge("eth0") def test_is_bond(self): """is_bond is True when /sys/net/devname/bonding exists.""" - self.assertFalse(net.is_bond("eth0")) + assert not net.is_bond("eth0") ensure_file(os.path.join(self.sysdir, "eth0", "bonding")) - self.assertTrue(net.is_bond("eth0")) + assert net.is_bond("eth0") def test_get_master(self): """get_master returns the path when /sys/net/devname/master exists.""" - self.assertIsNone(net.get_master("enP1s1")) + assert net.get_master("enP1s1") is None master_path = os.path.join(self.sysdir, "enP1s1", "master") ensure_file(master_path) - self.assertEqual(master_path, net.get_master("enP1s1")) + assert master_path == net.get_master("enP1s1") def test_master_is_bridge_or_bond(self): bridge_mac = "aa:bb:cc:aa:bb:cc" @@ -173,8 +176,8 @@ def test_master_is_bridge_or_bond(self): write_file(os.path.join(self.sysdir, "eth1", "address"), bridge_mac) write_file(os.path.join(self.sysdir, "eth2", "address"), bond_mac) - self.assertFalse(net.master_is_bridge_or_bond("eth1")) - self.assertFalse(net.master_is_bridge_or_bond("eth2")) + assert not net.master_is_bridge_or_bond("eth1") + assert not net.master_is_bridge_or_bond("eth2") # masters without bridge/bonding => False write_file(os.path.join(self.sysdir, "br0", "address"), bridge_mac) @@ -183,15 +186,15 @@ def test_master_is_bridge_or_bond(self): os.symlink("../br0", os.path.join(self.sysdir, "eth1", "master")) os.symlink("../bond0", os.path.join(self.sysdir, "eth2", "master")) - self.assertFalse(net.master_is_bridge_or_bond("eth1")) - self.assertFalse(net.master_is_bridge_or_bond("eth2")) + assert not net.master_is_bridge_or_bond("eth1") + assert not net.master_is_bridge_or_bond("eth2") # masters with bridge/bonding => True write_file(os.path.join(self.sysdir, "br0", "bridge"), "") write_file(os.path.join(self.sysdir, "bond0", "bonding"), "") - self.assertTrue(net.master_is_bridge_or_bond("eth1")) - self.assertTrue(net.master_is_bridge_or_bond("eth2")) + assert net.master_is_bridge_or_bond("eth1") + assert net.master_is_bridge_or_bond("eth2") def test_master_is_openvswitch(self): ovs_mac = "bb:cc:aa:bb:cc:aa" @@ -199,7 +202,7 @@ def test_master_is_openvswitch(self): # No master => False write_file(os.path.join(self.sysdir, "eth1", "address"), ovs_mac) - self.assertFalse(net.master_is_bridge_or_bond("eth1")) + assert not net.master_is_bridge_or_bond("eth1") # masters without ovs-system => False write_file(os.path.join(self.sysdir, "ovs-system", "address"), ovs_mac) @@ -208,7 +211,7 @@ def test_master_is_openvswitch(self): "../ovs-system", os.path.join(self.sysdir, "eth1", "master") ) - self.assertFalse(net.master_is_openvswitch("eth1")) + assert not net.master_is_openvswitch("eth1") # masters with ovs-system => True os.symlink( @@ -216,15 +219,15 @@ def test_master_is_openvswitch(self): os.path.join(self.sysdir, "eth1", "upper_ovs-system"), ) - self.assertTrue(net.master_is_openvswitch("eth1")) + assert net.master_is_openvswitch("eth1") def test_is_vlan(self): """is_vlan is True when /sys/net/devname/uevent has DEVTYPE=vlan.""" ensure_file(os.path.join(self.sysdir, "eth0", "uevent")) - self.assertFalse(net.is_vlan("eth0")) + assert not net.is_vlan("eth0") content = "junk\nDEVTYPE=vlan\njunk\n" write_file(os.path.join(self.sysdir, "eth0", "uevent"), content) - self.assertTrue(net.is_vlan("eth0")) + assert net.is_vlan("eth0") class TestGenerateFallbackConfig(CiTestCase): @@ -1457,134 +1460,121 @@ def test_runtime_error_on_unknown_netcfg_version(self): net.extract_physdevs({"version": 3, "awesome_config": []}) -class TestNetFailOver(CiTestCase): - def setUp(self): - super(TestNetFailOver, self).setUp() - self.add_patch("cloudinit.net.util", "m_util") - self.add_patch("cloudinit.net.read_sys_net", "m_read_sys_net") - self.add_patch("cloudinit.net.device_driver", "m_device_driver") +class TestNetFailOver: + @pytest.fixture(autouse=True) + def setup(self, mocker): + mocker.patch("cloudinit.net.util") + self.device_driver = mocker.patch("cloudinit.net.device_driver") + self.read_sys_net = mocker.patch("cloudinit.net.read_sys_net") def test_get_dev_features(self): - devname = self.random_string() - features = self.random_string() - self.m_read_sys_net.return_value = features + devname = random_string() + features = random_string() + self.read_sys_net.return_value = features - self.assertEqual(features, net.get_dev_features(devname)) - self.assertEqual(1, self.m_read_sys_net.call_count) - self.assertEqual( - mock.call(devname, "device/features"), - self.m_read_sys_net.call_args_list[0], - ) + assert features == net.get_dev_features(devname) + assert 1 == self.read_sys_net.call_count + self.read_sys_net.assert_called_once_with(devname, "device/features") def test_get_dev_features_none_returns_empty_string(self): - devname = self.random_string() - self.m_read_sys_net.side_effect = Exception("error") - self.assertEqual("", net.get_dev_features(devname)) - self.assertEqual(1, self.m_read_sys_net.call_count) - self.assertEqual( - mock.call(devname, "device/features"), - self.m_read_sys_net.call_args_list[0], - ) + devname = random_string() + self.read_sys_net.side_effect = Exception("error") + assert "" == net.get_dev_features(devname) + assert 1 == self.read_sys_net.call_count + self.read_sys_net.assert_called_once_with(devname, "device/features") @mock.patch("cloudinit.net.get_dev_features") def test_has_netfail_standby_feature(self, m_dev_features): - devname = self.random_string() + devname = random_string() standby_features = ("0" * 62) + "1" + "0" m_dev_features.return_value = standby_features - self.assertTrue(net.has_netfail_standby_feature(devname)) + assert net.has_netfail_standby_feature(devname) @mock.patch("cloudinit.net.get_dev_features") def test_has_netfail_standby_feature_short_is_false(self, m_dev_features): - devname = self.random_string() - standby_features = self.random_string() + devname = random_string() + standby_features = random_string() m_dev_features.return_value = standby_features - self.assertFalse(net.has_netfail_standby_feature(devname)) + assert not net.has_netfail_standby_feature(devname) @mock.patch("cloudinit.net.get_dev_features") def test_has_netfail_standby_feature_not_present_is_false( self, m_dev_features ): - devname = self.random_string() + devname = random_string() standby_features = "0" * 64 m_dev_features.return_value = standby_features - self.assertFalse(net.has_netfail_standby_feature(devname)) + assert not net.has_netfail_standby_feature(devname) @mock.patch("cloudinit.net.get_dev_features") def test_has_netfail_standby_feature_no_features_is_false( self, m_dev_features ): - devname = self.random_string() + devname = random_string() standby_features = None m_dev_features.return_value = standby_features - self.assertFalse(net.has_netfail_standby_feature(devname)) + assert not net.has_netfail_standby_feature(devname) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_master(self, m_exists, m_standby): - devname = self.random_string() + devname = random_string() driver = "virtio_net" m_exists.return_value = False # no master sysfs attr m_standby.return_value = True # has standby feature flag - self.assertTrue(net.is_netfail_master(devname, driver)) + assert net.is_netfail_master(devname, driver) @mock.patch("cloudinit.net.sys_dev_path") def test_is_netfail_master_checks_master_attr(self, m_sysdev): - devname = self.random_string() + devname = random_string() driver = "virtio_net" - m_sysdev.return_value = self.random_string() - self.assertFalse(net.is_netfail_master(devname, driver)) - self.assertEqual(1, m_sysdev.call_count) - self.assertEqual( - mock.call(devname, path="master"), m_sysdev.call_args_list[0] - ) + m_sysdev.return_value = random_string() + assert not net.is_netfail_master(devname, driver) + assert 1 == m_sysdev.call_count + m_sysdev.assert_called_once_with(devname, path="master") @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_master_wrong_driver(self, m_exists, m_standby): - devname = self.random_string() - driver = self.random_string() - self.assertFalse(net.is_netfail_master(devname, driver)) + devname = random_string() + driver = random_string() + assert not net.is_netfail_master(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_master_has_master_attr(self, m_exists, m_standby): - devname = self.random_string() + devname = random_string() driver = "virtio_net" m_exists.return_value = True # has master sysfs attr - self.assertFalse(net.is_netfail_master(devname, driver)) + assert not net.is_netfail_master(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_master_no_standby_feat(self, m_exists, m_standby): - devname = self.random_string() + devname = random_string() driver = "virtio_net" m_exists.return_value = False # no master sysfs attr m_standby.return_value = False # no standby feature flag - self.assertFalse(net.is_netfail_master(devname, driver)) + assert not net.is_netfail_master(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") @mock.patch("cloudinit.net.sys_dev_path") def test_is_netfail_primary(self, m_sysdev, m_exists, m_standby): - devname = self.random_string() - driver = self.random_string() # device not virtio_net - master_devname = self.random_string() + devname = random_string() + driver = random_string() # device not virtio_net + master_devname = random_string() m_sysdev.return_value = "%s/%s" % ( - self.random_string(), + random_string(), master_devname, ) m_exists.return_value = True # has master sysfs attr - self.m_device_driver.return_value = "virtio_net" # master virtio_net + self.device_driver.return_value = "virtio_net" # master virtio_net m_standby.return_value = True # has standby feature flag - self.assertTrue(net.is_netfail_primary(devname, driver)) - self.assertEqual(1, self.m_device_driver.call_count) - self.assertEqual( - mock.call(master_devname), self.m_device_driver.call_args_list[0] - ) - self.assertEqual(1, m_standby.call_count) - self.assertEqual( - mock.call(master_devname), m_standby.call_args_list[0] - ) + assert net.is_netfail_primary(devname, driver) + self.device_driver.assert_called_once_with(master_devname) + assert 1 == m_standby.call_count + m_standby.assert_called_once_with(master_devname) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") @@ -1592,18 +1582,18 @@ def test_is_netfail_primary(self, m_sysdev, m_exists, m_standby): def test_is_netfail_primary_wrong_driver( self, m_sysdev, m_exists, m_standby ): - devname = self.random_string() + devname = random_string() driver = "virtio_net" - self.assertFalse(net.is_netfail_primary(devname, driver)) + assert not net.is_netfail_primary(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") @mock.patch("cloudinit.net.sys_dev_path") def test_is_netfail_primary_no_master(self, m_sysdev, m_exists, m_standby): - devname = self.random_string() - driver = self.random_string() # device not virtio_net + devname = random_string() + driver = random_string() # device not virtio_net m_exists.return_value = False # no master sysfs attr - self.assertFalse(net.is_netfail_primary(devname, driver)) + assert not net.is_netfail_primary(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") @@ -1611,16 +1601,16 @@ def test_is_netfail_primary_no_master(self, m_sysdev, m_exists, m_standby): def test_is_netfail_primary_bad_master( self, m_sysdev, m_exists, m_standby ): - devname = self.random_string() - driver = self.random_string() # device not virtio_net - master_devname = self.random_string() + devname = random_string() + driver = random_string() # device not virtio_net + master_devname = random_string() m_sysdev.return_value = "%s/%s" % ( - self.random_string(), + random_string(), master_devname, ) m_exists.return_value = True # has master sysfs attr - self.m_device_driver.return_value = "XXXX" # master not virtio_net - self.assertFalse(net.is_netfail_primary(devname, driver)) + self.device_driver.return_value = "XXXX" # master not virtio_net + assert not net.is_netfail_primary(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") @@ -1628,77 +1618,77 @@ def test_is_netfail_primary_bad_master( def test_is_netfail_primary_no_standby( self, m_sysdev, m_exists, m_standby ): - devname = self.random_string() - driver = self.random_string() # device not virtio_net - master_devname = self.random_string() + devname = random_string() + driver = random_string() # device not virtio_net + master_devname = random_string() m_sysdev.return_value = "%s/%s" % ( - self.random_string(), + random_string(), master_devname, ) m_exists.return_value = True # has master sysfs attr - self.m_device_driver.return_value = "virtio_net" # master virtio_net + self.device_driver.return_value = "virtio_net" # master virtio_net m_standby.return_value = False # master has no standby feature flag - self.assertFalse(net.is_netfail_primary(devname, driver)) + assert not net.is_netfail_primary(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_standby(self, m_exists, m_standby): - devname = self.random_string() + devname = random_string() driver = "virtio_net" m_exists.return_value = True # has master sysfs attr m_standby.return_value = True # has standby feature flag - self.assertTrue(net.is_netfail_standby(devname, driver)) + assert net.is_netfail_standby(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_standby_wrong_driver(self, m_exists, m_standby): - devname = self.random_string() - driver = self.random_string() - self.assertFalse(net.is_netfail_standby(devname, driver)) + devname = random_string() + driver = random_string() + assert not net.is_netfail_standby(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_standby_no_master(self, m_exists, m_standby): - devname = self.random_string() + devname = random_string() driver = "virtio_net" m_exists.return_value = False # has master sysfs attr - self.assertFalse(net.is_netfail_standby(devname, driver)) + assert not net.is_netfail_standby(devname, driver) @mock.patch("cloudinit.net.has_netfail_standby_feature") @mock.patch("cloudinit.net.os.path.exists") def test_is_netfail_standby_no_standby_feature(self, m_exists, m_standby): - devname = self.random_string() + devname = random_string() driver = "virtio_net" m_exists.return_value = True # has master sysfs attr m_standby.return_value = False # has standby feature flag - self.assertFalse(net.is_netfail_standby(devname, driver)) + assert not net.is_netfail_standby(devname, driver) @mock.patch("cloudinit.net.is_netfail_standby") @mock.patch("cloudinit.net.is_netfail_primary") def test_is_netfailover_primary(self, m_primary, m_standby): - devname = self.random_string() - driver = self.random_string() + devname = random_string() + driver = random_string() m_primary.return_value = True m_standby.return_value = False - self.assertTrue(net.is_netfailover(devname, driver)) + assert net.is_netfailover(devname, driver) @mock.patch("cloudinit.net.is_netfail_standby") @mock.patch("cloudinit.net.is_netfail_primary") def test_is_netfailover_standby(self, m_primary, m_standby): - devname = self.random_string() - driver = self.random_string() + devname = random_string() + driver = random_string() m_primary.return_value = False m_standby.return_value = True - self.assertTrue(net.is_netfailover(devname, driver)) + assert net.is_netfailover(devname, driver) @mock.patch("cloudinit.net.is_netfail_standby") @mock.patch("cloudinit.net.is_netfail_primary") def test_is_netfailover_returns_false(self, m_primary, m_standby): - devname = self.random_string() - driver = self.random_string() + devname = random_string() + driver = random_string() m_primary.return_value = False m_standby.return_value = False - self.assertFalse(net.is_netfailover(devname, driver)) + assert not net.is_netfailover(devname, driver) class TestOpenvswitchIsInstalled: diff --git a/tests/unittests/sources/helpers/test_openstack.py b/tests/unittests/sources/helpers/test_openstack.py index ac8e2a35418..4d85ec3c61f 100644 --- a/tests/unittests/sources/helpers/test_openstack.py +++ b/tests/unittests/sources/helpers/test_openstack.py @@ -3,14 +3,13 @@ from unittest import mock from cloudinit.sources.helpers import openstack -from tests.unittests import helpers as test_helpers @mock.patch( "cloudinit.net.is_openvswitch_internal_interface", mock.Mock(return_value=False), ) -class TestConvertNetJson(test_helpers.CiTestCase): +class TestConvertNetJson: def test_phy_types(self): """Verify the different known physical types are handled.""" # network_data.json example from @@ -54,11 +53,8 @@ def test_phy_types(self): for t in openstack.KNOWN_PHYSICAL_TYPES: net_json["links"][0]["type"] = t - self.assertEqual( - expected, - openstack.convert_net_json( - network_json=net_json, known_macs=macs - ), + assert expected == openstack.convert_net_json( + network_json=net_json, known_macs=macs ) def test_subnet_dns(self): @@ -113,9 +109,6 @@ def test_subnet_dns(self): for t in openstack.KNOWN_PHYSICAL_TYPES: net_json["links"][0]["type"] = t - self.assertEqual( - expected, - openstack.convert_net_json( - network_json=net_json, known_macs=macs - ), + assert expected == openstack.convert_net_json( + network_json=net_json, known_macs=macs )