dodil/erp-brick-inventory

DODIL ERP brick: inventory

tests/test_replenishment.py

"""Replenishment: INV-44 (propose, accept, dismiss, auto) and INV-54 (one purchase request per accepted proposal)."""
from decimal import Decimal

import fakes
import world
from conftest import MANAGER, OPERATOR, OWNER, SYSTEM, api, must, refused

SITE1 = {"product_id": "COLA-075", "site_id": "WH1", "valid_from": "2026-01-01", "procurement_type": "buy", "planning_method": "min_max",
         "reorder_point": "360", "safety_stock": "90", "max_stock": "1440", "purchase_unit_code": "PK24", "base_unit_code": "LTR",
         "lot_size_method": "none", "rounding_profile_code": "PALLET-LAYER"}
SITE2 = {"product_id": "COLA-075", "site_id": "WH2", "valid_from": "2026-02-01", "procurement_type": "transfer", "supplying_site_id": "WH1",
         "planning_method": "min_max", "reorder_point": "72", "max_stock": "500", "lot_size_method": "fixed", "fixed_lot_size": "144",
         "base_unit_code": "LTR", "planned_delivery_days": 2}


def plant():
    world.catalogue()
    world.warehouse()
    world.warehouse("WH2", bins=("A/01",), extra_bins=[("QC", {"stock_status": "quality"})])
    world.receive("COLA-075", "20", "PK24", price="21.60", location="WH1/A/01")                     # 360 LTR
    world.receive("COLA-075", "3", "PK24", price="21.60", location="WH2/A/01")                      # 54 LTR
    must(api("POST", "/reservations/sales-orders/SO-9", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [
        {"line_no": 1, "product_id": "COLA-075", "unit_code": "LTR", "quantity": "60", "unit_numerator": 1, "unit_denominator": 1}]}))
    fakes.SITES.extend([dict(SITE1), dict(SITE2)])
    fakes.ROUNDING[("COLA-075", "WH1")] = lambda q: Decimal(60) if q == 64 else q


def run():
    return must(api("POST", "/replenishment/run", SYSTEM, {}))


def test_inv44_low_projected_stock_is_proposed_up_to_the_max_rounded_to_how_we_order_and_a_repeat_writes_nothing():
    plant()
    got = run()
    by = {p["warehouse_id"]: p for p in got["proposals"]}
    w1, w2 = by["WH1"], by["WH2"]
    assert w1["figures"]["on_hand_base"] == "360" and w1["figures"]["reserved_base"] == "60" and w1["figures"]["projected_base"] == "300"
    assert w1["figures"]["proposed_base"] == "1140" and (w1["rounded"], w1["unit_code"], w1["procurement"]) == ("60", "PK24", "buy")
    assert w1["site_row_valid_from"] == "2026-01-01" and w1["request_ref"] == "inv:repl:WH1:COLA-075:2026-11-20"
    assert (w2["procurement"], w2["supplying_site_id"], w2["rounded"], w2["figures"]["projected_base"]) == ("transfer", "WH1", "144", "54")
    assert run()["proposals"] == []                                                                    # a repeat writes nothing more
    assert len(must(api("GET", "/replenishment-proposals", MANAGER))["proposals"]) == 2
    assert fakes.REQUESTS == {}                                                                         # suggest mode: nothing sent without a person
    assert api("POST", "/replenishment/run", MANAGER, {})[0] == 403                                     # a machine grant


def test_inv44_accepting_sends_one_purchase_request_and_the_other_becomes_a_draft_transfer():
    plant()
    run()
    base = "/replenishment-proposals/WH1/COLA-075/2026-11-20"
    sent = must(api("POST", f"{base}/accept", MANAGER, {}))
    assert sent["lifecycle"] == "sent" and sent["purchase_request_id"] == "PR-1"
    (req,) = fakes.REQUESTS.values()
    assert req["body"]["quantity"] == "60" and req["body"]["unit_code"] == "PK24" and req["body"]["warehouse_id"] == "WH1"
    again = must(api("POST", f"{base}/accept", MANAGER, {}))
    assert again["replay"] is True and len(fakes.REQUESTS) == 1                                          # INV-54: one request
    t = must(api("POST", "/replenishment-proposals/WH2/COLA-075/2026-11-20/accept", MANAGER, {}))
    doc = must(api("GET", f"/transfers/{t['transfer_document_id']}", OPERATOR))
    assert doc["lifecycle"] == "draft" and doc["lines"][0]["quantity_base"] == "144" and doc["lines"][0]["to_location_id"] == "WH2/A/01"
    refused(api("POST", f"{base}/dismiss", MANAGER, {"reason": "late"}), 409, "already sent")
    assert api("POST", f"{base}/accept", OPERATOR, {})[0] == 403


