Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[action] [PR:15972] Update tests to handle PSUs with multiple PDUs #16319

Merged
merged 1 commit into from
Jan 3, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions tests/common/helpers/psu_helpers.py
Original file line number Diff line number Diff line change
@@ -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
95 changes: 51 additions & 44 deletions tests/platform_tests/test_platform_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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 = {}
Expand All @@ -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()
Expand All @@ -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],
Expand Down
60 changes: 28 additions & 32 deletions tests/snmp/test_snmp_phy_entity.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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
Loading