dodil/erp-brick-time-off
DODIL ERP brick: time off
tests/test_invariants.py
"""The recurring bug classes, each tried to break: one claim key per person/day operation; a decision and its ledger entry in ONE transaction (nothing
leaves before the commit - this slice has no outbound write); states never move backwards; nobody decides their own time off; the business date; bounds
(422, never 500); results marked stale; the balance is a pure sum; no row of one person changes another's answer."""
import random
from datetime import date
from decimal import Decimal
import pytest
from conftest import (
EDINBURGH,
EMPLOYEE,
HR_ADMIN,
MANAGER,
MANAGER2,
SECOND_LOGIN_OF_2280,
api,
approve,
ask,
balance,
decided,
entries_of,
get,
must,
on_day,
person,
put_entry,
refused,
request,
)
from sqlalchemy import func, select
from brick_time_off.application import use_cases
from brick_time_off.domain import request as rq
from brick_time_off.infrastructure import tables as t
def claim_keys(scope=None):
with use_cases.repository() as repo:
q = select(t.tof_claim.c.claim_scope, t.tof_claim.c.claim_key, t.tof_claim.c.n)
if scope:
q = q.where(t.tof_claim.c.claim_scope == scope)
return sorted(tuple(r) for r in repo.conn.execute(q))
def count(table):
with use_cases.repository() as repo:
return repo.conn.execute(select(func.count()).select_from(table)).scalar()
# ── one claim key per person/day operation ──────────────────────────────────────────────────────
def test_every_operation_takes_exactly_one_claim_and_each_half_day_exactly_one_row():
a = decided("2026-12-21", "2026-12-22")["request_id"]
b = request("2026-12-23")["request_id"]
must(api("POST", f"/leave-requests/{b}/withdraw", EMPLOYEE))
must(api("POST", f"/leave-requests/{a}/ask-cancellation", EMPLOYEE))
must(api("POST", f"/leave-requests/{a}/refuse-cancellation", MANAGER, {"reason": "no"}))
must(api("POST", f"/leave-requests/{a}/ask-cancellation", EMPLOYEE))
must(api("POST", f"/leave-requests/{a}/cancel", MANAGER, {"reason": "ok"}))
assert claim_keys() == sorted([("series", "UK01-2026", 1), ("series", "UK01-2026", 2), ("decide", f"{a}:r1", 1), ("decide", f"{b}:r1", 1),
("cancel_ask", a, 1), ("cancel_ask", a, 2), ("cancel_refuse", f"{a}:1", 1), ("cancel", f"{a}:r1", 1)])
assert count(t.tof_day_claim) == 0 # all released
c = request("2026-12-21", "2026-12-22")["request_id"] # and a fresh request re-claims each half exactly once
assert count(t.tof_day_claim) == 4 and claim_keys("series")[-1] == ("series", "UK01-2026", 3) and c.endswith("00003")
def test_a_lost_race_on_a_decision_is_a_409_and_writes_nothing_more():
rid = request("2026-12-21")["request_id"]
with use_cases.repository() as repo: # the other approver got there first: THEIR claim row is committed
repo.claims.once("decide", f"{rid}:r1", use_cases._now(), "someone.else")
refused(approve(rid), 409, "Someone else did the same")
refused(api("POST", f"/leave-requests/{rid}/reject", MANAGER, {"reason": "x"}), 409, "Someone else did the same")
refused(api("POST", f"/leave-requests/{rid}/withdraw", EMPLOYEE), 409, "Someone else did the same")
assert entries_of(4410) == [] and get(rid)["status"] == "requested" and count(t.tof_day_claim) == 2
def test_two_approvals_of_one_request_leave_exactly_one_take_entry():
rid = request("2026-12-21")["request_id"]
must(approve(rid))
refused(approve(rid), 409, "approved")
refused(approve(rid, MANAGER2), 404, "no leave request")
assert entries_of(4410) == [(f"req:{rid}:r1", "take", "-1.0000")]
with use_cases.repository() as repo, pytest.raises(Exception): # the ledger key itself refuses a second entry of the same ref
repo.entries.add(f"req:{rid}:r1", 4410, "ANNUAL", 2026, "take", Decimal("-1"), "days", date(2026, 12, 21), "x", "y", "", use_cases._now())
def test_request_numbers_come_from_the_companys_yearly_series():
assert [request(d)["request_id"] for d in ("2026-12-21", "2026-12-22")] == ["UK01-TOF-2026-00001", "UK01-TOF-2026-00002"]
with on_day("2027-01-04T09:00:00+00:00"):
assert request("2027-01-11")["request_id"] == "UK01-TOF-2027-00001"
# ── a decision and its ledger entry are one transaction ─────────────────────────────────────────
def test_a_failing_approval_leaves_no_entry_no_decision_and_no_claim_taken(monkeypatch):
rid = request("2026-12-21")["request_id"]
real = t.RequestRepository.save
def broken(self, r, at):
if r.status == "approved":
raise RuntimeError("the store fell over mid-approval")
return real(self, r, at)
monkeypatch.setattr(t.RequestRepository, "save", broken)
with pytest.raises(Exception):
api("POST", f"/leave-requests/{rid}/approve", MANAGER)
assert entries_of(4410) == [] and claim_keys("decide") == [] and get(rid)["status"] == "requested" # the entry went with the transaction
monkeypatch.setattr(t.RequestRepository, "save", real)
assert must(approve(rid))["status"] == "approved" and len(entries_of(4410)) == 1
def test_a_cancellation_reverses_the_entry_and_releases_the_days_together(monkeypatch):
rid = decided("2026-12-21")["request_id"]
monkeypatch.setattr(t.DayClaimRepository, "release", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("fell over")))
with pytest.raises(Exception):
api("POST", f"/leave-requests/{rid}/cancel", MANAGER, {"reason": "x"})
assert [e[0] for e in entries_of(4410)] == [f"req:{rid}:r1"] and get(rid)["status"] == "approved" and count(t.tof_day_claim) == 2
# ── states never move backwards ─────────────────────────────────────────────────────────────────
def test_every_arrow_goes_forward_and_every_reverse_edge_is_refused():
order = {"requested": 0, "approved": 1, "rejected": 2, "withdrawn": 2, "cancelled": 2}
for key, (starts, end) in rq.MOVES.items():
for s in starts:
assert order[end] >= order[s], (key, s, end) # nothing returns to an earlier state
for terminal in rq.TERMINAL:
assert not any(terminal in starts for starts, _ in rq.MOVES.values()), terminal # a rejected, withdrawn or cancelled request is terminal
assert "requested" not in [end for _, end in rq.MOVES.values()] # nothing moves back to requested
@pytest.mark.parametrize("final", ["rejected", "withdrawn", "cancelled", "approved"])
def test_a_decided_request_takes_no_further_decision_through_the_doors(final):
rid = request("2026-12-21")["request_id"]
if final == "rejected":
must(api("POST", f"/leave-requests/{rid}/reject", MANAGER, {"reason": "no"}))
elif final == "withdrawn":
must(api("POST", f"/leave-requests/{rid}/withdraw", EMPLOYEE))
else:
must(approve(rid))
if final == "cancelled":
must(api("POST", f"/leave-requests/{rid}/cancel", MANAGER, {"reason": "no"}))
before = (get(rid)["status"], entries_of(4410), count(t.tof_day_claim))
attempts = [(MANAGER, "approve", None), (MANAGER, "reject", {"reason": "x"}), (EMPLOYEE, "withdraw", None), (MANAGER, "cancel", {"reason": "x"}),
(EMPLOYEE, "ask-cancellation", None), (MANAGER, "refuse-cancellation", {"reason": "x"}), (EMPLOYEE, "recompute", None)]
allowed = {"approved": {"cancel", "ask-cancellation"}, "rejected": set(), "withdrawn": set(), "cancelled": set()}[final]
for who, action, body in attempts:
if action in allowed:
continue
code, out = api("POST", f"/leave-requests/{rid}/{action}", who, body)
assert code == 409, (final, action, code, out)
assert (get(rid)["status"], entries_of(4410), count(t.tof_day_claim)) == before
# ── nobody decides their own time off ───────────────────────────────────────────────────────────
def test_nobody_decides_their_own_request_under_either_login_not_even_hr():
own_hr = request("2026-12-21", who=HR_ADMIN)["request_id"]
refused(approve(own_hr, HR_ADMIN), 409, "you cannot decide it") # HR sees everything and still not their own
refused(api("POST", f"/leave-requests/{own_hr}/reject", HR_ADMIN, {"reason": "x"}), 409, "cannot decide")
other_email_same_sub = person("someone.else@uk.example", ["tof:request:approve", "tof:request:read_all", "tof:request:read_own"], "sub-3100")
same_email_other_sub = person("hr3100@uk.example", ["tof:request:approve", "tof:request:read_all", "tof:request:read_own"], "a-different-sub")
refused(approve(own_hr, other_email_same_sub), 409, "cannot decide") # the same sub under another email
refused(approve(own_hr, same_email_other_sub), 409, "cannot decide") # the same email under another sub
assert get(own_hr, HR_ADMIN)["status"] == "requested" and entries_of(3100) == []
rid = decided("2026-12-21", who=MANAGER, by=MANAGER2)["request_id"]
must(api("POST", f"/leave-requests/{rid}/ask-cancellation", MANAGER))
refused(api("POST", f"/leave-requests/{rid}/cancel", SECOND_LOGIN_OF_2280, {"reason": "x"}), 409, "cannot decide") # nor cancel their own
refused(api("POST", f"/leave-requests/{rid}/refuse-cancellation", SECOND_LOGIN_OF_2280, {"reason": "x"}), 409, "cannot decide")
def test_the_approver_is_fixed_when_the_request_is_made(hr):
rid = request("2026-12-21")["request_id"]
hr.managers[4410][0] = hr.as_manager(1904) # hr changes the manager later: the request keeps its approver
assert get(rid)["approver_person_id"] == "2280"
refused(approve(rid, MANAGER2), 404, "no leave request")
must(approve(rid))
def test_with_no_manager_in_hr_the_request_waits_for_hr_alone(hr):
hr.managers[4410] = [None, None]
rid = request("2026-12-21")["request_id"]
assert get(rid)["approver_person_id"] is None
refused(approve(rid, MANAGER), 404, "no leave request")
assert must(approve(rid, HR_ADMIN))["status"] == "approved"
# ── the business date ───────────────────────────────────────────────────────────────────────────
def test_a_past_day_is_judged_by_the_date_in_business_timezone_not_the_utc_date():
with on_day("2026-07-10T09:00:00+00:00"):
rid = decided("2026-07-14")["request_id"]
with on_day("2026-07-14T23:30:00+00:00"): # 00:30 on the 15th in London (BST): the 14th is already past
refused(api("POST", f"/leave-requests/{rid}/ask-cancellation", EMPLOYEE), 409, "2026-07-14 is already past")
with on_day("2026-07-14T22:30:00+00:00"): # 23:30 on the 14th in London: today is not past
assert must(api("POST", f"/leave-requests/{rid}/ask-cancellation", EMPLOYEE))["cancellation_asked"] is True
def test_how_far_ahead_a_request_may_start_counts_from_the_business_date():
with on_day("2026-07-14T23:30:00+00:00"): # the business date is 2026-07-15
assert request("2027-07-15")["status"] == "requested" # exactly 365 days ahead
body = refused(ask("2027-07-16"), 422, "more than 365 days ahead")
assert body["gate"] == "max_days_ahead"
def test_a_settings_row_dated_tomorrow_does_not_take_over_early():
with use_cases.repository() as repo:
repo.settings.put("max_days_ahead", "10", date(2026, 11, 3), "owner", use_cases._now())
assert request("2027-06-01")["status"] == "requested" # the 10-day row starts tomorrow
with on_day("2026-11-03T09:00:00+00:00"):
refused(ask("2027-06-01"), 422, "more than 10 days ahead")
# ── bounds: 422 {gate: bounds}, never a 500 ─────────────────────────────────────────────────────
BAD = [
{"from_date": "not-a-date"}, {"from_date": "2026-12-21", "to_date": "2026-12-20"}, {"from_date": "9999-12-31"}, {"from_date": "1999-12-31"},
{"from_date": "2026-12-21", "to_date": "2027-12-31"}, {"from_date": "2026-01-05", "to_date": "2027-02-05"},
{"from_date": "2026-12-21", "to_date": "2026-12-22", "from_part": "am"}, {"from_date": "2026-12-21", "to_date": "2026-12-22", "to_part": "pm"},
{"from_date": "2026-12-21", "from_part": "am", "to_part": "pm"}, {"from_date": "2026-12-21", "to_date": "2026-12-21", "from_part": "pm", "to_part": "full"},
{"from_date": "2026-12-21", "from_part": "noon"}, {"from_date": ""}, {"from_date": "2026-12-21", "note": "x" * 501},
{"from_date": "2026-12-21\x00"}, {"from_date": "2026-12-21", "leave_type_code": "x" * 65},
]
@pytest.mark.parametrize("body", BAD)
def test_a_value_out_of_bounds_is_a_422_never_a_500_and_writes_nothing(body):
code, out = api("POST", "/leave-requests", EMPLOYEE, body)
assert code == 422, (body, code, out)
assert count(t.tof_request) == 0 and count(t.tof_day_claim) == 0 and claim_keys() == []
def test_a_request_longer_than_a_year_is_a_bounds_refusal_here_never_hrs_422_passed_on_as_a_502():
body = refused(ask("2026-01-05", "2027-02-05"), 422, "at most 366 days")
assert body["gate"] == "bounds"
def test_a_bounds_refusal_names_its_gate_and_an_unknown_leave_type_is_a_409():
assert refused(ask("2026-12-21", "2026-12-20"), 422, "before the first")["gate"] == "bounds"
assert refused(ask("2026-12-21", leave_type_code="NO_SUCH"), 409, "No leave type NO_SUCH")["gate"] == "leave_type"
assert refused(ask("2026-12-21", leave_type_code="SICK"), 409, "recorded by your manager")["gate"] == "leave_type"
rid = request("2026-12-21")["request_id"]
for who, action in ((MANAGER, "reject"), (MANAGER, "cancel")):
assert refused(api("POST", f"/leave-requests/{rid}/{action}", who, {"reason": ""}), 422, "reason is required")["gate"] == "bounds"
for path in ("/balance", "/persons/4410/balance"):
code, out = api("GET", path, HR_ADMIN if path == "/balance" else EMPLOYEE, {"on_date": "x", "on": "x"})
assert code in (403, 422), (path, code, out)
def test_a_leaver_notice_out_of_bounds_is_a_422():
from conftest import HR_CLIENT
for body in ({"last_day": "never", "source_ref": "hr:left:4410:1"}, {"last_day": "2026-10-16", "source_ref": "x" * 300},
{"last_day": "9999-01-01", "source_ref": "hr:left:4410:1"}, {"last_day": "2026-10-16", "source_ref": "hr:left:4410:x"}):
assert api("POST", "/employees/4410/leaver", HR_CLIENT, body)[0] == 422, body
assert api("POST", "/employees/99999999999999999999999/leaver", HR_CLIENT, {"last_day": "2026-10-16", "source_ref": "hr:left:1:1"})[0] == 422
# ── settings the owner holds ────────────────────────────────────────────────────────────────────
def test_allow_negative_days_is_read_from_the_dated_row():
put_entry(4410, "adj:4410:ANNUAL:2026:1", "adjust", "-19.00") # 6.00 left
refused(ask("2026-12-21", "2026-12-31"), 422, "7.00 asked, 6.00 available")
with use_cases.repository() as repo:
repo.settings.put("allow_negative_days", "2", date(2026, 1, 1), "owner", use_cases._now())
assert request("2026-12-21", "2026-12-31")["cost"] == "7.00" # 7.00 asked, 6.00 available, 2.00 allowed below zero
assert balance()["available"] == "-1.00"
def test_min_booking_full_day_refuses_a_half_day():
with use_cases.repository() as repo:
repo.settings.put("min_booking", "full_day", date(2026, 1, 1), "owner", use_cases._now())
refused(ask("2026-12-21", from_part="am"), 422, "whole days only")
assert request("2026-12-21")["cost"] == "1.00"
# ── results marked stale ────────────────────────────────────────────────────────────────────────
def test_a_changed_calendar_flags_the_request_and_refuses_approval_until_it_is_recomputed(hr):
rid = request("2026-12-21", "2026-12-24")["request_id"]
assert get(rid)["cost"] == "4.00" and balance()["pending"] == "4.00"
hr.people[4410]["calendar"] = {**hr.people[4410]["calendar"], "2026-12-23": "Company day"} # hr loads a new holiday
body = refused(approve(rid), 409, "the cost is now 3.00, it was 4.00")
assert body["gate"] == "stale" and get(rid)["stale"] is True and get(rid)["stale_cost"] == "3.00" # the flag is committed though the approval is refused
refused(approve(rid), 409, "recompute the cost first") # flagged: refused again, no hr call needed
refused(api("POST", f"/leave-requests/{rid}/recompute", MANAGER), 404, "no leave request") # only the requester recomputes
got = must(api("POST", f"/leave-requests/{rid}/recompute", EMPLOYEE))
assert got["cost"] == "3.00" and got["stale"] is False and len(got["days"]) == 6 and count(t.tof_day_claim) == 6
assert must(approve(rid))["status"] == "approved" and entries_of(4410) == [(f"req:{rid}:r1", "take", "-3.0000")]
refused(api("POST", f"/leave-requests/{rid}/recompute", EMPLOYEE), 409, "only a request that waits")
def test_an_employment_that_ended_before_approval_marks_the_request_stale_too(hr):
rid = request("2026-12-21", "2026-12-24")["request_id"]
hr.people[4410]["end"] = date(2026, 12, 22)
refused(approve(rid), 409, "hr's calendar or pattern changed")
refused(api("POST", f"/leave-requests/{rid}/recompute", EMPLOYEE), 409, "not employed on 2026-12-23")
# ── the balance is a pure sum and no one else's row changes it ──────────────────────────────────
def test_the_balance_is_order_independent():
rows = [("carry:4410:ANNUAL:2025", "carry", "2.00"), ("adj:4410:ANNUAL:2026:1", "adjust", "-1.50"), ("adj:4410:ANNUAL:2026:2", "adjust", "3.25"),
("lapse:4410:ANNUAL:2025", "lapse", "-0.75"), ("open:4410:ANNUAL:2026", "opening", "1.00")]
answers = set()
for k in range(5):
random.Random(k).shuffle(rows)
with use_cases.repository() as repo:
repo.conn.execute(t.tof_balance_entry.delete())
for ref, kind, q in rows:
put_entry(4410, ref, kind, q)
answers.add((balance()["available"], balance()["adjustments"], balance()["carried"], balance()["lapsed"]))
assert answers == {("29.00", "2.75", "2.00", "-0.75")} # 25 + 2 carried + (-1.50 + 3.25 + 1.00) - 0.75 lapsed
def test_the_ledger_is_the_sum_of_what_was_taken_and_given_back_whatever_the_history():
a = decided("2026-12-21", "2026-12-23")["request_id"]
b = decided("2026-12-29", "2026-12-30")["request_id"]
must(api("POST", f"/leave-requests/{a}/cancel", MANAGER, {"reason": "x"}))
bal = balance()
spent = -sum(Decimal(q) for _, k, q in entries_of(4410) if k in ("take", "reverse"))
assert (bal["taken"], bal["booked"]) == ("0.00", "2.00") and spent == Decimal("2.0000") and bal["available"] == "23.00"
assert get(b)["status"] == "approved" and get(a)["status"] == "cancelled"
def seed_others(n, first=0):
"""n unrelated people with a request, its half days and a ledger entry each - straight into the tables."""
at = use_cases._now()
with use_cases.repository() as repo:
for k in range(first, first + n):
pid, rid = 6000 + k, f"UK01-TOF-2026-9{k:04d}"
who = rq.Identity(f"p{k}@uk.example", f"sub-{pid}")
r = rq.Request(rid, pid, "UK01", who, "ANNUAL", "UK-ANNUAL", 2026, date(2026, 12, 21), "full", date(2026, 12, 22), "full", Decimal("2"), "days",
2280, rq.Identity(*("mgr2280@uk.example", "sub-2280")))
repo.requests.create(r, at)
from brick_time_off.domain.calendar import Slot
repo.days.add(pid, rid, [Slot(date(2026, 12, d), p, 225, Decimal("0.5")) for d in (21, 22) for p in ("am", "pm")], at)
repo.entries.add(f"req:{rid}:r1", pid, "ANNUAL", 2026, "take", Decimal("-2"), "days", date(2026, 12, 21), "m", "s", "", at)
def test_other_peoples_rows_change_none_of_my_answers():
rid = request("2026-12-21", "2026-12-23")["request_id"]
before = (balance(), get(rid), must(api("GET", "/leave-requests", EMPLOYEE)), must(api("GET", "/leave-requests", MANAGER, {"scope": "waiting"})))
seed_others(10)
after = (balance(), get(rid), must(api("GET", "/leave-requests", EMPLOYEE)))
assert after == before[:3]
assert len(must(api("GET", "/leave-requests", MANAGER, {"scope": "waiting"}))["requests"]) == len(before[3]["requests"]) + 10 # theirs: the same approver
assert len(must(api("GET", "/leave-requests", EMPLOYEE))["requests"]) == 1
assert request("2026-12-29")["status"] == "requested" # days of others never collide with mine
assert balance(EDINBURGH)["taken"] == "0.00"
def test_a_half_day_of_one_person_never_blocks_another_person_the_same_day():
request("2026-12-21")
assert request("2026-12-21", who=EDINBURGH)["status"] == "requested"
assert count(t.tof_day_claim) == 4