test(cron): cover cron_list/status/tick/create CLI helpers

Salvaged from #40430; re-verified on main, tightened, tested.

Co-authored-by: xuezhaolan <xuezhaolan@users.noreply.github.com>
This commit is contained in:
teknium1 2026-06-06 08:50:13 -07:00 committed by Teknium
parent 7e7e3af5b0
commit 3ba5ba89c2

View file

@ -1,10 +1,12 @@
"""Tests for hermes_cli.cron command handling."""
from argparse import Namespace
from types import SimpleNamespace
import pytest
from cron.jobs import create_job, get_job, list_jobs
from hermes_cli import cron as cron_cli
from hermes_cli.cron import cron_command
@ -56,6 +58,11 @@ class TestCronCommandLifecycle:
skill=None,
skills=["maps", "blogwatcher"],
clear_skills=False,
add_skills=None,
remove_skills=None,
script=None,
workdir=None,
no_agent=None,
)
)
updated = get_job(job["id"])
@ -76,6 +83,11 @@ class TestCronCommandLifecycle:
skill=None,
skills=None,
clear_skills=True,
add_skills=None,
remove_skills=None,
script=None,
workdir=None,
no_agent=None,
)
)
cleared = get_job(job["id"])
@ -96,6 +108,9 @@ class TestCronCommandLifecycle:
repeat=None,
skill=None,
skills=["blogwatcher", "maps"],
script=None,
workdir=None,
no_agent=False,
)
)
out = capsys.readouterr().out
@ -265,3 +280,125 @@ class TestExternalCronProviderStatus:
out = capsys.readouterr().out
assert "Created job" in out
assert "Gateway is not running" not in out
def test_cron_list_warns_when_gateway_not_running(monkeypatch, capsys):
monkeypatch.setattr(
"cron.jobs.list_jobs",
lambda include_disabled=False: [
{
"id": "job-1",
"name": "Nightly docs",
"schedule_display": "every day",
"state": "scheduled",
"enabled": True,
"next_run_at": "2026-06-01T00:00:00Z",
"deliver": ["local"],
}
],
)
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [])
monkeypatch.setattr(cron_cli, "_active_cron_provider_name", lambda: "builtin")
cron_cli.cron_list()
out = capsys.readouterr().out
assert "Gateway is not running" in out
assert "Nightly docs" in out
def test_cron_status_reports_running_gateway(monkeypatch, capsys):
monkeypatch.setattr(cron_cli, "_active_cron_provider_name", lambda: "builtin")
monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [1234, 5678])
monkeypatch.setattr(
"cron.jobs.list_jobs",
lambda include_disabled=False: [
{"next_run_at": "2026-06-01T00:00:00Z"},
{"next_run_at": "2026-05-31T12:00:00Z"},
],
)
cron_cli.cron_status()
out = capsys.readouterr().out
assert "Gateway is running" in out
assert "1234, 5678" in out
assert "2 active job(s)" in out
assert "2026-05-31T12:00:00Z" in out
def test_cron_tick_invokes_scheduler_tick_with_verbose(monkeypatch):
calls = []
monkeypatch.setattr("cron.scheduler.tick", lambda verbose=False: calls.append(verbose))
cron_cli.cron_tick()
assert calls == [True]
def test_cron_create_success_prints_job_details(monkeypatch, capsys):
monkeypatch.setattr(
cron_cli,
"_cron_api",
lambda **kwargs: {
"success": True,
"job_id": "job-1",
"name": "Nightly docs",
"schedule": "every day",
"skills": ["docs"],
"next_run_at": "2026-06-01T00:00:00Z",
"job": {
"script": "scripts/build_docs.py",
"no_agent": True,
"workdir": "/tmp/repo",
},
},
)
monkeypatch.setattr(cron_cli, "_warn_if_gateway_not_running", lambda: None)
args = SimpleNamespace(
schedule="every day",
prompt="refresh docs",
name="Nightly docs",
deliver=None,
repeat=None,
skill="docs",
skills=None,
script="scripts/build_docs.py",
workdir="/tmp/repo",
no_agent=True,
)
rc = cron_cli.cron_create(args)
out = capsys.readouterr().out
assert rc == 0
assert "Created job: job-1" in out
assert "Skills: docs" in out
assert "Script: scripts/build_docs.py" in out
assert "Mode: no-agent" in out
assert "Workdir: /tmp/repo" in out
assert "Next run: 2026-06-01T00:00:00Z" in out
def test_cron_create_failure_returns_nonzero(monkeypatch, capsys):
monkeypatch.setattr(cron_cli, "_cron_api", lambda **kwargs: {"success": False, "error": "boom"})
args = SimpleNamespace(
schedule="every day",
prompt="refresh docs",
name=None,
deliver=None,
repeat=None,
skill=None,
skills=None,
script=None,
workdir=None,
no_agent=False,
)
rc = cron_cli.cron_create(args)
out = capsys.readouterr().out
assert rc == 1
assert "Failed to create job: boom" in out