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).

The one I'd want resolved before merge is on is_test_leak(): it has no callers, so an injected leak is indistinguishable from a physical one everywhere downstream, and system_critical_leak_action defaults to power_off. 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.

🤖 Generated with Claude Code

[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]

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]

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]

This is the one I'd want resolved before merge. is_test_leak() is new and currently has no callers anywhere. Injection sets the sensor's leaking and leak_severity, so an injected leak is published through get_leak_sensor_status() exactly like a physical one, and nothing downstream distinguishes the two.

That matters because the leak policy has teeth: system_critical_leak_action defaults to power_off (see show/platform.py and config/liquid_cool.py in sonic-utilities). So the "non-destructive" property test_injection_is_non_destructive asserts holds inside the Python object graph, but on a running system the drill would drive the same mitigation a real leak does. Is the policy enforcer being updated to consult is_test_leak() in the same release? If not, injection probably needs a gate rather than relying on every future consumer to remember to filter.

Separately, the annotation is -> bool, but self.is_leak() and self.test_leak returns whatever is_leak() returned when that's falsy — 'N/A' or None on a vendor read failure, not False. assert sensor.is_test_leak() == False in the conformance suite then fails on None with a message pointing at the test rather than the sensor read error. Worth wrapping in bool().

[nikamirrr]

This is a pytest-shaped suite living inside the runtime package, so setup.py's packages list installs 173 lines of test code into site-packages on every device.

More importantly, every check is a bare assert. Under python -O or with PYTHONOPTIMIZE set, all of them are stripped and the entire conformance suite passes vacuously — a platform shipping a completely broken set_test_leak() would still come back green. Publishing this via a tests extra or a separate package, and/or using explicit raises, would avoid both.

[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]

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]

leakage_sensor_test_base.py says platforms with conditional support (e.g. dependent on a BMC feature) may override is_leak_test_supported() to return False. This asserts it's True, so such a platform hard-fails here with no skip path — and the other seven tests fail too, since injection genuinely isn't available.

Either skip the suite when it returns False (pytest.skip, or a class-level opt-out), or drop the method, since as written it can never legally be False.

[nikamirrr]

This requires a withdrawal on an already-cleared sensor to return True, but the abstract docstring only says "True if the test leak state was applied, False otherwise". A vendor reading that literally returns False when there was nothing to withdraw — a defensible implementation of the written contract — and fails here. Same implied requirement for clear_test_leaks() on line 172 when nothing is injected.

The idempotency requirement should probably be stated in the base docstring rather than only enforced by the test.

[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.

[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.