dodil/erp-brick-time-off

DODIL ERP brick: time off

tests/test_sickness.py

"""Slice B item 2: sickness is RECORDED (TOF-07), a fit note is asked from the 8th day (TOF-08), and payroll reads the facts for statutory sick pay
(qualifying days, periods of incapacity, linked periods) - never the money (TOF-20)."""
import base64
from datetime import date

from conftest import (
    EMPLOYEE,
    HR_ADMIN,
    MANAGER,
    NOBODY,
    OWNER,
    PAYROLL_CLIENT,
    PDF,
    api,
    as_role,
    decided,
    entries_of,
    must,
    on_day,
    put_entry,
    refused,
    request,
    tool,
    tools,
)

from brick_time_off.application import use_cases

DOC = base64.b64encode(PDF).decode()
OTHER_DOC = base64.b64encode(PDF + b"second").decode()
D10, D13 = "2026-11-10T09:00:00+00:00", "2026-11-13T09:00:00+00:00"


def record(first, back, who=None, **kw):
    return api("POST", "/sickness", who or EMPLOYEE, {"first_day": first, "expected_return": back, **kw})


def sick(first="2026-11-10", back="2026-11-13", who=None, at=D10, **kw):
    with on_day(at):
        return must(record(first, back, who, **kw), 201)


def confirm(aid, last, who=None, at=D13):
    with on_day(at):
        return api("POST", f"/sickness/{aid}/confirm-return", who or MANAGER, {"last_day_off": last})


def fit_note(aid, a, b, who=None, doc=DOC, late=False, at=D10, **kw):
    with on_day(at):
        return api("POST", f"/sickness/{aid}/{'late-fit-note' if late else 'fit-note'}", who or EMPLOYEE,
                   {"covers_from": a, "covers_to": b, "content_type": "application/pdf", "content_base64": doc, **kw})


def payroll(first="2026-11-01", last="2026-11-30", **q):
    return api("GET", "/absences", PAYROLL_CLIENT, {"company_code": "UK01", "from": first, "to": last, **q})


def newcomer(hr, pid, email):
    hr.person(pid, email, manager=hr.as_manager(2280), manager2=hr.as_manager(1904))
    return as_role("employee", email, pid)


def test_TOF_07_sickness_is_recorded_then_closed_by_the_manager_with_its_working_days(hr):
    s = sick()
    assert (s["status"], s["absence_id"].startswith("UK01-TOF-2026-"), s["self_certified"], s["pay_treatment"]) == ("open", True, True, "sick")
    assert [d["date"] for d in s["days"]] == ["2026-11-10"] * 2 + ["2026-11-11"] * 2 + ["2026-11-12"] * 2       # claimed up to the day before the return
    closed = must(confirm(s["absence_id"], "2026-11-12"))
    assert (closed["status"], closed["revision"], closed["last_day_off"], closed["confirmed_by"]) == ("closed", 2, "2026-11-12", "mgr2280@uk.example")
    assert closed["payroll"] == {"qualifying_days": 3, "piw": False, "linked_to": None, "linked_group": s["absence_id"]}
    assert len(closed["days"]) == 6
    assert entries_of(4410) == []                                                           # sickness never touches the leave balance
    assert must(api("GET", "/balance", EMPLOYEE))["available"] == "25.00"


def test_TOF_07_confirming_an_earlier_last_day_releases_the_days_after_it(hr):
    s = sick()
    closed = must(confirm(s["absence_id"], "2026-11-11"))
    assert [d["date"] for d in closed["days"]] == ["2026-11-10"] * 2 + ["2026-11-11"] * 2
    assert must(api("POST", "/leave-requests", EMPLOYEE, {"from_date": "2026-11-12"}), 201)["status"] == "requested"     # the freed day is bookable


