dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

tests/test_seams.py

"""Every neighbour call is checked against the CALLEE's signed spec (tests/conftest.py); these tests show the check is live, that the
fakes answer what the spec says, and which gaps in the neighbours' specs the conftest works around - they fail the day a spec closes a
gap, so the workaround goes with it."""
import json
import pathlib

import pytest
import yaml
from conftest import METHOD_DIR, SYSTEM, api, close, complete, get_order, issue, must, output, released_order

from brick_manufacturing.infrastructure import seams

MODULES = pathlib.Path(METHOD_DIR) / "modules"


def _spec(name):
    return yaml.load((MODULES / name / "module.yaml").read_text(), Loader=yaml.CSafeLoader)


def test_a_call_the_callee_would_refuse_is_caught_at_the_call(contract):
    with pytest.raises(AssertionError, match="breaks the callee's spec"):
        seams.INVENTORY.post("/goods-issues/from-source", {"source_kind": "production_issue", "source_id": "X", "lines": []})
    assert len(contract.violations) == 1
    contract.violations.clear()                 # that one was deliberate


def test_the_calls_a_real_run_makes_all_pass_the_contract_check(nb):
    oid = released_order()
    must(issue(oid))
    must(output(oid))
    must(complete(oid))
    must(close(oid))
    assert [k for k, _ in nb.calls] == ["issue", "receipt", "receipt", "journal"]        # each checked as it was sent
    assert get_order(oid)["status"] == "closed"


def test_each_outgoing_route_is_listed_in_the_callees_spec_for_manufacturing():
    want = {("inventory", "POST /goods-issues/from-source"), ("inventory", "POST /receipts"), ("inventory", "POST /valuation-adjustments"),
            ("gl", "POST /journal-entry/journals")}
    got = set()
    for callee in ("inventory", "gl"):
        for s in _spec(callee)["seams"]:
            if s.get("with") == "manufacturing":
                got.add((callee, s["route"].split("?")[0].replace('"', "")))
    assert want <= got, want - got


# ── GAPS in the neighbours' specs (reported to their owners) ─────────────────────────────────────────────
def test_gap_product_lists_no_manufacturing_entry_for_get_product_or_convert():
    routes = [s["route"] for s in _spec("product")["seams"] if s.get("with") == "manufacturing"]
    assert not any(r.startswith("GET /products/{product_id}") and "product-sites" not in r for r in routes), (
        "product now lists manufacturing's calls: drop `UNCHECKED = ('product',)` in tests/conftest.py")


def test_gap_inventory_lists_availability_and_valuation_only_for_sales_order():
    for route in ("GET /availability/{product_id}?warehouse_id=", "GET /valuation/products/{product_id}?company_code=&as_of="):
        callers = {s["with"] for s in _spec("inventory")["seams"] if s["route"].strip('"') == route}
        assert "manufacturing" not in callers and "sales-order" in callers, (
            f"inventory now lists {route} for manufacturing: drop AS_OTHER_CALLER in tests/conftest.py")


def test_gap_inventory_lists_the_lot_trace_only_for_crm_inside_a_serials_row():
    rows = [s for s in _spec("inventory")["seams"] if "/lots/{product_id}/{lot_code}/trace" in s["route"]]
    assert rows and all(s.get("with") != "manufacturing" and " and " in s["route"] for s in rows), (
        "inventory now lists the lot trace for manufacturing: drop UNCHECKED_PATHS in tests/conftest.py")


def test_gap_the_gl_journal_id_is_derived_from_the_document_and_the_event_not_from_the_settlement_number():
    entry = next(s for s in _spec("gl")["seams"] if s.get("with") == "manufacturing")
    jid = next(f for f in entry["request_schema"] if f["field"] == "journal_id")
    assert jid["derived"] == "derived_id(source_document, event)" and "settle:<n>" in entry["request"]    # prose says settle:<n>
    # so slice A sends derived_id(order, "settle"): one settlement per order, the post-close settle:2 needs gl to say how


def test_the_fake_inventory_answers_what_the_spec_says_it_answers(nb):
    body = {"source_kind": "production_issue", "source_id": "UK01-MO-2026-00001", "company_code": "UK01", "location_id": "ROASTERY/RAW",
            "issued_on": "2026-11-10", "account_purpose": "wip",
            "lines": [{"move_ref": "mfg:UK01-MO-2026-00001:issue:1:1", "product_id": "GREEN-ETH", "quantity": "1.000", "unit_code": "KGM"}]}
    first, again = seams.INVENTORY.post("/goods-issues/from-source", body), seams.INVENTORY.post("/goods-issues/from-source", body)
    assert (first.status, again.status) == (201, 200)                                    # 201 first, 200 for a known move_ref
    assert first.data["moves"][0]["value"] == again.data["moves"][0]["value"] == "6.20" and "issue_id" in first.data
    assert json.dumps(first.data) == json.dumps(again.data)
    assert api("GET", "/process/bpmn", SYSTEM)[0] == 200