dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

tests/test_depreciation_runs.py

"""FA-12 (simulate), FA-13 (approve and post), FA-14 (repeat, catch-up, reverse), FA-15 (close the year) — Given/When/Then from
the signed spec, against a gl stand-in that enforces gl's journal_id rule and refuses a closed period."""
from decimal import Decimal as D

import pytest
from conftest import (
    ACCOUNTANT,
    APPROVER,
    APPROVER2,
    OWNER,
    ROLES,
    acquire,
    api,
    create_asset,
    must,
    on_day,
    person,
    setup_policy,
)

OCT, NOV = "2026-10-31T09:00:00+00:00", "2026-11-30T09:00:00+00:00"


@pytest.fixture
def world(gl):
    setup_policy(classes=("VEH",))
    return gl


def sim(period="2026-10", who=ACCOUNTANT):
    return api("POST", "/depreciation-runs", who, {"company_code": "UK01", "period_id": period})


def approve(run_id, who=APPROVER):
    return api("POST", f"/depreciation-runs/{run_id}/approve", who)


def run_of(period, when, who_sim=ACCOUNTANT, who_app=APPROVER):
    with on_day(when):
        r = must(sim(period, who_sim))
        return must(approve(r["run_id"], who_app))


def dep_journals(gl, event="depreciate"):
    return [j for j in gl.journals.values() if j["event"] == event]


# ── FA-12 ───────────────────────────────────────────────────────────────────────────────────

def test_a_simulation_lists_planned_minus_posted_per_area_and_class_and_posts_nothing(world):
    van = create_asset(cost_centre="CC-WHS")
    must(acquire(van["asset_id"], "12000.00", ref="pur:B-1:1"))
    before = len(world.journals)
    r = must(sim("2026-10"))
    assert r["status"] == "simulated" and r["totals"] == {"01": "200.00", "30": "200.00"}
    assert {(g["area_code"], g["class_code"], g["cost_centre"], g["amount"]) for g in r["groups"]} == {
        ("01", "VEH", "CC-WHS", "200.00"), ("30", "VEH", "CC-WHS", "200.00")}
    names = {g["gate"]: g["met"] for g in r["gates"]}
    assert names["usage_missing"] and names["account_determination"] and names["period_open:01"] and names["period_open:30"]
    assert len(world.journals) == before                                       # nothing is posted
    again = must(sim("2026-10"))
    assert again["run_id"] == r["run_id"] and again["totals"] == r["totals"]    # the same figures until a row changes


def test_no_depreciation_in_advance(world):
    s, b = sim("2026-12")
    assert s == 409 and "not_in_advance" in str(b)


def test_a_simulation_names_the_gates_a_missing_reading_and_a_closed_period_raise(world):
    with on_day("2026-01-01T09:00:00+00:00"):
        must(api("POST", "/settings/keys", OWNER, {"key_code": "UOP", "method": "units_of_production"}))
        must(api("POST", "/settings/classes", OWNER, {"class_code": "MACH", "name": "Machines",
                                                      "areas": [{"area_code": "01", "key_code": "UOP"}]}))
        must(api("POST", "/settings/account-determination", OWNER, {"class_code": "MACH", "area_code": "01",
              "cost_account_id": 150, "accumulated_depreciation_account_id": 151, "depreciation_expense_account_id": 771}))
    m = create_asset("r-m", cls="MACH", description="CNC", areas=[{"area_code": "01", "total_units": "30000"}])
    must(acquire(m["asset_id"], "90000.00", ref="pur:B-3:1"))
    world.closed.add(("20", "2026-10"))
    r = must(sim("2026-10"))
    gates = {g["gate"]: g for g in r["gates"]}
    assert gates["usage_missing"]["met"] is False and m["asset_id"] in gates["usage_missing"]["reason"]
    assert gates["period_open:30"]["met"] is False and "30" in gates["period_open:30"]["reason"]
    s, b = approve(r["run_id"], APPROVER2)
    assert s == 409 and "usage_missing" in str(b)


# ── FA-13 ───────────────────────────────────────────────────────────────────────────────────

@pytest.fixture
def october(world):
    van = create_asset(cost_centre="CC-WHS")
    must(acquire(van["asset_id"], "12000.00", ref="pur:B-1:1"))
    return world, van["asset_id"]


def test_approving_posts_one_journal_per_area_to_its_ledger_keyed_from_the_run(october):
    world, van = october
    r = must(sim("2026-10"))
    done = must(approve(r["run_id"], APPROVER2))
    assert done["status"] == "posted" and done["approved_by_email"] == "second.approver@uk.example"
    assert done["approved_on"] == "2026-11-20"
    js = dep_journals(world)
    assert sorted(j["ledger_id"] for j in js) == ["10", "20"]
    for j in js:
        assert j["source_document"] == f"fa:{r['run_id']}" and j["journal_date"] == "2026-10-31"
        debit = next(ln for ln in j["lines"] if "debit" in ln)
        credit = next(ln for ln in j["lines"] if "credit" in ln)
        assert (debit["account_id"], debit["debit"], debit["cost_centre"]) == ("771", "200.00", "CC-WHS")
        assert (credit["account_id"], credit["credit"]) == ("151", "200.00")
    v = next(x for x in must(api("GET", f"/assets/{van}", ACCOUNTANT))["values"] if x["area_code"] == "01")
    assert v["posted_to_date"] == "200.00"                                    # the posted amount per asset is recorded


