Add vendor-independent leak test API

[chinmoy-nexthop]

Add a common API for injecting a simulated leak into the leak detection path, so the reporting chain can be validated without wetting hardware.

Injection is non-destructive: an injected leak is published like any other leak, and is additionally flagged through LeakageSensorBase is_test_leak() so consumers must not take a mitigation action on it.

• leakage_sensor_test_base.py: new LeakageSensorTestBase defining is_leak_test_supported(), set_test_leak(), is_test_leak_enabled() and clear_test_leaks()
• liquid_cooling_base.py: add is_test_leak() on LeakageSensorBase and get_leak_sensor_test() on LiquidCoolingBase, defaulting to False and None so platforms without injection support are unaffected
• tests/leak_test_api_base.py: LeakTestApiBase, a reusable conformance suite a platform subclasses to validate its implementation against the common contract

Description

Motivation and Context

How Has This Been Tested?

Additional Information (Optional)

[mssonicbld]

/azp run

[azure-pipelines]

Azure Pipelines:
There may be pipelines that require an authorized user to comment /azp run to run.

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[judyjoseph]

@fraserg-arista to review as well, this could be first step on sonic-net/SONiC#2441

[nikamirrr]

(Duplicate submission from a retried API call — superseded by the review below, which carries all 15 inline comments. Please disregard this entry.)

[nikamirrr]

Automated review pass focused on correctness. Fifteen inline comments below.

Findings 1, 2, 6, 7, 8, 9, 10, 11, 14 and 15 were confirmed by executing the code rather than
by reading it — including running the full LeakTestApiBase suite against both a correct and a
deliberately-sloppy implementation.

The through-line is the PR's central safety promise: an injected leak is published like a real one
but must never trigger a mitigation action. Three independent gaps each break it on their own:

1. is_test_leak() is decoupled from is_leak(), so a stale flag exempts a real leak from
   mitigation.
2. The reference clear_test_leaks() erases real leaks on sensors that were never injected.
3. The flag reaches no consumer and no STATE_DB field, while set_test_leak defaults to CRITICAL
   and system_critical_leak_action defaults to power_off — so a "non-destructive" test can power
   the switch off.

The conformance suite cannot catch any of the three: I built an implementation that is wrong in
exactly the dangerous way and it passes 8/8.

Also worth a look, below the cut: a shared mutable default leakage_sensors_list=[] in
LiquidCoolingBase.__init__; test_injection_is_non_destructive being a tautology that passes when
injection is a no-op; and @abstractmethod ... pass against this repo's own written rule in
.github/copilot-instructions.md ("Abstract methods: Raise NotImplementedError in base class"),
which also makes super().set_test_leak(...) silently return a falsy None.

[mssonicbld]

This PR has backport request label(s) for branch(es): msft-202608, but is missing required test information. Please make sure you tick the tested branch(es) in the Tested branch section and provide test evidence (e.g., 202608: <test result>) in the Test result section as well in your PR description.

_{---Powered by SONiC BuildBot}

[mssonicbld]

/azp run

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[chinmoy-nexthop]

(Duplicate submission from a retried API call — superseded by the review below, which carries all 15 inline comments. Please disregard this entry.)

@nikamirrr Thank you.
It is absolutely fine. Maybe your agent was a little too verbose. I have looked into your feedback and taken care of a few things that I felt were appropriate.

Addressed the following:-

• test_leak is now a class attribute, so is_test_leak() is safe on subclasses that don't chain super().init()
• is_test_leak() only reports True while the sensor is actually leaking, so a stale flag can't mask a real leak
• get_leak_severity() returns None when there's no leak, as documented
• Fixed get_profile() — it was using getattr() on a dict and could never find a profile
• Added a TYPE_CHECKING import so the LeakageSensorTestBase annotation resolves
• Moved the conformance suite into sonic_platform_base/ so it ships in the wheel, and added an in-repo subclass so it runs in CI
• Suite now clears injected leaks on teardown even when an assertion fails, fails loudly if SENSOR_NAMES isn't set, and verifies is_test_leak() is False after withdrawal
• Reference clear_test_leaks() only touches injected sensors, restores prior sensor state, and returns the aggregated result

[mssonicbld]

Hi, there are workflow run(s) waiting for approval, you may be first-time contributor. I will notify maintainers to help approve once PR is approved. Thanks!

_{---Powered by SONiC BuildBot}

[nikamirrr]

Code review

