dodil/erp-brick-time-off
DODIL ERP brick: time off
tests/test_migration.py
"""Slice B item 5: the Odoo slice (TOF-23, workflow tof_migration_run): validated leaves as approved requests, allocations as opening entries, reconciled to
Odoo and signed by the owner. Odoo is read through a port; here it answers fake rows (no live Odoo)."""
from datetime import date
import pytest
from conftest import EMPLOYEE, HR_ADMIN, MANAGER, OWNER, PARTTIMER, PAYROLL_CLIENT, SYSTEM, api, balance, entries_of, must, refused, request, tools
from brick_time_off.infrastructure import odoo as odoo_port
DB = "acme"
RUN = "odoo:acme:2026-10-31"
EMPLOYEES = {"11": 4410, "12": 5120}
TYPES = {"1": "ANNUAL", "2": "UNPAID"}
def leave(lid, emp, typ, a, b=None, days=1, state="validate", hours=0, half=False, period=False, in_hours=False):
return {"id": lid, "employee_id": [emp, f"Employee {emp}"], "holiday_status_id": [typ, "T"], "request_date_from": a, "request_date_to": b or a,
"request_unit_half": half, "request_date_from_period": period, "request_unit_hours": in_hours, "number_of_days": days, "number_of_hours": hours, "state": state}
def alloc(aid, emp, typ, days, start="2026-01-01"):
return {"id": aid, "employee_id": [emp, f"Employee {emp}"], "holiday_status_id": [typ, "T"], "number_of_days": days, "date_from": start, "date_to": False,
"state": "validate"}
class FakeOdoo:
def __init__(self):
self.down = None
self.type_rows = [{"id": 1, "name": "Paid Time Off"}, {"id": 2, "name": "Unpaid"}, {"id": 3, "name": "Unused"}]
self.leave_rows = [
leave(1, 11, 1, "2026-03-02", "2026-03-06", 5), # a week of annual leave
leave(2, 11, 1, "2026-04-07", days=0.5, half=True, period="am"), # a half day
leave(3, 11, 2, "2026-05-11", days=1), # unpaid: no balance here
leave(4, 12, 1, "2026-06-01", "2026-06-03", 3, hours=22.5), # the part-timer books in hours
]
self.alloc_rows = [alloc(1, 11, 1, 28), alloc(2, 12, 1, 20)]
self.waiting_rows = []
def _check(self):
if self.down:
raise odoo_port.OdooUnavailable(502, self.down)
def leave_types(self, db):
self._check()
return self.type_rows
def leaves(self, db, years):
self._check()
return [x for x in self.leave_rows if int(x["request_date_from"][:4]) in years]
def allocations(self, db, years):
self._check()
return [x for x in self.alloc_rows if int(x["date_from"][:4]) in years]
def waiting(self, db, years):
self._check()
return self.waiting_rows
@pytest.fixture
def odoo():
fake = FakeOdoo()
odoo_port.ports.reader = fake
yield fake
odoo_port.ports.reader = None
def load_body(**kw):
return {"odoo_db": DB, "cutover_date": "2026-10-31", "company_code": "UK01", "years": [2026], "employee_map": EMPLOYEES, "leave_types": TYPES,
"upstream_signed": True, "waiting": "finished_in_odoo", **kw}
def load(who=None, **kw):
return api("POST", "/migration/odoo/load", who or SYSTEM, load_body(**kw))
def reconcile(who=None, run=RUN):
return api("POST", f"/migration/odoo/{run}/reconcile", who or SYSTEM)
def req(rid, who=None):
return api("GET", f"/leave-requests/odoo:{DB}:hr.leave:{rid}", who or HR_ADMIN)
def test_TOF_23_validated_leaves_become_approved_requests_and_allocations_opening_entries(hr, odoo):
got = must(load())
assert (got["run_id"], got["status"]) == (RUN, "loaded")
assert got["stats"] == {"leaves": 4, "leaves_written": 4, "leaves_unchanged": 0, "openings_written": 2, "allocations": 2, "waiting_not_carried": [], "days_written": 19}
one = must(req(1))
assert (one["status"], one["cost"], one["unit"], one["leave_type_code"], one["from_date"], one["to_date"], one["approved_by"]) == \
("approved", "5.00", "days", "ANNUAL", "2026-03-02", "2026-03-06", "odoo:acme")
assert len(one["days"]) == 10 and one["leave_year"] == 2026
half = must(req(2))
assert (half["cost"], [(d["date"], d["part"]) for d in half["days"]]) == ("0.50", [("2026-04-07", "am")])
assert must(req(4))["unit"] == "hours" and must(req(4))["cost"] == "22.50" # Odoo's own number, in the person's unit
mine = dict((e[0], e[2]) for e in entries_of(4410))
assert mine["req:odoo:acme:hr.leave:1:r1"] == "-5.0000" and mine["open:4410:ANNUAL:2026"] == "3.0000" # 28 allocated - the 25 derived
def test_TOF_23_available_days_equal_odoos_to_the_hundredth_for_days_and_hours(hr, odoo):
must(load())
assert balance()["available"] == "22.50" # Odoo: 28 allocated - 5.5 taken
pt = balance(who=PARTTIMER)
assert (pt["unit"], pt["available"]) == ("hours", "127.50") # 20 days x 7.5 hours - 22.5
assert [e for e in balance()["entries"] if e["kind"] == "opening"][0]["reason"] == "opening balance from Odoo's allocations"
def test_TOF_23_migrated_days_are_held_so_nobody_books_them_and_payroll_reads_them(hr, odoo):
must(load())
refused(api("POST", "/leave-requests", EMPLOYEE, {"from_date": "2026-03-04"}), 409, "day_free")
rows = must(api("GET", "/absences", PAYROLL_CLIENT, {"company_code": "UK01", "from": "2026-03-01", "to": "2026-03-31"}))["absences"]
assert [(r["absence_ref"], r["status"], r["pay_treatment"], len(r["days"])) for r in rows] == [("odoo:acme:hr.leave:1", "approved", "paid", 10)]
def test_TOF_23_a_second_load_writes_nothing_new(hr, odoo):
must(load())
before = (entries_of(), len(must(api("GET", "/migration/odoo", OWNER))["runs"]))
again = must(load())
assert again["stats"]["leaves_written"] == 0 and again["stats"]["leaves_unchanged"] == 4 and again["stats"]["openings_written"] == 0
assert (entries_of(), len(must(api("GET", "/migration/odoo", OWNER))["runs"])) == before
def test_TOF_23_a_leave_odoo_gained_after_the_first_load_is_added_and_the_run_is_reconciled_again(hr, odoo):
must(load())
must(reconcile())
odoo.leave_rows.append(leave(5, 11, 1, "2026-07-06", "2026-07-07", 2))
got = must(load())
assert got["status"] == "loaded" and got["stats"]["leaves_written"] == 1 and got["stats"]["openings_written"] == 0
assert must(reconcile())["checks"][0]["equal"] is True and balance()["available"] == "20.50"
def test_TOF_23_reconciled_equal_then_signed_by_the_owner_and_frozen(hr, odoo):
must(load())
got = must(reconcile())
assert got["status"] == "reconciled" and [c["equal"] for c in got["checks"]] == [True, True]
c1 = got["checks"][0]
assert c1["detail"]["11:ANNUAL:2026"] == {"odoo": "22.50", "ours": "22.50"} and c1["detail"]["12:ANNUAL:2026"] == {"odoo": "17.00", "ours": "17.00"}
assert got["checks"][1]["detail"]["2026-06"] == {"odoo": "3.00", "ours": "3.00"} # the hours request is 3 days at Odoo's day
refused(api("POST", f"/migration/odoo/{RUN}/sign", HR_ADMIN), 403, "tof:migration:sign")
signed = must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))
assert (signed["status"], signed["signed_by"]) == ("signed", "owner@uk.example")
assert must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))["status"] == "signed" # twice is one signature
refused(load(), 409, "is signed")
refused(reconcile(), 409, "is signed")
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "oops"}), 409, "is signed")
def test_TOF_23_a_difference_must_be_fixed_or_explained_before_signing(hr, odoo):
must(load())
odoo.leave_rows[0]["number_of_days"] = 4.5 # Odoo says the week cost 4.5 (it counted a half day more)
got = must(reconcile())
assert [c["equal"] for c in got["checks"]] == [False, False] and got["readiness"][0]["met"] is False
body = refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "checks_reconciled")
assert body["gate"] == "checks_reconciled"
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "approved_days_per_month", "reason": " "}), 422, "reason")
for check in ("available_per_person_and_type", "approved_days_per_month"):
must(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": check, "reason": "Odoo counted a half day more over Easter"}))
again = must(reconcile()) # the explanation survives a second reconcile
assert all(c["explained"] for c in again["checks"])
assert must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))["status"] == "signed"
def test_TOF_23_nothing_to_explain_when_equal_and_only_a_reconciled_run_is_explained(hr, odoo):
must(load())
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "approved_days_per_month", "reason": "x"}), 409, "only a reconciled run")
refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "reconcile it first")
must(reconcile())
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "approved_days_per_month", "reason": "x"}), 409, "nothing to explain")
def test_TOF_23_the_doors_are_answered_for_everything_odoo_holds_or_nothing_is_loaded(hr, odoo):
refused(load(employee_map={"11": 4410}), 409, "Odoo employee 12 has no person in employee_map")
refused(load(leave_types={"1": "ANNUAL"}), 409, "Odoo leave type 2 has no leave_type_code")
refused(load(leave_types={**TYPES, "9": "ANNUAL"}), 409, "Odoo has no leave type 9")
refused(load(leave_types={**TYPES, "2": "NO_SUCH"}), 409, "NO_SUCH")
refused(load(upstream_signed=False), 409, "hr's migration must be signed first")
odoo.waiting_rows = [leave(7, 11, 1, "2026-11-02", state="confirm"), leave(8, 11, 1, "2026-11-09", state="validate1")]
refused(load(), 409, "2 requests are still waiting for approval in Odoo (7, 8)")
assert entries_of() == [] # nothing was written by any refusal
got = must(load(waiting="re_enter"))
assert got["stats"]["waiting_not_carried"] == [7, 8] and got["stats"]["leaves_written"] == 4
def test_TOF_23_what_cannot_be_carried_is_named_and_nothing_is_loaded(hr, odoo):
odoo.leave_rows.append(leave(5, 11, 1, "2026-09-07", days=0.25, in_hours=True)) # an hours-of-the-day request
odoo.leave_rows.append(leave(6, 11, 1, "2026-09-12", days=1)) # a Saturday: no working day
request("2026-03-03") # a request of their own on a migrated day
body = refused(load(), 409, "cannot be carried")
text = str(body)
assert "leave 5 is booked in hours of the day" in text and "leave 6: hr says there is no working day" in text and "overlaps" in text
assert [e for e in entries_of() if e[0].startswith(("open:", "req:odoo"))] == []
def test_TOF_23_a_leave_on_a_day_the_person_was_not_employed_is_named(hr, odoo):
hr.people[4410]["start"] = date(2026, 3, 4)
refused(load(), 409, "hr says the person was not employed on 2026-03-02")
def test_TOF_23_no_odoo_connection_or_odoo_down_is_a_reply(hr):
refused(load(), 409, "no Odoo connection is configured")
def test_TOF_23_odoo_down_loads_nothing(hr, odoo):
odoo.down = "Odoo did not answer (TimeoutError)"
refused(load(), 502, "Odoo did not answer")
assert entries_of() == []
def test_TOF_23_a_loaded_run_is_discarded_and_the_leaves_can_be_loaded_again(hr, odoo):
must(load())
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": " "}), 422, "reason")
refused(api("POST", f"/migration/odoo/{RUN}/discard", HR_ADMIN, {"reason": "wrong"}), 403)
got = must(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "wrong door"}))
assert (got["status"], got["discard_reason"]) == ("discarded", "wrong door")
assert [e for e in entries_of() if e[0].startswith(("open:", "req:odoo"))] == []
refused(req(1), 404)
assert balance()["available"] == "25.00"
again = must(load())
assert again["run_id"] == f"{RUN}:2" and again["stats"]["leaves_written"] == 4 and balance()["available"] == "22.50"
refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "is discarded")
def test_TOF_23_a_run_whose_leaves_have_moved_since_the_load_is_not_discarded(hr, odoo):
odoo.leave_rows.append(leave(5, 11, 1, "2026-12-14", "2026-12-15", 2)) # booked after the cut-over: still cancellable
must(load())
must(api("POST", "/leave-requests/odoo:acme:hr.leave:5/cancel", HR_ADMIN, {"reason": "plans changed"}))
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "wrong door"}), 409, "nothing_moved")
def test_TOF_23_the_machine_loads_and_reconciles_the_owner_signs_and_nobody_else_does(hr, odoo):
for who in (EMPLOYEE, MANAGER, HR_ADMIN, OWNER):
refused(load(who=who), 403, "tof:migration:load")
must(load())
for who in (EMPLOYEE, MANAGER, HR_ADMIN, OWNER):
refused(reconcile(who=who), 403, "tof:migration:reconcile")
for who in (EMPLOYEE, MANAGER, SYSTEM):
refused(api("GET", f"/migration/odoo/{RUN}", who), 403, "tof:migration:read")
assert must(api("GET", f"/migration/odoo/{RUN}", HR_ADMIN))["status"] == "loaded"
assert must(api("GET", f"/migration/odoo/{RUN}", OWNER))["doors"]["leave_types"] == TYPES
refused(api("GET", "/migration/odoo/nope", OWNER), 404)
def test_TOF_23_the_migration_map_answers_only_what_was_migrated(hr, odoo):
must(load())
got = must(api("GET", f"/migration-map/{DB}/hr.leave", HR_ADMIN, {"ids": "1,2,99"}))
assert got == [{"odoo_model": "hr.leave", "odoo_id": "1", "ours": "odoo:acme:hr.leave:1"}, {"odoo_model": "hr.leave", "odoo_id": "2", "ours": "odoo:acme:hr.leave:2"}]
allocs = must(api("GET", f"/migration-map/{DB}/hr.leave.allocation", HR_ADMIN, {"ids": "1"}))
assert allocs == [{"odoo_model": "hr.leave.allocation", "odoo_id": "1", "ours": "open:4410:ANNUAL:2026"}]
for bad in ("", "x", "1;2", "1" * 19):
assert api("GET", f"/migration-map/{DB}/hr.leave", HR_ADMIN, {"ids": bad})[0] == 422, bad
refused(api("GET", f"/migration-map/{DB}/hr.leave", EMPLOYEE, {"ids": "1"}), 403)
def test_TOF_23_bounds_are_422_never_500(hr, odoo):
for kw in ({"years": []}, {"years": [1999]}, {"years": [2026] * 7}, {"cutover_date": "soon"}, {"odoo_db": ""}, {"waiting": "maybe"},
{"employee_map": {"11": "x"}}, {"leave_types": {"1": ""}}):
got = load(**kw)
assert got[0] == 422, (kw, got)
def test_TOF_23_the_agent_door_has_the_four_transition_tools_and_the_map(hr, odoo):
for name in ("load_tof_from_odoo", "reconcile_tof_migration", "sign_tof_migration", "discard_tof_migration", "get_tof_migration_map"):
assert name in tools(OWNER), name
def test_TOF_23_an_allocation_odoo_gained_later_is_mapped_but_the_opening_is_written_once_and_reconcile_shows_the_difference(hr, odoo):
must(load())
odoo.alloc_rows.append(alloc(3, 11, 1, 2, start="2026-02-01")) # Odoo grants two more days after the load
got = must(load())
assert got["stats"]["openings_written"] == 0 and balance()["available"] == "22.50"
assert must(api("GET", f"/migration-map/{DB}/hr.leave.allocation", HR_ADMIN, {"ids": "3"}))[0]["ours"] == "open:4410:ANNUAL:2026"
assert must(reconcile())["checks"][0]["detail"]["11:ANNUAL:2026"] == {"odoo": "24.50", "ours": "22.50"} # the opening is not rewritten: it is shown