From bee21594f507ac3c8a30c4d0574f92fd2dbd079b Mon Sep 17 00:00:00 2001 From: Marius Mutu Date: Tue, 21 Apr 2026 07:15:00 +0000 Subject: [PATCH] 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/. --- tests/test_cron_jobs_schema.py | 163 +++++++++++++++++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 tests/test_cron_jobs_schema.py diff --git a/tests/test_cron_jobs_schema.py b/tests/test_cron_jobs_schema.py new file mode 100644 index 0000000..582571d --- /dev/null +++ b/tests/test_cron_jobs_schema.py @@ -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}" + )