Reviewed the new leak test API. Most of this is contract and robustness feedback rather than anything CI will catch — I ran tests/liquid_cooling_base_test.py and tests/leakage_sensor_test_base_test.py against 9ef9f23 and they pass (43 tests, all 8 conformance tests collected and green). Details inline.

One finding isn't on a changed line, so I couldn't leave it inline. LiquidCoolingBase.__init__ takes mutable default arguments and assigns them straight to instance state:

sonic-platform-common/sonic_platform_base/liquid_cooling_base.py

Lines 155 to 161 in 9ef9f23

	def __init__(self,
	leakage_sensors_num: int = 0,
	leakage_sensors_list: List[LeakageSensorBase] = [],
	*,
	profiles: List[LeakSensorProfileBase] = []):
	self.leakage_sensors: List[LeakageSensorBase] = leakage_sensors_list

Every instance constructed without explicit lists shares one list object. This is pre-existing, but the new test_get_profile is the first thing to exercise it — it does LiquidCoolingBase(profiles=[profile]), whose self.leakage_sensors is the shared default. Anything that later appends to a get_all_leak_sensors() result would leak sensors into every other instance in the process. = None plus or [] closes it.

[nikamirrr]

This is the only LeakageSensorBase attribute declared as a class-level default — leaking, leak_sensor_ok, leak_type, leak_location and leak_severity are all set in __init__. The justification in the comment (safe on vendor subclasses that don't chain super().__init__()) applies equally to those five, and the new test_is_test_leak_without_init_chaining has to hand-set all of them precisely because they aren't class defaults. That leaves two initialization conventions in one class, and it means LeakageSensorBase.test_leak = True (class rather than instance) flips every sensor in the process. Setting it in __init__ alongside the rest would be more consistent.

[nikamirrr]

Double checked, please consider adding it to init as well

[nikamirrr]

No objection to returning None — the previous docstring already promised it, so this makes the implementation match. The concern is that get_leak_severity() now calls is_leak(), which on vendor subclasses is a hardware read rather than an attribute lookup:

• platform/mellanox/mlnx-platform-api/sonic_platform/liquid_cooling.py reads the sysfs file and mutates self.leaking as a side effect.
• platform/aspeed/sonic-platform-modules-nvidia-bmc/ast2700/sonic_platform/leakage_sensor.py re-reads the channel input file on every call.

Two consequences. A consumer doing get_leak_sensor_status() and then get_leak_severity() re-reads hardware between the two, so if the leak cleared in between it gets None and .value on the result raises. And Mellanox's is_leak() returns the string 'N/A' on a failed read, which is truthy — so an unreadable sensor now reports its last-known severity rather than None. Reading self.leaking preserves the intent without the extra I/O or the truthiness trap.

[nikamirrr]

Concrete suggestion, since the cleanest fix touches the is_leak() contract rather than just this line.

What the three implementations do today:

	is_leak()	keeps self.leaking in sync?	get_leak_severity()	is_test_leak()
LeakageSensorBase	return self.leaking	n/a	base	base
mellanox LeakageSensor	sysfs read	yes on the 1/0 branches, no on the error branch (returns 'N/A')	base	base
aspeed LeakageSensor	sysfs read	never	overridden	base

Option 1 - tighten the is_leak() contract, then read the cached state.

    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
        """
        return self.leaking

    def get_leak_severity(self) -> LeakSeverity|None:
        return self.leak_severity if self.leaking else None

    def is_test_leak(self) -> bool:
        return bool(self.leaking and self.test_leak)

get_leak_sensor_status() already calls is_leak() on every sensor, so that call is the refresh and everything downstream then reads a consistent snapshot. This removes the extra hardware read, the TOCTOU window and the 'N/A'-is-truthy trap in one go, and makes the -> bool annotation true.

The cost is that it needs two one-line vendor follow-ups landing alongside it, because neither implementation satisfies the tightened contract today:

# mellanox .../sonic_platform/liquid_cooling.py - return a bool on read failure
        else:
            logger.error(f"Failed to read leakage sensor {self.name} value: {content}")
            self.leaking = False
            return False

# aspeed .../ast2700/sonic_platform/leakage_sensor.py - record the reading
    def is_leak(self):
        self.leaking, _, _ = self._check_channel_value()
        return self.leaking

Without the aspeed one, is_test_leak() would read a self.leaking that aspeed never sets and always return False - a silent regression versus the code as it stands here.

Option 2 - if you would rather not couple this PR to the vendor repos: drop the state dependence and revert to return self.leak_severity, with the docstring describing it as a static property of the sensor rather than of the current leak. get_leak_sensor_status() already tells a consumer which sensors are leaking, so the severity of a sensor obtained from it is unambiguous. Zero extra I/O, no vendor coupling, and consistent with get_leak_sensor_type() / get_leak_sensor_location(), which are all plain attribute reads.

Option 1 is the more correct model, option 2 is the smaller change - either resolves what I flagged.

[nikamirrr]

Double checked, please consider either approach to keep self.leaking and self.leak_severity in sync

[nikamirrr]

Same read-amplification point as above: with both this and get_leak_severity() routing through is_leak(), classifying a single leak now costs three hardware reads on aspeed where it previously cost one — and the three readings aren't guaranteed consistent with each other. Caching one evaluation per poll, or reading self.leaking, avoids it.

[nikamirrr]

same as above. Double checked, please consider either approach to keep self.leaking and self.leak_severity in sync

[nikamirrr]

The annotation is -> bool, but self.is_leak() and self.test_leak returns whatever is_leak() returned when that value is falsy. Neither vendor implementation trips this today - mellanox returns a real False on its no-leak branch and aspeed returns a bool - so this is latent rather than an active bug. But a subclass whose is_leak() returns a falsy non-bool (None, 0) would make is_test_leak() return that value, and assert sensor.is_test_leak() == False in the conformance suite then fails on None with a message pointing at the test rather than at the sensor read error. Wrapping in bool() costs nothing.

(To correct an earlier version of this comment: mellanox's 'N/A' is truthy, so it yields self.test_leak here rather than escaping as a non-bool. The truthiness problem shows up in get_leak_severity() instead, which I flagged above.)

[nikamirrr]

More like a nit, make sure it returns the correct bool, set the return type of is_leak() etc

[nikamirrr]

Verification in this module is 23 bare assert statements against a single raise. assert is the one statement the compiler is allowed to delete: under python -O, or with PYTHONOPTIMIZE set in the environment, asserts are not compiled into the bytecode at all. Running a reduced copy of assert_no_leaks() against a deliberately broken sensor:

normal interpreter   ->  FAILED (assertion fired)
python -O            ->  PASSED
PYTHONOPTIMIZE=1     ->  PASSED

assert_no_leaks: 21 bytecode instructions normally, 3 under -O

The helper's body is gone. Since assert_no_leaks() is the shared check behind several tests, most of the suite's coverage goes with it, and a platform whose set_test_leak() does nothing at all would still produce a green conformance run.

A file under tests/ is only ever run by a developer typing pytest, which never sets -O. This one ships in the wheel and runs inside vendor build systems and CI harnesses, where an optimized interpreter is entirely plausible - so it is worth not depending on that.

Suggested fix - one helper, and route the checks through it:

    @staticmethod
    def _check(condition, message):
        '''
        Raises on failure. Deliberately not an assert: this module runs in
        vendor environments where the interpreter may be started with -O,
        which strips assert statements entirely.
        '''
        if not condition:
            raise AssertionError(message)

so that, for example:

    def assert_no_leaks(self):
        self._check(self.liquid_cooling.get_leak_sensor_status() == [],
                    "sensors report a leak before injection")
        for sensor in self.liquid_cooling.get_all_leak_sensors():
            self._check(sensor.is_leak() == False,
                        f"{sensor.get_name()} reports a leak")
            self._check(sensor.is_test_leak() == False,
                        f"{sensor.get_name()} reports a test leak")
        for name in self.SENSOR_NAMES:
            self._check(self.leak_test.is_test_leak_enabled(name) == False,
                        f"{name} has an armed injection")

raise survives -O, and naming the sensor in the message beats a bare assert's repr when a platform is debugging its own implementation.

[nikamirrr]

Double checked, please see if need to WA the bare assert to make sure the error is raised even in the optimized python environment

[nikamirrr]

No check that this returned a non-None object. The base get_leak_sensor_test() returns None (liquid_cooling_base.py:228), and teardown_method on line 74 already guards if self.leak_test is not None — so the case is known to be reachable.

A platform that subclasses this without overriding get_leak_sensor_test() gets six 'NoneType' object has no attribute ... failures with no indication of the actual cause. An assert here, next to the SENSOR_NAMES one, would say what's actually wrong.

[nikamirrr]

Consider asserting that self.leak_test is not None

[nikamirrr]

The return value is discarded. If clear_test_leaks() returns False because one sensor's cleanup failed, the suite still reports success, the next test's setup_method runs against a switch with an injected leak still latched, and after the last test the platform is left reporting a CRITICAL leak.

The docstring claims "a failing test cannot leave the platform reporting a leak" — asserting this return value (or at minimum logging it) is what makes that true.

[nikamirrr]

Double checked, please consider asserting the returned value

[nikamirrr]

leakage_sensor_test_base.py says a platform with conditional support may override is_leak_test_supported() to return False. This asserts it is True, so such a platform hard-fails here - and so do the other seven tests, since injection genuinely is not available.

Correcting my own earlier suggestion of pytest.skip: this module deliberately does not import pytest. unittest.SkipTest is stdlib and pytest honours it natively (verified on 9.1.1), so raising it from setup_method skips all eight in one place:

        if not self.leak_test.is_leak_test_supported():
            raise unittest.SkipTest(
                "platform reports leak test injection is not supported")

test_leak_test_interface_exposed then keeps only its isinstance check. teardown_method correctly does not run on a skip, and raise survives -O.

[nikamirrr]

Double checked. Consider using s unittest.SkipTest rather than asserting is_leak_test_supported is True

[nikamirrr]

Line 161 is a second withdrawal on an already-clear sensor. The abstract contract says only "True if the test leak state was applied", and "applied" reads two ways: post-condition (no-op returns True, what this requires) or transition (no-op returns False, what if sensor.test_leak == enable: return False gives). The docstring is explicit about the neighbouring unknown-sensor case, so the omission looks deliberate to a vendor.

Correcting myself: I said the same applied to clear_test_leaks() at line 172, but that call does have injections to clear. The no-op case is in teardown_method, which passes today only because it discards the return value - so this couples to my line 74 comment.

Suggested wording in leakage_sensor_test_base.py:

        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 is already injected, or withdrawing from one that is already
        clear, must return True.

and for clear_test_leaks(): "True if no test leak remains injected on any sensor, including when none was injected to begin with."

[nikamirrr]

Double checked. Please resolve the ambiguity, most likely it the docstring,
True -> if the final state of the sensor is what is requested
False -> if not

Not indicating the sensor state change

[nikamirrr]

The injection state is split across two objects: test_leak lives on the sensor, _saved lives on the PlatformLeakTest. Nothing in get_leak_sensor_test()'s contract requires returning a stable instance, so a platform that builds a fresh LiquidCooling (and thus a fresh leak-test object) over cached or shared sensor objects — e.g. Chassis().get_liquid_cooling() — reaches this pop with test_leak set and an empty _saved, and the KeyError propagates out of LeakTestApiBase.teardown_method.

Since this class is presented as the reference that platforms will copy, the split-state design propagates with it. Keeping the saved state on the sensor, or self._saved.pop(sensor_name, None), would close it.

[nikamirrr]

Double checked, please consider moving _saved to the sensor, rather then keeping in PlatformLeakTest

[judyjoseph]

Hi @chinmoy-nexthop I feel we really don't need the test infra which you are defining here in this class LeakTestApiBase. This can be defined and used as needed from the sonic-mgmt test framework - can leave it to platform test writer to how to call the platform APIs to test leak. Let me know your thoughts

The other changes in sonic_platform_base/leak_sensor_test_base.py, sonic_platform_base/liquid_cooling_base.py is defined as we discussed in community REF: sonic-net/SONiC#2441.

[chinmoy-nexthop]

Absolutely!
Make sense @judyjoseph we should use this sonic-net/SONiC#2441 .

[parvathi-nexthop]

Thanks for the changes. The capability model in #2441 looks right to me, and there's one case worth making sure it keeps covering: platforms where the injected severity isn't selectable. On those, the severity a sensor reports is a fixed property of the sensor, so injection can assert a leak but not choose the severity it comes back as. #2441 handles this already — "a fixed-severity binary sensor can expose only LEAK_INJECTION" — but the set_test_leak() signature in this PR can't express it, since severity defaults to CRITICAL and is applied unconditionally. Keeping the capability negotiation would be worth it for that alone.

One case the design doesn't cover yet: injection mechanisms that can hold only one injection at a time. get_test_capabilities() is per-sensor with no concurrency dimension, so there's no defined answer for "inject on B while A is injected" — it would either need advertising, or a defined failure. Worth pinning down before platforms implement it, since test code that assumes it can inject on several sensors at once would silently mean different things on different hardware. Nexthop expects to have a platform in this shape which is why I'm rather than leaving theoretical.