Codex Release Bot
52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
"""Schedule next-run computation and cron stepping."""
|
|
|
|
from datetime import datetime, timezone
|
|
from types import SimpleNamespace
|
|
|
|
from server.background.schedule_runner import (
|
|
_compute_next_cron_run,
|
|
compute_schedule_next_run,
|
|
)
|
|
|
|
|
|
def test_compute_next_cron_run_beijing_wall_0200():
|
|
"""cron 0 2 * * * = Beijing 02:00, not UTC 02:00."""
|
|
after = datetime(2026, 6, 7, 17, 59, tzinfo=timezone.utc) # Beijing 2026-06-08 01:59
|
|
nxt = _compute_next_cron_run("0 2 * * *", after)
|
|
assert nxt == datetime(2026, 6, 7, 18, 0, 0)
|
|
|
|
|
|
def test_compute_schedule_next_run_once_future():
|
|
schedule = SimpleNamespace(
|
|
enabled=True,
|
|
run_mode="once",
|
|
fire_at=datetime(2026, 12, 1, 10, 0),
|
|
last_run_at=None,
|
|
cron_expr=None,
|
|
)
|
|
now = datetime(2026, 6, 7, 12, 0, tzinfo=timezone.utc)
|
|
assert compute_schedule_next_run(schedule, now) == schedule.fire_at
|
|
|
|
|
|
def test_compute_schedule_next_run_once_already_ran():
|
|
schedule = SimpleNamespace(
|
|
enabled=True,
|
|
run_mode="once",
|
|
fire_at=datetime(2026, 12, 1, 10, 0),
|
|
last_run_at=datetime(2026, 12, 1, 10, 1),
|
|
cron_expr=None,
|
|
)
|
|
now = datetime(2026, 6, 7, 12, 0, tzinfo=timezone.utc)
|
|
assert compute_schedule_next_run(schedule, now) is None
|
|
|
|
|
|
def test_compute_schedule_next_run_cron_disabled():
|
|
schedule = SimpleNamespace(
|
|
enabled=False,
|
|
run_mode="cron",
|
|
cron_expr="*/5 * * * *",
|
|
last_run_at=None,
|
|
fire_at=None,
|
|
)
|
|
assert compute_schedule_next_run(schedule) is None
|