def test_the_starter_cannot_approve_under_either_identity_and_a_second_approval_changes_nothing(october):
    world, _ = october
    r = must(sim("2026-10"))
    same_email_other_sub = person("asset_accountant@uk.example", ROLES["finance_approver"], sub="someone-else")
    same_sub_other_email = person("other.mail@uk.example", ROLES["finance_approver"], sub="asset_accountant@uk.example")
    for who in (same_email_other_sub, same_sub_other_email):
        s, b = approve(r["run_id"], who)
        assert s == 409 and "second_person" in str(b)
    assert not dep_journals(world)
    must(approve(r["run_id"], APPROVER))
    n = len(world.journals)
    again = must(approve(r["run_id"], APPROVER2))
    assert again["status"] == "posted" and len(world.journals) == n


def test_a_closed_ledger_period_refuses_the_approval_and_names_the_area(october):
    world, _ = october
    r = must(sim("2026-10"))
    world.closed.add(("20", "2026-10"))
    s, b = approve(r["run_id"], APPROVER)
    assert s == 409 and "period_open" in str(b) and "area 30" in str(b)
    assert not dep_journals(world) and must(api("GET", f"/depreciation-runs/{r['run_id']}", ACCOUNTANT))["status"] == "simulated"


def test_a_machine_gl_failure_is_resumed_not_doubled(october):
    world, _ = october
    r = must(sim("2026-10"))
    world.fail_next = (503, {"detail": "gl is down"})
    s, _ = approve(r["run_id"], APPROVER)
    assert s >= 400 and not dep_journals(world)
    again = must(approve(r["run_id"], APPROVER))                                # the retry finishes the posting
    assert again["status"] == "posted" and len(dep_journals(world)) == 2


def test_the_accountant_posts_without_a_second_person_only_when_the_owner_switched_it_off(october):
    world, _ = october
    r = must(sim("2026-10"))
    s, b = api("POST", f"/depreciation-runs/{r['run_id']}/post", ACCOUNTANT)
    assert s == 409 and "depreciation_needs_approval" in str(b)
    must(api("POST", "/settings/controls", OWNER, {"setting_key": "depreciation_needs_approval", "value": "false"}))
    done = must(api("POST", f"/depreciation-runs/{r['run_id']}/post", ACCOUNTANT))
    assert done["status"] == "posted" and done["posted_by"] and len(dep_journals(world)) == 2


# ── FA-14 ───────────────────────────────────────────────────────────────────────────────────

def test_a_late_bill_is_caught_up_in_the_next_run_dated_in_that_month(world):
    van1 = create_asset("r1")
    with on_day("2026-10-25T09:00:00+00:00"):
        must(acquire(van1["asset_id"], "12000.00", ref="pur:B-1:1"))
    run_of("2026-10", OCT)
    with on_day("2026-11-03T09:00:00+00:00"):
        van2 = create_asset("r2", description="Van 2")
        must(acquire(van2["asset_id"], "12000.00", value_date="2026-10-20", ref="pur:B-2:1"))
    nov = run_of("2026-11", NOV)
    assert nov["totals"] == {"01": "600.00", "30": "600.00"}                    # van 1: 200; van 2: October catch-up 200 + November 200
    detail = must(api("GET", f"/depreciation-runs/{nov['run_id']}", ACCOUNTANT, {"detail": True}))
    assert {ln["asset_id"]: ln["amount"] for ln in detail["lines"] if ln["area_code"] == "01"} == {
        van1["asset_id"]: "200.00", van2["asset_id"]: "400.00"}
    oct_dates = {j["journal_date"] for j in dep_journals(world) if "2026-10" in j["source_document"]}
    nov_dates = {j["journal_date"] for j in dep_journals(world) if "2026-11" in j["source_document"]}
    assert oct_dates == {"2026-10-31"} and nov_dates == {"2026-11-30"}         # no back-posting into the posted period


def test_a_period_is_never_posted_twice_for_the_same_planned_amount(world):
    van = create_asset()
    with on_day("2026-10-25T09:00:00+00:00"):
        must(acquire(van["asset_id"], "12000.00", ref="pur:B-1:1"))
    run_of("2026-10", OCT)
    n = len(world.journals)
    second = run_of("2026-10", OCT, who_app=APPROVER2)
    assert second["run_no"] == 2 and second["totals"] == {} and len(world.journals) == n


