dodil/erp-platform
tests/test_work.py
"""Tracked work: due times in the person's zone (DST included), recompute, the board, chasing, the demo clock."""
import logging
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
import pytest
from erp_platform import work
from erp_platform.work import ChaseCadence, TimingRule, TrackedStep, board, next_chase, recompute
UTC = timezone.utc
def z(s: str) -> datetime:
return datetime.fromisoformat(s.replace("Z", "+00:00"))
# ── due_at ──────────────────────────────────────────────────────────────────────────────────
def test_nine_in_london_is_eight_utc_in_summer_and_nine_in_winter():
rule = TimingRule("start_date", 0, "09:00", "Europe/London")
assert rule.due_at({"start_date": date(2026, 7, 1)}) == z("2026-07-01T08:00Z")
assert rule.due_at({"start_date": date(2026, 12, 1)}) == z("2026-12-01T09:00Z")
def test_the_offset_is_counted_in_local_days_across_a_clock_change():
"""Clocks go forward in London on 2026-03-29 and back on 2026-10-25. A rule 'the day after, 09:00'
anchored the day before each change must land on 09:00 local, not 24h later in UTC."""
rule = TimingRule("start_date", 1, "09:00", "Europe/London")
assert rule.due_at({"start_date": date(2026, 3, 28)}) == z("2026-03-29T08:00Z") # BST began
assert rule.due_at({"start_date": date(2026, 10, 24)}) == z("2026-10-25T09:00Z") # GMT again
# A datetime anchor: 10:00 BST the day before, +1 day, keeping its own time → 10:00 GMT = 10:00Z.
keep = TimingRule("run.created", 1, None, "Europe/London")
assert keep.due_at({"run.created": z("2026-10-24T09:00Z")}) == z("2026-10-25T10:00Z")
def test_a_time_that_does_not_exist_or_happens_twice_is_stated_not_guessed():
gap = TimingRule("d", 0, "01:30", "Europe/London") # 01:30 does not exist on 2026-03-29
assert gap.due_at({"d": date(2026, 3, 29)}) == z("2026-03-29T01:30Z") # = 02:30 BST
twice = TimingRule("d", 0, "01:30", "Europe/London") # 01:30 happens twice on 2026-10-25
assert twice.due_at({"d": date(2026, 10, 25)}) == z("2026-10-25T00:30Z") # the first
def test_negative_offsets_and_naive_datetimes_and_unknown_anchors():
before = TimingRule("start_date", -10, "09:00", "Europe/Stockholm")
assert before.due_at({"start_date": date(2026, 11, 2)}) == z("2026-10-23T07:00Z") # still CEST
assert TimingRule("run.created", 0, None).due_at({"run.created": datetime(2026, 1, 5, 12)}) == z("2026-01-05T12:00Z")
assert before.due_at({}) is None and before.due_at({"start_date": None}) is None
assert before.due_at({"start_date": date(2026, 11, 2)}).tzinfo is UTC
@pytest.mark.parametrize("kw", [dict(anchor=""), dict(anchor="has space"), dict(anchor="a", offset_days=1.5),
dict(anchor="a", time_of_day="9am"), dict(anchor="a", time_of_day="24:00"),
dict(anchor="a", timezone="Mars/Olympus")])
def test_a_rule_that_cannot_be_evaluated_is_refused_when_built(kw):
with pytest.raises(ValueError):
TimingRule(**kw)
# ── recompute ───────────────────────────────────────────────────────────────────────────────
def test_recompute_moves_open_steps_and_leaves_done_ones_alone():
london = "Europe/London"
steps = [TrackedStep(1, TimingRule("start_date", -10, "09:00", london), done=True),
TrackedStep(2, TimingRule("start_date", 0, "09:00", london)),
TrackedStep(3, TimingRule(work.step_anchor(2), 1, "09:00", london)),
TrackedStep(4, TimingRule("last_day", 0, "17:00", london))]
moved = recompute(steps, {"start_date": date(2026, 12, 7), "step:2.done": None})
assert moved == {2: z("2026-12-07T09:00Z"), 3: None, 4: None} # 1 is history
# ── the board ───────────────────────────────────────────────────────────────────────────────
@dataclass
class Item:
name: str
due_at: object
done: bool = False
def test_the_board_is_the_persons_today_and_a_timed_item_is_late_once_its_time_passes():
# 22:30Z on 1 July is already 2 July in Stockholm (CEST, +2), still 1 July in London (BST, +1).
now = z("2026-07-01T22:30Z")
items = [Item("late-by-date", date(2026, 7, 1)), Item("late-by-time", z("2026-07-01T22:00Z")),
Item("today-later", z("2026-07-02T15:00Z")), Item("today-date", date(2026, 7, 2)),
Item("week", z("2026-07-08T08:00Z")), Item("too-far", date(2026, 7, 20)),
Item("unknown", None), Item("done", date(2026, 6, 1), done=True)]
b = board(items, now=now, timezone="Europe/Stockholm", is_open=lambda i: not i.done)
names = {k: [i.name for i in v] for k, v in b.items()}
assert names == {"overdue": ["late-by-date", "late-by-time"], "today": ["today-date", "today-later"],
"this_week": ["week"]}
# The same moment in London is still 1 July: the date item is today's, not late.
assert [i.name for i in board(items, now=now, timezone="Europe/London",
is_open=lambda i: not i.done)["today"]] == ["late-by-date"]
# ── chasing ─────────────────────────────────────────────────────────────────────────────────
def test_chasing_every_three_days_at_most_twice():
cadence = ChaseCadence(every_days=3, max_count=2, timezone="Europe/London")
sent = z("2026-10-22T09:00Z") # 10:00 BST
first = next_chase(cadence, sent)
assert first == work.ChaseState(0, z("2026-10-25T10:00Z"), False) # 10:00 GMT: a local day, not 72h
assert not first.due(z("2026-10-25T09:59Z")) and first.due(z("2026-10-25T10:00Z"))
second = next_chase(cadence, sent, [z("2026-10-25T10:00Z")])
assert second.count == 1 and second.next_at == z("2026-10-28T10:00Z")
done = next_chase(cadence, sent, [z("2026-10-25T10:00Z"), z("2026-10-28T10:00Z")])
assert done.exhausted and done.next_at is None and not done.due(z("2027-01-01T00:00Z"))
assert next_chase(ChaseCadence(2, 3, first_after_days=0), sent).next_at == sent
assert next_chase(ChaseCadence(2, 0), sent).exhausted
def test_a_cadence_that_makes_no_sense_is_refused():
for kw in (dict(every_days=0, max_count=1), dict(every_days=1, max_count=-1),
dict(every_days=1, max_count=1, first_after_days=-1), dict(every_days=1.0, max_count=1)):
with pytest.raises(ValueError):
ChaseCadence(**kw)
# ── the demo clock ──────────────────────────────────────────────────────────────────────────
@pytest.fixture
def clock(monkeypatch):
real = z("2026-10-17T12:00Z")
monkeypatch.setattr(work, "_real_now", lambda: real)
monkeypatch.delenv(work.OFFSET_ENV, raising=False)
monkeypatch.delenv("DODIL_ENV", raising=False)
work._target_offsets.clear()
return real
def test_the_clock_is_real_unless_moved(clock, monkeypatch, caplog):
assert work.now() == clock
monkeypatch.setenv(work.OFFSET_ENV, "+3d")
monkeypatch.setenv("DODIL_ENV", "dev")
with caplog.at_level(logging.WARNING, logger="erp_platform.work"):
assert work.now() == clock + timedelta(days=3)
assert work.now() == clock + timedelta(days=3)
assert len([r for r in caplog.records if "demo_clock" in r.getMessage()]) == 2 # EVERY time
monkeypatch.setenv(work.OFFSET_ENV, "-2h30m")
assert work.now() == clock - timedelta(hours=2, minutes=30)
def test_an_iso_target_is_where_the_clock_starts_and_it_keeps_ticking(clock, monkeypatch):
monkeypatch.setenv(work.OFFSET_ENV, "2026-11-01T09:00:00Z")
assert work.now() == z("2026-11-01T09:00Z")
monkeypatch.setattr(work, "_real_now", lambda: clock + timedelta(minutes=5))
assert work.now() == z("2026-11-01T09:05Z")
@pytest.mark.parametrize("env", ["prod", "live", "PROD"])
def test_the_demo_clock_is_impossible_in_a_live_deployment(clock, monkeypatch, env):
monkeypatch.setenv(work.OFFSET_ENV, "+3d")
monkeypatch.setenv("DODIL_ENV", env)
with pytest.raises(work.DemoClockRefused, match="never applied to a live deployment"):
work.now()
def test_a_malformed_offset_says_so(clock, monkeypatch):
monkeypatch.setenv(work.OFFSET_ENV, "three days")
with pytest.raises(ValueError, match="neither an offset"):
work.now()
# ── working days ────────────────────────────────────────────────────────────────────────────
from erp_platform.work import WorkCalendar
# England 2026: Good Friday 3 April, Easter Monday 6 April — the consumer's rows, not ours.
EASTER = WorkCalendar(holidays=frozenset({date(2026, 4, 3), date(2026, 4, 6)}))
def test_working_days_skip_the_weekend_and_the_bank_holidays_both_ways():
fwd = TimingRule("start_date", 1, "09:00", "Europe/London", offset_unit="working_days")
assert fwd.due_at({"start_date": date(2026, 4, 2)}, EASTER) == z("2026-04-07T08:00Z") # Thu → Tue
back = TimingRule("start_date", -3, "09:00", "Europe/London", offset_unit="working_days")
assert back.due_at({"start_date": date(2026, 4, 8)}, EASTER) == z("2026-04-01T08:00Z") # Wed → Wed before
assert EASTER.add_working_days(date(2026, 4, 7), -1) == date(2026, 4, 2)
# Calendar days, for contrast: the same +1 is Good Friday itself.
assert TimingRule("start_date", 1, "09:00", "Europe/London").due_at({"start_date": date(2026, 4, 2)}) == z("2026-04-03T08:00Z")
def test_working_days_still_land_on_local_time_across_a_clock_change():
rule = TimingRule("start_date", 1, "09:00", "Europe/London", offset_unit="working_days")
assert rule.due_at({"start_date": date(2026, 3, 27)}, WorkCalendar()) == z("2026-03-30T08:00Z") # Fri GMT → Mon BST
def test_zero_working_days_is_the_anchor_itself_even_on_a_weekend():
rule = TimingRule("start_date", 0, "09:00", "UTC", offset_unit="working_days")
assert rule.due_at({"start_date": date(2026, 4, 4)}, EASTER) == z("2026-04-04T09:00Z") # a Saturday
def test_working_days_need_the_consumers_calendar():
rule = TimingRule("start_date", 1, offset_unit="working_days")
with pytest.raises(ValueError, match="WorkCalendar"):
rule.due_at({"start_date": date(2026, 4, 2)})
assert recompute([TrackedStep(1, rule)], {"start_date": date(2026, 4, 2)}, EASTER) == {1: z("2026-04-07T09:00Z")}
for bad in (dict(offset_unit="weeks"),):
with pytest.raises(ValueError):
TimingRule("a", **bad)
with pytest.raises(ValueError):
WorkCalendar(working_weekdays=frozenset({7}))
def test_chasing_every_two_working_days():
cadence = ChaseCadence(every_days=2, max_count=3, timezone="Europe/London", unit="working_days")
sent = z("2026-04-02T09:00Z") # Thu 10:00 BST
assert next_chase(cadence, sent, calendar=EASTER).next_at == z("2026-04-08T09:00Z") # Tue, Wed
assert next_chase(cadence, sent, [z("2026-04-08T09:00Z")], EASTER).next_at == z("2026-04-10T09:00Z")
with pytest.raises(ValueError):
ChaseCadence(1, 1, unit="fortnights")