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")