From 84f98f5e92c0f437477faa54922ce02c8eec9bd4 Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Mon, 3 Aug 2026 15:06:12 +0800 Subject: [PATCH] test(cron): cover invalid schedule expressions --- nanobot/cron/service.py | 6 ++++-- tests/cron/test_cron_service.py | 27 +++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/nanobot/cron/service.py b/nanobot/cron/service.py index b9833307c..ce0df35b4 100644 --- a/nanobot/cron/service.py +++ b/nanobot/cron/service.py @@ -80,12 +80,14 @@ def _validate_schedule_for_add(schedule: CronSchedule) -> None: raise ValueError("cron schedule requires a non-empty 'expr'") try: from croniter import croniter + croniter(schedule.expr) - except Exception as e: - raise ValueError(f"invalid cron expression '{schedule.expr}': {e}") from None + except Exception as exc: + raise ValueError(f"invalid cron expression '{schedule.expr}': {exc}") from None if schedule.tz: try: from zoneinfo import ZoneInfo + ZoneInfo(schedule.tz) except Exception: raise ValueError(f"unknown timezone '{schedule.tz}'") from None diff --git a/tests/cron/test_cron_service.py b/tests/cron/test_cron_service.py index f91bf2b89..5e18c6450 100644 --- a/tests/cron/test_cron_service.py +++ b/tests/cron/test_cron_service.py @@ -141,6 +141,33 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None: assert job.state.next_run_at_ms is not None +@pytest.mark.parametrize("expr", [None, "", " "]) +def test_add_job_rejects_missing_cron_expression(tmp_path, expr: str | None) -> None: + service = CronService(tmp_path / "cron" / "jobs.json") + + with pytest.raises(ValueError, match="requires a non-empty 'expr'"): + service.add_job( + name="missing expression", + schedule=CronSchedule(kind="cron", expr=expr), + message="hello", + ) + + assert service.list_jobs(include_disabled=True) == [] + + +def test_add_job_rejects_invalid_cron_expression_before_persisting(tmp_path) -> None: + service = CronService(tmp_path / "cron" / "jobs.json") + + with pytest.raises(ValueError, match="invalid cron expression"): + service.add_job( + name="bad expression", + schedule=CronSchedule(kind="cron", expr="not a cron expression"), + message="hello", + ) + + assert service.list_jobs(include_disabled=True) == [] + + def test_write_run_record_uses_cron_runs_dir(tmp_path) -> None: service = CronService(tmp_path / "cron" / "jobs.json")