|
# retoor <retoor@molodetz.nl>
|
|
|
|
import pytest
|
|
|
|
from devplacepy.services.devii.tasks.guards import task_columns
|
|
from devplacepy.services.devii.tasks.schedule import (
|
|
DEFAULT_MAX_RUNS,
|
|
MAX_LIFETIME_DAYS,
|
|
MAX_MAX_RUNS,
|
|
MIN_INTERVAL_SECONDS,
|
|
Schedule,
|
|
from_iso,
|
|
now_utc,
|
|
)
|
|
|
|
|
|
def test_interval_below_the_floor_is_rejected():
|
|
with pytest.raises(ValueError):
|
|
Schedule(kind="interval", every_seconds=MIN_INTERVAL_SECONDS - 1)
|
|
with pytest.raises(ValueError):
|
|
Schedule(kind="interval", every_seconds=180)
|
|
|
|
|
|
def test_interval_at_the_floor_is_accepted():
|
|
schedule = Schedule(kind="interval", every_seconds=MIN_INTERVAL_SECONDS)
|
|
assert schedule.every_seconds == MIN_INTERVAL_SECONDS
|
|
|
|
|
|
def test_cron_firing_faster_than_the_floor_is_rejected():
|
|
for expression in ("* * * * *", "*/5 * * * *", "*/14 * * * *"):
|
|
with pytest.raises(ValueError):
|
|
Schedule(kind="cron", cron=expression)
|
|
|
|
|
|
def test_cron_with_an_uneven_step_is_rejected_for_its_boundary_hop():
|
|
with pytest.raises(ValueError):
|
|
Schedule(kind="cron", cron="*/18 * * * *")
|
|
|
|
|
|
def test_cron_at_or_above_the_floor_is_accepted():
|
|
for expression in ("*/15 * * * *", "0 * * * *", "0 22 * * *"):
|
|
assert Schedule(kind="cron", cron=expression).cron == expression
|
|
|
|
|
|
def test_max_runs_ceiling_is_enforced():
|
|
with pytest.raises(ValueError):
|
|
Schedule(kind="interval", every_seconds=900, max_runs=MAX_MAX_RUNS + 1)
|
|
with pytest.raises(ValueError):
|
|
Schedule(kind="interval", every_seconds=900, max_runs=0)
|
|
|
|
|
|
def test_recurring_task_without_max_runs_gets_the_default():
|
|
reference = now_utc()
|
|
columns = task_columns(Schedule(kind="interval", every_seconds=900), reference)
|
|
assert columns["max_runs"] == DEFAULT_MAX_RUNS
|
|
|
|
|
|
def test_one_shot_task_keeps_no_run_limit():
|
|
reference = now_utc()
|
|
columns = task_columns(Schedule(kind="once", delay_seconds=60), reference)
|
|
assert columns["max_runs"] is None
|
|
|
|
|
|
def test_every_task_gets_an_expiry_within_the_ceiling():
|
|
reference = now_utc()
|
|
columns = task_columns(Schedule(kind="interval", every_seconds=900), reference)
|
|
expiry = from_iso(columns["expires_at"])
|
|
assert expiry > reference
|
|
assert (expiry - reference).days <= MAX_LIFETIME_DAYS
|
|
|
|
|
|
def test_a_caller_cannot_extend_the_expiry_past_the_ceiling():
|
|
reference = now_utc()
|
|
far = reference.replace(year=reference.year + 5)
|
|
columns = task_columns(
|
|
Schedule(kind="interval", every_seconds=900, expires_at=far), reference
|
|
)
|
|
expiry = from_iso(columns["expires_at"])
|
|
assert (expiry - reference).days <= MAX_LIFETIME_DAYS
|