235 lines
8.7 KiB
Python
235 lines
8.7 KiB
Python
# ============================================================
|
||
# OR-01 订单管理黄金测试（moduleId: golden-order-management, 发版门禁 §14.3）
|
||
# 固化订单池写入的最小闭环：
|
||
# O1 新增订单进入后续排产输入
|
||
# O2 取消订单从后续排产输入排除
|
||
# O3 P2 确认卡摘要可生成并校验成品物料
|
||
# O4 审计事件同时满足 at/ts 契约且哈希链可校验
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from server.agent_core.audit import write_audit
|
||
from server.agent_core.registry import verify_audit_chain
|
||
from server.aps_domain.orders import apply_order_action, confirmation_for_order_action
|
||
from server.engines import get_engine
|
||
from server.engines.base import EngineParams
|
||
from server.state.seed import seed_world
|
||
from server.timeutil import add_minutes, fmt_date, today0
|
||
|
||
|
||
def _next_id_factory(world):
|
||
"""按当前世界最大 ID 起号，避免新增订单与种子 ID 冲突。"""
|
||
tables = {
|
||
"audit": "auditEvents",
|
||
"conflict": "conflicts",
|
||
"log": "logs",
|
||
"productionOrder": "productionOrders",
|
||
"salesOrder": "salesOrders",
|
||
"scheduleVersion": "scheduleVersions",
|
||
"workOrder": "workOrders",
|
||
}
|
||
counters: dict[str, int] = {}
|
||
|
||
def next_id(kind: str) -> int:
|
||
if kind not in counters:
|
||
rows = world.get(tables.get(kind, kind + "s"), [])
|
||
counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
|
||
counters[kind] += 1
|
||
return counters[kind]
|
||
|
||
return next_id
|
||
|
||
|
||
def _run(world):
|
||
"""执行一次规则引擎排产。"""
|
||
params = EngineParams(
|
||
orderIds=[],
|
||
engineType="RULE",
|
||
strategyTemplate="COMPREHENSIVE",
|
||
planningHorizonDays=14,
|
||
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
||
)
|
||
return get_engine("RULE").solve(world, params, _next_id_factory(world))
|
||
|
||
|
||
def test_order_cancel_excludes_from_scheduling():
|
||
"""取消订单后，该订单不再生成 PO，参与排产订单数从 7 降为 6。"""
|
||
world = seed_world()
|
||
target = world["salesOrders"][0]
|
||
apply_order_action(world, _next_id_factory(world), "order.cancel", {"id": target["id"]})
|
||
|
||
result = _run(world)
|
||
|
||
assert result.orderCount == 6
|
||
assert all(po["salesOrderId"] != target["id"] for po in world["productionOrders"])
|
||
assert target["status"] == "CANCELLED"
|
||
assert all(item["status"] == "CANCELLED" for item in target["items"])
|
||
|
||
|
||
def test_order_upsert_create_enters_scheduling():
|
||
"""新增已确认订单后，下一版排产应把它纳入订单池。"""
|
||
world = seed_world()
|
||
payload = {
|
||
"customerName": "试点客户",
|
||
"customerLevel": "A",
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 9 * 24 * 60)),
|
||
"priority": 2,
|
||
"status": "CONFIRMED",
|
||
"productId": 1,
|
||
"quantity": 120,
|
||
"isRush": False,
|
||
"specialRequirements": "OR-01 golden",
|
||
}
|
||
applied = apply_order_action(world, _next_id_factory(world), "order.upsert", payload)
|
||
order = applied["order"]
|
||
|
||
result = _run(world)
|
||
|
||
assert applied["created"] is True
|
||
assert len(world["salesOrders"]) == 8
|
||
assert result.orderCount == 8
|
||
assert any(po["salesOrderId"] == order["id"] for po in world["productionOrders"])
|
||
assert order["orderNo"].startswith("SO")
|
||
|
||
|
||
def test_order_confirmation_summary_validates_product():
|
||
"""订单确认卡必须可解释影响范围，并拒绝非成品物料。"""
|
||
world = seed_world()
|
||
title, lines = confirmation_for_order_action(world, "order.upsert", {
|
||
"customerName": "试点客户",
|
||
"customerLevel": "VIP",
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 8 * 24 * 60)),
|
||
"priority": 1,
|
||
"status": "CONFIRMED",
|
||
"productId": 1,
|
||
"quantity": 80,
|
||
})
|
||
summary = "\n".join(lines)
|
||
|
||
assert "新增" in title
|
||
assert "智能控制器A型" in summary
|
||
assert "P2" in summary
|
||
with pytest.raises(ValueError):
|
||
confirmation_for_order_action(world, "order.upsert", {
|
||
"customerName": "错误物料客户",
|
||
"deliveryDate": fmt_date(add_minutes(today0(), 8 * 24 * 60)),
|
||
"productId": 10,
|
||
"quantity": 1,
|
||
})
|
||
|
||
|
||
def test_audit_event_exposes_at_and_ts_contract():
|
||
"""新增审计事件应同时提供 at/ts，前端治理页与旧数据均可读取。"""
|
||
world = seed_world()
|
||
event = write_audit(
|
||
world,
|
||
_next_id_factory(world),
|
||
actor="tester",
|
||
category="GATE",
|
||
action="order.upsert.stage",
|
||
target={"type": "SALES_ORDER", "id": None},
|
||
power="P2",
|
||
rationale={"confirmId": "unit-test"},
|
||
)
|
||
|
||
assert event["at"] == event["ts"]
|
||
assert verify_audit_chain(world["auditEvents"])["ok"] is True
|
||
|
||
|
||
def test_order_delete_removes_from_pool():
|
||
"""删除订单后从订单池物理移除，且不再参与排产。"""
|
||
world = seed_world()
|
||
target = world["salesOrders"][0]
|
||
oid = target["id"]
|
||
before = len(world["salesOrders"])
|
||
title, lines = confirmation_for_order_action(world, "order.delete", {"id": oid})
|
||
assert "删除" in title
|
||
assert any("物理" in s or "移除" in s for s in lines)
|
||
apply_order_action(world, _next_id_factory(world), "order.delete", {"id": oid})
|
||
assert len(world["salesOrders"]) == before - 1
|
||
assert all(so["id"] != oid for so in world["salesOrders"])
|
||
result = _run(world)
|
||
assert result.orderCount == 6
|
||
assert all(po["salesOrderId"] != oid for po in world["productionOrders"]
|
||
if po.get("schedulingVersionId") == world["scheduleVersions"][-1]["id"])
|
||
with pytest.raises(ValueError):
|
||
confirmation_for_order_action(world, "order.delete", {"id": oid})
|
||
|
||
|
||
def test_import_orders_and_materials():
|
||
"""Pi 已结构化的订单/物料行通过确定性校验并写入世界。"""
|
||
from server.aps_domain.intake import apply_import, parse_import
|
||
|
||
world = seed_world()
|
||
parsed = parse_import(
|
||
"orders",
|
||
world,
|
||
rows=[
|
||
{
|
||
"customerName": "比亚迪", "customerLevel": "VIP",
|
||
"productCode": "CTRL-C", "quantity": 200,
|
||
"deliveryDate": "2026-08-01",
|
||
},
|
||
{
|
||
"customerName": "华为", "productCode": "CTRL-A",
|
||
"quantity": 100, "deliveryDate": "2026-08-15",
|
||
},
|
||
],
|
||
)
|
||
assert parsed["kind"] == "orders" and len(parsed["rows"]) == 2
|
||
assert parsed["errors"] == []
|
||
before = len(world["salesOrders"])
|
||
applied = apply_import(world, _next_id_factory(world), parsed)
|
||
assert applied["count"] == 2
|
||
assert len(world["salesOrders"]) == before + applied["count"]
|
||
|
||
parsed2 = parse_import(
|
||
"materials",
|
||
rows=[{
|
||
"code": "SCR-NEW", "name": "不锈钢螺丝",
|
||
"type": "RAW_MATERIAL", "unit": "包", "stock": 800,
|
||
}],
|
||
)
|
||
assert parsed2["kind"] == "materials" and parsed2["rows"] and not parsed2["errors"]
|
||
apply_import(world, _next_id_factory(world), parsed2)
|
||
assert any(m["code"] == "SCR-NEW" for m in world["materials"])
|
||
|
||
missing = parse_import(None, world)
|
||
assert missing["rows"] == []
|
||
assert "kind" in missing["errors"][0]
|
||
|
||
legacy_text = parse_import("导入订单\n比亚迪,CTRL-C,200,2026-08-01,VIP", world)
|
||
assert legacy_text["rows"] == []
|
||
assert "kind" in legacy_text["errors"][0]
|
||
|
||
|
||
def test_order_and_master_clear():
|
||
"""一键清理：订单池清空；主数据工艺板块回种。"""
|
||
from server.aps_domain.masterdata import apply_master_action, confirmation_for_master_action
|
||
|
||
world = seed_world()
|
||
title, _ = confirmation_for_order_action(world, "order.clear", {})
|
||
assert "清理" in title
|
||
applied = apply_order_action(world, _next_id_factory(world), "order.clear", {})
|
||
assert applied["clearedOrders"] >= 1
|
||
assert world["salesOrders"] == []
|
||
assert world["purchaseOrders"] == []
|
||
|
||
world2 = seed_world()
|
||
apply_master_action(world2, _next_id_factory(world2), "master.material.upsert", {
|
||
"code": "TMP-CLR", "name": "临时料", "type": "RAW_MATERIAL", "unit": "件", "stock": 1,
|
||
})
|
||
assert any(m["code"] == "TMP-CLR" for m in world2["materials"])
|
||
t2, lines = confirmation_for_master_action(world2, "master.clear", {"scope": "process"})
|
||
assert "清理" in t2 and any("清空" in s for s in lines)
|
||
apply_master_action(world2, _next_id_factory(world2), "master.clear", {"scope": "process"})
|
||
assert world2["materials"] == []
|
||
assert world2["boms"] == []
|
||
|
||
world3 = seed_world()
|
||
assert world3["lines"]
|
||
apply_master_action(world3, _next_id_factory(world3), "master.clear", {"scope": "resource"})
|
||
assert world3["factories"] == [] and world3["lines"] == []
|