diff --git a/sonic_platform_base/liquid_cooling_base.py b/sonic_platform_base/liquid_cooling_base.py new file mode 100644 index 000000000..e1873c138 --- /dev/null +++ b/sonic_platform_base/liquid_cooling_base.py @@ -0,0 +1,88 @@ +""" + liquid_cooling_base.py + + Abstract base class for implementing a platform-specific class with which + to interact with a liquid cooling module in SONiC +""" + +from . import device_base +from .sensor_base import SensorBase +import sys + +class LeakageSensorBase(SensorBase): + def __init__(self, name): + self.name = name + self.leaking = False + + def get_name(self): + """ + Retrieves the name of the leakage sensor + + Returns: + tring: the name of the leakage sensor + """ + return self.name + + def is_leak(self): + """ + Retrieves the leak status of the sensor + + Returns: + bool: True if leak is detected, False if not + """ + return self.leaking + +class LiquidCoolingBase(device_base.DeviceBase): + """ + Base class for implementing liquid cooling system + """ + + def __init__(self, leakage_sensors_num = 0, leakage_sensors_list = None): + self.leakage_sensors_num = leakage_sensors_num + self.leakage_sensors = leakage_sensors_list if leakage_sensors_list else [] + + def get_num_leak_sensors(self): + """ + Retrieves the number of leakage sensors + + Returns: + int: The number of leakage sensors + """ + return self.leakage_sensors_num + + def get_leak_sensor(self, index): + """ + Retrieves the leakage sensor by index + """ + sensor = None + + try: + sensor = self.leakage_sensors[index] + except IndexError: + sys.stderr.write("Leakage sensor index {} out of range (0-{})\n".format( + index, len(self.leakage_sensors)-1)) + + return sensor + + def get_all_leak_sensors(self): + """ + Retrieves the list of leakage sensors + + Returns: + list: A list of leakage sensor names + """ + return self.leakage_sensors + + def get_leak_sensor_status(self): + """ + Retrieves the leak status of the sensors + + Returns: + list: A list of leakage sensor names that are leaking, empty list if no leakage + """ + leaking_sensors = [] + for sensor in self.leakage_sensors: + if sensor.is_leak(): + leaking_sensors.append(sensor) + return leaking_sensors + diff --git a/tests/liquid_cooling_base_test.py b/tests/liquid_cooling_base_test.py new file mode 100644 index 000000000..2f6e078fc --- /dev/null +++ b/tests/liquid_cooling_base_test.py @@ -0,0 +1,62 @@ +''' +Test liquid_cooling_base module +''' +import unittest +import sys +from io import StringIO +from sonic_platform_base.liquid_cooling_base import LeakageSensorBase +from sonic_platform_base.liquid_cooling_base import LiquidCoolingBase + +class TestLeakageSensorBase: + ''' + Collection of LeakageSensorBase test methods + ''' + + @staticmethod + def test_leakage_sensor_base_init(): + ''' + Test leakage sensor base default implementation + ''' + leakage_sensor = LeakageSensorBase("test_sensor") + + assert leakage_sensor.get_name() == "test_sensor" + assert leakage_sensor.is_leak() == False + +class TestLiquidCoolingBase(): + ''' + Collection of LiquidCoolingBase test methods + ''' + + @staticmethod + def test_liquid_cooling_base_init(): + ''' + Test liquid cooling base default implementation + ''' + liquid_cooling = LiquidCoolingBase() + + assert liquid_cooling.get_num_leak_sensors() == 0 + assert liquid_cooling.get_all_leak_sensors() == [] + assert liquid_cooling.get_leak_sensor_status() == [] + + @staticmethod + def test_get_leak_sensor_out_of_range(): + ''' + Test get_leak_sensor method with an out-of-range index + ''' + liquid_cooling = LiquidCoolingBase() + liquid_cooling.leakage_sensors = [LeakageSensorBase("Sensor1"), + LeakageSensorBase("Sensor2"), + LeakageSensorBase("Sensor3")] + + # Redirect stderr to capture error message + captured_output = StringIO() + sys.stderr = captured_output + + # Call get_leak_sensor and check the return value + sensor = liquid_cooling.get_leak_sensor(5) + assert sensor == None + + captured_output.seek(0) + error_message = captured_output.read() + assert "Leakage sensor index 5 out of range (0-2)" in error_message +