From d7d5c8c3c09139e25cec8a638d4dfa3c1eb5c03f Mon Sep 17 00:00:00 2001 From: stanley31huang Date: Mon, 9 Sep 2024 17:41:26 +0800 Subject: [PATCH] restructure the setup and teardown functions for network.py scripts (Bugfix) (#1342) * restructure the setup and teardown functions restructure the setup and teardown functions with context manager library. Another improvement is not bring up the network interfaces if it was down before testing. * fix wait_for_iface_up did not return status fix issue that the wait_for_iface_up function will not return correct status * correct function name for checking underspeed correct the function name for checking the underspeed issue * fix the return value of check_underspeed func fix the ruturn value of check_underspeed function * fix check_underspeed fix check_underspeed * update error messages update error messages * correct shutdown nets condition correct the condition to shutdown all other netifs * correct conduit network interface correct conduit network interface * update variable name update variable name * revise variable name revise variable name * fix calledprocess error fix calledprocess error * add unittest for network.py add unittest for network.py * fix unittest failed on python3.5 fix unittest failed on python3.5 * add more unit tests add more unit tests * fix unit test fix unit test * Update providers/base/bin/network.py Co-authored-by: Massimiliano * Fix issue fix issue * fix unit tests fix unit tests * fix black issue fix black issue --------- Co-authored-by: Massimiliano --- providers/base/bin/network.py | 387 +++++++++++------ providers/base/tests/test_network.py | 620 ++++++++++++++++++++++++++- 2 files changed, 875 insertions(+), 132 deletions(-) diff --git a/providers/base/bin/network.py b/providers/base/bin/network.py index 6ac375cf45..66318f01f2 100755 --- a/providers/base/bin/network.py +++ b/providers/base/bin/network.py @@ -34,7 +34,16 @@ import subprocess import tempfile import threading -from subprocess import CalledProcessError, check_call, check_output, STDOUT +from subprocess import ( + CalledProcessError, + check_call, + check_output, + STDOUT, + DEVNULL, +) +from subprocess import run as sp_run +from contextlib import contextmanager, suppress +from pathlib import Path import sys import time @@ -584,6 +593,18 @@ def status(self): def device_name(self): return self._read_data("device/label") + @property + def ifindex(self): + return self._read_data("ifindex") + + @property + def iflink(self): + return self._read_data("iflink") + + @property + def phys_switch_id(self): + return self._read_data("phys_switch_id") + def get_test_parameters(args, environ): # Decide the actual values for test parameters, which can come @@ -613,12 +634,11 @@ def can_ping(the_interface, test_target): working_interface = True try: - with open(os.devnull, "wb") as DEVNULL: - check_call( - ["ping", "-I", the_interface, "-c", "1", test_target], - stdout=DEVNULL, - stderr=DEVNULL, - ) + sp_run( + ["ping", "-I", the_interface, "-c", "1", test_target], + stdout=DEVNULL, + stderr=DEVNULL, + ) except CalledProcessError: working_interface = False @@ -753,24 +773,203 @@ def make_target_list(iface, test_targets, log_warnings): # Wait until the specified interface comes up, or until iface_timeout. def wait_for_iface_up(iface, timeout): - isdown = True deadline = time.time() + timeout - while (time.time() < deadline) and isdown: - try: - link_status = check_output( - ["ip", "link", "show", "dev", iface] - ).decode("utf-8") - except CalledProcessError as interface_failure: - logging.error("Failed to check %s:%s", iface, interface_failure) - return 1 - if "state UP" in link_status: + + net_if = Interface(iface) + while time.time() < deadline: + if net_if.status == "up": logging.debug("Interface {} is up!".format(iface)) - isdown = False - else: - logging.debug("Interface {} not yet up; waiting....".format(iface)) - # Sleep whether or not interface is up because sometimes the IP - # address gets assigned after "ip" claims it's up. - time.sleep(5) + return True + logging.debug("Interface {} not yet up; waiting....".format(iface)) + # Sleep whether or not interface is up because sometimes the IP + # address gets assigned after "ip" claims it's up. + time.sleep(5) + + return False + + +def get_network_ifaces(): + network_info = {} + + for iface in Path("/sys/class/net/").glob("*"): + if ( + iface.name == "lo" + or iface.name.startswith("virbr") + or iface.name.startswith("lxdbr") + ): + continue + logging.debug("Retrieve the network attribute for %s interface", iface) + network_if = Interface(iface) + network_info[iface.name] = { + "status": network_if.status, + "phys_switch_id": network_if.phys_switch_id, + "iflink": network_if.iflink, + "ifindex": network_if.ifindex, + } + + return network_info + + +def turn_down_network(iface): + logging.debug("Shutting down interface:%s", iface) + try: + check_call(["ip", "link", "set", "dev", iface, "down"]) + return True + except CalledProcessError as interface_failure: + logging.error("Failed to shut down %s:%s", iface, interface_failure) + return False + + +def turn_up_network(iface, timeout): + logging.debug("Restoring interface:%s", iface) + try: + check_call(["ip", "link", "set", "dev", iface, "up"]) + if not wait_for_iface_up(iface, timeout): + return False + return True + except CalledProcessError as interface_failure: + logging.error("Failed to restore %s:%s", iface, interface_failure) + return False + + +def check_underspeed(iface): + # Check for an underspeed link and abort if found, + # UNLESS --underspeed-ok option was used or max_speed is 0 + # (which indicates a probable WiFi link) + network_if = Interface(iface) + if ( + network_if.link_speed < network_if.max_speed + and network_if.max_speed != 0 + ): + logging.error( + "Detected link speed ({}) is lower than detected max " + "speed ({})".format(network_if.link_speed, network_if.max_speed) + ) + logging.error("Check your device configuration and try again.") + logging.error( + "If you want to override and test despite this " + "under-speed link, use" + ) + logging.error("the --underspeed-ok option.") + return True + return False + + +def setup_network_ifaces( + network_info, target_network, underspeed_ok, toggle_status, timeout +): + logging.debug("Setup network interface") + + target_if_attrs = network_info.pop(target_network) + # bring up target interface + if target_if_attrs["status"] == "down" and not turn_up_network( + target_network, timeout + ): + raise SystemExit( + "Failed to bring up {} interface".format(target_network) + ) + + if not underspeed_ok and check_underspeed(target_network): + raise SystemExit( + "the network speed is incorrect for {} interface".format( + target_network + ) + ) + + if ( + target_if_attrs["phys_switch_id"] is not None + and target_if_attrs["ifindex"] != target_if_attrs["iflink"] + ): + # Means it's not a Physical network, but a Linux DSA network + # Reference: + # https://www.kernel.org/doc/Documentation/ABI/testing/sysfs-class-net + conduit_net = [ + iface + for iface, attr in network_info.items() + if attr["ifindex"] == target_if_attrs["iflink"] + ] + if not conduit_net: + raise SystemExit("Conduit network interface not found") + + conduit_net = conduit_net[0] + conduit_if_attrs = network_info.pop(conduit_net) + + # bring up conduit network interface as well + if conduit_if_attrs["status"] == "down" and not turn_up_network( + conduit_net, timeout + ): + raise SystemExit( + "Failed to bring up {} conduit interface".format(conduit_net) + ) + + if toggle_status: + # Shutdown other network interfaces + for iface, attrs in network_info.items(): + if attrs["status"] == "up" and not turn_down_network(iface): + raise SystemExit( + "Failed to shutdown {} interface".format(iface) + ) + + +def restore_network_ifaces(cur_network_info, origin_network_info, timeout): + status = True + + logging.debug("Restoring interface") + for iface, attrs in origin_network_info.items(): + if attrs["status"] != cur_network_info[iface]["status"]: + if attrs["status"] == "up" and not turn_up_network(iface, timeout): + status = False + elif attrs["status"] == "down" and not turn_down_network(iface): + status = False + + return status + + +@contextmanager +def interface_test_initialize( + target_dev, underspeed_ok, dont_toggle_ifaces, recover_timeout +): + tempfile_route = tempfile.TemporaryFile() + try: + network_info = get_network_ifaces() + # Back up routing table, since network down/up process + # tends to trash it.... + logging.debug("Backup routing table") + check_call( + ["ip", "route", "save", "table", "all"], stdout=tempfile_route + ) + setup_network_ifaces( + network_info, + target_dev, + underspeed_ok, + not dont_toggle_ifaces, + recover_timeout, + ) + + yield + + finally: + cur_network_info = get_network_ifaces() + recover_success = restore_network_ifaces( + cur_network_info, network_info, recover_timeout + ) + + # Restore routing table to original state + logging.debug("Restore routing table") + with suppress(CalledProcessError): + # Harmless "RTNETLINK answers: File exists" messages on stderr + + # This always errors out -- but it works! + # The problem is virbr0, which has the "linkdown" flag, which the + # "ip route restore" command can't handle. + sp_run( + ["ip", "route", "restore"], + stdin=tempfile_route, + stderr=DEVNULL, + ) + + if not recover_success: + raise CalledProcessError(3, "restore network failed") def interface_test(args): @@ -808,114 +1007,48 @@ def interface_test(args): sys.exit(1) # Testing begins here! - # - # Make sure that the interface is indeed connected - try: - check_call(["ip", "link", "set", "dev", args.interface, "up"]) - except CalledProcessError as interface_failure: - logging.error("Failed to use %s:%s", args.interface, interface_failure) - return 1 - - # Check for an underspeed link and abort if found, UNLESS --underspeed-ok - # option was used or max_speed is 0 (which indicates a probable WiFi link) - iface = Interface(args.interface) - if ( - iface.link_speed < iface.max_speed - and iface.max_speed != 0 - and not args.underspeed_ok - ): - logging.error( - "Detected link speed ({}) is lower than detected max " - "speed ({})".format(iface.link_speed, iface.max_speed) - ) - logging.error("Check your device configuration and try again.") - logging.error( - "If you want to override and test despite this " - "under-speed link, use" - ) - logging.error("the --underspeed-ok option.") - sys.exit(1) - - # Back up routing table, since network down/up process - # tends to trash it.... - temp = tempfile.TemporaryFile() try: - check_call(["ip", "route", "save", "table", "all"], stdout=temp) - except CalledProcessError as route_error: - logging.warning("Unable to save routing table: %s", route_error) - - error_number = 0 - # Stop all other interfaces - if not args.dont_toggle_ifaces: - extra_interfaces = [ - iface - for iface in os.listdir("/sys/class/net") - if iface != "lo" - and iface != args.interface - and not iface.startswith("virbr") - and not iface.startswith("lxdbr") - ] - - for iface in extra_interfaces: - logging.debug("Shutting down interface:%s", iface) - try: - check_call(["ip", "link", "set", "dev", iface, "down"]) - except CalledProcessError as interface_failure: - logging.error( - "Failed to shut down %s:%s", iface, interface_failure - ) - error_number = 3 - - if error_number == 0: - start_time = datetime.datetime.now() - first_loop = True - # Keep testing until a success or we run out of both targets and time - while test_targets_list: - test_target = test_targets_list.pop().strip() - error_number = run_test(args, test_target) - elapsed_seconds = (datetime.datetime.now() - start_time).seconds - if ( - elapsed_seconds > args.scan_timeout and not first_loop - ) or not error_number: - break - if not test_targets_list: - logging.warning( - " Exhausted test target list; trying again ".center( - 60, "=" + error_number = 1 + with interface_test_initialize( + args.interface, + args.underspeed_ok, + args.dont_toggle_ifaces, + args.iface_timeout, + ): + logging.debug( + "Start Iperf testing with %s iperf server", test_targets_list + ) + start_time = datetime.datetime.now() + first_loop = True + # Keep testing until a success + # or we run out of both targets and time + while test_targets_list: + test_target = test_targets_list.pop().strip() + error_number = run_test(args, test_target) + elapsed_seconds = ( + datetime.datetime.now() - start_time + ).seconds + if ( + elapsed_seconds > args.scan_timeout and not first_loop + ) or not error_number: + break + if not test_targets_list: + logging.warning( + " Exhausted test target list; trying again ".center( + 60, "=" + ) ) - ) - test_targets_list = make_target_list( - args.interface, test_targets, False - ) - time.sleep(30) # Wait to give server(s) time to come online - first_loop = False - - if not args.dont_toggle_ifaces: - for iface in extra_interfaces: - logging.debug("Restoring interface:%s", iface) - try: - check_call(["ip", "link", "set", "dev", iface, "up"]) - wait_for_iface_up(iface, args.iface_timeout) - except CalledProcessError as interface_failure: - logging.error( - "Failed to restore %s:%s", iface, interface_failure - ) - error_number = 3 + test_targets_list = make_target_list( + args.interface, test_targets, False + ) + time.sleep( + 30 + ) # Wait to give server(s) time to come online + first_loop = False - # Restore routing table to original state - temp.seek(0) - try: - # Harmless "RTNETLINK answers: File exists" messages on stderr - with open(os.devnull, "wb") as DEVNULL: - check_call(["ip", "route", "restore"], stdin=temp, stderr=DEVNULL) + return error_number except CalledProcessError: - # This always errors out -- but it works! - # The problem is virbr0, which has the "linkdown" flag, which the - # "ip route restore" command can't handle. - pass - temp.close() - - return error_number + return 3 def interface_info(args): diff --git a/providers/base/tests/test_network.py b/providers/base/tests/test_network.py index 6ee1ca08d4..ec94861461 100644 --- a/providers/base/tests/test_network.py +++ b/providers/base/tests/test_network.py @@ -13,25 +13,635 @@ # along with this program. If not, see . import unittest -from unittest.mock import patch, mock_open +import subprocess +from unittest.mock import patch, mock_open, Mock, call +from contextlib import redirect_stdout, redirect_stderr +from io import StringIO +from pathlib import Path +from subprocess import CalledProcessError +from argparse import Namespace -from network import IPerfPerformanceTest +import network class IPerfPerfomanceTestTests(unittest.TestCase): def test_find_numa_reports_node(self): with patch("builtins.open", mock_open(read_data="1")) as mo: - returned = IPerfPerformanceTest.find_numa(None, "device") + returned = network.IPerfPerformanceTest.find_numa(None, "device") self.assertEqual(returned, 1) def test_find_numa_minus_one_from_sysfs(self): with patch("builtins.open", mock_open(read_data="-1")) as mo: - returned = IPerfPerformanceTest.find_numa(None, "device") + returned = network.IPerfPerformanceTest.find_numa(None, "device") self.assertEqual(returned, -1) def test_find_numa_numa_node_not_found(self): with patch("builtins.open", mock_open()) as mo: mo.side_effect = FileNotFoundError - returned = IPerfPerformanceTest.find_numa(None, "device") + returned = network.IPerfPerformanceTest.find_numa(None, "device") self.assertEqual(returned, -1) + + +class NetworkTests(unittest.TestCase): + @patch("network.Interface") + @patch("pathlib.Path.glob") + def test_get_network_ifaces(self, mock_glob, mock_intf): + path_list = [Path("eth0"), Path("eth1"), Path("eth2")] + mock_glob.return_value = path_list + mock_intf.side_effect = [ + Mock(status="up", phys_switch_id=None, iflink="2", ifindex="2"), + Mock(status="up", phys_switch_id=None, iflink="3", ifindex="3"), + Mock(status="down", phys_switch_id=None, iflink="4", ifindex="4"), + ] + + expected_result = { + "eth0": { + "status": "up", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "2", + }, + "eth1": { + "status": "up", + "phys_switch_id": None, + "iflink": "3", + "ifindex": "3", + }, + "eth2": { + "status": "down", + "phys_switch_id": None, + "iflink": "4", + "ifindex": "4", + }, + } + + with redirect_stdout(StringIO()): + result = network.get_network_ifaces() + + self.assertDictEqual(result, expected_result) + mock_glob.assert_called_with("*") + mock_intf.assert_called_with(path_list[-1]) + self.assertEqual(mock_intf.call_count, 3) + + @patch("network.check_call") + def test_turn_down_network_success(self, mock_call): + + mock_call.return_value = 0 + with redirect_stdout(StringIO()): + self.assertTrue(network.turn_down_network("test_if")) + + mock_call.assert_called_with( + ["ip", "link", "set", "dev", "test_if", "down"] + ) + + @patch("network.check_call") + def test_turn_down_network_fail(self, mock_call): + + mock_call.side_effect = CalledProcessError(1, "command failed") + + with redirect_stdout(StringIO()): + self.assertFalse(network.turn_down_network("test_if")) + + mock_call.assert_called_with( + ["ip", "link", "set", "dev", "test_if", "down"] + ) + + @patch("network.wait_for_iface_up") + @patch("network.check_call") + def test_turn_up_network_success(self, mock_call, mock_iface_up): + + mock_call.return_value = 0 + mock_iface_up.return_value = True + with redirect_stdout(StringIO()): + self.assertTrue(network.turn_up_network("test_if", 30)) + + mock_call.assert_called_with( + ["ip", "link", "set", "dev", "test_if", "up"] + ) + + @patch("network.wait_for_iface_up") + @patch("network.check_call") + def test_turn_up_network_fail(self, mock_call, mock_iface_up): + + mock_call.side_effect = CalledProcessError(1, "command failed") + mock_iface_up.return_value = True + with redirect_stdout(StringIO()): + self.assertFalse(network.turn_up_network("test_if", 30)) + + mock_call.assert_called_with( + ["ip", "link", "set", "dev", "test_if", "up"] + ) + + @patch("network.wait_for_iface_up") + @patch("network.check_call") + def test_turn_up_network_iface_up_timeout(self, mock_call, mock_iface_up): + + mock_call.return_value = 0 + mock_iface_up.return_value = False + with redirect_stdout(StringIO()): + self.assertFalse(network.turn_up_network("test_if", 30)) + + mock_call.assert_called_with( + ["ip", "link", "set", "dev", "test_if", "up"] + ) + + @patch("logging.error") + @patch("network.Interface") + def test_check_is_underspeed(self, mock_intf, mock_logging): + mock_intf.return_value = Mock( + status="up", link_speed=100, max_speed=1000 + ) + + self.assertTrue(network.check_underspeed("test_if")) + mock_intf.assert_called_with("test_if") + self.assertEqual(mock_logging.call_count, 4) + + @patch("network.Interface") + def test_check_is_not_underspeed(self, mock_intf): + mock_intf.return_value = Mock( + status="up", link_speed=1000, max_speed=1000 + ) + + with redirect_stderr(StringIO()): + self.assertFalse(network.check_underspeed("test_if")) + + mock_intf.assert_called_with("test_if") + + @patch("time.sleep") + @patch("network.Interface") + def test_wait_for_iface_up_success(self, mock_intf, mock_sleep): + mock_intf.return_value = Mock(status="up") + + with redirect_stderr(StringIO()): + self.assertTrue(network.wait_for_iface_up("test_if", 2)) + + mock_intf.assert_called_with("test_if") + mock_sleep.assert_not_called() + + @patch("time.sleep") + @patch("network.Interface") + def test_wait_for_iface_up_fail(self, mock_intf, mock_sleep): + mock_intf.return_value = Mock( + status="down", link_speed=100, max_speed=1000 + ) + + with redirect_stderr(StringIO()): + self.assertFalse(network.wait_for_iface_up("test_if", 0.1)) + + mock_intf.assert_called_with("test_if") + mock_sleep.assert_called_with(5) + + @patch("network.check_underspeed") + @patch("network.turn_down_network") + @patch("network.turn_up_network") + def test_setup_network_ifaces_success( + self, mock_net_up, mock_net_down, mock_speed + ): + network_info = { + "eth0": { + "status": "down", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "2", + }, + "lan1": { + "status": "down", + "phys_switch_id": "00000000", + "iflink": "2", + "ifindex": "3", + }, + "lan2": { + "status": "up", + "phys_switch_id": "00000000", + "iflink": "2", + "ifindex": "4", + }, + } + mock_net_up.return_value = True + mock_net_down.return_value = True + mock_speed.return_value = False + + network.setup_network_ifaces(network_info, "lan1", False, True, 10) + mock_net_up.assert_called_with("eth0", 10) + self.assertEqual(mock_net_up.call_count, 2) + mock_net_down.assert_called_with("lan2") + mock_speed.assert_called_with("lan1") + + @patch("network.turn_up_network") + def test_setup_network_ifaces_targetif_not_up(self, mock_net_up): + network_info = { + "eth0": { + "status": "down", + "phys_switch_id": None, + "iflink": "3", + "ifindex": "3", + } + } + mock_net_up.return_value = False + + with self.assertRaises(SystemExit): + network.setup_network_ifaces(network_info, "eth0", False, True, 10) + + mock_net_up.assert_called_with("eth0", 10) + + @patch("network.check_underspeed") + def test_setup_network_ifaces_targetif_low_speed(self, mock_underspeed): + network_info = { + "eth0": { + "status": "up", + "phys_switch_id": None, + "iflink": "3", + "ifindex": "3", + } + } + mock_underspeed.return_value = True + + with self.assertRaises(SystemExit): + network.setup_network_ifaces(network_info, "eth0", False, True, 10) + + mock_underspeed.assert_called_with("eth0") + + def test_setup_network_ifaces_conduit_network_not_found(self): + network_info = { + "eth0": { + "status": "up", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "2", + }, + "lan1": { + "status": "up", + "phys_switch_id": "00000000", + "iflink": "3", + "ifindex": "4", + }, + } + + with self.assertRaises(SystemExit) as context: + network.setup_network_ifaces(network_info, "lan1", True, True, 10) + + self.assertEqual( + str(context.exception), "Conduit network interface not found" + ) + + @patch("network.turn_up_network") + def test_setup_network_ifaces_conduit_network_not_up(self, mock_net_up): + network_info = { + "eth0": { + "status": "down", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "2", + }, + "lan1": { + "status": "up", + "phys_switch_id": "00000000", + "iflink": "2", + "ifindex": "3", + }, + } + mock_net_up.return_value = False + + with self.assertRaises(SystemExit) as context: + network.setup_network_ifaces(network_info, "lan1", True, True, 10) + + # print(context.exception) + self.assertEqual( + str(context.exception), "Failed to bring up eth0 conduit interface" + ) + mock_net_up.assert_called_with("eth0", 10) + + @patch("network.turn_down_network") + def test_setup_network_ifaces_shutdown_other_netif_failed( + self, mock_net_down + ): + network_info = { + "eth0": { + "status": "up", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "2", + }, + "eth1": { + "status": "up", + "phys_switch_id": None, + "iflink": "3", + "ifindex": "3", + }, + } + mock_net_down.return_value = False + + with self.assertRaises(SystemExit) as context: + network.setup_network_ifaces(network_info, "eth0", True, True, 10) + + # print(context.exception) + self.assertEqual( + str(context.exception), "Failed to shutdown eth1 interface" + ) + mock_net_down.assert_called_with("eth1") + + @patch("network.turn_up_network") + @patch("network.turn_down_network") + def test_restore_network_ifaces_success(self, mock_net_down, mock_net_up): + origin_network_info = { + "eth0": { + "status": "down", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + }, + "eth1": { + "status": "up", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + }, + } + cur_network_info = { + "eth0": { + "status": "up", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + }, + "eth1": { + "status": "down", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + }, + } + mock_net_down.return_value = True + mock_net_up.return_value = True + + self.assertTrue( + network.restore_network_ifaces( + cur_network_info, origin_network_info, 5 + ) + ) + + @patch("network.turn_down_network") + def test_restore_network_ifaces_turn_down_failed(self, mock_net_down): + origin_network_info = { + "eth1": { + "status": "down", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + }, + } + cur_network_info = { + "eth1": { + "status": "up", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + } + } + mock_net_down.return_value = False + + self.assertFalse( + network.restore_network_ifaces( + cur_network_info, origin_network_info, 5 + ) + ) + + @patch("network.turn_up_network") + @patch("network.turn_down_network") + def test_restore_network_ifaces_turn_up_failed( + self, mock_net_down, mock_net_up + ): + origin_network_info = { + "eth1": { + "status": "up", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + }, + } + cur_network_info = { + "eth1": { + "status": "down", + "phys_switch_id": None, + "iflink": "2", + "ifindex": "3", + }, + } + mock_net_down.return_value = False + + self.assertTrue( + network.restore_network_ifaces( + cur_network_info, origin_network_info, 5 + ) + ) + + @patch("network.restore_network_ifaces") + @patch("network.setup_network_ifaces") + @patch("network.sp_run") + @patch("network.check_call") + @patch("network.get_network_ifaces") + @patch("tempfile.TemporaryFile") + def test_interface_test_initilize_run_completed( + self, + mock_temp_file, + mock_net_ifs, + mock_check_call, + mock_sp_run, + mock_net_setup, + mock_net_restore, + ): + + mock_temp_file.return_value = "fake-file" + mock_net_ifs.return_value = {} + + with network.interface_test_initialize("eth0", False, False, 10): + pass + + mock_check_call.assert_called_with( + ["ip", "route", "save", "table", "all"], stdout="fake-file" + ) + mock_sp_run.assert_called_with( + ["ip", "route", "restore"], + stdin="fake-file", + stderr=subprocess.DEVNULL, + ) + self.assertEqual(mock_temp_file.call_count, 1) + self.assertEqual(mock_net_ifs.call_count, 2) + mock_net_setup.assert_called_with({}, "eth0", False, True, 10) + mock_net_restore.assert_called_with({}, {}, 10) + + @patch("network.restore_network_ifaces") + @patch("network.sp_run") + @patch("network.check_call") + @patch("network.get_network_ifaces") + @patch("tempfile.TemporaryFile") + def test_interface_test_initilize_backup_route_failed( + self, + mock_temp_file, + mock_net_ifs, + mock_check_call, + mock_sp_run, + mock_net_restore, + ): + mock_check_call.side_effect = CalledProcessError(1, "command failed") + mock_temp_file.return_value = "fake-file" + mock_net_ifs.return_value = {} + + with self.assertRaises(CalledProcessError): + with network.interface_test_initialize("eth0", False, False, 10): + pass + + mock_check_call.assert_called_with( + ["ip", "route", "save", "table", "all"], stdout="fake-file" + ) + mock_sp_run.assert_called_with( + ["ip", "route", "restore"], + stdin="fake-file", + stderr=subprocess.DEVNULL, + ) + self.assertEqual(mock_temp_file.call_count, 1) + self.assertEqual(mock_net_ifs.call_count, 2) + mock_net_restore.assert_called_with({}, {}, 10) + + @patch("network.suppress") + @patch("network.restore_network_ifaces") + @patch("network.setup_network_ifaces") + @patch("network.sp_run") + @patch("network.check_call") + @patch("network.get_network_ifaces") + @patch("tempfile.TemporaryFile") + def test_interface_test_initilize_restore_network_failed( + self, + mock_temp_file, + mock_net_ifs, + mock_check_call, + mock_sp_run, + mock_net_setup, + mock_net_restore, + mock_suppress, + ): + + mock_temp_file.return_value = "fake-file" + mock_net_ifs.return_value = {} + mock_net_restore.return_value = False + + with self.assertRaises(CalledProcessError) as context: + with network.interface_test_initialize("eth0", False, False, 10): + pass + + mock_check_call.assert_called_with( + ["ip", "route", "save", "table", "all"], stdout="fake-file" + ) + mock_sp_run.assert_called_with( + ["ip", "route", "restore"], + stdin="fake-file", + stderr=subprocess.DEVNULL, + ) + mock_suppress.assert_called_with(subprocess.CalledProcessError) + self.assertEqual(mock_temp_file.call_count, 1) + self.assertEqual(mock_net_ifs.call_count, 2) + mock_net_setup.assert_called_with({}, "eth0", False, True, 10) + mock_net_restore.assert_called_with({}, {}, 10) + self.assertIn( + ( + "Command 'restore network failed' " + "returned non-zero exit status 3" + ), + str(context.exception), + ) + + @patch("time.sleep") + @patch("network.run_test") + @patch("network.interface_test_initialize") + @patch("network.make_target_list") + @patch("network.get_test_parameters") + def test_interface_test_run_completed( + self, + mock_get_test_params, + mock_mk_targets, + mock_net_init, + mock_run, + mock_sleep, + ): + args = Namespace( + test_type="iperf", + interface="eth0", + scan_timeout=4, + underspeed_ok=True, + dont_toggle_ifaces=True, + iface_timeout=1, + ) + mock_mk_targets.side_effect = [["192.168.1.1"], ["192.168.1.1"]] + mock_run.side_effect = [1, 0] + mock_get_test_params.return_value = {"test_target_iperf": "127.0.0.1"} + + with redirect_stderr(StringIO()): + result = network.interface_test(args) + mock_net_init.assert_called_with("eth0", True, True, 1) + mock_mk_targets.assert_called_with("eth0", "127.0.0.1", False) + mock_sleep.assert_called_with(30) + self.assertEqual(mock_run.call_count, 2) + self.assertEqual(mock_sleep.call_count, 1) + self.assertEqual(mock_mk_targets.call_count, 2) + self.assertEqual(result, 0) + + def test_interface_test_no_test_type(self): + args = Namespace() + self.assertIsNone(network.interface_test(args)) + + @patch("logging.error") + @patch("network.make_target_list") + @patch("network.get_test_parameters") + def test_interface_test_no_target_list( + self, mock_get_test_params, mock_mk_targets, mock_logging + ): + mock_mk_targets.return_value = [] + args = Namespace(test_type="iperf", interface="eth0") + + with self.assertRaises(SystemExit) as context: + network.interface_test(args) + self.assertEqual(context.exception.code, 1) + self.assertEqual(mock_logging.call_count, 7) + + @patch("logging.error") + @patch("network.make_target_list") + @patch("network.get_test_parameters") + def test_interface_test_with_example_target_list( + self, mock_get_test_params, mock_mk_targets, mock_logging + ): + mock_mk_targets.return_value = ["192.168.1.1"] + mock_get_test_params.return_value = { + "test_target_iperf": "example.com" + } + args = Namespace(test_type="iperf", interface="eth0") + + with self.assertRaises(SystemExit) as context: + network.interface_test(args) + self.assertEqual(context.exception.code, 1) + self.assertEqual(mock_logging.call_count, 7) + + +class InterfaceClassTest(unittest.TestCase): + + @patch("network.Interface.__init__", Mock(return_value=None)) + def setUp(self): + self.obj_intf = network.Interface("eth0") + + @patch("network.Interface._read_data") + def test_ifindex(self, mock_read): + mock_read.return_value = "test" + + self.assertEqual(self.obj_intf.ifindex, "test") + + @patch("network.Interface._read_data") + def test_iflink(self, mock_read): + mock_read.return_value = "test" + + self.assertEqual(self.obj_intf.iflink, "test") + + @patch("network.Interface._read_data") + def test_phys_switch_id(self, mock_read): + mock_read.return_value = "test" + + self.assertEqual(self.obj_intf.phys_switch_id, "test")