Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 17 additions & 1 deletion nerve/cron/jobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -201,9 +201,25 @@ def load_jobs(jobs_file: Path) -> list[CronJob]:
logger.error("Failed to load cron jobs from %s: %s", jobs_file, e)
return []

jobs_data = data.get("jobs", [])
if isinstance(data, list):
jobs_data = data
elif isinstance(data, dict):
jobs_data = data.get("jobs", [])
else:
logger.warning(
"Invalid cron jobs file %s: expected a mapping or list, got %s",
jobs_file,
type(data).__name__,
)
return []

if not isinstance(jobs_data, list):
logger.warning(
"Invalid cron jobs file %s: expected 'jobs' to be a list, got %s",
jobs_file,
type(jobs_data).__name__,
)
return []

jobs = []
for item in jobs_data:
Expand Down
40 changes: 40 additions & 0 deletions tests/test_cron.py
Original file line number Diff line number Diff line change
Expand Up @@ -972,6 +972,46 @@ async def test_and_semantics_one_gate_blocks(self, cron_service):
cron_service.engine.run_cron.assert_not_called()


# ---------------------------------------------------------------------------
# Job loading
# ---------------------------------------------------------------------------

class TestLoadJobs:
def test_accepts_top_level_list(self, tmp_path):
from nerve.cron.jobs import load_jobs

yaml_file = tmp_path / "jobs.yaml"
yaml_file.write_text(
"- id: hourly-review\n"
" schedule: 1h\n"
" prompt: Review pending tasks\n",
encoding="utf-8",
)

jobs = load_jobs(yaml_file)

assert len(jobs) == 1
assert jobs[0].id == "hourly-review"
assert jobs[0].schedule == "1h"
assert jobs[0].prompt == "Review pending tasks"

@pytest.mark.parametrize(
("content", "message"),
[
("invalid\n", "expected a mapping or list, got str"),
("jobs: invalid\n", "expected 'jobs' to be a list, got str"),
],
)
def test_rejects_invalid_structure(self, tmp_path, caplog, content, message):
from nerve.cron.jobs import load_jobs

yaml_file = tmp_path / "jobs.yaml"
yaml_file.write_text(content, encoding="utf-8")

assert load_jobs(yaml_file) == []
assert message in caplog.text


# ---------------------------------------------------------------------------
# Prompt files (prompt_file)
# ---------------------------------------------------------------------------
Expand Down