[action] [PR:735] Add vendor-independent leak test API

sonic_platform_base/leakage_sensor_test_base.py

@@ -0,0 +1,97 @@
+"""
+    leakage_sensor_test_base.py
+
+    Abstract base class for implementing a platform-specific leak test, used
+    to inject a simulated leak into the leak detection path in SONiC
+"""
+
+from .liquid_cooling_base import LeakSeverity
+
+
+class LeakageSensorTestBase:
+    """
+    Platform-specific leak test interface.
+
+    Provides a vendor-independent way to inject a simulated leak so that the
+    leak reporting path can be validated without wetting hardware. Injection is
+    non-destructive: a leak reported through this interface is flagged as a
+    test leak and must not cause any mitigation action to be taken.
+
+    Reached through LiquidCoolingBase.get_leak_sensor_test().
+
+    Injected state must live wherever the platform's LeakageSensorBase reads
+    its leak status from, such as a hardware test bit, the BMC or sysfs, so
+    that is_leak(), get_leak_severity() and is_test_leak() on the sensor all
+    report the injection and a separate process polling the platform observes
+    it. Withdrawing an injection must return the sensor to reporting its
+    hardware state rather than restoring a snapshot taken at injection time,
+    since a genuine leak may have started in the meantime.
+
+    Sensor names passed to this interface must be unique on the platform.
+    """
+
+    def is_leak_test_supported(self) -> bool:
+        """
+        Retrieves whether this platform supports leak test injection.
+
+        The base implementation returns True: reaching an instance through
+        LiquidCoolingBase.get_leak_sensor_test() already implies support, and
+        a platform without support signals it by returning None from
+        get_leak_sensor_test(). Platforms with conditional support (e.g.
+        dependent on a BMC feature) may override.
+
+        Returns:
+            bool: True if leak test injection is supported, False otherwise
+        """
+        return True
+
+    def set_test_leak(self, sensor_name: str, enable: bool, *,
+                      severity: LeakSeverity = LeakSeverity.MINOR) -> bool:
+        """
+        Injects or withdraws a test leak on the given sensor.
+
+        Idempotent: the return value reports whether the sensor is left in the
+        requested state, not whether a change was made. Injecting on a sensor
+        that already carries an injection, or withdrawing from one that is
+        already clear, returns True.
+
+        Args:
+            sensor_name: name of the leak sensor to inject the leak on
+            enable: True to inject the test leak, False to withdraw it
+            severity: severity the injected leak is reported with. Keyword
+                only, so that a platform override declaring it positionally
+                cannot change the calling convention, and defaulting to MINOR
+                so an injection that omits it does not select the severity
+                mapped to the most destructive mitigation action. Ignored
+                when enable is False.
+
+        Returns:
+            bool: True if the sensor is left in the requested test leak state,
+            False otherwise. An unknown sensor_name must return False rather
+            than raise.
+        """
+        raise NotImplementedError
+
+    def is_test_leak_enabled(self, sensor_name: str) -> bool:
+        """
+        Retrieves whether a test leak is currently injected on the given sensor.
+
+        Args:
+            sensor_name: name of the leak sensor
+
+        Returns:
+            bool: True if a test leak is injected on the sensor, False
+            otherwise. An unknown sensor_name must return False rather than
+            raise.
+        """
+        raise NotImplementedError
+
+    def clear_test_leaks(self) -> bool:
+        """
+        Withdraws every injected test leak on the platform.
+
+        Returns:
+            bool: True if no test leak remains injected on any sensor,
+            including when none was injected to begin with, False otherwise
+        """
+        raise NotImplementedError

sonic_platform_base/liquid_cooling_base.py

@@ -7,12 +7,17 @@
 
 from abc import ABC, abstractmethod
 from enum import Enum
-from typing import Dict, List
+from typing import Dict, List, TYPE_CHECKING
 
 from . import device_base
 from .sensor_base import SensorBase
 import sys
 