def test_TOF_07_the_reason_is_seen_by_the_sick_person_and_holders_of_read_detail_only(hr):
    s = sick(reason_detail="migraine")
    aid = s["absence_id"]
    assert s["reason"] == "migraine"
    assert must(api("GET", f"/sickness/{aid}", EMPLOYEE))["reason"] == "migraine"
    assert must(api("GET", f"/sickness/{aid}", HR_ADMIN))["reason"] == "migraine"
    assert must(api("GET", f"/sickness/{aid}", MANAGER))["reason"] == "sick"
    assert must(api("GET", f"/sickness/{aid}", OWNER))["reason"] == "sick"                  # the owner reads every record, not the reason
    status, got = tool(MANAGER, "get_time_off_sickness", {"absence_id": aid})
    assert status == "ok" and got["reason"] == "sick"
    refused(api("GET", f"/sickness/{aid}", NOBODY), 403, "missing permission")
    stranger = as_role("employee", "stranger@uk.example", 4999)
    refused(api("GET", f"/sickness/{aid}", stranger), 404, "no such sickness record")


def test_TOF_07_a_manager_or_hr_records_for_a_person_and_a_stranger_may_not(hr):
    with on_day(D10):
        got = must(record("2026-11-10", "2026-11-12", MANAGER, person_id="4410"), 201)
        assert (got["person_id"], got["recorded_by"], got["reason"]) == ("4410", "mgr2280@uk.example", "sick")
        hr.person(4600, "emp4600@uk.example", manager=hr.as_manager(1904))                   # managed by 1904, not 2280
        refused(record("2026-11-10", "2026-11-12", MANAGER, person_id="4600"), 403, "people you manage")
        assert must(record("2026-11-10", "2026-11-12", HR_ADMIN, person_id="4600"), 201)["person_id"] == "4600"
        refused(record("2026-11-10", "2026-11-12", EMPLOYEE, person_id="5120"), 403)         # an employee records their own only


def test_TOF_07_a_manager_records_only_while_the_company_lets_managers(hr):
    with use_cases.repository() as repo:
        repo.settings.put("manager_records_sickness", "false", date(2026, 1, 1), "owner", use_cases._now())
    with on_day(D10):
        refused(record("2026-11-10", "2026-11-12", MANAGER, person_id="4410"), 403)
        assert must(record("2026-11-10", "2026-11-12", HR_ADMIN, person_id="4410"), 201)


def test_TOF_07_a_day_already_in_leave_or_sickness_is_refused_and_nothing_is_claimed_twice(hr):
    decided("2026-11-11")                                                                   # 4410 approved off on the 11th
    with on_day(D10):
        refused(record("2026-11-10", "2026-11-13"), 409, "2026-11-11 am is already in request")
    s = sick("2026-11-12", "2026-11-13", at="2026-11-12T09:00:00+00:00")
    with on_day("2026-11-12T09:00:00+00:00"):
        body = refused(record("2026-11-12", "2026-11-13"), 409, "day_free")
        assert s["absence_id"] in str(body)


def test_TOF_07_the_person_must_be_employed_and_work_a_day_of_it(hr):
    with on_day("2026-11-14T09:00:00+00:00"):                                                # a Saturday: no working day
        refused(record("2026-11-14", "2026-11-16"), 422, "empty")
    hr.people[4410]["end"] = date(2026, 11, 10)
    with on_day(D10):
        refused(record("2026-11-10", "2026-11-13"), 409, "not employed on 2026-11-11")


def test_TOF_07_the_business_date_is_the_date_in_london_never_the_utc_date(hr):
    with on_day("2026-07-01T23:30:00+00:00"):                                                # 00:30 on 2 July in London (BST)
        assert must(record("2026-07-02", "2026-07-03"), 201)["first_day"] == "2026-07-02"
        refused(record("2026-07-03", "2026-07-04"), 422, "bounds")                           # tomorrow in London


