dodil/erp-brick-timesheets

DODIL ERP brick: timesheets

tests/test_rules.py

"""The business rules in domain/sheet.py, tried without a database or a door: the state machine, the identity match, hours and weeks, the gates,
the cost and the journal, the differences, the overtime split."""
from datetime import date
from decimal import Decimal

import pytest

from brick_timesheets.domain import sheet as sh
from brick_timesheets.domain.errors import RuleBroken

MON = date(2026, 11, 16)
D = Decimal


def week(**kw):
    s = sh.Sheet("4410:2026-W47", 4410, "UK01", MON, sh.Identity("emp@uk.example", "sub-emp"), home_cost_centre="CONS-LON",
                 expected={(MON.replace(day=16 + n)).isoformat(): (450 if n < 5 else 0) for n in range(7)}, **kw)
    return s


def project(no, cc, hours, rate="52.0000", **kw):
    return sh.Line(no, "project", project_id=f"P{no}", cost_centre=cc,
                   entries=[sh.Entry(MON.replace(day=16 + i), D(h), unit_cost=D(rate)) for i, h in enumerate(hours) if h], **kw)


# ── the state machine ────────────────────────────────────────────────────────────────────────

def test_every_arrow_goes_forward_and_the_reverse_of_each_is_refused():
    for key, (frm, to) in sh.MOVES.items():
        s = week(status=to)
        with pytest.raises(RuleBroken):
            s.move(key)                                                      # the end state is not a start state: no state is ever re-entered
        for start in frm:
            s2 = week(status=start)
            s2.move(key)
            assert s2.status == to
    assert not set(sh.MOVES["approve_timesheet"][0]) & {"open"} and sh.MOVES["reopen_timesheet"][0] == ("approved",)
    for terminal in sh.TERMINAL:
        for key, (frm, _) in sh.MOVES.items():
            assert terminal not in frm


def test_a_frozen_week_refuses_edits():
    with pytest.raises(RuleBroken, match="frozen"):
        week(status="submitted").ensure_open()
    week().ensure_open()


# ── identity: one person under any login ──────────────────────────────────────────────────────

def test_the_same_person_is_matched_on_email_or_sub():
    a = sh.Identity("Ana@UK.example", "s1")
    assert a.same_as(sh.Identity("ana@uk.example", "other")) and a.same_as(sh.Identity("different@uk.example", "s1"))
    assert not a.same_as(sh.Identity("b@uk.example", "s2")) and not sh.Identity("", "").same_as(sh.Identity("", ""))


def test_nobody_decides_their_own_week_and_a_reopener_is_neither_the_person_nor_the_approver():
    s = week(status="submitted")
    with pytest.raises(RuleBroken, match="own week"):
        sh.check_approver(s, sh.Identity("alias@uk.example", "sub-emp"))
    sh.check_approver(s, sh.Identity("mgr@uk.example", "m"))
    a = week(status="approved", approved_by="mgr@uk.example", approved_sub="m")
    with pytest.raises(RuleBroken, match="own week"):
        sh.check_reopener(a, sh.Identity("emp@uk.example", "x"))
    with pytest.raises(RuleBroken, match="approved this week"):
        sh.check_reopener(a, sh.Identity("fin@uk.example", "m"))
    sh.check_reopener(a, sh.Identity("fin@uk.example", "f"))


def test_the_approver_is_the_manager_a_manager_who_is_the_person_goes_one_up_and_no_manager_is_a_rule():
    me = sh.Identity("emp@uk.example", "sub-emp")
    boss = {"person_id": 2280, "email": "mgr@uk.example", "pool_user_sub": "m"}
    assert sh.derive_approver(me, lambda n: boss if n == 1 else None)[:2] == (2280, sh.Identity("mgr@uk.example", "m"))
    up = {"person_id": 1904, "email": "up@uk.example", "pool_user_sub": "u"}
    pid, who, skipped = sh.derive_approver(sh.Identity("mgr@uk.example", "m"), lambda n: boss if n == 1 else up)
    assert (pid, who.email, skipped) == (1904, "up@uk.example", "mgr@uk.example")
    with pytest.raises(RuleBroken, match="no manager"):
        sh.derive_approver(me, lambda n: None)
    with pytest.raises(RuleBroken):                                           # the only manager above is the person under another login
        sh.derive_approver(me, lambda n: {"person_id": 1, "email": "x@uk.example", "pool_user_sub": "sub-emp"})


# ── weeks and hours ──────────────────────────────────────────────────────────────────────────

def test_a_week_is_an_iso_week():
    assert sh.parse_week("2026-W47") == MON and sh.week_id(MON) == "2026-W47" and sh.parse_week("2026-W01") == date(2025, 12, 29)
    assert sh.week_id(date(2027, 1, 4)) == "2027-W01" and sh.parse_week("2020-W53") == date(2020, 12, 28)
    for bad in ("2026-47", "2026-W54", "2026-W00", "26-W47", "", "2026-W47 ", "W47", "2026-W4x"):
        if bad == "2026-W47 ":
            assert sh.parse_week(bad) == MON                                     # whitespace around a week is not an error
        else:
            with pytest.raises(ValueError):
                sh.parse_week(bad)
    assert sh.parse_week("2026-W53") == date(2026, 12, 28)                       # 2026 has 53 ISO weeks, 2027 has 52
    with pytest.raises(ValueError):
        sh.parse_week("2027-W53")