+if TYPE_CHECKING:
+    # Imported for type annotations only: a runtime import would be circular,
+    # as leakage_sensor_test_base imports LeakSeverity from this module.
+    from .leakage_sensor_test_base import LeakageSensorTestBase
+
 
 class LeakSeverity(Enum):
     MINOR    = "MINOR"
@@ -35,6 +40,7 @@ def __init__(self,
         self.leak_type: str = type or 'unknown'
         self.leak_location: str = location or 'unknown'
         self.leak_severity = severity
+        self.test_leak = False
 
     def get_name(self) -> str:
         """
@@ -50,6 +56,11 @@ def is_leak(self) -> bool:
         Retrieves the leak status of the sensor.
         The platform should apply debounce logic before reporting/clearing leak.
 
+        Implementations that read hardware must record the result in
+        self.leaking and must return a bool. The other accessors on this class
+        read self.leaking rather than calling is_leak() again, so one is_leak()
+        call per poll gives a consistent view of the sensor.
+
         Returns:
             bool: True if leak is detected, False if not
         """
@@ -82,21 +93,37 @@ def get_leak_sensor_location(self) -> str:
         """
         return self.leak_location
 
-    def get_leak_severity(self) -> LeakSeverity:
+    def get_leak_severity(self) -> LeakSeverity|None:
         """
         Retrieves the severity of leak
 
         Returns:
-            LeakSeverity: LeakSeverity.CRITICAL or LeakSeverity.MINOR, or None if no leak
+            LeakSeverity: LeakSeverity.CRITICAL or LeakSeverity.MINOR, or None
+            if no leak
+        """
+        return self.leak_severity if self.leaking else None
+
+    def is_test_leak(self) -> bool:
+        """
+        Retrieves whether the leak currently reported by this sensor originates
+        from a test injection rather than from the hardware.
+
+        A test leak is observable end to end (it is published like any other
+        leak) but must never be used to trigger a mitigation action.
+
+        Returns:
+            bool: True if the reported leak was injected by a leak test,
+                  False otherwise
         """
-        return self.leak_severity
+        return bool(self.leaking and self.test_leak)
 
     def get_leak_profile(self):
         """
         Returns the leak sensor profile associated with this sensor.
         """
         raise NotImplementedError
 
+
 class LeakSensorProfileBase(ABC):
     """
     Platform-specific leak sensor profile, which defines APIs pre leaksensor type
@@ -130,12 +157,13 @@ class LiquidCoolingBase(device_base.DeviceBase):
 
     def __init__(self,
                  leakage_sensors_num: int = 0,
-                 leakage_sensors_list: List[LeakageSensorBase] = [],
+                 leakage_sensors_list: List[LeakageSensorBase]|None = None,
                  *,
-                 profiles: List[LeakSensorProfileBase] = []):
-        self.leakage_sensors: List[LeakageSensorBase] = leakage_sensors_list
+                 profiles: List[LeakSensorProfileBase]|None = None):
+        self.leakage_sensors: List[LeakageSensorBase] = \
+            leakage_sensors_list if leakage_sensors_list is not None else []
         self.profiles: Dict[str, LeakSensorProfileBase] = {
-            p.get_type(): p for p in profiles
+            p.get_type(): p for p in (profiles or [])
         }
 
     def get_num_leak_sensors(self) -> int:
@@ -191,11 +219,22 @@ def get_all_profiles(self) -> List[LeakSensorProfileBase]:
         """
         return list(self.profiles.values())
 
+    def get_leak_sensor_test(self) -> 'LeakageSensorTestBase|None':
+        """
+        Retrieves the leak test interface of this platform.
+
+        Returns:
+            LeakageSensorTestBase: the leak test interface defined in
+            leakage_sensor_test_base, or None if the platform does not support
+            leak test injection
+        """
+        return None
+
     def get_profile(self, type: str) -> LeakSensorProfileBase|None:
         """
         Retrives the profile with the given name.
         """
-        profile = getattr(self.profiles, type, None)
+        profile = self.profiles.get(type)
 
         if profile is None:
             sys.stderr.write(f"Leakage sensor profile {type} doesn't exist")