5d0f55cac4
Add an optional 'script' parameter to cron jobs that references a Python script. The script runs before each agent turn, and its stdout is injected into the prompt as context. This enables stateful monitoring — the script handles data collection and change detection, the LLM analyzes and reports. - cron/jobs.py: add script field to create_job(), stored in job dict - cron/scheduler.py: add _run_job_script() executor with timeout handling, inject script output/errors into _build_job_prompt() - tools/cronjob_tools.py: add script to tool schema, create/update handlers, _format_job display - hermes_cli/cron.py: add --script to create/edit, display in list/edit output - hermes_cli/main.py: add --script argparse for cron create/edit subcommands - tests/cron/test_cron_script.py: 20 tests covering job CRUD, script execution, path resolution, error handling, prompt injection, tool API Script paths can be absolute or relative (resolved against ~/.hermes/scripts/). Scripts run with a 120s timeout. Failures are injected as error context so the LLM can report the problem. Empty string clears an attached script.
301 lines
9.7 KiB
Python
301 lines
9.7 KiB
Python
"""Tests for cron job script injection feature.
|
|
|
|
Tests cover:
|
|
- Script field in job creation / storage / update
|
|
- Script execution and output injection into prompts
|
|
- Error handling (missing script, timeout, non-zero exit)
|
|
- Path resolution (absolute, relative to HERMES_HOME/scripts/)
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import sys
|
|
import textwrap
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
# Ensure project root is importable
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
|
|
@pytest.fixture
|
|
def cron_env(tmp_path, monkeypatch):
|
|
"""Isolated cron environment with temp HERMES_HOME."""
|
|
hermes_home = tmp_path / ".hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "cron").mkdir()
|
|
(hermes_home / "cron" / "output").mkdir()
|
|
(hermes_home / "scripts").mkdir()
|
|
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
|
|
|
# Clear cached module-level paths
|
|
import cron.jobs as jobs_mod
|
|
monkeypatch.setattr(jobs_mod, "HERMES_DIR", hermes_home)
|
|
monkeypatch.setattr(jobs_mod, "CRON_DIR", hermes_home / "cron")
|
|
monkeypatch.setattr(jobs_mod, "JOBS_FILE", hermes_home / "cron" / "jobs.json")
|
|
monkeypatch.setattr(jobs_mod, "OUTPUT_DIR", hermes_home / "cron" / "output")
|
|
|
|
return hermes_home
|
|
|
|
|
|
class TestJobScriptField:
|
|
"""Test that the script field is stored and retrieved correctly."""
|
|
|
|
def test_create_job_with_script(self, cron_env):
|
|
from cron.jobs import create_job, get_job
|
|
|
|
job = create_job(
|
|
prompt="Analyze the data",
|
|
schedule="every 30m",
|
|
script="/path/to/monitor.py",
|
|
)
|
|
assert job["script"] == "/path/to/monitor.py"
|
|
|
|
loaded = get_job(job["id"])
|
|
assert loaded["script"] == "/path/to/monitor.py"
|
|
|
|
def test_create_job_without_script(self, cron_env):
|
|
from cron.jobs import create_job
|
|
|
|
job = create_job(prompt="Hello", schedule="every 1h")
|
|
assert job.get("script") is None
|
|
|
|
def test_create_job_empty_script_normalized_to_none(self, cron_env):
|
|
from cron.jobs import create_job
|
|
|
|
job = create_job(prompt="Hello", schedule="every 1h", script=" ")
|
|
assert job.get("script") is None
|
|
|
|
def test_update_job_add_script(self, cron_env):
|
|
from cron.jobs import create_job, update_job
|
|
|
|
job = create_job(prompt="Hello", schedule="every 1h")
|
|
assert job.get("script") is None
|
|
|
|
updated = update_job(job["id"], {"script": "/new/script.py"})
|
|
assert updated["script"] == "/new/script.py"
|
|
|
|
def test_update_job_clear_script(self, cron_env):
|
|
from cron.jobs import create_job, update_job
|
|
|
|
job = create_job(prompt="Hello", schedule="every 1h", script="/some/script.py")
|
|
assert job["script"] == "/some/script.py"
|
|
|
|
updated = update_job(job["id"], {"script": None})
|
|
assert updated.get("script") is None
|
|
|
|
|
|
class TestRunJobScript:
|
|
"""Test the _run_job_script() function."""
|
|
|
|
def test_successful_script(self, cron_env):
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "test.py"
|
|
script.write_text('print("hello from script")\n')
|
|
|
|
success, output = _run_job_script(str(script))
|
|
assert success is True
|
|
assert output == "hello from script"
|
|
|
|
def test_script_relative_path(self, cron_env):
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "relative.py"
|
|
script.write_text('print("relative works")\n')
|
|
|
|
success, output = _run_job_script("relative.py")
|
|
assert success is True
|
|
assert output == "relative works"
|
|
|
|
def test_script_not_found(self, cron_env):
|
|
from cron.scheduler import _run_job_script
|
|
|
|
success, output = _run_job_script("/nonexistent/script.py")
|
|
assert success is False
|
|
assert "not found" in output.lower()
|
|
|
|
def test_script_nonzero_exit(self, cron_env):
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "fail.py"
|
|
script.write_text(textwrap.dedent("""\
|
|
import sys
|
|
print("partial output")
|
|
print("error info", file=sys.stderr)
|
|
sys.exit(1)
|
|
"""))
|
|
|
|
success, output = _run_job_script(str(script))
|
|
assert success is False
|
|
assert "exited with code 1" in output
|
|
assert "error info" in output
|
|
|
|
def test_script_empty_output(self, cron_env):
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "empty.py"
|
|
script.write_text("# no output\n")
|
|
|
|
success, output = _run_job_script(str(script))
|
|
assert success is True
|
|
assert output == ""
|
|
|
|
def test_script_timeout(self, cron_env, monkeypatch):
|
|
from cron import scheduler as sched_mod
|
|
from cron.scheduler import _run_job_script
|
|
|
|
# Use a very short timeout
|
|
monkeypatch.setattr(sched_mod, "_SCRIPT_TIMEOUT", 1)
|
|
|
|
script = cron_env / "scripts" / "slow.py"
|
|
script.write_text("import time; time.sleep(30)\n")
|
|
|
|
success, output = _run_job_script(str(script))
|
|
assert success is False
|
|
assert "timed out" in output.lower()
|
|
|
|
def test_script_json_output(self, cron_env):
|
|
"""Scripts can output structured JSON for the LLM to parse."""
|
|
from cron.scheduler import _run_job_script
|
|
|
|
script = cron_env / "scripts" / "json_out.py"
|
|
script.write_text(textwrap.dedent("""\
|
|
import json
|
|
data = {"new_prs": [{"number": 42, "title": "Fix bug"}]}
|
|
print(json.dumps(data, indent=2))
|
|
"""))
|
|
|
|
success, output = _run_job_script(str(script))
|
|
assert success is True
|
|
parsed = json.loads(output)
|
|
assert parsed["new_prs"][0]["number"] == 42
|
|
|
|
|
|
class TestBuildJobPromptWithScript:
|
|
"""Test that script output is injected into the prompt."""
|
|
|
|
def test_script_output_injected(self, cron_env):
|
|
from cron.scheduler import _build_job_prompt
|
|
|
|
script = cron_env / "scripts" / "data.py"
|
|
script.write_text('print("new PR: #123 fix typo")\n')
|
|
|
|
job = {
|
|
"prompt": "Report any notable changes.",
|
|
"script": str(script),
|
|
}
|
|
prompt = _build_job_prompt(job)
|
|
assert "## Script Output" in prompt
|
|
assert "new PR: #123 fix typo" in prompt
|
|
assert "Report any notable changes." in prompt
|
|
|
|
def test_script_error_injected(self, cron_env):
|
|
from cron.scheduler import _build_job_prompt
|
|
|
|
job = {
|
|
"prompt": "Report status.",
|
|
"script": "/nonexistent/script.py",
|
|
}
|
|
prompt = _build_job_prompt(job)
|
|
assert "## Script Error" in prompt
|
|
assert "not found" in prompt.lower()
|
|
assert "Report status." in prompt
|
|
|
|
def test_no_script_unchanged(self, cron_env):
|
|
from cron.scheduler import _build_job_prompt
|
|
|
|
job = {"prompt": "Simple job."}
|
|
prompt = _build_job_prompt(job)
|
|
assert "## Script Output" not in prompt
|
|
assert "Simple job." in prompt
|
|
|
|
def test_script_empty_output_noted(self, cron_env):
|
|
from cron.scheduler import _build_job_prompt
|
|
|
|
script = cron_env / "scripts" / "noop.py"
|
|
script.write_text("# nothing\n")
|
|
|
|
job = {
|
|
"prompt": "Check status.",
|
|
"script": str(script),
|
|
}
|
|
prompt = _build_job_prompt(job)
|
|
assert "no output" in prompt.lower()
|
|
assert "Check status." in prompt
|
|
|
|
|
|
class TestCronjobToolScript:
|
|
"""Test the cronjob tool's script parameter."""
|
|
|
|
def test_create_with_script(self, cron_env, monkeypatch):
|
|
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
result = json.loads(cronjob(
|
|
action="create",
|
|
schedule="every 1h",
|
|
prompt="Monitor things",
|
|
script="/home/user/monitor.py",
|
|
))
|
|
assert result["success"] is True
|
|
assert result["job"]["script"] == "/home/user/monitor.py"
|
|
|
|
def test_update_script(self, cron_env, monkeypatch):
|
|
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
create_result = json.loads(cronjob(
|
|
action="create",
|
|
schedule="every 1h",
|
|
prompt="Monitor things",
|
|
))
|
|
job_id = create_result["job_id"]
|
|
|
|
update_result = json.loads(cronjob(
|
|
action="update",
|
|
job_id=job_id,
|
|
script="/new/script.py",
|
|
))
|
|
assert update_result["success"] is True
|
|
assert update_result["job"]["script"] == "/new/script.py"
|
|
|
|
def test_clear_script(self, cron_env, monkeypatch):
|
|
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
create_result = json.loads(cronjob(
|
|
action="create",
|
|
schedule="every 1h",
|
|
prompt="Monitor things",
|
|
script="/some/script.py",
|
|
))
|
|
job_id = create_result["job_id"]
|
|
|
|
update_result = json.loads(cronjob(
|
|
action="update",
|
|
job_id=job_id,
|
|
script="",
|
|
))
|
|
assert update_result["success"] is True
|
|
assert "script" not in update_result["job"]
|
|
|
|
def test_list_shows_script(self, cron_env, monkeypatch):
|
|
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
|
|
from tools.cronjob_tools import cronjob
|
|
|
|
cronjob(
|
|
action="create",
|
|
schedule="every 1h",
|
|
prompt="Monitor things",
|
|
script="/path/to/script.py",
|
|
)
|
|
|
|
list_result = json.loads(cronjob(action="list"))
|
|
assert list_result["success"] is True
|
|
assert len(list_result["jobs"]) == 1
|
|
assert list_result["jobs"][0]["script"] == "/path/to/script.py"
|