test(cron): pin lock-first liveness + harden lock-probe failure

Follow-ups to the salvaged #95947 cron commit:

- Wrap the lock probe in its own try/except: a crashing probe is
  'unknown', not 'dead' — the pid scan still decides instead of the
  whole tri-state collapsing to None.
- Regression tests (shape adapted from #94155 by @liuhao1024): lock
  held + empty pid scan -> alive (the reported false alarm); lock
  inactive -> pid-scan fallback both ways; crashing lock probe still
  falls back.
- patch_liveness now pins the lock probe inactive by default so the
  pre-existing pid-scan tests stay deterministic on machines where a
  real gateway holds the real lock.
- contributors mapping for magnus.lundstedt@infidyne.com.

Co-authored-by: liuhao1024 <sunsky.lau@gmail.com>
This commit is contained in:
kshitijk4poor
2026-08-27 14:14:30 +05:30
committed by kshitij
parent b71c3cc13c
commit f2d043eb08
3 changed files with 94 additions and 7 deletions
@@ -0,0 +1 @@
magnuslundstedt
+9 -3
View File
@@ -83,10 +83,16 @@ def _builtin_gateway_liveness() -> Optional[bool]:
# — and inside the gateway process it short-circuits to True, so the in-gateway
# cron tool never emits a false "gateway not running" (find_gateway_pids can
# transiently miss the gateway just after a restart).
from gateway.status import is_gateway_runtime_lock_active
try:
from gateway.status import is_gateway_runtime_lock_active
if is_gateway_runtime_lock_active():
return True
if is_gateway_runtime_lock_active():
return True
except Exception:
# A crashing lock probe is "unknown", not "dead" — let the pid
# scan below still decide instead of collapsing the whole
# tri-state to None.
pass
from hermes_cli.gateway import find_gateway_pids
return bool(find_gateway_pids())
@@ -162,11 +162,20 @@ from contextlib import ExitStack
class _LivenessPatches:
"""Context manager patching the provider/gateway-pid probes."""
"""Context manager patching the provider/gateway-pid probes.
def __init__(self, *, provider, pids):
Also pins the gateway runtime lock probe to *inactive* by default so
these tests are deterministic even when a real gateway (holding the
real lock) runs on the developer's machine — the lock-first check in
``_builtin_gateway_liveness`` would otherwise short-circuit to True
and mask the pid-scan behavior under test. Pass ``lock_active=True``
to exercise the lock-first path itself.
"""
def __init__(self, *, provider, pids, lock_active=False):
self._provider = provider
self._pids = pids
self._lock_active = lock_active
def __enter__(self):
from unittest.mock import patch
@@ -190,11 +199,82 @@ class _LivenessPatches:
return_value=list(self._pids),
)
)
self._stack.enter_context(
patch(
"gateway.status.is_gateway_runtime_lock_active",
return_value=self._lock_active,
)
)
return self
def __exit__(self, *exc):
return self._stack.__exit__(*exc)
def patch_liveness(*, provider, pids):
return _LivenessPatches(provider=provider, pids=pids)
def patch_liveness(*, provider, pids, lock_active=False):
return _LivenessPatches(provider=provider, pids=pids, lock_active=lock_active)
class TestRuntimeLockFirstLiveness:
"""The gateway runtime lock is the primary liveness signal (#95947).
``find_gateway_pids`` can transiently return empty while the gateway is
up (right after a restart) and excludes the current PID by design
(#13242), so a single-process gateway probed as dead while its own
ticker was firing (#94143 class). The lock is held for exactly the
gateway's lifetime and short-circuits to True before the pid scan.
"""
def test_lock_active_reports_alive_despite_empty_pid_scan(self, hermes_env):
"""The reported false alarm: lock held, pid scan empty → alive."""
_create_job()
with patch_liveness(provider="builtin", pids=[], lock_active=True):
from tools.cronjob_tools import cronjob
result = json.loads(cronjob(action="list"))
assert result["success"] is True
assert result["gateway_running"] is True
assert "warning" not in result
def test_lock_inactive_falls_back_to_pid_scan(self):
from unittest.mock import patch
import hermes_cli.cron as cron_cli
with (
patch("hermes_cli.cron._active_cron_provider_name", return_value="builtin"),
patch("gateway.status.is_gateway_runtime_lock_active", return_value=False),
patch("hermes_cli.gateway.find_gateway_pids", return_value=[424242]),
):
assert cron_cli._builtin_gateway_liveness() is True
def test_no_lock_no_pids_is_false(self):
from unittest.mock import patch
import hermes_cli.cron as cron_cli
with (
patch("hermes_cli.cron._active_cron_provider_name", return_value="builtin"),
patch("gateway.status.is_gateway_runtime_lock_active", return_value=False),
patch("hermes_cli.gateway.find_gateway_pids", return_value=[]),
):
assert cron_cli._builtin_gateway_liveness() is False
def test_lock_probe_failure_still_falls_back(self):
"""A crashing lock probe must not poison the tri-state helper —
the pid scan still decides (the outer except returns None only
when both probes fail)."""
from unittest.mock import patch
import hermes_cli.cron as cron_cli
with (
patch("hermes_cli.cron._active_cron_provider_name", return_value="builtin"),
patch(
"gateway.status.is_gateway_runtime_lock_active",
side_effect=OSError("lock probe failed"),
),
patch("hermes_cli.gateway.find_gateway_pids", return_value=[424242]),
):
assert cron_cli._builtin_gateway_liveness() is True