dodil/erp-brick-fixed-assets

DODIL ERP brick: fixed assets

tests/test_reports_counts.py

"""FA-23 (history sheet), FA-24 (reconciliation to gl), FA-25 and FA-26 (counts)."""
from decimal import Decimal as D

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


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


def run(period, when):
    with on_day(when):
        r = must(api("POST", "/depreciation-runs", ACCOUNTANT, {"company_code": "UK01", "period_id": period}))
        return must(api("POST", f"/depreciation-runs/{r['run_id']}/approve", APPROVER))


def hist(year, area="01", **extra):
    return must(api("GET", "/reports/history", ACCOUNTANT, {"company_code": "UK01", "area_code": area, "fiscal_year": year, **extra}))


def van(world, price="12000.00"):
    v = create_asset("van")
    must(acquire(v["asset_id"], price, value_date="2026-10-20", ref="pur:V-1:1"))
    run("2026-12", "2026-12-31T09:00:00+00:00")                      # October to December: 3 x 200.00
    return v


# ── FA-23 ───────────────────────────────────────────────────────────────────────────────────

def test_each_column_is_the_sum_of_transactions_and_one_years_closing_is_the_next_years_opening(world):
    v = van(world)
    run("2027-03", "2027-03-31T09:00:00+00:00")
    with on_day("2027-03-31T10:00:00+00:00"):
        t = must(api("POST", f"/assets/{v['asset_id']}/retirements", ACCOUNTANT,
                     {"retirement_date": "2027-03-31", "percent": "100", "proceeds": "9000.00"}))
        must(api("POST", f"/asset-transactions/{t['transaction_id']}/approve", APPROVER))
    y26, y27 = hist(2026)["classes"][0], hist(2027)["classes"][0]
    assert (y26["cost_start"], y26["additions"], y26["cost_end"], y26["charge"], y26["depreciation_end"], y26["nbv_end"]) == (
        "0.00", "12000.00", "12000.00", "600.00", "600.00", "11400.00")
    assert y27["cost_start"] == y26["cost_end"] and y27["depreciation_start"] == y26["depreciation_end"]
    assert y27["nbv_start"] == y26["nbv_end"] == "11400.00"
    assert (y27["charge"], y27["retirements"], y27["depreciation_on_retirements"]) == ("600.00", "-12000.00", "-1200.00")
    assert (y27["cost_end"], y27["depreciation_end"], y27["nbv_end"]) == ("0.00", "0.00", "0.00")
    for y in (y26, y27):                                          # every closing is the opening plus the movements in rows
        movements = sum(D(y[c]) for c in ("additions", "retirements", "transfers_in", "transfers_out", "revaluations"))
        assert D(y["cost_end"]) == D(y["cost_start"]) + movements
        dep = sum(D(y[c]) for c in ("charge", "impairments", "impairment_reversals", "depreciation_on_retirements",
                                    "depreciation_on_transfers", "depreciation_on_revaluations"))
        assert D(y["depreciation_end"]) == D(y["depreciation_start"]) + dep
        assert D(y["nbv_end"]) == D(y["cost_end"]) - D(y["depreciation_end"])


def test_the_sheet_exports_as_csv_and_takes_any_period_range_and_any_area(world):
    van(world)
    csv = hist(2026, format="csv")["csv"].splitlines()
    assert csv[0].startswith("class_code,cost_start,additions") and csv[1].startswith("VEH,0.00,12000.00") and csv[-1].startswith("TOTAL,")
    part = must(api("GET", "/reports/history", ACCOUNTANT, {"company_code": "UK01", "area_code": "30", "from_period": "2026-10",
                                                            "to_period": "2026-12"}))
    assert part["classes"][0]["charge"] == "600.00" and part["classes"][0]["cost_start"] == "0.00"
    assert api("GET", "/reports/history", ACCOUNTANT, {"company_code": "UK01", "area_code": "01"})[0] == 422


# ── FA-24 ───────────────────────────────────────────────────────────────────────────────────

def test_the_register_is_reconciled_to_the_ledger_and_a_manual_journal_is_a_named_break(world):
    van(world)
    bal = must(api("GET", "/assets/balances", GL_CLIENT, {"company_code": "UK01", "as_of": "2026-12-31", "ledger_id": "10"}))
    assert {a["account_id"]: a["balance"] for a in bal["accounts"]} == {"150": "12000.00", "151": "600.00"}
    assert bal["clearing"]["open_acquisitions"] == []                # every bill was handed over: clearing is zero
    same = must(api("POST", "/reports/reconcile", ACCOUNTANT, {"company_code": "UK01", "as_of": "2026-12-31", "ledger_id": "10",
                                                               "gl_balances": [{"account_id": "150", "balance": "12000.00"},
                                                                               {"account_id": "151", "balance": "600.00"}]}))
    assert all(D(a["difference"]) == 0 for a in same["accounts"])
    off = must(api("POST", "/reports/reconcile", ACCOUNTANT, {"company_code": "UK01", "as_of": "2026-12-31", "ledger_id": "10",
                                                              "gl_balances": [{"account_id": "150", "balance": "12500.00"},
                                                                              {"account_id": "151", "balance": "600.00"}]}))
    row = next(a for a in off["accounts"] if a["account_id"] == "150")
    assert row["register"] == "12000.00" and row["gl"] == "12500.00" and row["difference"] == "500.00"
    assert row["fix"] == "posted to an asset account outside the register: reverse it and post a manual acquisition"
    assert api("GET", "/assets/balances", ACCOUNTANT, {"company_code": "UK01", "as_of": "2026-12-31"})[0] == 200
    assert api("GET", "/assets/balances", GL_CLIENT, {"company_code": "UK01", "as_of": "2026-12-31", "ledger_id": "99"})[0] == 409