def test_TOF_07_bounds_are_422_never_500(hr):
    with on_day(D10):
        for body in ({"first_day": "2026-11-11", "expected_return": "2026-11-13"},          # a future start
                     {"first_day": "2026-11-10", "expected_return": "2026-11-10"},          # back the day it starts
                     {"first_day": "2026-11-10", "expected_return": "2027-12-01"},          # over a year
                     {"first_day": "not a date", "expected_return": "2026-11-13"},
                     {"first_day": "1999-01-01", "expected_return": "2026-11-13"},
                     {"first_day": "2026-11-10", "expected_return": "2026-11-13", "reason_detail": "x" * 501}):
            got, out = api("POST", "/sickness", EMPLOYEE, body)
            assert got == 422, (body, got, out)


def test_TOF_07_nobody_confirms_their_own_return_under_either_login(hr):
    s = sick()
    refused(api("POST", f"/sickness/{s['absence_id']}/confirm-return", EMPLOYEE, {"last_day_off": "2026-11-12"}), 409, "second_person")
    own_hr = as_role("hr_admin", "hr3100@uk.example", 3100)
    mine = sick("2026-11-10", "2026-11-12", own_hr)
    refused(confirm(mine["absence_id"], "2026-11-11", own_hr), 409, "second_person")                                    # HR does not confirm their own
    second_login = as_role("hr_admin", "hr3100.alt@uk.example", 3100)                                                    # the same sub, another email
    refused(confirm(mine["absence_id"], "2026-11-11", second_login), 409, "second_person")
    assert must(confirm(mine["absence_id"], "2026-11-11", as_role("hr_admin", "hr-other@uk.example", 3199)))["status"] == "closed"


def test_TOF_07_a_return_cannot_be_confirmed_in_the_future_before_the_start_or_past_the_expected_day(hr):
    s = sick()
    for last, status, words in (("2026-11-09", 422, "bounds"), ("2026-11-14", 422, "bounds")):
        refused(confirm(s["absence_id"], last), status, words)
    with on_day("2026-11-11T09:00:00+00:00"):
        refused(api("POST", f"/sickness/{s['absence_id']}/confirm-return", MANAGER, {"last_day_off": "2026-11-12"}), 422, "bounds")   # not yet


def test_TOF_07_extending_adds_days_bumps_the_revision_and_payroll_sees_what_changed(hr):
    s = sick()
    first = must(payroll())
    row = first["absences"][0]
    assert (row["absence_ref"], row["revision"], row["status"], row["pay_treatment"], len(row["days"])) == (s["absence_id"], 1, "recorded", "sick", 6)
    with on_day(D13):
        got = must(api("POST", f"/sickness/{s['absence_id']}/extend", EMPLOYEE, {"expected_return": "2026-11-17"}))
    assert (got["revision"], got["expected_return"], got["payroll"]["qualifying_days"], len(got["days"])) == (2, "2026-11-17", 5, 10)
    assert got["self_certified"] is True and got["payroll"]["piw"] is True and got["evidence_ok"] is True      # 7 calendar days: still no fit note
    again = must(payroll(changed_since="2026-11-10T09:30:00Z"))["absences"][0]
    assert (again["revision"], len(again["days"]), again["sickness"]["qualifying_days"]) == (2, 10, 5)
    assert must(payroll(changed_since="2026-11-20T00:00:00Z"))["absences"] == []                                # nothing changed since
    with on_day(D13):
        refused(api("POST", f"/sickness/{s['absence_id']}/extend", EMPLOYEE, {"expected_return": "2026-11-17"}), 422, "bounds")   # not an extension
        busy = decided("2026-11-17")                                                                              # the 17th is booked leave
        refused(api("POST", f"/sickness/{s['absence_id']}/extend", EMPLOYEE, {"expected_return": "2026-11-19"}), 409, "day_free")
        assert busy["status"] == "approved"


