dodil/erp-brick-gl
DODIL ERP brick: general ledger
tests/test_cockpit.py
"""GL-57 the month-end checklist, GL-58 the second person who signs the close."""
from conftest import ACCOUNTANT, ACCOUNTANT2, BOOKKEEPER, OWNER, api, must, on_day, refused
from world import books, close_through, invoice
TASKS = [
{"task_id": "recurring", "title": "Recurring entries run", "kind_of_task": "system", "gate": "recurring_posted", "due_business_day": 1},
{"task_id": "accruals", "title": "Accruals run", "kind_of_task": "system", "gate": "accruals_posted", "due_business_day": 1, "depends_on": ["recurring"]},
{"task_id": "payroll", "title": "Payroll journal received", "kind_of_task": "manual", "owner_role": "bookkeeper", "due_business_day": 2},
{"task_id": "review", "title": "Review the result", "kind_of_task": "manual", "owner_role": "accountant", "due_business_day": 3, "depends_on": ["payroll"]},
]
def cockpit_books():
books()
close_through("2026-09")
must(api("POST", "/controls", OWNER, {"key": "close_cockpit", "value": True, "company_code": "UK01"}))
must(api("POST", "/recurring", ACCOUNTANT, {"recurring_id": "RENT", "first_period": "2026-01", "frequency": "monthly",
"lines": [{"account_id": __import__("world").account_id("6100").__str__(), "debit": "100.00"}, {"account_id": __import__("world").account_id("2100").__str__(), "credit": "100.00"}]}))
must(api("POST", "/recurring/RENT/approve-recurring", ACCOUNTANT2, {}))
must(api("POST", "/close-task-lists", ACCOUNTANT, {"task_list_id": "MONTH-UK", "kind": "month", "tasks": TASKS}))
def test_a_run_instantiates_tasks_with_owner_due_day_and_dependencies():
books()
close_through("2026-09")
must(api("POST", "/close-task-lists", ACCOUNTANT, {"task_list_id": "MONTH-UK", "kind": "month", "tasks": TASKS}))
run = must(api("POST", "/periods/2026-10/start-close-run", ACCOUNTANT, {}))
assert run["run_id"] == "UK01:2026-10:month" and run["status"] == "in_progress"
by = {t["task_id"]: t for t in run["tasks"]}
assert by["payroll"]["due_on"] == "2026-11-03" and by["recurring"]["due_on"] == "2026-11-02" and by["review"]["due_on"] == "2026-11-04"
assert by["review"]["blocked_by"] == ["payroll"] and by["payroll"]["owner_role"] == "bookkeeper"
assert by["recurring"]["status"] == "done" # nothing recurring is due: its gate is met on read
assert must(api("POST", "/periods/2026-10/start-close-run", ACCOUNTANT, {}))["run_id"] == run["run_id"] # starting again answers the run
with on_day("2026-11-05T09:00:00+00:00"):
late = must(api("GET", f"/close-runs/{run['run_id']}", ACCOUNTANT))["late"]
assert sorted(late) == ["payroll", "review"]
def test_the_system_ticks_what_it_can_check_and_a_manual_task_waits_for_what_it_depends_on():
cockpit_books()
run = must(api("POST", "/periods/2026-10/start-close-run", ACCOUNTANT, {}))
by = {t["task_id"]: t for t in run["tasks"]}
assert by["recurring"]["status"] == "open" and "RENT" in by["recurring"]["reason"]
refused(api("POST", f"/close-runs/{run['run_id']}/complete-close-task", ACCOUNTANT, {"task_id": "recurring", "note": "done"}), 409, "system task")
refused(api("POST", f"/close-runs/{run['run_id']}/complete-close-task", BOOKKEEPER, {"task_id": "review", "note": "ok"}), 409, "[dependencies_done]", "payroll")
must(api("POST", "/periods/2026-10/run-recurring-entries", ACCOUNTANT, {}))
now = {t["task_id"]: t for t in must(api("GET", f"/close-runs/{run['run_id']}", ACCOUNTANT))["tasks"]}
assert now["recurring"]["status"] == "done" and now["accruals"]["status"] == "done"
refused(api("POST", f"/close-runs/{run['run_id']}/complete-close-task", BOOKKEEPER, {"task_id": "payroll"}), 422, "note or an evidence key")
refused(api("POST", f"/close-runs/{run['run_id']}/complete-close-task", BOOKKEEPER, {"task_id": "payroll", "evidence_key": "other/payroll.pdf"}), 422, "raw/close/")
out = must(api("POST", f"/close-runs/{run['run_id']}/complete-close-task", BOOKKEEPER, {"task_id": "payroll", "evidence_key": "raw/close/2026-10/payroll.pdf"}))
assert {t["task_id"]: t for t in out["tasks"]}["payroll"]["done_by"] == "bookkeeper@firm.example"
must(api("POST", f"/close-runs/{run['run_id']}/skip-close-task", ACCOUNTANT, {"task_id": "review", "reason": "nothing to review"}))
def finish_run():
cockpit_books()
run = must(api("POST", "/periods/2026-10/start-close-run", ACCOUNTANT, {}))
must(api("POST", "/periods/2026-10/run-recurring-entries", ACCOUNTANT, {}))
must(api("POST", f"/close-runs/{run['run_id']}/complete-close-task", BOOKKEEPER, {"task_id": "payroll", "note": "received"}))
must(api("POST", f"/close-runs/{run['run_id']}/skip-close-task", ACCOUNTANT, {"task_id": "review", "reason": "n/a"}))
return run["run_id"]
def test_the_preparer_cannot_sign_and_close_waits_for_the_second_persons_signature():
rid = finish_run()
ready = must(api("POST", f"/close-runs/{rid}/mark-close-ready", __import__("conftest").SYSTEM, {}))
assert ready["status"] == "ready" and ready["tb_digest"]
refused(api("POST", "/periods/2026-10/close-period", ACCOUNTANT, {}), 409, "[close_signed_off]")
refused(api("POST", f"/close-runs/{rid}/sign-off-close", ACCOUNTANT, {}), 409, "[second_person]")
assert api("POST", f"/close-runs/{rid}/sign-off-close", BOOKKEEPER, {})[0] == 403
signed = must(api("POST", f"/close-runs/{rid}/sign-off-close", ACCOUNTANT2, {}))
assert signed["status"] == "signed" and signed["signed_by"] == "accountant2@firm.example" and signed["tb_digest"]
assert must(api("POST", "/periods/2026-10/close-period", ACCOUNTANT, {}))["status"] == "closed"
def test_a_late_posting_makes_the_signoff_stale_and_the_close_waits_for_a_new_one():
rid = finish_run()
must(api("POST", f"/close-runs/{rid}/sign-off-close", ACCOUNTANT2, {}))
assert invoice("INV-LATE", on="2026-10-21")[0] == 200
stale = must(api("GET", f"/close-runs/{rid}", ACCOUNTANT))
assert stale["status"] == "in_progress" and "posting landed" in stale["stale_reason"]
refused(api("POST", "/periods/2026-10/close-period", ACCOUNTANT, {}), 409, "[close_signed_off]", "stale")
again = must(api("POST", f"/close-runs/{rid}/sign-off-close", ACCOUNTANT2, {}))
assert again["status"] == "signed"
must(api("POST", "/periods/2026-10/close-period", ACCOUNTANT, {}))
def test_someone_who_completed_a_task_cannot_sign():
from conftest import person
cockpit_books()
run = must(api("POST", "/periods/2026-10/start-close-run", ACCOUNTANT, {}))
must(api("POST", "/periods/2026-10/run-recurring-entries", ACCOUNTANT, {}))
both = person("sam@firm.example", ["gl:close:task", "gl:close:signoff", "gl:close:manage", "gl:period:read", "gl:company:read"], sub="sub-sam")
must(api("POST", f"/close-runs/{run['run_id']}/complete-close-task", both, {"task_id": "payroll", "note": "received"}))
must(api("POST", f"/close-runs/{run['run_id']}/skip-close-task", ACCOUNTANT, {"task_id": "review", "reason": "n/a"}))
twin = person("sam.work@firm.example", ["gl:close:signoff", "gl:period:read", "gl:company:read"], sub="sub-sam")
refused(api("POST", f"/close-runs/{run['run_id']}/sign-off-close", twin, {}), 409, "[second_person]")
def test_signing_before_the_tasks_are_done_and_without_a_task_list():
books()
close_through("2026-09")
refused(api("POST", "/periods/2026-10/start-close-run", ACCOUNTANT, {}), 409, "[task_list]")
must(api("POST", "/close-task-lists", ACCOUNTANT, {"task_list_id": "M", "kind": "month", "tasks": TASKS}))
rid = must(api("POST", "/periods/2026-10/start-close-run", ACCOUNTANT, {}))["run_id"]
refused(api("POST", f"/close-runs/{rid}/sign-off-close", ACCOUNTANT2, {}), 409, "[tasks_done]", "payroll")
assert api("POST", "/close-task-lists", ACCOUNTANT, {"task_list_id": "X", "kind": "month", "tasks": [{"task_id": "a", "title": "a", "kind_of_task": "system", "gate": "nope"}]})[0] == 422
def test_the_year_end_list_runs_for_the_last_period():
books()
must(api("POST", "/close-task-lists", ACCOUNTANT, {"task_list_id": "YEAR-UK", "kind": "year", "tasks": [
{"task_id": "adjustments", "title": "Audit adjustments posted", "kind_of_task": "manual", "owner_role": "accountant"}]}))
run = must(api("POST", "/periods/2026-12/start-close-run", ACCOUNTANT, {}))
assert run["kind"] == "year" and run["run_id"] == "UK01:2026-12:year"