# ── FA-25, FA-26 ────────────────────────────────────────────────────────────────────────────

SITE = "SITE-A"
CUSTODIAN_EMAIL = "asset_custodian@uk.example"                        # who the conftest's CUSTODIAN persona signs in as
OTHER_CUSTODIAN = person("someone.else@uk.example", ROLES["asset_custodian"])


def site_assets(n=4):
    out = []
    for i in range(n):
        a = create_asset(f"site-{i}", description=f"Item {i}", inventory_no=f"TAG-{i:04d}", serial_no=f"SN-{i}",
                         location_id=SITE, custodian_email=CUSTODIAN_EMAIL)
        must(acquire(a["asset_id"], "1000.00", value_date="2026-10-20", ref=f"pur:S-{i}:1"))
        out.append(a)
    return out


def open_count(when="2026-11-10"):
    return api("POST", "/asset-counts", ACCOUNTANT, {"location_id": SITE, "count_date": when, "custodian_email": CUSTODIAN_EMAIL})


def test_a_count_lists_the_sites_assets_is_keyed_and_fixed_when_it_opens(world):
    site_assets(4)
    c = must(open_count())
    assert [(x["tag"], x["description"], x["serial_no"]) for x in c["lines"]] == [
        (f"TAG-{i:04d}", f"Item {i}", f"SN-{i}") for i in range(4)]
    late = create_asset("late", description="Moved in later", inventory_no="TAG-LATE", location_id=SITE)
    again = must(open_count())
    assert [x["tag"] for x in again["lines"]] == [x["tag"] for x in c["lines"]]      # the same count, the list fixed
    assert api("POST", "/asset-counts", ACCOUNTANT, {"location_id": SITE, "count_date": "2026-11-21",
                                                      "custodian_email": CUSTODIAN_EMAIL})[0] == 409     # the date is a fact
    del late


def test_the_custodian_records_results_and_only_a_complete_decided_count_closes(world):
    assets = site_assets(4)
    must(open_count())
    path = f"/asset-counts/{SITE}/2026-11-10"
    s, b = api("POST", path + "/finish", CUSTODIAN)
    assert s == 409 and "all_counted" in str(b), b                               # nothing counted yet, each line named
    results = [{"inventory_no": "TAG-0000", "result": "found"}, {"inventory_no": "TAG-0001", "result": "found"},
               {"inventory_no": "TAG-0002", "result": "not_found"}, {"inventory_no": "TAG-0003", "result": "damaged"},
               {"result": "found_unlisted", "description": "Pressure washer"}]
    got = must(api("POST", path + "/results", CUSTODIAN, {"results": results}))
    assert {x["result"] for x in got["lines"]} == {"found", "not_found", "damaged", "found_unlisted"}
    assert all(x["counted_by_email"] == CUSTODIAN_EMAIL and x["counted_on"] == "2026-11-10" for x in got["lines"])
    assert api("POST", path + "/results", OTHER_CUSTODIAN, {"results": results})[0] == 404       # not their count
    assert api("GET", path, OTHER_CUSTODIAN)[0] == 404
    s, b = api("POST", path + "/close", ACCOUNTANT)
    assert s == 409, b                                                                              # not handed back yet
    must(api("POST", path + "/finish", CUSTODIAN))
    s, b = api("POST", path + "/close", ACCOUNTANT)
    assert s == 409 and "differences_decided" in str(b), b                                         # three differences, no decision
    lines = must(api("GET", f"/manage{path}", ACCOUNTANT))["lines"]
    n = len(ACCOUNTANT_PROPOSALS(lines))
    assert n == 3
    for ln in lines:
        if ln["result"] in ("not_found", "damaged"):
            body = {"line_no": ln["line_no"], "decision": "retire"} if ln["result"] == "not_found" else \
                {"line_no": ln["line_no"], "decision": "note", "note": "repair booked"}
            must(api("POST", path + "/decisions", ACCOUNTANT, body))
        elif ln["result"] == "found_unlisted":
            must(api("POST", path + "/decisions", ACCOUNTANT, {"line_no": ln["line_no"], "decision": "new_asset",
                                                               "class_code": "VEH", "company_code": "UK01"}))
    closed = must(api("POST", path + "/close", ACCOUNTANT))
    assert closed["status"] == "closed"
    seen = {a["asset_id"]: must(api("GET", f"/assets/{a['asset_id']}", ACCOUNTANT))["last_counted_on"] for a in assets}
    assert [seen[a["asset_id"]] for a in assets] == ["2026-11-10", "2026-11-10", None, "2026-11-10"]   # found (and damaged but there)
    props = must(api("GET", "/asset-transactions", ACCOUNTANT, {"asset_id": assets[2]["asset_id"]}))
    assert [t["status"] for t in props["transactions"] if t["kind"] == "retirement"] == ["proposed"]                              # nothing posted
    assert not [j for j in world.journals.values() if j["event"] == "retire"]


def ACCOUNTANT_PROPOSALS(lines):
    return [x for x in lines if x["result"] in ("not_found", "damaged", "found_unlisted")]


def test_a_count_date_in_the_future_is_refused_for_results(world):
    site_assets(1)
    must(open_count())
    assert api("POST", f"/asset-counts/{SITE}/2026-11-21/results", CUSTODIAN, {"results": []})[0] in (404, 409)