diff --git a/tests/common/helpers/psu_helpers.py b/tests/common/helpers/psu_helpers.py new file mode 100644 index 00000000000..b471487eafb --- /dev/null +++ b/tests/common/helpers/psu_helpers.py @@ -0,0 +1,60 @@ +import logging +from tests.common.helpers.assertions import pytest_assert +from tests.common.utilities import wait_until + + +def turn_on_all_outlets(pdu_controller): + """Turns on all outlets through SNMP and confirms they are turned on successfully + + Args: + pdu_controller (BasePduController): Instance of PDU controller + + Returns: + None + """ + logging.info("Turning on all outlets/PDUs") + outlet_status = pdu_controller.get_outlet_status() + for outlet in outlet_status: + if not outlet['outlet_on']: + pdu_controller.turn_on_outlet(outlet) + + for outlet in outlet_status: + pytest_assert(wait_until(60, 5, 0, check_outlet_status, + pdu_controller, outlet, True), + "Outlet {} did not turn on".format(outlet['pdu_name'])) + + +def check_outlet_status(pdu_controller, outlet, expect_status=True): + """Check if a given PDU matches the expected status + + Args: + pdu_controller (BasePduController): Instance of PDU controller + outlet (RPS outlet): Outlet whose status is to be checked + expect_status (boolean): Expected status in True/False (On/Off) + + Returns: + boolean: True if the outlet matches expected status, False otherwise + """ + status = pdu_controller.get_outlet_status(outlet) + return 'outlet_on' in status[0] and status[0]['outlet_on'] == expect_status + + +def get_grouped_pdus_by_psu(pdu_controller): + """Returns a grouping of PDUs associated with a PSU in dictionary form + + Args: + pdu_controller (BasePduController): Instance of PDU controller + + Returns: + dict: {PSU: array of PDUs} where PDUs are associated with PSU + """ + # Group outlets/PDUs by PSU + outlet_status = pdu_controller.get_outlet_status() + psu_to_pdus = {} + for outlet in outlet_status: + if outlet['psu_name'] not in psu_to_pdus: + psu_to_pdus[outlet['psu_name']] = [outlet] + else: + psu_to_pdus[outlet['psu_name']].append(outlet) + + return psu_to_pdus diff --git a/tests/platform_tests/test_platform_info.py b/tests/platform_tests/test_platform_info.py index 2a3c0906848..b20ecb160a7 100644 --- a/tests/platform_tests/test_platform_info.py +++ b/tests/platform_tests/test_platform_info.py @@ -11,6 +11,7 @@ from retry.api import retry_call from tests.common.helpers.assertions import pytest_assert, pytest_require +from tests.common.helpers.psu_helpers import turn_on_all_outlets, get_grouped_pdus_by_psu from tests.common.plugins.loganalyzer.loganalyzer import LogAnalyzer from tests.common.utilities import wait_until, get_sup_node_or_random_node from tests.common.platform.device_utils import get_dut_psu_line_pattern @@ -28,6 +29,7 @@ CMD_PLATFORM_TEMPER = "show platform temperature" PDU_WAIT_TIME = 20 +MODULAR_CHASSIS_PDU_WAIT_TIME = 60 THERMAL_CONTROL_TEST_WAIT_TIME = 65 THERMAL_CONTROL_TEST_CHECK_INTERVAL = 5 @@ -205,17 +207,6 @@ def check_vendor_specific_psustatus(dut, psu_status_line, psu_line_pattern): check_psu_sysfs(dut, psu_id, psu_status) -def turn_all_outlets_on(pdu_ctrl): - all_outlet_status = pdu_ctrl.get_outlet_status() - pytest_require(all_outlet_status and len(all_outlet_status) >= 2, - 'Skip the test, cannot to get at least 2 outlet status: {}'.format(all_outlet_status)) - for outlet in all_outlet_status: - if not outlet["outlet_on"]: - pdu_ctrl.turn_on_outlet(outlet) - time.sleep(5) - time.sleep(5) - - def check_all_psu_on(dut, psu_test_results): """ @summary: check all PSUs are in 'OK' status. @@ -273,7 +264,7 @@ def test_turn_on_off_psu_and_check_psustatus(duthosts, logging.info( "To avoid DUT being shutdown, need to turn on PSUs that are not powered") - turn_all_outlets_on(pdu_ctrl) + turn_on_all_outlets(pdu_ctrl) logging.info("Initialize test results") psu_test_results = {} @@ -282,7 +273,13 @@ def test_turn_on_off_psu_and_check_psustatus(duthosts, pytest_assert( len(list(psu_test_results.keys())) == psu_num, - "In consistent PSU number output by '%s' and '%s'" % (CMD_PLATFORM_PSUSTATUS, "sudo psuutil numpsus")) + "Inconsistent PSU number output by '%s' and '%s'" % (CMD_PLATFORM_PSUSTATUS, "sudo psuutil numpsus")) + + # Increase pdu_wait_time for modular chassis + pdu_wait_time = PDU_WAIT_TIME + is_modular_chassis = duthosts[0].get_facts().get("modular_chassis") + if is_modular_chassis: + pdu_wait_time = MODULAR_CHASSIS_PDU_WAIT_TIME logging.info("Start testing turn off/on PSUs") all_outlet_status = pdu_ctrl.get_outlet_status() @@ -292,39 +289,49 @@ def test_turn_on_off_psu_and_check_psustatus(duthosts, all_outlet_status = all_outlet_status[0:-2] logging.info( "DUT is MgmtTsToR, the last 2 outlets are reserved for Console Switch and are not visible from DUT.") - for outlet in all_outlet_status: - psu_under_test = None - if outlet['outlet_on'] is False: - continue - logging.info("Turn off outlet {}".format(outlet)) - pdu_ctrl.turn_off_outlet(outlet) - time.sleep(PDU_WAIT_TIME) + # Group outlets/PDUs by PSU and toggle PDUs by PSU + psu_to_pdus = get_grouped_pdus_by_psu(pdu_ctrl) - cli_psu_status = duthost.command(CMD_PLATFORM_PSUSTATUS) - for line in cli_psu_status["stdout_lines"][2:]: - psu_match = psu_line_pattern.match(line) - pytest_assert(psu_match, "Unexpected PSU status output") - # also make sure psustatus is not 'NOT PRESENT', which cannot be turned on/off - if psu_match.group(2) != "OK" and psu_match.group(2) != "NOT PRESENT": - psu_under_test = psu_match.group(1) - check_vendor_specific_psustatus(duthost, line, psu_line_pattern) - pytest_assert(psu_under_test is not None, "No PSU is turned off") - - logging.info("Turn on outlet {}".format(outlet)) - pdu_ctrl.turn_on_outlet(outlet) - time.sleep(PDU_WAIT_TIME) - - cli_psu_status = duthost.command(CMD_PLATFORM_PSUSTATUS) - for line in cli_psu_status["stdout_lines"][2:]: - psu_match = psu_line_pattern.match(line) - pytest_assert(psu_match, "Unexpected PSU status output") - if psu_match.group(1) == psu_under_test: - pytest_assert(psu_match.group(2) == "OK", - "Unexpected PSU status after turned it on") - check_vendor_specific_psustatus(duthost, line, psu_line_pattern) - - psu_test_results[psu_under_test] = True + try: + for psu in psu_to_pdus.keys(): + outlets = psu_to_pdus[psu] + psu_under_test = None + + logging.info("Turning off {} PDUs connected to {}".format(len(outlets), psu)) + for outlet in outlets: + pdu_ctrl.turn_off_outlet(outlet) + time.sleep(pdu_wait_time) + + # Check that PSU is turned off + cli_psu_status = duthost.command(CMD_PLATFORM_PSUSTATUS) + + for line in cli_psu_status["stdout_lines"][2:]: + psu_match = psu_line_pattern.match(line) + pytest_assert(psu_match, "Unexpected PSU status output") + # also make sure psustatus is not 'NOT PRESENT', which cannot be turned on/off + if psu_match.group(2) != "OK" and psu_match.group(2) != "NOT PRESENT": + psu_under_test = psu_match.group(1) + check_vendor_specific_psustatus(duthost, line, psu_line_pattern) + pytest_assert(psu_under_test is not None, "No PSU is turned off") + + for outlet in outlets: + logging.info("Turn on outlet {}".format(outlet)) + pdu_ctrl.turn_on_outlet(outlet) + time.sleep(pdu_wait_time) + + cli_psu_status = duthost.command(CMD_PLATFORM_PSUSTATUS) + for line in cli_psu_status["stdout_lines"][2:]: + psu_match = psu_line_pattern.match(line) + pytest_assert(psu_match, "Unexpected PSU status output") + if psu_match.group(1) == psu_under_test: + pytest_assert(psu_match.group(2) == "OK", + "Unexpected PSU status after turned it on") + check_vendor_specific_psustatus(duthost, line, psu_line_pattern) + + psu_test_results[psu_under_test] = True + finally: + turn_on_all_outlets(pdu_ctrl) for psu in psu_test_results: pytest_assert(psu_test_results[psu], diff --git a/tests/snmp/test_snmp_phy_entity.py b/tests/snmp/test_snmp_phy_entity.py index 18421fbb693..9a9650f56d5 100644 --- a/tests/snmp/test_snmp_phy_entity.py +++ b/tests/snmp/test_snmp_phy_entity.py @@ -3,10 +3,13 @@ import pytest import re import time +import random from enum import Enum, unique from tests.common.utilities import wait_until from tests.common.helpers.assertions import pytest_require from tests.common.helpers.snmp_helpers import get_snmp_facts +from tests.common.helpers.assertions import pytest_assert +from tests.common.helpers.psu_helpers import turn_on_all_outlets, check_outlet_status, get_grouped_pdus_by_psu from tests.common.helpers.thermal_control_test_helper import mocker_factory # noqa F401 pytestmark = [ @@ -634,31 +637,36 @@ def test_turn_off_psu_and_check_psu_info(duthosts, enum_supervisor_dut_hostname, pdu_controller = get_pdu_controller(duthost) if not pdu_controller: pytest.skip('psu_controller is None, skipping this test') + outlet_status = pdu_controller.get_outlet_status() if len(outlet_status) < 2: pytest.skip( 'At least 2 PSUs required for rest of the testing in this case') - # turn on all PSU - for outlet in outlet_status: - if not outlet['outlet_on']: - pdu_controller.turn_on_outlet(outlet) - time.sleep(5) - - outlet_status = pdu_controller.get_outlet_status() - for outlet in outlet_status: - if not outlet['outlet_on']: - pytest.skip( - 'Not all outlet are powered on, skip rest of the testing in this case') - - # turn off the first PSU - first_outlet = outlet_status[0] - pdu_controller.turn_off_outlet(first_outlet) - assert wait_until(30, 5, 0, check_outlet_status, - pdu_controller, first_outlet, False) - # wait for psud update the database - assert wait_until(180, 20, 5, _check_psu_status_after_power_off, - duthost, localhost, creds_all_duts) + # Turn on all PDUs + logging.info("Turning all outlets on before test") + turn_on_all_outlets(pdu_controller) + + psu_to_pdus = get_grouped_pdus_by_psu(pdu_controller) + try: + logging.info("Turning off PDUs connected to a random PSU") + # Get a random PSU's related PDUs to turn off + off_psu = random.choice(list(psu_to_pdus.keys())) + outlets = psu_to_pdus[off_psu] + logging.info("Toggling {} PDUs connected to {}".format(len(outlets), off_psu)) + for outlet in outlets: + pdu_controller.turn_off_outlet(outlet) + pytest_assert(wait_until(30, 5, 0, check_outlet_status, + pdu_controller, outlet, False), + "Outlet {} did not turn off".format(outlet['pdu_name'])) + + logging.info("Checking that turning off these outlets affects PSUs") + # wait for psud update the database + pytest_assert(wait_until(900, 20, 5, _check_psu_status_after_power_off, + duthost, localhost, creds_all_duts), + "No PSUs turned off") + finally: + turn_on_all_outlets(pdu_controller) def _check_psu_status_after_power_off(duthost, localhost, creds_all_duts): @@ -844,15 +852,3 @@ def is_null_str(value): :return: True if a string is None or 'None' or 'N/A' """ return not value or value == str(None) or value == 'N/A' - - -def check_outlet_status(pdu_controller, outlet, expect_status): - """ - Check if a given PSU is at expect status - :param pdu_controller: PDU controller - :param outlet: PDU outlet - :param expect_status: Expect bool status, True means on, False means off - :return: True if a given PSU is at expect status - """ - status = pdu_controller.get_outlet_status(outlet) - return 'outlet_on' in status[0] and status[0]['outlet_on'] == expect_status