dodil/erp-brick-inventory

DODIL ERP brick: inventory

tests/test_valuation.py

"""Value: INV-31 average to zero, INV-32 postings, INV-48 / INV-56 standard cost, INV-57 delivery-cost differences."""
import fakes
import world
from conftest import ACCOUNTANT, MANAGER, OPERATOR, OWNER, SYSTEM, api, must, on_day, refused


def receive_line(ref, qty, cost, on="2026-11-20", loc="WH1/A/01", unit="PK24", product="COLA-075", **kw):
    return must(api("POST", "/receipts", SYSTEM, {"move_ref": f"pur:{ref}:1", "source_id": ref, "product_id": product, "quantity": qty, "unit_code": unit,
                                                   "location_id": loc, "received_on": on, "unit_cost": cost, **kw}))


def deliver(order, qty, loc="WH1/A/01", unit="C62", product="COLA-075", num=3, den=4):
    fakes.add_order(order, {1: {"product_id": product, "unit_code": unit, "quantity": qty, "unit_numerator": num, "unit_denominator": den}})
    doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": order, "warehouse_id": "WH1"}))["document_id"]
    must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [{"line_no": 1, "location_id": loc, "quantity": qty, "unit_code": unit}]}))
    return doc


def val(**kw):
    return must(api("GET", "/valuation/products/COLA-075", ACCOUNTANT, {"company_code": "UK1", **kw}))


def test_sc_inv03_average_cost_to_zero_and_four_journals_each_once():
    world.catalogue()
    world.warehouse()
    receive_line("R-1", "8", "21.60")
    receive_line("R-2", "4", "24.30")
    deliver("SO-1", "120")                                                                            # 90 LTR
    assert (val()["quantity_base"], val()["value_amount"], val()["average_cost"]) == ("126", "157.50", "1.2500")
    deliver("SO-2", "168")                                                                            # the last 126 LTR
    last = val()
    assert (last["quantity_base"], last["value_amount"]) == ("0", "0.00") and last["sum_of_move_values"] == "0.00"
    rows = [(r["quantity_after"], r["value_after"], r["average_cost_after"]) for r in last["valuation_rows"]]
    assert rows == [("144", "172.80", "1.2000"), ("216", "270.00", "1.2500"), ("126", "157.50", "1.2500"), ("0", "0.00", "1.2500")]
    assert len(fakes.JOURNALS) == 4
    net = {}
    for j in fakes.JOURNALS.values():
        for l in j["lines"]:
            if l["purpose"] == "stock_valuation":
                net[j["source_document"]] = Decimal_(l.get("debit", "0")) - Decimal_(l.get("credit", "0"))
    assert sum(net.values()) == 0                                                                          # the stock account nets to zero, no residue


def Decimal_(x):
    from decimal import Decimal
    return Decimal(x)


def test_inv32_a_refused_posting_is_pending_with_the_ledgers_reason_and_retries_land_it_once():
    world.catalogue()
    world.warehouse()
    fakes.GL_REFUSE.update(gate="period_open", reason="the period is closed", status=409)
    receive_line("R-1", "8", "21.60")
    assert must(api("GET", "/stock/COLA-075", MANAGER))["on_hand"] == "144"                                # the stock move stands
    (p,) = must(api("GET", "/postings", ACCOUNTANT, {"status": "pending"}))["postings"]
    assert p["event"] == "receipt:1" and "the period is closed" in p["reason"] and "reopen the period" in p["reason"] and p["attempts"] == 1
    assert fakes.JOURNALS == {}
    fakes.GL_REFUSE.clear()
    assert must(api("POST", "/postings/retry", SYSTEM, {})) == {"tried": 1, "posted": 1, "still_pending": 0}
    must(api("POST", "/postings/retry", SYSTEM, {}))
    (j,) = fakes.JOURNALS.values()
    assert j["source_module"] == "inventory" and j["journal_id"] == str(p["journal_id"]) and j["company_code"] == "UK1"
    assert j["currency"] == "GBP"                          # INV-24: the company's currency, read from gl GET /companies/{company_code}
    assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j["lines"]} == {"stock_valuation": ("172.80", None), "stock_received_not_billed": (None, "172.80")}
    assert api("POST", "/postings/retry", MANAGER, {})[0] == 403