def test_TOF_08_a_fit_note_is_asked_from_the_8th_day_and_the_upload_is_stored_as_an_object(hr, store):
    s = sick("2026-11-02", "2026-11-21", at="2026-11-02T09:00:00+00:00", reason_detail="flu")
    aid = s["absence_id"]
    assert (s["self_certified"], s["evidence_ok"], s["fit_note_due_from"], s["gates"][0]["met"]) == (False, False, "2026-11-09", False)
    assert must(api("GET", "/sickness", MANAGER, {"scope": "team"}))["sickness"][0]["fit_note_missing"] is True
    part = must(fit_note(aid, "2026-11-02", "2026-11-12"))
    assert part["evidence_ok"] is False                                                                     # covers to the 12th, the spell runs to the 20th
    got = must(fit_note(aid, "2026-11-02", "2026-11-20", doc=OTHER_DOC))
    assert got["evidence_ok"] is True and got["gates"][0]["met"] is True and len(got["fit_notes"]) == 2
    key = got["fit_notes"][1]["object_key"]
    assert key.startswith(f"raw/time-off/{aid}/") and key.endswith(".pdf") and key in store.objects
    assert must(api("GET", "/sickness", MANAGER, {"scope": "team"}))["sickness"][0]["fit_note_missing"] is False
    assert must(payroll("2026-11-01", "2026-11-30"))["absences"][0]["evidence_ok"] is True


def test_TOF_08_the_same_document_twice_is_one_object_and_one_row(hr, store):
    s = sick("2026-11-02", "2026-11-21", at="2026-11-02T09:00:00+00:00")
    fit_note(s["absence_id"], "2026-11-02", "2026-11-20")
    again = must(fit_note(s["absence_id"], "2026-11-02", "2026-11-20"))
    assert len(again["fit_notes"]) == 1 and store.puts == 1


def test_TOF_08_a_fit_note_is_uploaded_by_the_sick_person_or_hr_and_is_checked(hr, store):
    s = sick("2026-11-02", "2026-11-21", at="2026-11-02T09:00:00+00:00")
    aid = s["absence_id"]
    refused(fit_note(aid, "2026-11-02", "2026-11-20", MANAGER), 403)                                       # the manager asks for it, the person supplies it
    assert must(fit_note(aid, "2026-11-02", "2026-11-20", HR_ADMIN, doc=OTHER_DOC))["evidence_ok"] is True
    for kw, status in (({"a": "2026-11-01", "b": "2026-11-05"}, 422),                                         # before the first day
                       ({"a": "2026-11-05", "b": "2026-11-04"}, 422),                                         # ends before it starts
                       ({"a": "2026-11-02", "b": "2026-11-05", "doc": base64.b64encode(b"not a pdf").decode()}, 422),
                       ({"a": "2026-11-02", "b": "2026-11-05", "doc": "!!!"}, 422),
                       ({"a": "2026-11-02", "b": "2026-11-05", "doc": ""}, 422)):
        refused(fit_note(aid, kw["a"], kw["b"], doc=kw.get("doc", DOC)), status)
    assert len(store.objects) == 1


def test_TOF_08_no_store_is_a_503_and_nothing_is_recorded(hr):
    s = sick("2026-11-02", "2026-11-21", at="2026-11-02T09:00:00+00:00")
    refused(fit_note(s["absence_id"], "2026-11-02", "2026-11-20"), 503, "evidence store")
    assert must(api("GET", f"/sickness/{s['absence_id']}", EMPLOYEE))["fit_notes"] == []


