Add new liquid cooling base object (#603)

Signed-off-by: Yuanzhe, Liu <[email protected]>

Author: yuazhe

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
+

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
+