101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
# ============================================================
|
|
# MD-05 SAP 集成桩黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
from pathlib import Path
|
|
|
|
from server.agent_core import harness
|
|
from server.aps_domain.flex import run_flex_schedule
|
|
from server.aps_domain.sap_sync import (
|
|
apply_inbound,
|
|
apply_outbound,
|
|
build_outbound_projection,
|
|
preview_inbound,
|
|
preview_outbound,
|
|
)
|
|
from server.integrations.sap_stub import reset_sap_client
|
|
from server.state.seed import seed_world
|
|
|
|
|
|
class _MemStore:
|
|
def __init__(self, data):
|
|
self.data = data
|
|
|
|
def next_id(self, kind: str) -> int:
|
|
key = f"_c_{kind}"
|
|
self.data[key] = self.data.get(key, 3000) + 1
|
|
return self.data[key]
|
|
|
|
def save(self):
|
|
pass
|
|
|
|
|
|
def test_inbound_creates_flex_orders(tmp_path: Path):
|
|
mirror = tmp_path / "sap_mirror.json"
|
|
reset_sap_client(mirror)
|
|
store = _MemStore(seed_world())
|
|
before = len(store.data.get("flexOrders") or [])
|
|
prev = preview_inbound(store.data)
|
|
assert len(prev["newOrders"]) >= 1
|
|
wire = next(m for m in store.data["flexMaterials"] if m["code"] == "WIRE-HV")
|
|
old_stock = wire["stock"]
|
|
r = apply_inbound(store, actor="test")
|
|
assert r["created"]
|
|
assert len(store.data["flexOrders"]) == before + len(r["created"])
|
|
assert all(o.get("source") == "SAP" for o in r["created"])
|
|
assert all(o.get("externalAufnr") for o in r["created"])
|
|
assert wire["stock"] != old_stock or "WIRE-HV" in r["stockUpdated"]
|
|
# 二次入站应跳过已映射订单
|
|
prev2 = preview_inbound(store.data)
|
|
assert all(s["reason"] == "已同步" for s in prev2["skipOrders"][:3])
|
|
assert len(prev2["newOrders"]) == 0
|
|
|
|
|
|
def test_outbound_idempotent(tmp_path: Path, monkeypatch):
|
|
mirror = tmp_path / "sap_mirror.json"
|
|
client = reset_sap_client(mirror)
|
|
store = _MemStore(seed_world())
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
prev = preview_outbound(store.data)
|
|
assert prev["items"]
|
|
monkeypatch.setattr(harness, "consume_execution_grant", lambda *_a, **_k: True)
|
|
|
|
class Checkpoints:
|
|
def __init__(self):
|
|
self.rows = {}
|
|
|
|
def add(self, pair_id: str) -> None:
|
|
self.rows[pair_id] = {"pairId": pair_id, "world": copy.deepcopy(store.data)}
|
|
|
|
def get(self, pair_id: str):
|
|
return self.rows.get(pair_id)
|
|
|
|
checkpoints = Checkpoints()
|
|
|
|
def execute(confirm_id: str, pair_id: str):
|
|
projection = build_outbound_projection(store.data, limit=50)
|
|
checkpoints.add(pair_id)
|
|
params = {**copy.deepcopy(projection), "beforeSnapshot": pair_id}
|
|
return apply_outbound(
|
|
store,
|
|
actor="test",
|
|
confirm_id=confirm_id,
|
|
execution_grant=f"grant-{confirm_id}",
|
|
approval_params=params,
|
|
requester={"userId": 1},
|
|
approvals=[{"userId": 2}, {"userId": 3}],
|
|
before_snapshot=pair_id,
|
|
checkpoint_store=checkpoints,
|
|
)
|
|
|
|
r1 = execute("confirm-1", "pair-1")
|
|
assert r1["pushed"]
|
|
assert len(r1["duplicates"]) == 0
|
|
n1 = client.status()["receiptCount"]
|
|
r2 = execute("confirm-2", "pair-2")
|
|
assert len(r2["pushed"]) == 0
|
|
assert len(r2["duplicates"]) == len(r1["pushed"])
|
|
assert client.status()["receiptCount"] == n1
|