hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix(cron): adapt delivery-notice tests to the return_job claim API

Browse Source 
Main grew claim_job_for_fire(job_id, return_job=True) — a claimed
snapshot dict instead of a bool — while this branch sat on an older
base. The merge-ref CI ran the hybrid: the wiring tests still mocked
return_value=True, which fails isinstance(claimed_job, dict) and fell
into the 'already being fired' branch, so every dispatch assert failed.

Mock the claim to return the job snapshot (the API's success shape),
read the summary's deliver from the claimed snapshot the run actually
executes, and keep the dispatch-result failure renderer. Rebased onto
current main; cron suite 710 passed.
This commit is contained in:
赵桂雄
2026-08-16 17:57:41 +08:00
committed by Teknium
parent fd387c15eb
commit 2f58cbfa7f
2 changed files with 38 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
tests/tools/test_cronjob_run_delivery_notice.py
View File

@@ -91,6 +91,19 @@ def _bound_session_key(key):
_approval_session_key.reset(token)
def _dispatch_diag(res) -> str:
"""Failure renderer for the wiring tests' dispatch asserts: the result
dict plus the scheduler running set, so a broken assert names the return
path that was actually taken instead of a bare KeyError."""
try:
from cron.scheduler import get_running_job_ids
running = sorted(get_running_job_ids())
except Exception as e: # pragma: no cover - diagnostic only
running = f"<unavailable: {e}>"
return f"dispatch result: {res!r}; running: {running}"
def _drain_completion_event(delegation_id):
"""Wait (bounded) for this delegation's completion event; requeue others.
@@ -175,9 +188,13 @@ class TestRunnerSummaryWiring:
def test_delivery_failure_surfaces_in_completion_summary(self):
from tools.cronjob_tools import _try_dispatch_background_run
job = _job("job-dn-01", "telegram")
with _bound_session_key("agent:main:telegram:dm:83993"):
with (
patch("tools.cronjob_tools.claim_job_for_fire", return_value=True),
patch(
"tools.cronjob_tools.claim_job_for_fire",
return_value=job, # claimed snapshot (return_job=True API)
),
patch("cron.scheduler.run_one_job", return_value=True),
patch(
"tools.cronjob_tools.get_job",
@@ -188,8 +205,8 @@ class TestRunnerSummaryWiring:
},
),
):
res = _try_dispatch_background_run(_job("job-dn-01", "telegram"))
assert res["dispatched"] is True
res = _try_dispatch_background_run(job)
assert res.get("dispatched") is True, _dispatch_diag(res)
evt = _drain_completion_event(res["delegation_id"])
assert evt is not None, "completion event never reached the queue"
summary = evt.get("summary") or ""
@@ -204,17 +221,21 @@ class TestRunnerSummaryWiring:
by a delivered-there claim."""
from tools.cronjob_tools import _try_dispatch_background_run
job = _job("job-dn-03", "")
with _bound_session_key("agent:main:telegram:dm:86622"):
with (
patch("tools.cronjob_tools.claim_job_for_fire", return_value=True),
patch(
"tools.cronjob_tools.claim_job_for_fire",
return_value=job, # claimed snapshot (return_job=True API)
),
patch("cron.scheduler.run_one_job", return_value=True),
patch(
"tools.cronjob_tools.get_job",
return_value={"last_status": "ok", "last_error": None},
),
):
res = _try_dispatch_background_run(_job("job-dn-03", ""))
assert res["dispatched"] is True
res = _try_dispatch_background_run(job)
assert res.get("dispatched") is True, _dispatch_diag(res)
evt = _drain_completion_event(res["delegation_id"])
assert evt is not None, "completion event never reached the queue"
summary = evt.get("summary") or ""
@@ -224,17 +245,21 @@ class TestRunnerSummaryWiring:
def test_delivery_success_wording_unchanged_in_completion_summary(self):
from tools.cronjob_tools import _try_dispatch_background_run
job = _job("job-dn-02", "telegram")
with _bound_session_key("agent:main:telegram:dm:83994"):
with (
patch("tools.cronjob_tools.claim_job_for_fire", return_value=True),
patch(
"tools.cronjob_tools.claim_job_for_fire",
return_value=job, # claimed snapshot (return_job=True API)
),
patch("cron.scheduler.run_one_job", return_value=True),
patch(
"tools.cronjob_tools.get_job",
return_value={"last_status": "ok", "last_error": None},
),
):
res = _try_dispatch_background_run(_job("job-dn-02", "telegram"))
assert res["dispatched"] is True
res = _try_dispatch_background_run(job)
assert res.get("dispatched") is True, _dispatch_diag(res)
evt = _drain_completion_event(res["delegation_id"])
assert evt is not None, "completion event never reached the queue"
summary = evt.get("summary") or ""

6
tools/cronjob_tools.py
View File

@@ -1360,10 +1360,12 @@ def _try_dispatch_background_run(
started_at = time.time()
# Canonicalize with the scheduler's own normalizer so the summary states
# the same target fire time will use: falsy ("", stored JSON null) reads
# "local", legacy list-form deliver flattens to its comma string.
# "local", legacy list-form deliver flattens to its comma string. Read
# from the claimed snapshot — the owner-bearing record the run actually
# executes — not the pre-claim `job` the tool loaded.
from cron.scheduler import _normalize_deliver_value
deliver = _normalize_deliver_value(job.get("deliver", "local"))
deliver = _normalize_deliver_value(claimed_job.get("deliver", "local"))
def _runner() -> Dict[str, Any]:
res = _run_claimed_job(claimed_job, extra_prompt=extra_prompt)