Nexus Agent
dfb0ea2d8f
修复 ScriptsPage 执行契约（script_id/command、历史过滤、长任务/凭据）及 SchedulesPage 缺失能力（target_path、next_run、push/script、cron/once、sync_mode）；后端补迁移与 schedule_runner 传参。
51 lines
1.5 KiB
Python
51 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_finds_next_minute():
|
|
after = datetime(2026, 6, 7, 1, 59, tzinfo=timezone.utc)
|
|
nxt = _compute_next_cron_run("0 2 * * *", after)
|
|
assert nxt == datetime(2026, 6, 7, 2, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
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
|