-
Notifications
You must be signed in to change notification settings - Fork 714
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Ying Xie <ying.xie@microsoft.com>
- Loading branch information
Showing
2 changed files
with
105 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,25 @@ | ||
""" | ||
Helper script for fanout switch operations | ||
""" | ||
|
||
import logging | ||
|
||
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 | ||
instaances. | ||
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.keys(): | ||
return fanout, fanout.host_to_fanout_port_map[dut_port] | ||
|
||
return None, None |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,80 @@ | ||
import logging | ||
|
||
import pytest | ||
|
||
from common.platform.device_utils import fanout_switch_port_lookup | ||
from common.utilities import wait_until | ||
|
||
class TestLinkFlap: | ||
def __init__(self): | ||
self.ports_shutdown_by_test = set() | ||
|
||
|
||
def __get_dut_if_facts(self, dut): | ||
interface_facts = dut.interface_facts() | ||
ansible_facts = interface_facts['ansible_facts'] | ||
if_facts = ansible_facts['ansible_interface_facts'] | ||
|
||
return if_facts | ||
|
||
|
||
def __check_if_status(self, dut, dut_port, exp_state): | ||
ifstate = self.__get_dut_if_facts(dut)[dut_port] | ||
return ifstate['active'] == exp_state | ||
|
||
|
||
def __toggle_one_port(self, dut, dut_port, fanout, fanout_port): | ||
logging.info("Testing port toggle on {}".format(dut_port)) | ||
|
||
ifstate = self.__get_dut_if_facts(dut)[dut_port] | ||
assert ifstate['active'], "dut port {} is down".format(dut_port) | ||
|
||
logging.debug("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, 0.2, self.__check_if_status, dut, dut_port, False) | ||
ifstate = self.__get_dut_if_facts(dut)[dut_port] | ||
logging.debug("Interface fact : {}".format(ifstate)) | ||
assert not ifstate['active'], "dut port {} didn't go down as expected".format(dut_port) | ||
|
||
logging.debug("Bring up fanout switch {} port {} connecting to {}".format(fanout.hostname, fanout_port, dut_port)) | ||
fanout.no_shutdown(fanout_port) | ||
wait_until(30, 0.2, self.__check_if_status, dut, dut_port, True) | ||
ifstate = self.__get_dut_if_facts(dut)[dut_port] | ||
logging.debug("Interface fact : {}".format(ifstate)) | ||
assert ifstate['active'], "dut port {} didn't come up as expected".format(dut_port) | ||
self.ports_shutdown_by_test.discard((fanout, fanout_port)) | ||
|
||
|
||
def __build_test_candidates(self, dut, fanouthosts): | ||
if_facts = self.__get_dut_if_facts(dut) | ||
candidates = [] | ||
for dut_port in if_facts.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)) | ||
else: | ||
candidates.append((dut_port, fanout, fanout_port)) | ||
|
||
return candidates | ||
|
||
|
||
def run_link_flap_test(self, dut, fanouthosts): | ||
candidates = self.__build_test_candidates(dut, fanouthosts) | ||
|
||
try: | ||
for dut_port, fanout, fanout_port in candidates: | ||
self.__toggle_one_port(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_agnostic | ||
@pytest.mark.platform_physical | ||
def test_link_flap(duthost, fanouthosts): | ||
tlf = TestLinkFlap() | ||
tlf.run_link_flap_test(duthost, fanouthosts) |