211 lines
9.6 KiB
Python
211 lines
9.6 KiB
Python
"""Tests for cronjob action='run' immediate execution (#41037).
|
|
|
|
Before this fix, `cronjob(action='run')` only set next_run_at=now and returned
|
|
success, relying on the scheduler ticker to actually run the job. With no
|
|
gateway/ticker active (e.g. a CLI-only Windows setup) the job never executed and
|
|
last_run_at stayed null forever. Now action='run' claims the job (at-most-once,
|
|
blocking a concurrent tick) and fires it inline via the shared run_one_job body.
|
|
|
|
#76502: the inline fire is synchronous, so while it runs it fires a heartbeat
|
|
into the calling agent's activity tracker — otherwise the gateway inactivity
|
|
watchdog kills the parent turn at ~1800s.
|
|
"""
|
|
import json
|
|
import sys
|
|
import threading
|
|
import time
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from tools.cronjob_tools import cronjob, _execute_job_now
|
|
from tools.environments.base import set_activity_callback
|
|
|
|
|
|
_JOB = {"id": "job-run-1", "name": "manual run", "prompt": "hi",
|
|
"schedule": {"kind": "cron", "expr": "0 9 * * *"}}
|
|
|
|
|
|
class TestCronjobRunExecutesImmediately:
|
|
def test_run_action_claims_and_fires_via_run_one_job(self):
|
|
"""action='run' must claim the job then fire it through run_one_job."""
|
|
ran = {"job": "after-run", "last_status": "ok", "last_error": None}
|
|
with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \
|
|
patch("tools.cronjob_tools.claim_job_for_fire", return_value=True) as m_claim, \
|
|
patch("cron.scheduler.run_one_job", return_value=True) as m_run, \
|
|
patch("tools.cronjob_tools.get_job", return_value=ran):
|
|
out = json.loads(cronjob(action="run", job_id="job-run-1"))
|
|
|
|
assert out["success"] is True
|
|
assert out["job"]["executed"] is True
|
|
assert out["job"]["execution_success"] is True
|
|
m_claim.assert_called_once_with("job-run-1") # at-most-once claim taken
|
|
m_run.assert_called_once() # fired via the shared body
|
|
|
|
|
|
def test_execute_job_now_bails_without_claim(self):
|
|
"""_execute_job_now never calls run_one_job when the claim is lost."""
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \
|
|
patch("cron.scheduler.run_one_job") as m_run:
|
|
res = _execute_job_now(dict(_JOB))
|
|
assert res["claimed"] is False
|
|
assert res["success"] is False
|
|
m_run.assert_not_called()
|
|
|
|
def test_execute_job_now_passes_live_gateway_context_to_delivery(self):
|
|
"""Manual runs must deliver on the live gateway adapter's owning loop."""
|
|
adapters = {"matrix": object()}
|
|
gateway_loop = object()
|
|
runner = SimpleNamespace(adapters=adapters, _gateway_loop=gateway_loop)
|
|
completed = {"id": "job-run-1", "last_status": "ok", "last_error": None}
|
|
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
|
|
patch("gateway.run._gateway_runner_ref", return_value=runner), \
|
|
patch("cron.scheduler.run_one_job", return_value=True) as m_run, \
|
|
patch("tools.cronjob_tools.get_job", return_value=completed):
|
|
res = _execute_job_now(dict(_JOB))
|
|
|
|
assert res["success"] is True
|
|
m_run.assert_called_once_with(
|
|
_JOB,
|
|
adapters=adapters,
|
|
loop=gateway_loop,
|
|
extra_prompt=None,
|
|
)
|
|
|
|
def test_execute_job_now_remains_standalone_without_gateway(self):
|
|
"""CLI-only runs retain the standalone delivery path."""
|
|
completed = {"id": "job-run-1", "last_status": "ok", "last_error": None}
|
|
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
|
|
patch.dict(sys.modules, {"gateway.run": None}), \
|
|
patch("cron.scheduler.run_one_job", return_value=True) as m_run, \
|
|
patch("tools.cronjob_tools.get_job", return_value=completed):
|
|
res = _execute_job_now(dict(_JOB))
|
|
|
|
assert res["success"] is True
|
|
m_run.assert_called_once_with(_JOB, adapters=None, loop=None, extra_prompt=None)
|
|
|
|
def test_execute_job_now_marks_failure_on_exception(self):
|
|
"""An exception during fire is captured, marked failed, not propagated."""
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
|
|
patch("cron.scheduler.run_one_job", side_effect=RuntimeError("boom")), \
|
|
patch("tools.cronjob_tools.mark_job_run") as m_mark, \
|
|
patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)):
|
|
res = _execute_job_now(dict(_JOB))
|
|
assert res["claimed"] is True
|
|
assert res["success"] is False
|
|
assert "boom" in res["error"]
|
|
m_mark.assert_called_once()
|
|
|
|
def test_execute_job_now_heartbeats_while_job_runs(self):
|
|
"""A manual run ticks the caller's activity tracker while the job
|
|
executes so the gateway inactivity watchdog doesn't kill the parent
|
|
turn (#76502)."""
|
|
touches = []
|
|
heartbeat_seen = threading.Event()
|
|
|
|
def record(desc):
|
|
touches.append(desc)
|
|
heartbeat_seen.set()
|
|
|
|
set_activity_callback(record)
|
|
try:
|
|
def slow_run(job, **kw):
|
|
# Deterministic: block until at least one heartbeat has fired
|
|
# (bounded so a broken heartbeat can't hang the test).
|
|
assert heartbeat_seen.wait(timeout=5.0), "no heartbeat within 5s"
|
|
return True
|
|
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
|
|
patch("tools.cronjob_tools._CRON_RUN_HEARTBEAT_INTERVAL", 0.05), \
|
|
patch("cron.scheduler.run_one_job", side_effect=slow_run) as m_run, \
|
|
patch("tools.cronjob_tools.get_job",
|
|
return_value={"last_status": "ok", "last_error": None}):
|
|
res = _execute_job_now(dict(_JOB))
|
|
|
|
m_run.assert_called_once()
|
|
assert res["success"] is True, res
|
|
assert any("cronjob: running job" in t for t in touches), touches
|
|
finally:
|
|
set_activity_callback(None)
|
|
|
|
def test_execute_job_now_without_callback_does_not_heartbeat(self):
|
|
"""No activity callback registered (direct callers, tests) → the
|
|
heartbeat thread is never started and behavior is unchanged."""
|
|
set_activity_callback(None)
|
|
try:
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
|
|
patch("cron.scheduler.run_one_job", return_value=True) as m_run, \
|
|
patch("tools.cronjob_tools.get_job",
|
|
return_value={"last_status": "ok", "last_error": None}), \
|
|
patch("tools.cronjob_tools.threading.Thread") as m_thread:
|
|
res = _execute_job_now(dict(_JOB))
|
|
assert res["success"] is True
|
|
m_run.assert_called_once()
|
|
m_thread.assert_not_called() # heartbeat thread truly never created
|
|
finally:
|
|
set_activity_callback(None)
|
|
|
|
def test_heartbeat_stops_at_ceiling_but_job_completes(self):
|
|
"""Past _CRON_RUN_HEARTBEAT_CEILING the heartbeat stops (so the
|
|
gateway watchdog regains authority over a wedged run) while the job
|
|
itself keeps running to completion."""
|
|
touches = []
|
|
first_beat = threading.Event()
|
|
|
|
def record(desc):
|
|
touches.append(desc)
|
|
first_beat.set()
|
|
|
|
set_activity_callback(record)
|
|
try:
|
|
def slow_run(job, **kw):
|
|
# Ceiling=0 → the very first wake stops the loop without
|
|
# touching. Give it a couple of cycles to prove silence.
|
|
time.sleep(0.2)
|
|
return True
|
|
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
|
|
patch("tools.cronjob_tools._CRON_RUN_HEARTBEAT_INTERVAL", 0.05), \
|
|
patch("tools.cronjob_tools._CRON_RUN_HEARTBEAT_CEILING", 0.0), \
|
|
patch("cron.scheduler.run_one_job", side_effect=slow_run), \
|
|
patch("tools.cronjob_tools.get_job",
|
|
return_value={"last_status": "ok", "last_error": None}):
|
|
res = _execute_job_now(dict(_JOB))
|
|
assert res["success"] is True, res
|
|
assert not first_beat.is_set(), touches # heartbeat never fired
|
|
finally:
|
|
set_activity_callback(None)
|
|
|
|
def test_heartbeat_survives_callback_exception(self):
|
|
"""One raising callback must not silently kill watchdog protection
|
|
for the rest of a long job — the loop continues heartbeating."""
|
|
calls = []
|
|
second_beat = threading.Event()
|
|
|
|
def flaky(desc):
|
|
calls.append(desc)
|
|
if len(calls) >= 2:
|
|
second_beat.set()
|
|
if len(calls) == 1:
|
|
raise RuntimeError("transient")
|
|
|
|
set_activity_callback(flaky)
|
|
try:
|
|
def slow_run(job, **kw):
|
|
# Block until a heartbeat AFTER the raising one has fired.
|
|
assert second_beat.wait(timeout=5.0), \
|
|
"heartbeat stopped after one callback exception"
|
|
return True
|
|
|
|
with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
|
|
patch("tools.cronjob_tools._CRON_RUN_HEARTBEAT_INTERVAL", 0.05), \
|
|
patch("cron.scheduler.run_one_job", side_effect=slow_run), \
|
|
patch("tools.cronjob_tools.get_job",
|
|
return_value={"last_status": "ok", "last_error": None}):
|
|
res = _execute_job_now(dict(_JOB))
|
|
assert res["success"] is True, res
|
|
assert len(calls) >= 2, calls
|
|
finally:
|
|
set_activity_callback(None)
|