mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 09:28:34 +00:00
test(cron): cover invalid schedule expressions
This commit is contained in:
parent
73a0080484
commit
84f98f5e92
@ -80,12 +80,14 @@ def _validate_schedule_for_add(schedule: CronSchedule) -> None:
|
|||||||
raise ValueError("cron schedule requires a non-empty 'expr'")
|
raise ValueError("cron schedule requires a non-empty 'expr'")
|
||||||
try:
|
try:
|
||||||
from croniter import croniter
|
from croniter import croniter
|
||||||
|
|
||||||
croniter(schedule.expr)
|
croniter(schedule.expr)
|
||||||
except Exception as e:
|
except Exception as exc:
|
||||||
raise ValueError(f"invalid cron expression '{schedule.expr}': {e}") from None
|
raise ValueError(f"invalid cron expression '{schedule.expr}': {exc}") from None
|
||||||
if schedule.tz:
|
if schedule.tz:
|
||||||
try:
|
try:
|
||||||
from zoneinfo import ZoneInfo
|
from zoneinfo import ZoneInfo
|
||||||
|
|
||||||
ZoneInfo(schedule.tz)
|
ZoneInfo(schedule.tz)
|
||||||
except Exception:
|
except Exception:
|
||||||
raise ValueError(f"unknown timezone '{schedule.tz}'") from None
|
raise ValueError(f"unknown timezone '{schedule.tz}'") from None
|
||||||
|
|||||||
@ -141,6 +141,33 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None:
|
|||||||
assert job.state.next_run_at_ms is not 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:
|
def test_write_run_record_uses_cron_runs_dir(tmp_path) -> None:
|
||||||
service = CronService(tmp_path / "cron" / "jobs.json")
|
service = CronService(tmp_path / "cron" / "jobs.json")
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user