test(cron): characterize in-process + desktop ticker contract before provider refactor

This commit is contained in:
Ben 2026-06-18 13:08:21 +10:00
parent c1f9eb0ec4
commit a657397769

View file

@ -0,0 +1,83 @@
"""Characterization tests for the cron trigger before/after the provider refactor.
These lock the CURRENT in-process-ticker contract (Phase 0 of the pluggable
CronScheduler plan, .hermes/plans/cron-scheduler-provider-interface.md). They
must pass unchanged on `main` now, and after every subsequent phase of the
refactor they are the regression harness that proves the built-in firing
behavior is byte-for-byte preserved when the ticker is moved behind the
CronScheduler provider interface.
No production code is exercised beyond the two ticker entry points:
- gateway/run.py::_start_cron_ticker (production gateway ticker)
- hermes_cli/web_server.py::_start_desktop_cron_ticker (desktop fallback)
Both call `cron.scheduler.tick(...)` on a loop and exit when their stop_event
is set. We patch `cron.scheduler.tick` (both tickers import it locally as
`cron_tick`, so the module-attribute patch is observed) and assert the loop
drives it and stops promptly.
"""
import threading
import time
from unittest.mock import patch
def test_ticker_calls_tick_at_least_once_then_stops():
"""The gateway in-process ticker loop calls cron.scheduler.tick repeatedly
and exits promptly once the stop_event is set."""
from gateway.run import _start_cron_ticker
calls = []
stop = threading.Event()
def fake_tick(*args, **kwargs):
calls.append(kwargs)
return 0
with patch("cron.scheduler.tick", side_effect=fake_tick):
# interval=0 keeps the loop tight; stop after a brief beat.
t = threading.Thread(
target=_start_cron_ticker,
args=(stop,),
kwargs={"interval": 0},
daemon=True,
)
t.start()
time.sleep(0.2)
stop.set()
t.join(timeout=5)
assert not t.is_alive(), "ticker did not exit after stop_event was set"
assert len(calls) >= 1, "ticker never called tick()"
# Contract: the ticker invokes tick with sync=False (fire-and-forget from
# the background thread, never the synchronous CLI path).
assert calls[0].get("sync") is False
def test_desktop_ticker_calls_tick_then_stops():
"""The desktop dashboard ticker loop calls cron.scheduler.tick and exits
once the stop_event is set. Desktop has no live adapters, so it ticks with
no adapters/loop."""
from hermes_cli.web_server import _start_desktop_cron_ticker
calls = []
stop = threading.Event()
def fake_tick(*args, **kwargs):
calls.append(kwargs)
return 0
with patch("cron.scheduler.tick", side_effect=fake_tick):
t = threading.Thread(
target=_start_desktop_cron_ticker,
args=(stop,),
kwargs={"interval": 0},
daemon=True,
)
t.start()
time.sleep(0.2)
stop.set()
t.join(timeout=5)
assert not t.is_alive(), "desktop ticker did not exit after stop_event was set"
assert len(calls) >= 1, "desktop ticker never called tick()"
assert calls[0].get("sync") is False