dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
tests/test_ai.py
"""The explain_variance AI step (ai_steps.explain_variance): suggest by default, the owner's mode, a person decides, everything audited, nothing
changed by the step, and the variance split alone as the fallback - never a 500, never a block on closing."""
import json
import re
from decimal import Decimal
import pytest
from conftest import (
ACCOUNTANT,
OPERATOR,
OWNER,
PLANNER,
api,
close,
complete,
issue,
must,
new_order,
on_day,
output,
refused,
release,
)
from erp_platform import capabilities, seam
from fastapi import FastAPI, Request, Response
from test_floor import confirm, routed_order
class Provider:
"""A stand-in for whoever provides ai.suggest in a composition: answers what the test says, records what it was asked."""
def __init__(self):
self.mode, self.calls = "ok", []
self.causes = [{"cause": "a lower yield on lot G-2611-ETH-07", "evidence": "82.50% against 84.00% planned; the lot is the only input"},
{"cause": "the roast ran 10 minutes long", "evidence": "160 minutes confirmed against 150 planned"},
{"cause": "a third cause", "evidence": "evidence three"}, {"cause": "a fourth cause", "evidence": "evidence four"}]
self.app = FastAPI()
me = self
@self.app.post("/capabilities/ai.suggest")
async def suggest(request: Request, response: Response):
body = await request.json()
me.calls.append(body)
if me.mode == "down":
response.status_code = 500
return {"detail": "down"}
if me.mode == "refuse":
response.status_code = 422
return {"detail": "bad question"}
causes = {"ok": me.causes, "empty": [], "garbage": ["nonsense", {"cause": "", "evidence": "x"}]}[me.mode]
return {"status": "done", "provider": "ai-fake", "output": {"suggestion": {"causes": causes}, "rationale": "the lot and the time explain it",
"confidence": 72, "model": "kimi-k2.6", "sources": ["mfg_order_movement"]}}
@pytest.fixture
def provider():
p = Provider()
seam.register_local("ai-fake", p.app)
capabilities.register_provider("ai.suggest", "ai-fake")
yield p
capabilities.reset_providers()
seam.unregister_local("ai-fake")
def completed_order(nb):
oid = routed_order()
must(issue(oid))
must(output(oid))
must(confirm(oid, machine="160", labour="160"), 201)
must(complete(oid))
nb.traces[("GREEN-ETH", "G-2611-ETH-07")] = {"history": [{"document_id": "PO-77/1", "bp_id": "SUPPLIER-ETH", "quantity_base": "500"}], "on_hand": [],
"delivered_to": []}
return oid
def explain(oid, who=None):
return api("POST", f"/production-orders/{oid}/explain-variance", who or ACCOUNTANT)
def test_a_suggestion_is_audited_capped_at_three_and_changes_nothing(nb, provider):
oid = completed_order(nb)
before = (len(nb.journals), len(nb.calls), must(api("GET", f"/production-orders/{oid}", ACCOUNTANT)))
got = must(explain(oid), 201)
assert got["explained"] is True and got["mode"] == "suggest" and got["finding_no"] == 1 and got["decision"] == "pending"
assert [c["cause"] for c in got["causes"]] == ["a lower yield on lot G-2611-ETH-07", "the roast ran 10 minutes long", "a third cause"] # up to three
assert got["variance"]["balance"] == "25.33" and got["model"] == "kimi-k2.6" and re.fullmatch(r"[0-9a-f]{64}", got["inputs_digest"])
ask = provider.calls[0]
assert ask["idempotency_key"].startswith(f"explain_variance|mfg_production_order:{oid}|") and ask["input"]["step"] == "explain_variance"
ctx = {k: (json.loads(v) if isinstance(v, str) and v[:1] in "[{" else v) for k, v in ask["input"]["context"].items()} # lists and maps arrive as JSON text
assert ctx["actual_yield_pct"] == "82.50" and ctx["planned_yield_pct"] == "84.0000" and ctx["input_lot_suppliers"] == {"GREEN-ETH/G-2611-ETH-07": ["SUPPLIER-ETH"]}
assert ctx["confirmations"][0]["machine_minutes"] == "160.0000" and ctx["planned_operations"][0]["run_minutes"] == "20" and ctx["last_orders"] == []
after = (len(nb.journals), len(nb.calls), must(api("GET", f"/production-orders/{oid}", ACCOUNTANT)))
assert before == after # nothing posted, nothing sent, the order untouched
audit = must(api("GET", f"/production-orders/{oid}/variance-findings", ACCOUNTANT))["findings"][0]
assert (audit["model"], audit["inputs_digest"], audit["decision"], audit["requested_by"], audit["decided_by"]) == (
"kimi-k2.6", got["inputs_digest"], "pending", "cost_accountant@roastery.example", None)
assert len(audit["causes"]) == 3
def test_a_person_decides_once_and_it_is_recorded(nb, provider):
oid = completed_order(nb)
must(explain(oid), 201)
refused(api("POST", f"/production-orders/{oid}/variance-findings/1/decide", OPERATOR, {"decision": "accepted"}), 403)
refused(api("POST", f"/production-orders/{oid}/variance-findings/1/decide", PLANNER, {"decision": "accepted"}), 403)
refused(api("POST", f"/production-orders/{oid}/variance-findings/9/decide", ACCOUNTANT, {"decision": "accepted"}), 404)
refused(api("POST", f"/production-orders/{oid}/variance-findings/1/decide", ACCOUNTANT, {"decision": "maybe"}), 422)
got = must(api("POST", f"/production-orders/{oid}/variance-findings/1/decide", ACCOUNTANT, {"decision": "accepted"}))
assert got["decision"] == "accepted" and got["decided_by"] == "cost_accountant@roastery.example" and got["decided_at"]
refused(api("POST", f"/production-orders/{oid}/variance-findings/1/decide", ACCOUNTANT, {"decision": "dismissed"}), 409, "already accepted", gate="state")
assert must(explain(oid), 201)["finding_no"] == 2 # a second question is a second audited finding
def test_the_owner_holds_the_mode_and_off_means_the_model_is_never_asked(nb, provider):
oid = completed_order(nb)
refused(api("POST", "/settings/ai_explain_variance_mode", PLANNER, {"value": "off", "valid_from": "2026-11-11"}), 403)
refused(api("POST", "/settings/ai_explain_variance_mode", OWNER, {"value": "auto"}), 422)
must(api("POST", "/settings/ai_explain_variance_mode", OWNER, {"value": "off"}))
got = explain(oid)
assert got[0] == 201 and got[1]["mode"] == "off" and got[1]["explained"] is False and got[1]["variance"]["balance"] == "25.33"
assert "off" in got[1]["reason"] and provider.calls == []
assert must(api("GET", f"/production-orders/{oid}/variance-findings", ACCOUNTANT))["findings"] == []
@pytest.mark.parametrize("mode,words", [("down", "failed"), ("refuse", "refused"), ("empty", "no usable cause"), ("garbage", "no usable cause")])
def test_a_provider_that_fails_or_answers_nonsense_leaves_the_variance_split_alone(nb, provider, mode, words):
oid = completed_order(nb)
provider.mode = mode
got = must(explain(oid), 201)
assert got["explained"] is False and got["causes"] == [] and got["variance"]["balance"] == "25.33" and words in got["reason"]
assert must(api("GET", f"/production-orders/{oid}/variance-findings", ACCOUNTANT))["findings"] == [] # nothing to audit: no finding exists
def test_no_provider_is_the_fallback_not_an_error(nb):
oid = completed_order(nb)
got = must(explain(oid), 201)
assert got["explained"] is False and "no explanation service" in got["reason"] and got["variance"]["material_usage_variance"] == "11.07"
def test_the_step_never_blocks_closing_and_needs_a_variance_to_explain(nb, provider):
oid = routed_order()
refused(explain(oid), 409, gate="state") # still released: no variance yet
must(issue(oid))
must(output(oid))
must(confirm(oid, machine="160", labour="160"), 201)
must(complete(oid))
must(explain(oid), 201) # a pending finding is not a gate
assert must(close(oid))["status"] == "closed"
assert must(explain(oid), 201)["explained"] is True # and a closed order can still be explained
refused(explain("NOPE"), 404)
refused(explain(oid, OPERATOR), 403)
def test_the_last_orders_of_the_same_product_are_part_of_what_the_model_sees(nb, provider):
first = completed_order(nb)
must(close(first))
second = new_order()
must(release(second))
with on_day("2026-11-16T09:00:00+00:00"):
must(api("POST", f"/production-orders/{second}/issue", OPERATOR, {"location_id": "ROASTERY/RAW", "lines": [{"product_id": "GREEN-ETH", "quantity": "100", "lot_code": "G2"}]}))
nb.available[("GREEN-ETH", "ROASTERY")] = Decimal("500")
must(output(second, lot="R2", chaff=None))
must(confirm(second, machine="150", labour="150"), 201)
must(complete(second))
must(explain(second), 201)
last = json.loads(provider.calls[-1]["input"]["context"]["last_orders"])
assert [(r["production_order_id"], r["material_usage_variance"], r["activity_variance"]) for r in last] == [(first, "11.07", "14.26")]