def test_a_run_posted_in_error_is_reversed_with_its_own_journals_and_the_period_plans_again(world):
    van = create_asset()
    with on_day("2026-10-25T09:00:00+00:00"):
        must(acquire(van["asset_id"], "12000.00", ref="pur:B-1:1"))
    done = run_of("2026-10", OCT)
    originals = {k: dict(v) for k, v in world.journals.items() if v["event"] == "depreciate"}
    with on_day(NOV):
        s, b = api("POST", f"/depreciation-runs/{done['run_id']}/reverse", ACCOUNTANT, {"reason": "wrong life"})
        assert s in (403, 409)                                                 # the starter / the accountant cannot reverse
        rev = must(api("POST", f"/depreciation-runs/{done['run_id']}/reverse", APPROVER2, {"reason": "wrong life"}))
        assert rev["reverses_run_id"] == done["run_id"] and rev["totals"] == {"01": "-200.00", "30": "-200.00"}
        reversals = dep_journals(world, "reverse")
        assert len(reversals) == 2 and {j["journal_date"] for j in reversals} == {"2026-11-30"}   # dated today
        for j in reversals:
            assert next(ln for ln in j["lines"] if "credit" in ln)["account_id"] == "771"          # the mirror image
        again = must(api("POST", f"/depreciation-runs/{done['run_id']}/reverse", APPROVER2, {"reason": "wrong life"}))
        assert again["run_id"] == rev["run_id"] and len(dep_journals(world, "reverse")) == 2        # twice posts once
    assert {k: world.journals[k] for k in originals} == originals               # the original journals are untouched
    assert must(api("GET", f"/depreciation-runs/{done['run_id']}", ACCOUNTANT))["status"] == "reversed"
    with on_day(NOV):
        replan = must(sim("2026-10"))
        assert replan["totals"] == {"01": "200.00", "30": "200.00"}              # the next run plans the period again


def test_a_reversal_is_refused_in_a_closed_ledger_period(world):
    van = create_asset()
    with on_day("2026-10-25T09:00:00+00:00"):
        must(acquire(van["asset_id"], "12000.00", ref="pur:B-1:1"))
    done = run_of("2026-10", OCT)
    world.closed.add(("10", "2026-11"))
    with on_day(NOV):
        s, b = api("POST", f"/depreciation-runs/{done['run_id']}/reverse", APPROVER2, {"reason": "x"})
    assert s == 409 and "period_open" in str(b)


# ── FA-15 ───────────────────────────────────────────────────────────────────────────────────

JAN = "2027-01-05T09:00:00+00:00"


def test_the_year_closes_only_when_every_period_is_posted_and_then_refuses_its_own_dates(world):
    van = create_asset()
    with on_day("2026-10-25T09:00:00+00:00"):
        must(acquire(van["asset_id"], "12000.00", ref="pur:B-1:1"))
    with on_day(JAN):
        s, b = api("POST", "/years/UK01/2026/close", APPROVER)
        assert s == 409 and "period 2026-01 has no posted depreciation run" in str(b)
        for m in range(1, 12):
            run_of(f"2026-{m:02d}", JAN)
        s, b = api("POST", "/years/UK01/2026/close", APPROVER)
        assert s == 409 and "2026-12" in str(b) and "2026-11" not in str(b)       # exactly the period still open is named
        run_of("2026-12", JAN)
        closed = must(api("POST", "/years/UK01/2026/close", APPROVER))
        assert closed["status"] == "closed"
        assert must(api("POST", "/years/UK01/2026/close", APPROVER))["status"] == "closed"        # twice, once
        s, b = acquire(van["asset_id"], "100.00", value_date="2026-12-01", ref="pur:B-9:1")
        assert s == 409 and b["detail"]["gate"] == "year_open" and "2026" in b["detail"]["reason"]
        s, b = api("POST", "/depreciation-runs/UK01:2026-12:1/reverse", APPROVER2, {"reason": "x"})
        assert s == 409 and "closed" in str(b)                                     # a closed year is not reopened here
        # next year's opening values are the closing values, derived: 2027 starts where 2026 ended
        h26 = must(api("GET", "/reports/history", ACCOUNTANT, {"company_code": "UK01", "area_code": "01", "fiscal_year": 2026}))
        h27 = must(api("GET", "/reports/history", ACCOUNTANT, {"company_code": "UK01", "area_code": "01", "fiscal_year": 2027}))
        assert h26["totals"]["cost_end"] == h27["totals"]["cost_start"] == "12000.00"
        assert h26["totals"]["depreciation_end"] == h27["totals"]["depreciation_start"] == "600.00"
        assert D(h26["totals"]["charge"]) == D("600.00")


def test_a_transaction_awaiting_approval_blocks_the_year_close_and_is_named(world):
    van = create_asset()
    with on_day("2026-10-25T09:00:00+00:00"):
        must(acquire(van["asset_id"], "12000.00", ref="pur:B-1:1"))
    with on_day(JAN):
        for m in range(1, 13):
            run_of(f"2026-{m:02d}", JAN)
        pending = must(api("POST", f"/assets/{van['asset_id']}/retirements", ACCOUNTANT,
                           {"retirement_date": "2026-12-15", "percent": "100", "proceeds": "5000.00"}))
        s, b = api("POST", "/years/UK01/2026/close", APPROVER)
        assert s == 409 and pending["transaction_id"] in str(b)