def test_inv44_dismissing_needs_a_reason_and_a_changed_quantity_must_meet_the_purchase_rule():
    plant()
    run()
    base = "/replenishment-proposals/WH1/COLA-075/2026-11-20"
    refused(api("POST", f"{base}/dismiss", MANAGER, {"reason": " "}), 422, "reason")
    assert must(api("POST", f"{base}/dismiss", MANAGER, {"reason": "promotion ended"}))["lifecycle"] == "dismissed"
    assert run()["proposals"] == []                                                                     # not proposed again that day
    refused(api("POST", f"{base}/accept", MANAGER, {}), 409, "dismissed")
    assert fakes.REQUESTS == {}


def test_inv54_purchasing_refusing_the_request_leaves_the_proposal_open_and_a_changed_quantity_goes_through():
    plant()
    run()
    base = "/replenishment-proposals/WH1/COLA-075/2026-11-20"
    fakes.REQUEST_REFUSAL.append(1)
    body = refused(api("POST", f"{base}/accept", MANAGER, {}), 409, "the product has no supplier")
    assert body["detail"]["gate"] == "purchasable"
    assert [p["lifecycle"] for p in must(api("GET", "/replenishment-proposals", MANAGER, {"warehouse_id": "WH1"}))["proposals"]] == ["proposed"]
    fakes.REQUEST_REFUSAL.clear()
    must(api("POST", f"{base}/accept", MANAGER, {"quantity": "50"}))
    (req,) = fakes.REQUESTS.values()
    assert req["body"]["quantity"] == "50"


def test_inv44_auto_mode_sends_without_a_person():
    plant()
    must(api("POST", "/settings", OWNER, {"key": "replenishment_mode", "value": "auto"}))
    got = run()
    assert {p["lifecycle"] for p in got["proposals"]} == {"sent"} and len(fakes.REQUESTS) == 1
    assert {p["decided_by"] for p in got["proposals"]} == {"system@shop.example"}
    assert run()["proposals"] == [] and len(fakes.REQUESTS) == 1


def test_inv44_a_proposal_never_counts_quality_stock():
    plant()
    must(api("POST", "/stock/change-type", MANAGER, {"location_id": "WH2/A/01", "product_id": "COLA-075", "quantity": "36", "unit_code": "LTR",
                                                    "from_type": "unrestricted", "to_type": "quality", "reason": "damaged"}))
    by = {p["warehouse_id"]: p for p in run()["proposals"]}
    assert by["WH2"]["figures"]["on_hand_base"] == "18" and by["WH2"]["figures"]["projected_base"] == "18"


def test_inv44_a_changed_quantity_below_the_purchase_rule_is_refused_and_enough_projected_stock_proposes_nothing():
    plant()
    run()
    fakes.MIN_ORDER["COLA-075"] = Decimal(30)
    body = refused(api("POST", "/replenishment-proposals/WH1/COLA-075/2026-11-20/accept", MANAGER, {"quantity": "10"}), 409, "the minimum order is 30")
    assert body["detail"]["gate"] == "purchase_rule" and fakes.REQUESTS == {}
    world.receive("COLA-075", "20", "PK24", price="21.60", location="WH1/A/02")                      # 720 litres more: projected 1020
    with __import__("conftest").on_day("2026-11-21T09:00:00+00:00"):
        assert [p["warehouse_id"] for p in run()["proposals"]] == ["WH2"]                              # WH1 is above its reorder point now