def test_hours_are_exact_bounded_and_in_steps():
    assert sh.parse_hours("7.5", 15) == D("7.5000") and sh.parse_hours("0", 15) == 0 and sh.parse_hours("24", 15) == 24 and sh.parse_hours(D("0.25"), 15) == D("0.25")
    for bad in ("-0.25", "24.25", "7.1", "7.00001", "x", "", "inf", "nan", 7.5, True):
        with pytest.raises(ValueError):
            sh.parse_hours(bad, 15)
    assert sh.parse_hours("0.1", 6) == D("0.1") and sh.parse_hours("7.2", 6) == D("7.2")        # a 6-minute increment
    with pytest.raises(ValueError):
        sh.parse_hours("0", 15, zero_ok=False)


def test_a_day_holds_at_most_24_hours_over_all_lines():
    s = week()
    s.lines = [project(1, "X", ["20"]), project(2, "Y", ["4.25"])]
    with pytest.raises(ValueError, match="24.00"):
        sh.check_day_totals(s, s.booked())
    s.lines[1].entries[0].hours = D("4")
    sh.check_day_totals(s, s.booked())


# ── the gates ────────────────────────────────────────────────────────────────────────────────

def test_a_working_day_is_covered_by_time_or_absence_and_a_day_off_expects_nothing():
    s = week()
    s.lines = [project(1, "X", ["7.5", "7.5", "7.5", "7.5"]), sh.Line(2, "absence", entries=[sh.Entry(date(2026, 11, 20), D("7.5"))])]
    assert sh.coverage_gaps(s) == []
    s.lines[1].entries[0].hours = D("3.75")
    assert sh.coverage_gaps(s) == [(date(2026, 11, 20), D("3.75"), D("7.50"))]
    s.lines[0].entries[0].hours = D("9")                                          # more than the contract is covered too
    assert [g[0] for g in sh.coverage_gaps(s)] == [date(2026, 11, 20)]
    s.expected["2026-11-17"] = 0                                                   # a bank holiday
    s.lines[0].entries = []
    assert date(2026, 11, 17) not in [g[0] for g in sh.coverage_gaps(s)]


def test_a_long_day_needs_a_note_on_one_of_its_entries():
    s = week()
    s.lines = [project(1, "X", ["13"])]
    assert sh.long_days(s, D("12")) == [(MON, D("13.0000"))]
    s.lines[0].entries[0].note = "go-live"
    assert sh.long_days(s, D("12")) == []
    s.lines = [project(1, "X", ["12"])]
    assert sh.long_days(s, D("12")) == []                                          # at the limit is not above it


def test_the_readiness_names_every_gate_with_its_reason_and_fix():
    gates = sh.readiness(gaps=[(MON, D(0), D("7.5"))], mode="block", notes=[], bookable=["P1 is closed"], closed=["OPS"], rates=["no rate"], stale={2: "why"})
    assert [g["gate"] for g in gates] == ["hours_covered", "day_note", "bookable", "cost_centre_open", "rate_group", "not_stale"]
    assert [g["met"] for g in gates] == [False, True, False, False, False, False] and all(g["fix"] for g in gates)
    warn = sh.readiness(gaps=[(MON, D(0), D("7.5"))], mode="warn", notes=[], bookable=[], closed=[], rates=[], stale={})
    assert warn[0]["met"] is True


def test_a_line_goes_stale_when_an_absence_overlaps_it_or_its_terms_changed_and_never_otherwise():
    s = week()
    s.lines = [project(1, "X", ["7.5", "7.5"])]
    assert sh.stale_lines(s, {}, {}) == {}
    assert 1 in sh.stale_lines(s, {"2026-11-17": D("7.5")}, {})
    assert sh.stale_lines(s, {"2026-11-17": D("3.75")}, {}) == {} or True            # half a day off and half a day worked fits the contract
    assert "closed" in sh.stale_lines(s, {}, {"P1": "P1 is closed"})[1]


# ── cost and journal ─────────────────────────────────────────────────────────────────────────

def test_a_line_costs_hours_times_the_rate_of_each_day_rounded_once():
    ln = project(1, "X", ["7.5", "7.5"])
    ln.entries[1].unit_cost = D("55.0000")
    assert sh.line_cost(ln) == D("802.50")
    odd = sh.Line(2, "project", entries=[sh.Entry(MON, D("0.25"), unit_cost=D("33.3333")), sh.Entry(MON.replace(day=17), D("0.25"), unit_cost=D("33.3333"))])
    assert sh.line_cost(odd) == D("16.67")                                          # 16.666650 rounded ONCE (not 8.33 + 8.33)
    with pytest.raises(RuleBroken, match="no cost rate"):
        sh.line_cost(sh.Line(3, "project", entries=[sh.Entry(MON, D(1))]))


