131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
"""Unit tests for the Chronos NAS-mediated cron provider (Phase 4D).
|
|
|
|
All NAS calls are mocked — ZERO live network. These prove:
|
|
- is_available is config-only (no network), false without config.
|
|
- one-shot arming sends the right provision payload (incl. sub-minute fires —
|
|
the agent owns the time, so there's no 1-minute floor).
|
|
- reconcile arms missing, cancels orphaned, skips paused.
|
|
- fire_due re-arms the next one-shot after a successful run, and repeat-N
|
|
(job gone) stops re-arming.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_home(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
|
yield tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def chronos(monkeypatch):
|
|
"""A ChronosCronScheduler with a fake NAS client capturing calls."""
|
|
from plugins.cron_providers.chronos import ChronosCronScheduler
|
|
|
|
class FakeClient:
|
|
def __init__(self):
|
|
self.provisions = []
|
|
self.cancels = []
|
|
self._armed = []
|
|
|
|
def provision(self, *, job_id, fire_at, agent_callback_url, dedup_key):
|
|
self.provisions.append({
|
|
"job_id": job_id, "fire_at": fire_at,
|
|
"agent_callback_url": agent_callback_url, "dedup_key": dedup_key,
|
|
})
|
|
return {"schedule_id": f"sched-{job_id}"}
|
|
|
|
def cancel(self, *, job_id):
|
|
self.cancels.append(job_id)
|
|
return {}
|
|
|
|
def list_armed(self):
|
|
return list(self._armed)
|
|
|
|
prov = ChronosCronScheduler()
|
|
fake = FakeClient()
|
|
prov._client = fake
|
|
# callback_url is read via _cfg; patch the module helper to avoid config.
|
|
monkeypatch.setattr("plugins.cron_providers.chronos._cfg",
|
|
lambda *k, default="": "https://agent.example/" if k[-1] == "callback_url" else "https://portal.test")
|
|
return prov, fake
|
|
|
|
|
|
# -- is_available -------------------------------------------------------------
|
|
|
|
def test_is_available_false_without_config(temp_home, monkeypatch):
|
|
from plugins.cron_providers.chronos import ChronosCronScheduler
|
|
|
|
monkeypatch.setattr("plugins.cron_providers.chronos._cfg", lambda *k, default="": "")
|
|
assert ChronosCronScheduler().is_available() is False
|
|
|
|
|
|
# -- arming -------------------------------------------------------------------
|
|
|
|
def test_arm_one_shot_sends_provision(chronos):
|
|
prov, fake = chronos
|
|
prov._arm_one_shot({"id": "j1", "next_run_at": "2026-06-18T12:00:00+00:00"})
|
|
|
|
assert len(fake.provisions) == 1
|
|
p = fake.provisions[0]
|
|
assert p["job_id"] == "j1"
|
|
assert p["fire_at"] == "2026-06-18T12:00:00+00:00"
|
|
assert p["dedup_key"] == "j1:2026-06-18T12:00:00+00:00"
|
|
assert p["agent_callback_url"] == "https://agent.example/"
|
|
|
|
|
|
def test_register_job_arms_only_the_created_job(chronos):
|
|
prov, fake = chronos
|
|
job = {"id": "created", "next_run_at": "2026-06-18T12:00:00+00:00"}
|
|
|
|
prov.register_job(job)
|
|
|
|
assert [p["job_id"] for p in fake.provisions] == ["created"]
|
|
|
|
|
|
def test_register_job_propagates_provision_failure(chronos):
|
|
prov, fake = chronos
|
|
|
|
def fail_provision(**kwargs):
|
|
raise RuntimeError("provision rejected")
|
|
|
|
fake.provision = fail_provision
|
|
|
|
with pytest.raises(RuntimeError, match="provision rejected"):
|
|
prov.register_job(
|
|
{"id": "created", "next_run_at": "2026-06-18T12:00:00+00:00"}
|
|
)
|
|
|
|
|
|
# -- reconcile ----------------------------------------------------------------
|
|
|
|
def test_reconcile_arms_all_enabled(temp_home, chronos, monkeypatch):
|
|
prov, fake = chronos
|
|
jobs = [
|
|
{"id": "a", "enabled": True, "next_run_at": "2026-06-18T12:00:00+00:00", "state": "scheduled"},
|
|
{"id": "b", "enabled": True, "next_run_at": "2026-06-18T12:05:00+00:00", "state": "scheduled"},
|
|
]
|
|
monkeypatch.setattr("cron.jobs.load_jobs", lambda: jobs)
|
|
monkeypatch.setattr("cron.jobs.get_job", lambda jid: next(j for j in jobs if j["id"] == jid))
|
|
|
|
prov.reconcile()
|
|
assert {p["job_id"] for p in fake.provisions} == {"a", "b"}
|
|
assert fake.cancels == []
|
|
|
|
|
|
# -- fire_due re-arm ----------------------------------------------------------
|
|
|
|
def test_fire_due_rearms_next_oneshot(chronos, monkeypatch):
|
|
prov, fake = chronos
|
|
# super().fire_due runs the job; stub the ABC default to "ran".
|
|
monkeypatch.setattr("cron.scheduler_provider.CronScheduler.fire_due",
|
|
lambda self, jid, **kw: True)
|
|
monkeypatch.setattr("cron.jobs.get_job",
|
|
lambda jid: {"id": jid, "enabled": True, "next_run_at": "2026-06-18T12:05:00+00:00"})
|
|
|
|
assert prov.fire_due("j1") is True
|
|
assert [p["job_id"] for p in fake.provisions] == ["j1"]
|
|
assert fake.provisions[0]["fire_at"] == "2026-06-18T12:05:00+00:00"
|
|
|