240 lines
16 KiB
Python
240 lines
16 KiB
Python
"""Read the user-named workbook and prove master edits reach actual trial output.
|
||
|
||
Run from an integrated Round87 checkout. All persistence is redirected before
|
||
server imports. The workbook is never modified and never skipped if missing.
|
||
Positive scenarios list their explicit, isolated test supplements; they do not
|
||
turn demo values or missing personnel into confirmed factory facts.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
|
||
class MemoryStore:
|
||
def __init__(self, data):
|
||
self.data, self.counters = data, {}
|
||
|
||
def next_id(self, kind):
|
||
self.counters[kind] = self.counters.get(kind, 10000) + 1
|
||
return self.counters[kind]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def verify(workbook: Path, expectations: dict | None = None) -> dict:
|
||
from server.aps_domain.flex import run_flex_schedule
|
||
from server.aps_domain.importers import apply_import_commit, preview_file
|
||
from server.aps_domain.masterdata import apply_master_action
|
||
from server.aps_domain.masterdata_consumption import as_datetime
|
||
from server.aps_domain.readiness import check_readiness
|
||
from server.state.seed import empty_world
|
||
from tests.workbook_acceptance import load_expectations, resolve_source
|
||
expected = expectations if expectations is not None else load_expectations()
|
||
workbook = resolve_source(workbook, required=True, expectations=expected)
|
||
trial, weekend = expected["trial"], expected["weekend"]
|
||
raw = workbook.read_bytes()
|
||
assert hashlib.sha256(raw).hexdigest() == expected["sourceSha256"], "必须使用已核对的用户指定原文件"
|
||
store = MemoryStore(empty_world())
|
||
preview = preview_file(workbook.name, raw, store.data)
|
||
assert preview["canCommit"], preview
|
||
apply_import_commit(store.data, store.next_id, preview["batches"])
|
||
world = store.data
|
||
counts = {key: len(world.get(key) or []) for key in (
|
||
"flexMaterials", "flexEquipment", "flexBom", "flexRoutings", "flexOrders",
|
||
"flexPersonnel", "flexWip", "flexMaintenance", "flexSandboxOrders")}
|
||
assert counts == expected["worldCounts"], counts
|
||
assert sum(m.get("inTransit", 0) for m in world["flexMaterials"]) == expected["inventory"]["inTransit"]
|
||
assert sum(bool(b.get("isKey")) for b in world["flexBom"]) == expected["inventory"]["keyBomRows"]
|
||
assert {o["orderNo"] for o in world["flexOrders"]} == set(expected["formalOrderNos"])
|
||
assert world["flexSandboxOrders"][0]["orderNo"] == expected["sandboxOrderNos"][0]
|
||
assert all(r.get("stdTimeSource") == expected["timeSource"] for r in world["flexRoutings"])
|
||
|
||
plan = expected["trialPlan"]
|
||
readiness = check_readiness(world)
|
||
baseline = run_flex_schedule(store, actor="round87-isolated-verification")
|
||
assert baseline["trialOnly"] and baseline["productionReady"] is False
|
||
# 资料检查仍是真实阻断（readiness 全部未就绪），试排则用显式假设排出草稿；
|
||
# 只有工作簿自己标注为完全短缺的订单保持未排出。
|
||
assert readiness["summary"]["blocked"] == expected["entityCounts"]["orders"]
|
||
assert baseline["woCount"] > 0 and baseline["solveStatus"] == plan["expectedSolveStatus"]
|
||
assert {row["orderNo"] for row in store.data["flexVirtualLines"]
|
||
if row["versionId"] == baseline["versionId"]} == set(plan["scheduledOrderNos"])
|
||
assert baseline["blockedOrderCount"] == len(plan["blockedOrderNos"])
|
||
baseline_conflicts = [c for c in store.data["flexConflicts"] if c["versionId"] == baseline["versionId"]]
|
||
assumptions = [c for c in baseline_conflicts if c["severity"] == plan["assumptionSeverity"]]
|
||
assert {c["conflictType"] for c in assumptions} == set(plan["assumptionTypes"])
|
||
assert all(str(c["description"]).startswith(plan["assumptionDescriptionPrefix"]) for c in assumptions)
|
||
assert {c["orderNo"] for c in baseline_conflicts if c["severity"] == "CRITICAL"} == set(plan["blockedOrderNos"])
|
||
assumed_paint = [row for row in store.data["flexWorkOrders"]
|
||
if row["versionId"] == baseline["versionId"] and row["operationCode"] == "PAINT"]
|
||
assert assumed_paint and all(not row.get("personCode") for row in assumed_paint), "试排假设不得虚构操作人员"
|
||
assert any(i["type"] == "NO_PERSONNEL_SKILL" for o in readiness["orders"] for i in o["issues"])
|
||
assert any(i["type"] == "WIP_RESOURCE_UNKNOWN" and o["orderNo"] == expected["unknownWip"]["orderNo"]
|
||
for o in readiness["orders"] for i in o["issues"])
|
||
after_import = copy.deepcopy(world)
|
||
reimport = apply_import_commit(world, store.next_id, preview["batches"])
|
||
assert reimport.get("unchanged") is True
|
||
assert world == after_import
|
||
|
||
# Positive cases select an order whose completed first operation is explicit.
|
||
# Inventory and a missing skill are deliberate test edits, not inferred facts.
|
||
target = next(o for o in world["flexOrders"] if o["orderNo"] == trial["orderNo"])
|
||
supplements = [{"kind": "test_personnel", **copy.deepcopy(trial["supplementPerson"]),
|
||
"reason": "仅在隔离验证中模拟用户补齐源文件缺失技能"}]
|
||
world["flexPersonnel"].append({**copy.deepcopy(trial["supplementPerson"]),
|
||
"sourceRef": {"kind": "isolated_test_supplement"}})
|
||
for material in world["materials"]:
|
||
if material["code"] not in {b["materialCode"] for b in world["flexBom"] if b["productCode"] == target["productCode"]}:
|
||
continue
|
||
supplements.append({"kind": "test_inventory", "code": material["code"],
|
||
"before": material.get("stock"), "after": trial["supplementStock"]})
|
||
apply_master_action(world, store.next_id, "master.material.upsert", {"id": material["id"], "stock": trial["supplementStock"]})
|
||
|
||
def schedule(start_date=None):
|
||
result = run_flex_schedule(store, order_ids=[target["id"]], start_date=start_date,
|
||
actor="round87-isolated-verification")
|
||
rows = [r for r in world["flexWorkOrders"] if r["versionId"] == result["versionId"]]
|
||
assert all(r["flexOrderNo"] not in expected["sandboxOrderNos"] for r in rows)
|
||
return result, rows
|
||
|
||
first, rows = schedule()
|
||
assert first["woCount"] == trial["workOrderCount"], (first, world["flexConflicts"][-10:])
|
||
assert all(row["operationCode"] != trial["completedOperation"] for row in rows), "已完成下料不能再次排产"
|
||
assert min(row["plannedStartTime"] for row in rows) >= trial["completedAt"]
|
||
assert next(row["plannedStartTime"] for row in rows if row["operationCode"] == trial["editOperation"]) == trial["firstEditedOperationStart"]
|
||
old_rows = copy.deepcopy(rows)
|
||
old_version = copy.deepcopy(world["flexScheduleVersions"][-1])
|
||
|
||
bend = next(r for r in world["flexRoutings"] if r["productCode"] == target["productCode"] and r["operationCode"] == trial["editOperation"])
|
||
route = next(r for r in world["routings"] if r["productId"] == next(m["id"] for m in world["materials"] if m["code"] == target["productCode"]))
|
||
classic_step = next(s for s in world["routingSteps"] if s["routingId"] == route["id"] and s["sequenceNo"] == bend["seq"])
|
||
before_minutes = next(r["runMin"] for r in rows if r["operationCode"] == trial["editOperation"])
|
||
apply_master_action(world, store.next_id, "master.routing.upsert",
|
||
{"stepId": classic_step["id"], "runTimePerUnit": bend["stdTimePerUnit"] * 2})
|
||
_changed, changed_rows = schedule()
|
||
after_minutes = next(r["runMin"] for r in changed_rows if r["operationCode"] == trial["editOperation"])
|
||
assert after_minutes > before_minutes * 1.9
|
||
assert next(v for v in world["flexScheduleVersions"] if v["id"] == old_version["id"]) == old_version
|
||
assert [r for r in world["flexWorkOrders"] if r["versionId"] == first["versionId"]] == old_rows
|
||
|
||
material_code = next(b["materialCode"] for b in world["flexBom"] if b["productCode"] == target["productCode"])
|
||
material_id = next(m["id"] for m in world["materials"] if m["code"] == material_code)
|
||
apply_master_action(world, store.next_id, "master.material.upsert", {"id": material_id, "stock": 0, "inTransit": 0})
|
||
shortage_readiness = check_readiness(world, [target["id"]])
|
||
assert shortage_readiness["summary"]["ready"] == 0 and shortage_readiness["summary"]["blocked"] == 1
|
||
shortage, shortage_rows = schedule()
|
||
assert shortage["woCount"] == 0 and not shortage_rows
|
||
assert any(c["versionId"] == shortage["versionId"] and c["conflictType"] == "MATERIAL_SHORTAGE" for c in world["flexConflicts"])
|
||
apply_master_action(world, store.next_id, "master.material.upsert", {"id": material_id, "stock": trial["supplementStock"]})
|
||
restored, restored_rows = schedule()
|
||
assert restored["woCount"] == trial["workOrderCount"]
|
||
|
||
chosen_code = restored_rows[0]["equipmentCode"]
|
||
classic_equipment = next(e for e in world["equipment"] if e["code"] == chosen_code)
|
||
apply_master_action(world, store.next_id, "master.equipment.upsert", {"id": classic_equipment["id"], "status": "INACTIVE"})
|
||
disabled, disabled_rows = schedule()
|
||
assert all(r["equipmentCode"] != chosen_code for r in disabled_rows)
|
||
apply_master_action(world, store.next_id, "master.equipment.upsert", {"id": classic_equipment["id"], "status": "RUNNING"})
|
||
|
||
# Verify imported maintenance and running WIP against every actual work segment.
|
||
for row in restored_rows:
|
||
for event in world["flexMaintenance"]:
|
||
if row["equipmentCode"] == event["equipmentCode"]:
|
||
for segment in row.get("workSegments") or []:
|
||
assert not (as_datetime(segment["start"]) < as_datetime(event["end"])
|
||
and as_datetime(segment["end"]) > as_datetime(event["start"]))
|
||
for wip in world["flexWip"]:
|
||
if wip["status"] == "RUNNING" and row["equipmentCode"] == wip["equipmentCode"]:
|
||
assert as_datetime(row["plannedStartTime"]) >= as_datetime(wip.get("expectedEnd") or wip["completionTime"])
|
||
|
||
_, before_weekend_rows = schedule(weekend["startDate"])
|
||
before_weekend_start = min(r["plannedStartTime"] for r in before_weekend_rows)
|
||
day_shift = next(s for s in world["shifts"] if s["code"] == expected["shifts"]["dayCode"])
|
||
line_ids = [line["id"] for line in world["lines"]]
|
||
template = apply_master_action(world, store.next_id, "master.calendar.template.create", {
|
||
"name": "仅验收：周末白班", "lineId": line_ids[0],
|
||
"shifts": [{"shiftId": day_shift["id"], "workdays": weekend["workdays"]}]})
|
||
copied = apply_master_action(world, store.next_id, "master.calendar.week.copy", {
|
||
"templateId": template["id"], "startDate": weekend["startDate"], "endDate": weekend["endDate"], "lineIds": line_ids})
|
||
assert copied["generatedCount"] > 0 and copied["skippedWeekendCount"] == 0
|
||
_, weekend_rows = schedule(weekend["startDate"])
|
||
weekend_start = min(r["plannedStartTime"] for r in weekend_rows)
|
||
assert weekend_start == weekend["expectedStart"] and before_weekend_start > weekend_start
|
||
supplements.append({"kind": "test_calendar", "dates": [weekend["startDate"], weekend["endDate"]],
|
||
"shiftCode": expected["shifts"]["dayCode"], "reason": "模拟用户明确启用周末白班，只用于隔离场景"})
|
||
# 导入的维护记录落在周末白班上，只有周末班启用后才真正占槽：移除后选机必须改变。
|
||
with_maintenance_bend = next(r for r in weekend_rows if r["operationCode"] == trial["editOperation"])
|
||
maintenance_rows = world["flexMaintenance"]
|
||
world["flexMaintenance"] = []
|
||
_, without_maintenance_rows = schedule(weekend["startDate"])
|
||
world["flexMaintenance"] = maintenance_rows
|
||
without_maintenance_bend = next(r for r in without_maintenance_rows if r["operationCode"] == trial["editOperation"])
|
||
assert with_maintenance_bend["equipmentCode"] != without_maintenance_bend["equipmentCode"]
|
||
supplements.append({"kind": "counterfactual_maintenance", "reason": "隔离副本暂时移除维护记录后对比选机，再恢复原记录"})
|
||
assert hashlib.sha256(workbook.read_bytes()).hexdigest() == expected["sourceSha256"]
|
||
return {"status": "PASS", "workbook": str(workbook), "sha256": expected["sourceSha256"],
|
||
"sourceUnchanged": True, "counts": counts, "sourceBaseline": {"readiness": readiness, "schedule": baseline},
|
||
"supplements": supplements,
|
||
"checks": {"importCounts": True, "sandboxExcluded": True,
|
||
"sourceBaselineTrialDraft": {"scheduled": plan["scheduledOrderNos"],
|
||
"blocked": plan["blockedOrderNos"],
|
||
"assumptions": sorted({c["conflictType"] for c in assumptions}),
|
||
"solveStatus": baseline["solveStatus"]},
|
||
"duplicateImportUnchanged": True, "completedOperationNotRepeated": True,
|
||
"runningCapacityReserved": True,
|
||
"maintenanceRespected": {"withMaintenance": with_maintenance_bend["equipmentCode"],
|
||
"withoutMaintenance": without_maintenance_bend["equipmentCode"]},
|
||
"editedDurationConsumed": {"beforeRunMinutes": before_minutes, "afterRunMinutes": after_minutes},
|
||
"editedStockConsumed": {"shortageWorkOrders": shortage["woCount"], "restoredWorkOrders": restored["woCount"],
|
||
"shortageReadinessBlocked": shortage_readiness["summary"]["blocked"]},
|
||
"disabledEquipmentNotAllocated": {"code": chosen_code, "newWorkOrders": disabled["woCount"]},
|
||
"weekendMaintenanceConsumed": {"beforeStart": before_weekend_start, "afterStart": weekend_start,
|
||
"generatedShifts": copied["generatedCount"]},
|
||
"oldVersionUnchanged": True},
|
||
"sampleSchedule": restored, "sampleWorkOrders": restored_rows,
|
||
"trialOnly": True, "productionReady": False,
|
||
"boundary": "原始输入存在人员与在制资料缺项；正向验证仅使用明确列出的隔离测试补充，不代表现场确认或正式下发"}
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--source", "--workbook", dest="source", type=Path,
|
||
default=os.environ.get("ROUND87_SOURCE") or None,
|
||
help="User-designated source workbook; alternatively set ROUND87_SOURCE")
|
||
parser.add_argument("--expectations", type=Path, help="Independent acceptance manifest (or ROUND87_EXPECTATIONS)")
|
||
parser.add_argument("--output", type=Path, required=True)
|
||
args = parser.parse_args()
|
||
if args.source is None:
|
||
parser.error("--source or ROUND87_SOURCE must identify the user-designated workbook")
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
with tempfile.TemporaryDirectory(prefix="aps-round87-e2e-") as temporary:
|
||
root = Path(temporary)
|
||
os.environ.pop("APS_DATABASE_URL", None)
|
||
for name in list(os.environ):
|
||
if name.startswith("APS_") and name.endswith(("_PATH", "_DIR")):
|
||
os.environ.pop(name, None)
|
||
os.environ.update(APS_HOME=str(root), APS_DATA_DIR=str(root / "data"),
|
||
APS_DB_PATH=str(root / "data" / "master.db"), APS_SEED_DEMO="0",
|
||
APS_APPROVAL_BACKEND="file", APS_AUTH_ENABLED="1")
|
||
from tests.workbook_acceptance import load_expectations
|
||
result = verify(args.source, load_expectations(args.expectations))
|
||
from server.db.database import reset_engine
|
||
reset_engine()
|
||
result["isolation"] = {"temporaryRoot": temporary, "userServiceTouched": False}
|
||
result["isolation"]["temporaryDirectoryRemoved"] = not root.exists()
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(json.dumps({"status": result["status"], "output": str(args.output), "checks": result["checks"]}, ensure_ascii=False))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|