diff --git a/ansible/library/show_interface.py b/ansible/library/show_interface.py index e56b9852a6d..4ca76d456a5 100644 --- a/ansible/library/show_interface.py +++ b/ansible/library/show_interface.py @@ -93,7 +93,7 @@ def collect_interface_status(self): rc, self.out, err = self.module.run_command(command, executable='/bin/bash', use_unsafe_shell=True) for line in self.out.split("\n"): line = line.strip() - if regex_int.match(line): + if regex_int.match(line) and interface == regex_int.match(line).group(1): self.int_status[interface]['name'] = regex_int.match(line).group(1) self.int_status[interface]['speed'] = regex_int.match(line).group(2) self.int_status[interface]['alias'] = regex_int.match(line).group(4) diff --git a/tests/common/platform/device_utils.py b/tests/common/platform/device_utils.py new file mode 100644 index 00000000000..e4978d3ea17 --- /dev/null +++ b/tests/common/platform/device_utils.py @@ -0,0 +1,23 @@ +""" +Helper script for fanout switch operations +""" + +def fanout_switch_port_lookup(fanout_switches, dut_port): + """ + look up the fanout switch instance and the fanout switch port + connecting to the dut_port + + Args: + fanout_switches (list FanoutHost): list of fanout switch + instances. + dut_port (str): port name on the DUT + + Returns: + None, None if fanout switch instance and port is not found + FanoutHost, Portname(str) if found + """ + for _, fanout in fanout_switches.items(): + if dut_port in fanout.host_to_fanout_port_map: + return fanout, fanout.host_to_fanout_port_map[dut_port] + + return None, None diff --git a/tests/platform/test_link_flap.py b/tests/platform/test_link_flap.py new file mode 100644 index 00000000000..3604f038a88 --- /dev/null +++ b/tests/platform/test_link_flap.py @@ -0,0 +1,81 @@ +import logging + +import pytest + +from common.platform.device_utils import fanout_switch_port_lookup +from common.utilities import wait_until + +class TestLinkFlap: + def __get_dut_if_status(self, dut, ifname=None): + if not ifname: + status = dut.show_interface(command='status')['ansible_facts']['int_status'] + else: + status = dut.show_interface(command='status', interfaces=[ifname])['ansible_facts']['int_status'] + + return status + + + def __check_if_status(self, dut, dut_port, exp_state, verbose=False): + status = self.__get_dut_if_status(dut, dut_port)[dut_port] + if verbose: + logging.debug("Interface status : {}".format(status)) + return status['oper_state'] == exp_state + + + def __toggle_one_link(self, dut, dut_port, fanout, fanout_port): + logging.info("Testing link flap on {}".format(dut_port)) + + assert self.__check_if_status(dut, dut_port, 'up', verbose=True), "Fail: dut port {}: link operational down".format(status) + + logging.info("Shutting down fanout switch {} port {} connecting to {}".format(fanout.hostname, fanout_port, dut_port)) + self.ports_shutdown_by_test.add((fanout, fanout_port)) + fanout.shutdown(fanout_port) + wait_until(30, 1, self.__check_if_status, dut, dut_port, 'down') + assert self.__check_if_status(dut, dut_port, 'down', verbose=True), "dut port {} didn't go down as expected".format(dut_port) + + logging.info("Bring up fanout switch {} port {} connecting to {}".format(fanout.hostname, fanout_port, dut_port)) + fanout.no_shutdown(fanout_port) + wait_until(30, 1, self.__check_if_status, dut, dut_port, 'up') + assert self.__check_if_status(dut, dut_port, 'up', verbose=True), "dut port {} didn't go down as expected".format(dut_port) + self.ports_shutdown_by_test.discard((fanout, fanout_port)) + + + def __build_test_candidates(self, dut, fanouthosts): + status = self.__get_dut_if_status(dut) + candidates = [] + + for dut_port in status.keys(): + fanout, fanout_port = fanout_switch_port_lookup(fanouthosts, dut_port) + + if not fanout or not fanout_port: + logging.info("Skipping port {} that is not found in connection graph".format(dut_port)) + elif status[dut_port]['admin_state'] == 'down': + logging.info("Skipping port {} that is admin down".format(dut_port)) + else: + candidates.append((dut_port, fanout, fanout_port)) + + return candidates + + + def run_link_flap_test(self, dut, fanouthosts): + self.ports_shutdown_by_test = set() + + candidates = self.__build_test_candidates(dut, fanouthosts) + if not candidates: + pytest.skip("Didn't find any port that is admin up and present in the connection graph") + + try: + for dut_port, fanout, fanout_port in candidates: + self.__toggle_one_link(dut, dut_port, fanout, fanout_port) + finally: + logging.info("Restoring fanout switch ports that were shut down by test") + for fanout, fanout_port in self.ports_shutdown_by_test: + logging.debug("Restoring fanout switch {} port {} shut down by test".format(fanout.hostname, fanout_port)) + fanout.no_shutdown(fanout_port) + + +@pytest.mark.topology('any') +@pytest.mark.platform('physical') +def test_link_flap(duthost, fanouthosts): + tlf = TestLinkFlap() + tlf.run_link_flap_test(duthost, fanouthosts) diff --git a/tests/pytest.ini b/tests/pytest.ini index a3f8ab4d42b..07a5e9240e3 100644 --- a/tests/pytest.ini +++ b/tests/pytest.ini @@ -7,3 +7,5 @@ markers: disable_loganalyzer: make to disable automatic loganalyzer broadcom: test specific to Broadcom platform sanity_check: override the default sanity check settings + topology: specify which topology testcase can be executed on: (t0, t1, ptf, etc) + platform: specify which platform testcase can be executed on: (physical, virtual, broadcom, mellanox, etc)