diff --git a/sonic_platform_base/leakage_sensor_test_base.py b/sonic_platform_base/leakage_sensor_test_base.py new file mode 100644 index 0000000..0a059c9 --- /dev/null +++ b/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 diff --git a/sonic_platform_base/liquid_cooling_base.py b/sonic_platform_base/liquid_cooling_base.py index 070d635..16fd756 100644 --- a/sonic_platform_base/liquid_cooling_base.py +++ b/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,14 +93,29 @@ 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): """ @@ -97,6 +123,7 @@ def get_leak_profile(self): """ 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") diff --git a/tests/leakage_sensor_test_base_test.py b/tests/leakage_sensor_test_base_test.py new file mode 100644 index 0000000..11f7069 --- /dev/null +++ b/tests/leakage_sensor_test_base_test.py @@ -0,0 +1,256 @@ +''' +Test leakage_sensor_test_base module +''' +import pytest +from sonic_platform_base.leakage_sensor_test_base import LeakageSensorTestBase +from sonic_platform_base.liquid_cooling_base import LeakageSensorBase +from sonic_platform_base.liquid_cooling_base import LeakSeverity +from sonic_platform_base.liquid_cooling_base import LiquidCoolingBase + + +class PlatformSensor(LeakageSensorBase): + ''' + Sensor whose leak state is computed from a "hardware" reading combined + with any injection, standing in for a platform that reads a test bit + alongside the physical sensor. + + Keeping the injected state on the sensor, next to the hardware state it + overlays, means withdrawal re-derives the reported state from hardware + rather than restoring a snapshot taken at injection time — so a genuine + leak that arrives while an injection is armed survives the withdrawal. + ''' + + def __init__(self, name, **kwargs): + super().__init__(name, **kwargs) + self.hw_leak = False + self.hw_severity = LeakSeverity.CRITICAL + self.test_severity = None + + def refresh(self): + ''' + Recomputes the reported state from the hardware reading and the + injection, as a platform's is_leak() would on each poll. + ''' + if self.test_leak: + self.leaking = True + self.leak_severity = self.test_severity + else: + self.leaking = self.hw_leak + self.leak_severity = self.hw_severity + + def is_leak(self): + self.refresh() + return self.leaking + + +class PlatformLeakTest(LeakageSensorTestBase): + ''' + Reference leak test implementation. + + Injection state lives on the sensor object itself, and withdrawal lets the + sensor re-derive its state from hardware — so sensors that never carried + an injection are untouched, and a genuine leak present before, during or + after the test keeps its severity. + ''' + + def __init__(self, sensors): + self._sensors = {s.get_name(): s for s in sensors} + + def set_test_leak(self, sensor_name, enable, *, + severity=LeakSeverity.MINOR): + sensor = self._sensors.get(sensor_name) + if sensor is None: + return False + sensor.test_leak = enable + sensor.test_severity = severity if enable else None + sensor.refresh() + return True + + def is_test_leak_enabled(self, sensor_name): + sensor = self._sensors.get(sensor_name) + return bool(sensor and sensor.test_leak) + + def clear_test_leaks(self): + # A list comprehension rather than a generator, so a failed withdrawal + # does not stop the remaining sensors from being attempted + return all([self.set_test_leak(name, False) + for name, sensor in self._sensors.items() + if sensor.test_leak]) + + +class PlatformLiquidCooling(LiquidCoolingBase): + ''' + Reference liquid cooling implementation exposing the leak test interface. + ''' + + def __init__(self, sensors): + super().__init__(len(sensors), sensors) + self._leak_test = PlatformLeakTest(sensors) + + def get_leak_sensor_test(self): + return self._leak_test + + +class TestLeakageSensorTestBase: + ''' + Collection of LeakageSensorTestBase test methods + ''' + + @staticmethod + def test_injection_methods_not_implemented(): + ''' + Test the injection methods raise NotImplementedError until the + platform provides them + ''' + leak_test = LeakageSensorTestBase() + + with pytest.raises(NotImplementedError): + leak_test.set_test_leak("sensor1", True) + with pytest.raises(NotImplementedError): + leak_test.is_test_leak_enabled("sensor1") + with pytest.raises(NotImplementedError): + leak_test.clear_test_leaks() + + @staticmethod + def test_severity_is_keyword_only(): + ''' + Test severity cannot be passed positionally, so a platform override + cannot change the calling convention for callers that omit it + ''' + with pytest.raises(TypeError): + LeakageSensorTestBase().set_test_leak("sensor1", True, + LeakSeverity.CRITICAL) + + sensors = [PlatformSensor("sensor1")] + leak_test = PlatformLiquidCooling(sensors).get_leak_sensor_test() + + with pytest.raises(TypeError): + leak_test.set_test_leak("sensor1", True, LeakSeverity.CRITICAL) + + @staticmethod + def test_default_severity_is_minor(): + ''' + Test an injection that omits severity reports MINOR, so the least + thought path does not select the severity mapped to the most + destructive mitigation action + ''' + sensors = [PlatformSensor("sensor1")] + leak_test = PlatformLiquidCooling(sensors).get_leak_sensor_test() + + assert leak_test.set_test_leak("sensor1", True) == True + assert sensors[0].get_leak_severity() is LeakSeverity.MINOR + + @staticmethod + def test_set_test_leak_is_idempotent(): + ''' + Test the return value reports the resulting state rather than whether + a change was made: a repeated injection or withdrawal returns True + ''' + sensors = [PlatformSensor("sensor1")] + leak_test = PlatformLiquidCooling(sensors).get_leak_sensor_test() + + assert leak_test.set_test_leak("sensor1", True) == True + assert leak_test.set_test_leak("sensor1", True) == True + assert leak_test.set_test_leak("sensor1", False) == True + assert leak_test.set_test_leak("sensor1", False) == True + assert leak_test.clear_test_leaks() == True + + @staticmethod + def test_unknown_sensor_returns_false(): + ''' + Test an unknown sensor name is rejected rather than raising + ''' + sensors = [PlatformSensor("sensor1")] + leak_test = PlatformLiquidCooling(sensors).get_leak_sensor_test() + + assert leak_test.set_test_leak("no_such_sensor", True) == False + assert leak_test.is_test_leak_enabled("no_such_sensor") == False + + @staticmethod + def test_is_leak_test_supported_default(): + ''' + Test is_leak_test_supported defaults to True: reaching an instance + through get_leak_sensor_test() already implies support, and the + unsupported signal is get_leak_sensor_test() returning None + ''' + assert LeakageSensorTestBase().is_leak_test_supported() == True + + @staticmethod + def test_get_leak_sensor_test_default_none(): + ''' + Test LiquidCoolingBase reports no leak test interface by default, so + platforms without injection support are unaffected + ''' + liquid_cooling = LiquidCoolingBase(leakage_sensors_list=[]) + assert liquid_cooling.get_leak_sensor_test() is None + + @staticmethod + def test_injected_leak_is_flagged(): + ''' + Test an injected leak is published like any other leak and is + additionally flagged as a test leak, on a healthy sensor + ''' + sensors = [PlatformSensor("sensor1"), PlatformSensor("sensor2")] + liquid_cooling = PlatformLiquidCooling(sensors) + leak_test = liquid_cooling.get_leak_sensor_test() + + assert leak_test.set_test_leak("sensor1", True, + severity=LeakSeverity.CRITICAL) == True + assert leak_test.is_test_leak_enabled("sensor1") == True + + leaking = liquid_cooling.get_leak_sensor_status() + assert len(leaking) == 1 + assert leaking[0].get_name() == "sensor1" + assert leaking[0].is_test_leak() == True + assert leaking[0].get_leak_severity() is LeakSeverity.CRITICAL + assert leaking[0].is_leak_sensor_ok() == True + + assert leak_test.clear_test_leaks() == True + assert liquid_cooling.get_leak_sensor_status() == [] + assert sensors[0].is_test_leak() == False + + @staticmethod + def test_clear_test_leaks_preserves_real_leaks(): + ''' + Test withdrawing test leaks only touches sensors that carry an + injection: a genuine leak on another sensor must survive + clear_test_leaks() with its severity intact + ''' + sensors = [PlatformSensor("sensor1"), PlatformSensor("sensor2")] + liquid_cooling = PlatformLiquidCooling(sensors) + leak_test = liquid_cooling.get_leak_sensor_test() + + # A genuine leak is present on sensor2 + sensors[1].hw_leak = True + sensors[1].hw_severity = LeakSeverity.CRITICAL + + leak_test.set_test_leak("sensor1", True, severity=LeakSeverity.MINOR) + assert leak_test.clear_test_leaks() == True + + assert sensors[1].is_leak() == True + assert sensors[1].get_leak_severity() is LeakSeverity.CRITICAL + assert sensors[1].is_test_leak() == False + assert sensors[0].is_leak() == False + assert leak_test.is_test_leak_enabled("sensor1") == False + + @staticmethod + def test_withdrawal_rereads_hardware(): + ''' + Test withdrawal returns the sensor to reporting hardware state rather + than restoring a snapshot taken at injection time: a genuine leak that + arrives while the injection is armed must survive the withdrawal + ''' + sensors = [PlatformSensor("sensor1")] + liquid_cooling = PlatformLiquidCooling(sensors) + leak_test = liquid_cooling.get_leak_sensor_test() + + leak_test.set_test_leak("sensor1", True, severity=LeakSeverity.MINOR) + + # A genuine leak arrives while the injection is armed + sensors[0].hw_leak = True + sensors[0].hw_severity = LeakSeverity.CRITICAL + + assert leak_test.clear_test_leaks() == True + assert sensors[0].is_leak() == True + assert sensors[0].get_leak_severity() is LeakSeverity.CRITICAL + assert sensors[0].is_test_leak() == False diff --git a/tests/liquid_cooling_base_test.py b/tests/liquid_cooling_base_test.py index 8ad1913..fa66c3a 100644 --- a/tests/liquid_cooling_base_test.py +++ b/tests/liquid_cooling_base_test.py @@ -25,30 +25,117 @@ def test_leakage_sensor_base_init(): assert leakage_sensor.get_name() == "test_sensor" assert leakage_sensor.get_leak_sensor_type() == 'unknown' assert leakage_sensor.get_leak_sensor_location() == 'unknown' - assert leakage_sensor.get_leak_severity() == LeakSeverity.CRITICAL + assert leakage_sensor.get_leak_severity() is None assert leakage_sensor.is_leak() == False leakage_sensor2 = LeakageSensorBase("test_sensor2", type="rope", location="drip tray", severity=LeakSeverity.MINOR) - + assert leakage_sensor2.get_name() == "test_sensor2" assert leakage_sensor2.get_leak_sensor_type() == "rope" assert leakage_sensor2.get_leak_sensor_location() == "drip tray" - assert leakage_sensor2.get_leak_severity() == LeakSeverity.MINOR + assert leakage_sensor2.get_leak_severity() is None assert leakage_sensor2.is_leak() == False + # The configured severity is reported once the sensor leaks + leakage_sensor2.leaking = True + assert leakage_sensor2.get_leak_severity() is LeakSeverity.MINOR + @staticmethod def test_severity_constants(): ''' - Test LEAK_SEVERITY_CRITICAL and LEAK_SEVERITY_MINOR are LeakSeverity enum members, - and that the LeakSeverity enum has the expected string values. + Test LEAK_SEVERITY_CRITICAL and LEAK_SEVERITY_MINOR are LeakSeverity + enum members, and that the LeakSeverity enum has the expected string + values. ''' assert LeakSeverity.CRITICAL.value == "CRITICAL" assert LeakSeverity.MINOR.value == "MINOR" assert LeakageSensorBase.LEAK_SEVERITY_CRITICAL is LeakSeverity.CRITICAL assert LeakageSensorBase.LEAK_SEVERITY_MINOR is LeakSeverity.MINOR + @staticmethod + def test_is_test_leak_default(): + ''' + Test is_test_leak defaults to False, so a leak is treated as physical + unless the platform explicitly flags it as injected + ''' + sensor = LeakageSensorBase("sensor1") + assert sensor.is_test_leak() == False + + @staticmethod + def test_is_test_leak_injected(): + ''' + Test a leaking sensor can be flagged as carrying an injected leak + ''' + sensor = LeakageSensorBase("sensor1") + sensor.leaking = True + sensor.test_leak = True + assert sensor.is_leak() == True + assert sensor.is_test_leak() == True + + @staticmethod + def test_is_test_leak_false_when_not_leaking(): + ''' + Test a test_leak flag left set on a non-leaking sensor is not + reported: is_test_leak() describes the leak currently reported by the + sensor, so with no leak there is nothing to attribute to an injection + ''' + sensor = LeakageSensorBase("sensor1") + sensor.test_leak = True + assert sensor.is_test_leak() == False + + @staticmethod + def test_is_test_leak_without_init_chaining(): + ''' + Test is_test_leak is safe on a vendor subclass that does not chain + super().__init__() and reports no leak, since the flag is only read + once the sensor is leaking + ''' + class NoSuper(LeakageSensorBase): + def __init__(self, name): + self.name = name + self.leaking = False + self.leak_sensor_ok = True + self.leak_type = 'rope' + self.leak_location = 'front' + self.leak_severity = LeakSeverity.CRITICAL + + sensor = NoSuper("sensor1") + assert sensor.is_test_leak() == False + + @staticmethod + def test_accessors_follow_recorded_leak_state(): + ''' + Test get_leak_severity and is_test_leak read self.leaking, which an + is_leak() override that reads hardware must record, so one is_leak() + call per poll gives a consistent view without extra hardware reads + ''' + class HardwareSensor(LeakageSensorBase): + def __init__(self, name): + super().__init__(name, severity=LeakSeverity.MINOR) + self.hw_leak = False + + def is_leak(self): + # Reads "hardware" state and records it, as documented + self.leaking = self.hw_leak + return self.leaking + + sensor = HardwareSensor("sensor1") + assert sensor.is_leak() == False + assert sensor.get_leak_severity() is None + assert sensor.is_test_leak() == False + + # A genuine leak surfacing through the override carries its severity + sensor.hw_leak = True + assert sensor.is_leak() == True + assert sensor.get_leak_severity() is LeakSeverity.MINOR + assert sensor.is_test_leak() == False + + # An injected leak recorded by the override is flagged + sensor.test_leak = True + assert sensor.is_test_leak() == True + @staticmethod def test_is_leak_sensor_ok_default(): ''' @@ -72,7 +159,7 @@ def test_get_leak_sensor_type_default(): Test get_leak_sensor_type default value is None ''' sensor = LeakageSensorBase("sensor1") - assert sensor.get_leak_sensor_type() is 'unknown' + assert sensor.get_leak_sensor_type() == 'unknown' @staticmethod def test_get_leak_sensor_type_set(): @@ -90,7 +177,7 @@ def test_get_leak_sensor_location_default(): Test get_leak_sensor_location default value is None ''' sensor = LeakageSensorBase("sensor1") - assert sensor.get_leak_sensor_location() is 'unknown' + assert sensor.get_leak_sensor_location() == 'unknown' @staticmethod def test_get_leak_sensor_location_set(): @@ -107,7 +194,7 @@ def test_get_leak_severity_default(): Test get_leak_severity default value is None ''' sensor = LeakageSensorBase("sensor1") - assert sensor.get_leak_severity() is LeakSeverity.CRITICAL + assert sensor.get_leak_severity() is None @staticmethod def test_get_leak_severity_critical(): @@ -115,6 +202,7 @@ def test_get_leak_severity_critical(): Test get_leak_severity returns LeakSeverity.CRITICAL enum member ''' sensor = LeakageSensorBase("sensor1") + sensor.leaking = True sensor.leak_severity = LeakSeverity.CRITICAL assert sensor.get_leak_severity() is LeakSeverity.CRITICAL assert sensor.get_leak_severity().value == "CRITICAL" @@ -125,6 +213,7 @@ def test_get_leak_severity_minor(): Test get_leak_severity returns LeakSeverity.MINOR enum member ''' sensor = LeakageSensorBase("sensor1") + sensor.leaking = True sensor.leak_severity = LeakSeverity.MINOR assert sensor.get_leak_severity() is LeakSeverity.MINOR assert sensor.get_leak_severity().value == "MINOR" @@ -258,7 +347,7 @@ class TestLiquidCoolingBase(): ''' Collection of LiquidCoolingBase test methods ''' - + @staticmethod def test_liquid_cooling_base_init(): ''' @@ -279,7 +368,7 @@ def test_get_leak_sensor_out_of_range(): "severity": LeakSeverity.MINOR} liquid_cooling = LiquidCoolingBase(0, [LeakageSensorBase("Sensor1", **common), - LeakageSensorBase("Sensor2", **common), + LeakageSensorBase("Sensor2", **common), LeakageSensorBase("Sensor3", **common)]) # Redirect stderr to capture error message @@ -378,6 +467,33 @@ def test_get_leak_sensor_status_faulty_sensor_ignored(): ) assert liquid_cooling.get_leak_sensor_status() == [] + @staticmethod + def test_get_profile(): + ''' + Test get_profile returns the registered profile by type, and None + for an unknown type — including dict method names, which must not be + reachable through the type lookup + ''' + class RopeProfile(LeakSensorProfileBase): + def get_type(self): return "rope" + def get_leak_max_minor_duration_sec(self): + return 300 + + profile = RopeProfile() + liquid_cooling = LiquidCoolingBase(profiles=[profile]) + + captured_output = StringIO() + sys.stderr = captured_output + try: + assert liquid_cooling.get_profile("rope") is profile + assert liquid_cooling.get_profile("flood") is None + assert liquid_cooling.get_profile("items") is None + finally: + sys.stderr = sys.__stderr__ + + captured_output.seek(0) + assert "flood" in captured_output.read() + @staticmethod def test_liquid_cooling_with_sensor_profiles(): ''' @@ -416,4 +532,3 @@ def get_leak_profile(self): assert leaking[0].get_leak_sensor_location() == "front" assert leaking[0].get_leak_severity() is LeakSeverity.MINOR assert leaking[0].get_leak_profile().get_leak_max_minor_duration_sec() == 300 -