def test_inv32_the_account_map_gives_the_ledger_account_ids_and_export_mode_posts_nothing():
    world.catalogue()
    world.warehouse()
    must(api("POST", "/account-map", ACCOUNTANT, {"company_code": "UK1", "purpose": "stock_valuation", "account_id": "1400"}))
    receive_line("R-1", "8", "21.60")
    (j,) = fakes.JOURNALS.values()
    assert [l for l in j["lines"] if l.get("account_id") == "1400"] and not [l for l in j["lines"] if l.get("purpose") == "stock_valuation"]
    assert [l for l in j["lines"] if l.get("purpose") == "stock_received_not_billed"][0].get("account_id") is None     # unmapped: the ledger maps the purpose
    refused(api("POST", "/account-map", ACCOUNTANT, {"company_code": "UK1", "purpose": "nonsense", "account_id": "1"}), 422, "purpose is one of")
    with on_day("2026-12-01T09:00:00+00:00"):
        pass
    must(api("POST", "/settings", OWNER, {"key": "ledger_mode", "value": "export", "valid_from": "2026-12-01"}))
    fakes.JOURNALS.clear()
    with on_day("2026-12-02T09:00:00+00:00"):
        receive_line("R-2", "1", "21.60", on="2026-12-02")
    assert fakes.JOURNALS == {}
    assert must(api("GET", "/postings", ACCOUNTANT, {"status": "exported"}))["postings"][0]["document_id"] == "rcpt:pur:R-2"


def test_inv48_standard_cost_revalues_the_stock_and_books_the_price_difference():
    world.catalogue()
    world.warehouse()
    receive_line("R-1", "8", "21.60")
    receive_line("R-2", "4", "24.30")
    deliver("SO-1", "120")                                                                                  # 126 LTR, 157.50 average
    must(api("POST", "/settings", OWNER, {"key": "valuation_method:UK1", "value": "standard", "valid_from": "2026-12-01"}))
    refused(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "COLA-075", "standard_price": "1.2", "valid_from": "2026-11-20"}),
            409, "starts tomorrow at the earliest")
    refused(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "COLA-075", "standard_price": "1.2", "valid_from": "2026-11-25"}),
            409, "values stock at average")
    must(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "COLA-075", "standard_price": "1.2000", "valid_from": "2026-12-01"}))
    refused(api("POST", "/standard-prices", MANAGER, {"company_code": "UK1", "product_id": "COLA-075", "standard_price": "1.2", "valid_from": "2026-12-01"}),
            403, "inv:valuation:write")
    with on_day("2026-12-01T09:00:00+00:00"):
        assert must(api("POST", "/standard-prices/apply", SYSTEM, {}))["revalued"] == 1
        assert must(api("POST", "/standard-prices/apply", SYSTEM, {}))["revalued"] == 0                       # once
        v = val()
        assert (v["quantity_base"], v["value_amount"], v["method"]) == ("126", "151.20", "standard")
        reval = [j for j in fakes.JOURNALS.values() if j["event"] == "revaluation"]
        assert len(reval) == 1 and {l["purpose"]: (l.get("debit"), l.get("credit")) for l in reval[0]["lines"]} == {
            "stock_valuation": (None, "6.30"), "revaluation": ("6.30", None)}
    with on_day("2026-12-03T09:00:00+00:00"):
        receive_line("R-3", "8", "24.30", on="2026-12-03")                                                    # 144 LTR at 1.35 = 194.40
        r = [j for j in fakes.JOURNALS.values() if j["source_document"] == "rcpt:pur:R-3"][0]
        assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in r["lines"]} == {
            "stock_valuation": ("172.80", None), "price_difference": ("21.60", None), "stock_received_not_billed": (None, "194.40")}
    with on_day("2026-12-04T09:00:00+00:00"):
        deliver("SO-2", "120")                                                                                # 90 LTR at 1.20
        v = val()
        assert (v["quantity_base"], v["value_amount"]) == ("180", "216.00")
        assert v["valuation_rows"][-1]["value_after"] == "216.00" and v["valuation_rows"][-1]["average_cost_after"] == "1.2000"
        d = must(api("GET", "/documents/dlv:so:SO-2:1", ACCOUNTANT))
        assert d["moves"][0]["value_amount"] == "-108.00"