def test_TOF_08_a_late_fit_note_changes_the_evidence_never_the_days(hr, store):
    s = sick("2026-11-02", "2026-11-14", at="2026-11-02T09:00:00+00:00")
    aid = s["absence_id"]
    refused(fit_note(aid, "2026-11-02", "2026-11-13", late=True), 409, "needs it to be closed")
    with on_day("2026-11-14T09:00:00+00:00"):
        must(api("POST", f"/sickness/{aid}/confirm-return", MANAGER, {"last_day_off": "2026-11-13"}))
    before = must(payroll())["absences"][0]
    assert (before["status"], before["evidence_ok"], len(before["days"])) == ("closed", False, 20)
    refused(fit_note(aid, "2026-11-02", "2026-11-13"), 409, "needs it to be open")
    late = must(fit_note(aid, "2026-11-02", "2026-11-13", late=True, at="2026-11-20T09:00:00+00:00"))
    assert late["evidence_ok"] is True and late["revision"] == before["revision"] and len(late["days"]) == 20
    after = must(payroll(changed_since="2026-11-19T00:00:00Z"))["absences"][0]
    assert after["evidence_ok"] is True and after["revision"] == before["revision"]


def test_the_sickness_machine_only_moves_forward(hr, store):
    s = sick()
    aid = s["absence_id"]
    must(confirm(aid, "2026-11-12"))
    for path, who, body in (("confirm-return", MANAGER, {"last_day_off": "2026-11-12"}), ("extend", EMPLOYEE, {"expected_return": "2026-11-20"}),
                            ("void", HR_ADMIN, {"reason": "mistake"})):
        with on_day(D13):
            refused(api("POST", f"/sickness/{aid}/{path}", who, body), 409, "needs it to be open")
    v = sick("2026-11-10", "2026-11-12", as_role("employee", "emp5120@uk.example", 5120))
    voided = must(api("POST", f"/sickness/{v['absence_id']}/void", HR_ADMIN, {"reason": "recorded for the wrong person"}))
    assert voided["status"] == "voided" and voided["void_reason"]
    with on_day(D13):
        for path, who, body in (("confirm-return", MANAGER, {"last_day_off": "2026-11-10"}), ("extend", as_role("employee", "emp5120@uk.example", 5120),
                                                                                         {"expected_return": "2026-11-20"})):
            refused(api("POST", f"/sickness/{v['absence_id']}/{path}", who, body), 409, "voided")
    with on_day(D10):
        assert must(record("2026-11-10", "2026-11-12", as_role("employee", "emp5120@uk.example", 5120)), 201)       # voiding released the days: sick again


def test_a_void_is_hrs_with_a_reason_and_releases_the_days(hr):
    s = sick()
    aid = s["absence_id"]
    refused(api("POST", f"/sickness/{aid}/void", MANAGER, {"reason": "mistake"}), 403)
    refused(api("POST", f"/sickness/{aid}/void", HR_ADMIN, {"reason": "  "}), 422, "reason")
    must(api("POST", f"/sickness/{aid}/void", HR_ADMIN, {"reason": "mistake"}))
    assert must(api("POST", "/leave-requests", EMPLOYEE, {"from_date": "2026-11-11"}), 201)["status"] == "requested"     # the day is free again
    assert [a["absence_ref"] for a in must(payroll())["absences"]] == []                                              # payroll no longer sees it...
    gone = must(payroll(changed_since="2026-11-01T00:00:00Z"))["absences"]
    assert [(a["absence_ref"], a["status"], a["days"]) for a in gone] == [(aid, "cancelled", [])]                    # ...unless it asks what changed


def test_TOF_07_sickness_links_to_an_earlier_period_within_eight_weeks_only(hr):
    first = sick("2026-11-02", "2026-11-07", at="2026-11-02T09:00:00+00:00")                                        # 5 calendar days: a PIW
    must(confirm(first["absence_id"], "2026-11-06", at="2026-11-07T09:00:00+00:00"))
    second = sick("2026-12-14", "2026-12-18", at="2026-12-14T09:00:00+00:00")                                       # 38 days after it ended
    assert (second["payroll"]["linked_to"], second["payroll"]["linked_group"]) == (first["absence_id"], first["absence_id"])
    must(confirm(second["absence_id"], "2026-12-17", at="2026-12-18T09:00:00+00:00"))
    third = sick("2027-02-15", "2027-02-17", at="2027-02-15T09:00:00+00:00")                                        # 60 days after the second ended
    assert (third["payroll"]["linked_to"], third["payroll"]["linked_group"]) == (None, third["absence_id"])
    short = sick("2026-11-02", "2026-11-04", who=as_role("employee", "emp5120@uk.example", 5120), at="2026-11-02T09:00:00+00:00")
    must(confirm(short["absence_id"], "2026-11-03", at="2026-11-04T09:00:00+00:00"))                                # 2 days: not a PIW
    after = sick("2026-11-16", "2026-11-18", who=as_role("employee", "emp5120@uk.example", 5120), at="2026-11-16T09:00:00+00:00")
    assert after["payroll"]["linked_to"] is None                                                                    # only a PIW links