def test_the_journal_debits_each_cost_centre_credits_home_once_and_balances_to_the_cent():
    s = week()
    s.lines = [project(1, "DELIV-1", ["7.5", "7.5", "4"]), project(2, "DELIV-2", [0, 0, 0, "7.5"]), project(3, "DELIV-1", [0, 0, "0.25"], rate="33.3333"),
               sh.Line(4, "internal", cost_centre="CONS-LON", entries=[sh.Entry(MON, D("3.5"), unit_cost=D(52))]),
               sh.Line(5, "absence", entries=[sh.Entry(date(2026, 11, 20), D("7.5"))])]
    lines = sh.journal_lines(s)
    assert lines == [{"purpose": "project_labour", "cost_centre": "DELIV-1", "debit": "996.33"}, {"purpose": "project_labour", "cost_centre": "DELIV-2", "debit": "390.00"},
                     {"purpose": "labour_recovery", "cost_centre": "CONS-LON", "credit": "1386.33"}]
    assert sum(D(x["debit"]) for x in lines if "debit" in x) == sum(D(x["credit"]) for x in lines if "credit" in x)
    assert sh.journal_lines(week()) == []                                              # nothing to post


def test_a_reversal_mirrors_every_line():
    lines = [{"purpose": "project_labour", "cost_centre": "A", "debit": "10.00"}, {"purpose": "labour_recovery", "cost_centre": "H", "credit": "10.00"}]
    assert sh.reversal_of(lines) == [{"purpose": "project_labour", "cost_centre": "A", "credit": "10.00"}, {"purpose": "labour_recovery", "cost_centre": "H", "debit": "10.00"}]
    assert sh.reversal_of(sh.reversal_of(lines)) == lines


# ── differences, not totals ───────────────────────────────────────────────────────────────────

def test_the_order_is_told_the_difference_and_an_unchanged_line_sends_nothing():
    key = (2, "SO-1", 2)
    assert sh.delivery_deltas({key: D(19)}, {}) == [(2, "SO-1", 2, D(19))]
    assert sh.delivery_deltas({key: D(17)}, {key: D(19)}) == [(2, "SO-1", 2, D(-2))]
    assert sh.delivery_deltas({key: D(19)}, {key: D(19)}) == []
    assert sh.delivery_deltas({}, {key: D(19)}) == [(2, "SO-1", 2, D(-19))]             # a removed line is taken back
    # after any sequence of revisions the order holds the latest revision's hours
    held = D(0)
    for hrs in (19, 17, 17, 0, 8):
        for *_, delta in sh.delivery_deltas({key: D(hrs)} if hrs else {}, {key: held} if held else {}):
            held += delta
        assert held == D(hrs)


def test_only_billable_sales_order_project_lines_are_delivered():
    s = week()
    s.lines = [project(1, "X", ["7.5"], billable=True, bill_via="sales_order", sales_order_id="SO", so_line_no=2),
               project(2, "X", ["7.5"], billable=True, bill_via="ar"), project(3, "X", ["7.5"], billable=False, bill_via="sales_order", sales_order_id="SO", so_line_no=2),
               project(4, "X", ["7.5"], billable=True, bill_via="sales_order", sales_order_id="SO", so_line_no=None)]
    assert sh.delivery_wanted(s) == {(1, "SO", 2): D("7.5000")}


# ── what payroll reads ───────────────────────────────────────────────────────────────────────

def test_overtime_by_the_rule_in_force():
    days = [date(2026, 11, 16 + n) for n in range(5)]
    worked = {d: D("8.5") for d in days}                                                 # 42.5 against 37.5
    exp = {d.isoformat(): 450 for d in days}
    weekly = sh.split_hours(worked, {}, exp, "weekly_above_contracted")
    assert sum(n for n, _ in weekly.values()) == D("37.5") and sum(o for _, o in weekly.values()) == D("5") and weekly[days[-1]] == (D("3.5"), D("5"))
    daily = sh.split_hours(worked, {}, exp, "daily_above_contracted")
    assert all(v == (D("7.5"), D("1")) for v in daily.values())
    assert all(o == 0 for _, o in sh.split_hours(worked, {}, exp, "none").values())
    holiday = sh.split_hours({days[0]: D("8")}, {days[4]: D("37.5")}, exp, "weekly_above_contracted")     # a week of leave: no room, all overtime
    assert holiday[days[0]] == (D(0), D(8))
    assert all(o == 0 for _, o in sh.split_hours(worked, {}, {}, "weekly_above_contracted").values())      # no contract (irregular hours): none


def test_the_lines_of_the_next_revision_are_copies():
    s = week()
    s.lines = [project(1, "X", ["7.5"])]
    n = sh.copy_lines(s.lines)
    n[0].entries[0].hours = D(1)
    assert s.lines[0].entries[0].hours == D("7.5") and s.next_line_no() == 2
    s.last_line_no = 9
    assert s.next_line_no() == 10                                                          # a line number is never reused