def test_inv56_a_new_standard_price_revalues_what_is_on_the_shelf_once():
    world.catalogue()
    world.warehouse()
    must(api("POST", "/settings", OWNER, {"key": "valuation_method:UK1", "value": "standard", "valid_from": "2026-12-01"}))
    must(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "COLA-075", "standard_price": "1.2000", "valid_from": "2026-12-01"}))
    with on_day("2026-12-02T09:00:00+00:00"):
        receive_line("R-1", "10", "21.60", on="2026-12-02")                                                    # 180 LTR
        must(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "COLA-075", "standard_price": "1.2500", "valid_from": "2026-12-20"}))
        refused(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "COLA-075", "standard_price": "1.3", "valid_from": "2026-12-02"}), 409, "starts tomorrow")
    with on_day("2026-12-20T09:00:00+00:00"):
        must(api("POST", "/standard-prices/apply", SYSTEM, {}))
        v = val()
        assert v["value_amount"] == "225.00" and v["quantity_base"] == "180"
        rev = [j for j in fakes.JOURNALS.values() if j["event"] == "revaluation" and j["journal_date"] == "2026-12-20"]
        assert len(rev) == 1 and rev[0]["lines"][0]["debit"] == "9.00"
        must(api("POST", "/standard-prices/apply", SYSTEM, {}))
        assert len([j for j in fakes.JOURNALS.values() if j["event"] == "revaluation"]) == 1                # the switch on a zero stock books nothing; this one books once


def test_a_company_on_standard_with_no_price_cannot_make_up_a_value():
    world.catalogue()
    world.warehouse()
    must(api("POST", "/settings", OWNER, {"key": "valuation_method:UK1", "value": "standard", "valid_from": "2026-12-01"}))
    with on_day("2026-12-02T09:00:00+00:00"):
        refused(api("POST", "/receipts", SYSTEM, {"move_ref": "pur:R-9:1", "source_id": "R-9", "product_id": "COLA-075", "quantity": "1", "unit_code": "PK24",
                                                  "location_id": "WH1/A/01", "received_on": "2026-12-02", "unit_cost": "21.6"}), 409, "no standard price")


def test_inv57_a_delivery_cost_difference_is_spread_over_what_is_still_on_hand_and_the_rest_is_price_difference():
    world.catalogue()
    world.warehouse()
    receive_line("R-77", "100", "18.00")                                                                       # 1800.00, 1800 LTR
    deliver("SO-1", "1920")                                                                                    # 1440 LTR out: 360 LTR (20 PK24) left, FIFO from this receipt
    adj = {"adjustment_ref": "pur:B-912:1", "source_kind": "purchase_delivery_cost", "receipt_id": "R-77", "product_id": "COLA-075", "amount": "30.00"}
    got = must(api("POST", "/valuation-adjustments", SYSTEM, adj))
    assert got["on_hand_amount"] == "6.00" and got["price_difference_amount"] == "24.00"                      # 20 of 100 on hand
    assert must(api("POST", "/valuation-adjustments", SYSTEM, adj))["replay"] is True
    j = [j for j in fakes.JOURNALS.values() if j["source_document"] == "vadj:pur:B-912:1"]
    assert len(j) == 1 and {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j[0]["lines"]} == {
        "stock_valuation": ("6.00", None), "price_difference": ("24.00", None), "stock_received_not_billed": (None, "30.00")}
    assert val()["value_amount"] == "366.00"
    assert api("POST", "/valuation-adjustments", MANAGER, adj)[0] == 403


def test_opening_stock_at_a_unit_cost_makes_the_first_sale_post_cost_of_goods():
    """A shop starting with stock on the shelf records it once through POST /receipts with `unit_cost` (a count of found items has no
    cost: it is valued at the average, which is zero). 40 LTR at 9.00, then a delivery of 10 LTR: COGS 90.00 to the ledger."""
    world.catalogue()
    world.warehouse()
    receive_line("OPEN-1", "40", "9.00", unit="LTR")
    deliver("SO-1", "10", unit="LTR", num=1, den=1)
    assert (val()["quantity_base"], val()["value_amount"]) == ("30", "270.00")
    cogs = [l for j in fakes.JOURNALS.values() for l in j["lines"] if l["purpose"] == "cost_of_goods_sold"]
    assert [Decimal_(l.get("debit", "0")) for l in cogs] == [Decimal_("90.00")]