def test_TOF_20_payroll_reads_approved_and_recorded_absences_of_its_company_in_order(hr):
    leave = decided("2026-11-16")["request_id"]
    waiting = request("2026-11-23")["request_id"]
    s = sick()
    rows = must(payroll())["absences"]
    assert [(r["absence_ref"], r["status"]) for r in rows] == sorted([(leave, "approved"), (s["absence_id"], "recorded")], key=lambda x: x[0])
    assert waiting not in str(rows)                                                                         # a request still waiting is never visible
    assert [d["part"] for d in next(r for r in rows if r["absence_ref"] == leave)["days"]] == ["am", "pm"]
    hr.person(7001, "uk2@uk.example", company="DE01")
    assert must(api("GET", "/absences", PAYROLL_CLIENT, {"company_code": "DE01", "from": "2026-11-01", "to": "2026-11-30"}))["absences"] == []
    assert must(payroll(person_id="5120"))["absences"] == []
    assert len(must(payroll("2026-11-12", "2026-11-12"))["absences"]) == 1 and must(payroll("2026-11-12", "2026-11-12"))["absences"][0]["days"][0]["date"] == "2026-11-12"


def test_TOF_20_a_cancelled_absence_appears_only_with_changed_since_and_bounds_are_422(hr):
    leave = decided("2026-12-14")["request_id"]
    must(api("POST", f"/leave-requests/{leave}/cancel", MANAGER, {"reason": "plans changed"}))
    assert must(payroll("2026-12-01", "2026-12-31"))["absences"] == []
    got = must(payroll("2026-12-01", "2026-12-31", changed_since="2026-11-01T00:00:00Z"))["absences"]
    assert [(r["absence_ref"], r["status"], r["days"]) for r in got] == [(leave, "cancelled", [])]
    for q in ({"from": "2026-11-01", "to": "2027-03-01"}, {"from": "2026-11-30", "to": "2026-11-01"}, {"from": "x", "to": "2026-11-01"},
              {"from": "2026-11-01", "to": "2026-11-30", "changed_since": "yesterday"}):
        refused(api("GET", "/absences", PAYROLL_CLIENT, {"company_code": "UK01", **q}), 422, "bounds")
    refused(api("GET", "/absences", EMPLOYEE, {"company_code": "UK01", "from": "2026-11-01", "to": "2026-11-30"}), 403)
    refused(api("GET", "/absences", HR_ADMIN, {"company_code": "UK01", "from": "2026-11-01", "to": "2026-11-30"}), 403)
    assert "list_tof_absences_for_payroll" not in tools(EMPLOYEE)


def test_flat_statements_the_absences_read_does_not_grow_with_other_companies_or_people(hr, engine):
    from erp_platform.testing import count_statements
    sick()
    with count_statements(engine) as few:
        must(payroll())
    for i in range(12):
        who = newcomer(hr, 7100 + i, f"x{i}@uk.example")
        sick("2026-11-10", "2026-11-12", who=who)
        put_entry(7100 + i, f"adj:{7100 + i}", "adjust", "1")
    with count_statements(engine) as many:
        rows = must(payroll())["absences"]
    assert len(rows) == 13 and many.statements == few.statements, (few.statements, many.statements)