Skip to content

Commit

Permalink
Refactor test_power function to improve readability
Browse files Browse the repository at this point in the history
  • Loading branch information
@smagarwal-arista
smagarwal-arista committed Oct 17, 2024
1 parent 6928d80 commit 8b85062
Showing 1 changed file with 26 additions and 45 deletions.
71 changes: 26 additions & 45 deletions tests/platform_tests/api/test_psu.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,16 @@ def skip_absent_psu(self, psu_num, platform_api_conn):
return True
return False

def get_psu_parameter(self, psu_info, psu_parameter, get_data, message):
data = None
is_supported = self.get_psu_facts(psu_info["duthost"], psu_info["psu_id"], True, psu_parameter)
if is_supported:
data = get_data(psu_info["api"], psu_info["psu_id"])
if self.expect(data is not None, "Failed to retrieve {} of PSU {}".format(message, psu_info["psu_id"])):
self.expect(isinstance(data, float), "PSU {} {} appears incorrect".format(psu_info["psu_id"], message))

return data

#
# Functions to test methods inherited from DeviceBase class
#
Expand Down Expand Up @@ -205,44 +215,24 @@ def test_power(self, duthosts, enum_rand_one_per_hwsku_hostname, localhost, plat
''' PSU power test '''
duthost = duthosts[enum_rand_one_per_hwsku_hostname]
skip_release_for_platform(duthost, ["202012", "201911", "201811"], ["arista"])

for psu_id in range(self.num_psus):
failure_count = self.get_len_failed_expectations()
failure_occured = False
psu_info = {
"duthost": duthost,
"api": platform_api_conn,
"psu_id": psu_id
}
for i in range(0, MAX_ATTEMPTS):
name = psu.get_name(platform_api_conn, psu_id)
if name in self.psu_skip_list:
logger.info("skipping check for {}".format(name))
else:
voltage = None
voltage_supported = self.get_psu_facts(duthost, psu_id, True, "voltage")
if voltage_supported:
voltage = psu.get_voltage(platform_api_conn, psu_id)
if self.expect(voltage is not None, "Failed to retrieve voltage of PSU {}".format(psu_id)):
self.expect(isinstance(voltage, float), "PSU {} voltage appears incorrect".format(psu_id))
current = None
current_supported = self.get_psu_facts(duthost, psu_id, True, "current")
if current_supported:
current = psu.get_current(platform_api_conn, psu_id)
if self.expect(current is not None, "Failed to retrieve current of PSU {}".format(psu_id)):
self.expect(isinstance(current, float), "PSU {} current appears incorrect".format(psu_id))
power = None
power_supported = self.get_psu_facts(duthost, psu_id, True, "power")
if power_supported:
power = psu.get_power(platform_api_conn, psu_id)
if self.expect(power is not None, "Failed to retrieve power of PSU {}".format(psu_id)):
self.expect(isinstance(power, float), "PSU {} power appears incorrect".format(psu_id))
max_supp_power = None
max_power_supported = self.get_psu_facts(duthost, psu_id, True, "max_power")
if max_power_supported:
max_supp_power = psu.get_maximum_supplied_power(platform_api_conn, psu_id)
if self.expect(max_supp_power is not None,
"Failed to retrieve maximum supplied power of PSU {}".format(psu_id)):
self.expect(isinstance(max_supp_power, float),
"PSU {} maximum supplied power appears incorrect".format(psu_id))
voltage = self.get_psu_parameter(psu_info, "voltage", psu.get_voltage, "voltage")
current = self.get_psu_parameter(psu_info, "current", psu.get_current, "current")
power = self.get_psu_parameter(psu_info, "power", psu.get_power, "power")

failure_occured = self.get_len_failed_expectations() > failure_count

if current is not None and voltage is not None and power is not None:
is_within_tolerance = abs(power - (voltage*current)) < power*0.1
if not failure_occured and not is_within_tolerance and i < MAX_ATTEMPTS - 1:
Expand All @@ -251,28 +241,19 @@ def test_power(self, duthosts, enum_rand_one_per_hwsku_hostname, localhost, plat
self.expect(is_within_tolerance, "PSU {} reading does not make sense \
(power:{}, voltage:{}, current:{})".format(psu_id, power, voltage, current))

self.get_psu_parameter(psu_info, "max_power", psu.get_maximum_supplied_power,
"maximum supplied power")

powergood_status = psu.get_powergood_status(platform_api_conn, psu_id)
if self.expect(powergood_status is not None,
"Failed to retrieve operational status of PSU {}".format(psu_id)):
self.expect(powergood_status is True, "PSU {} is not operational".format(psu_id))

high_threshold = None
voltage_high_threshold_supported = self.get_psu_facts(duthost, psu_id, True,
"voltage_high_threshold")
if voltage_high_threshold_supported:
high_threshold = psu.get_voltage_high_threshold(platform_api_conn, psu_id)
if self.expect(high_threshold is not None,
"Failed to retrieve the high voltage threshold of PSU {}".format(psu_id)):
self.expect(isinstance(high_threshold, float),
"PSU {} voltage high threshold appears incorrect".format(psu_id))
low_threshold = None
voltage_low_threshold_supported = self.get_psu_facts(duthost, psu_id, True, "voltage_low_threshold")
if voltage_low_threshold_supported:
low_threshold = psu.get_voltage_low_threshold(platform_api_conn, psu_id)
if self.expect(low_threshold is not None,
"Failed to retrieve the low voltage threshold of PSU {}".format(psu_id)):
self.expect(isinstance(low_threshold, float),
"PSU {} voltage low threshold appears incorrect".format(psu_id))
high_threshold = self.get_psu_parameter(psu_info, "voltage_high_threshold",
psu.get_voltage_high_threshold, "high voltage threshold")
low_threshold = self.get_psu_parameter(psu_info, "voltage_low_threshold",
psu.get_voltage_low_threshold, "low voltage threshold")

if high_threshold is not None and low_threshold is not None:
self.expect(voltage < high_threshold and voltage > low_threshold,
"Voltage {} of PSU {} is not in between {} and {}"
Expand Down

0 comments on commit 8b85062

Please sign in to comment.