test(cron): validate jobs.json schema per kind

Loads cron/jobs.json and asserts: unique names, valid cron expressions
(APScheduler parseable), bool enabled field; kind:"shell" entries must
have non-empty channel, non-empty command list of strings, valid
report_on, and timeout within [1, 3600] when present; claude entries
must have non-empty prompt, valid model, list-typed allowed_tools.
Sanity-checks that shell commands reference existing scripts in the
repo and that no imported claude prompt still points at /home/moltbot/clawd/.
This commit is contained in:
2026-04-21 07:15:00 +00:00
parent dd8f40774f
commit bee21594f5

View File

@@ -0,0 +1,163 @@
"""Validate cron/jobs.json schema per kind.
Every job must be either `kind: "shell"` (with command/report_on/channel)
or Claude-style (prompt/model/channel). Required fields are enforced
per kind. Cron expressions must parse.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from apscheduler.triggers.cron import CronTrigger
PROJECT_ROOT = Path(__file__).resolve().parents[1]
JOBS_FILE = PROJECT_ROOT / "cron" / "jobs.json"
VALID_REPORT_ON = {"always", "changes", "never"}
# Keep in sync with src.claude_session.VALID_MODELS (informational — we don't
# import there to avoid pulling the whole module into the schema test).
VALID_MODELS = {"haiku", "sonnet", "opus"}
@pytest.fixture(scope="module")
def jobs():
text = JOBS_FILE.read_text(encoding="utf-8")
data = json.loads(text)
assert isinstance(data, list), "jobs.json must be a JSON array"
return data
def test_jobs_file_exists_and_parses(jobs):
assert isinstance(jobs, list)
def test_names_are_unique(jobs):
names = [j["name"] for j in jobs]
assert len(names) == len(set(names)), f"duplicate names: {names}"
def test_every_job_has_name_and_cron(jobs):
for j in jobs:
assert "name" in j and isinstance(j["name"], str) and j["name"]
assert "cron" in j and isinstance(j["cron"], str) and j["cron"]
def test_every_cron_expression_parses(jobs):
for j in jobs:
try:
CronTrigger.from_crontab(j["cron"])
except Exception as exc: # pragma: no cover — diagnostic
pytest.fail(f"job {j['name']!r} has invalid cron {j['cron']!r}: {exc}")
def test_enabled_is_bool(jobs):
for j in jobs:
assert isinstance(j.get("enabled"), bool), (
f"job {j['name']!r} missing/non-bool enabled"
)
def test_shell_jobs_have_required_fields(jobs):
shell_jobs = [j for j in jobs if j.get("kind") == "shell"]
assert shell_jobs, "expected at least one shell job"
for j in shell_jobs:
name = j["name"]
# channel
assert isinstance(j.get("channel"), str) and j["channel"], (
f"shell job {name!r} has empty/missing channel"
)
# command
cmd = j.get("command")
assert isinstance(cmd, list) and cmd, (
f"shell job {name!r} has empty/missing command"
)
assert all(isinstance(c, str) and c.strip() for c in cmd), (
f"shell job {name!r} has non-string entry in command: {cmd!r}"
)
# report_on
report_on = j.get("report_on")
assert report_on in VALID_REPORT_ON, (
f"shell job {name!r} has invalid report_on={report_on!r}"
)
# timeout (optional but if present must be a positive int)
to = j.get("timeout")
if to is not None:
assert isinstance(to, int) and not isinstance(to, bool), (
f"shell job {name!r} has non-int timeout"
)
assert 1 <= to <= 3600, (
f"shell job {name!r} timeout out of range: {to}"
)
def test_claude_jobs_have_required_fields(jobs):
claude_jobs = [j for j in jobs if j.get("kind", "claude") == "claude"]
assert claude_jobs, "expected at least one claude job"
for j in claude_jobs:
name = j["name"]
# channel
assert isinstance(j.get("channel"), str) and j["channel"], (
f"claude job {name!r} has empty/missing channel"
)
# prompt
prompt = j.get("prompt")
assert isinstance(prompt, str) and prompt.strip(), (
f"claude job {name!r} has empty/missing prompt"
)
# model
model = j.get("model")
assert model in VALID_MODELS, (
f"claude job {name!r} has invalid model={model!r}"
)
# allowed_tools
allowed = j.get("allowed_tools")
assert isinstance(allowed, list), (
f"claude job {name!r} has non-list allowed_tools"
)
for t in allowed:
assert isinstance(t, str), (
f"claude job {name!r} has non-string tool: {t!r}"
)
def test_shell_jobs_reference_existing_scripts(jobs):
"""Every shell command's first path-looking argument must refer to a file
that exists in the repo (so we don't ship a jobs.json that runs missing
scripts on day one)."""
shell_jobs = [j for j in jobs if j.get("kind") == "shell"]
for j in shell_jobs:
cmd = j["command"]
# Find the first argument that looks like a relative path
# (contains "/" and doesn't start with "-").
script = next(
(a for a in cmd[1:] if "/" in a and not a.startswith("-")),
None,
)
if script is None:
continue
script_path = PROJECT_ROOT / script
assert script_path.exists(), (
f"shell job {j['name']!r} references missing script: {script}"
)
def test_no_clawd_paths_remain_in_claude_prompts(jobs):
"""Sanity check: no imported prompt still points at /home/moltbot/clawd/."""
offenders: list[str] = []
for j in jobs:
if j.get("kind", "claude") != "claude":
continue
prompt = j.get("prompt", "")
if "/home/moltbot/clawd/" in prompt:
offenders.append(j["name"])
if "cd ~/clawd " in prompt or prompt.endswith("cd ~/clawd") or "cd ~/clawd\n" in prompt:
offenders.append(j["name"])
assert not offenders, (
f"these claude jobs still reference /home/moltbot/clawd/ or cd ~/clawd: "
f"{offenders}"
)