294 lines
15 KiB
Python
294 lines
15 KiB
Python
|
"""Planner edits must reach the next solve while historical evidence stays immutable."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import copy
|
||
|
|
|
||
|
|
import openpyxl
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from server.aps_domain.masterdata import (
|
||
|
|
apply_master_action,
|
||
|
|
confirmation_for_master_action,
|
||
|
|
master_overview,
|
||
|
|
)
|
||
|
|
from server.aps_domain.masterdata_sync import reconcile_imported_masterdata
|
||
|
|
from server.state.seed import empty_world
|
||
|
|
from tests.workbook_acceptance import load_expectations, source_for_test
|
||
|
|
|
||
|
|
EXPECTED = load_expectations()
|
||
|
|
|
||
|
|
|
||
|
|
def _ids():
|
||
|
|
counters = {}
|
||
|
|
|
||
|
|
def next_id(kind):
|
||
|
|
counters[kind] = counters.get(kind, 0) + 1
|
||
|
|
return counters[kind]
|
||
|
|
return next_id
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def imported_world():
|
||
|
|
world = empty_world()
|
||
|
|
world["flexMaterials"] = [
|
||
|
|
{"code": "P1", "name": "成品", "type": "FINISHED_PRODUCT", "unit": "PCS", "stock": 5},
|
||
|
|
{"code": "M1", "name": "钢板", "type": "RAW_MATERIAL", "unit": "kg", "stock": 5,
|
||
|
|
"inTransit": 30, "expectedArrivalDate": "2026-08-07", "sourceRef": {"sheet": "库存与在途"}},
|
||
|
|
]
|
||
|
|
world["flexRoutings"] = [{"productCode": "P1", "operationCode": "CUT", "operationName": "切割",
|
||
|
|
"seq": 10, "setupTime": 3, "stdTimePerUnit": 5,
|
||
|
|
"stdTimeSource": "演示标准工时", "sourceRef": {"sheet": "工艺路线", "excelRow": 2}}]
|
||
|
|
world["flexBom"] = [{"productCode": "P1", "materialCode": "M1", "quantity": 2,
|
||
|
|
"isKey": True, "lossRate": 0.02}]
|
||
|
|
world["flexEquipment"] = [{"code": "EQ1", "name": "切割机", "status": "RUNNING",
|
||
|
|
"operations": ["CUT"], "zoneCode": "A", "efficiencyFactor": 0.9}]
|
||
|
|
world["shifts"] = [{"id": 1, "code": "D", "name": "白班", "startTime": "08:00", "endTime": "17:00"}]
|
||
|
|
return world
|
||
|
|
|
||
|
|
|
||
|
|
def test_material_edit_updates_consumed_stock_and_keeps_supply_metadata(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
material = next(m for m in w["materials"] if m["code"] == "M1")
|
||
|
|
apply_master_action(w, _ids(), "master.material.upsert", {"id": material["id"], "stock": 123})
|
||
|
|
flex = next(m for m in w["flexMaterials"] if m["code"] == "M1")
|
||
|
|
assert material["stock"] == flex["stock"] == 123
|
||
|
|
assert flex["inTransit"] == 30 and flex["expectedArrivalDate"] == "2026-08-07"
|
||
|
|
assert flex["unit"] == "kg" and flex["sourceRef"] == {"sheet": "库存与在途"}
|
||
|
|
|
||
|
|
|
||
|
|
def test_old_flex_only_view_has_stable_editable_ids_and_reads_do_not_write(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
original = copy.deepcopy(w)
|
||
|
|
view = master_overview(w)
|
||
|
|
step_id = view["routings"][0]["steps"][0]["id"]
|
||
|
|
confirmation_for_master_action(w, "master.routing.upsert", {"stepId": step_id, "runTimePerUnit": 17})
|
||
|
|
assert w == original
|
||
|
|
apply_master_action(w, _ids(), "master.routing.upsert", {"stepId": step_id, "runTimePerUnit": 17})
|
||
|
|
assert w["routingSteps"][0]["id"] == step_id
|
||
|
|
assert w["routingSteps"][0]["runTimePerUnit"] == w["flexRoutings"][0]["stdTimePerUnit"] == 17
|
||
|
|
assert w["flexRoutings"][0]["stdTimeSource"] == "演示标准工时"
|
||
|
|
|
||
|
|
|
||
|
|
def test_rejected_edit_does_not_materialize_or_partially_modify_source(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
source = copy.deepcopy(w)
|
||
|
|
step_id = master_overview(w)["routings"][0]["steps"][0]["id"]
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
apply_master_action(w, _ids(), "master.routing.upsert", {"stepId": step_id, "runTimePerUnit": -5})
|
||
|
|
assert w == source
|
||
|
|
|
||
|
|
|
||
|
|
def test_linked_material_and_equipment_codes_cannot_be_silently_renamed(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
before = copy.deepcopy(w)
|
||
|
|
for action, payload in [("master.material.upsert", {"id": w["materials"][0]["id"], "code": "NEW"}),
|
||
|
|
("master.equipment.upsert", {"id": w["equipment"][0]["id"], "code": "NEW"})]:
|
||
|
|
with pytest.raises(ValueError, match="编码不能直接修改"):
|
||
|
|
apply_master_action(w, _ids(), action, payload)
|
||
|
|
assert w == before
|
||
|
|
|
||
|
|
|
||
|
|
def test_repeated_import_keeps_ids_edits_and_unrelated_objects(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
source = copy.deepcopy(w)
|
||
|
|
w["materials"].append({"id": 200, "code": "OTHER", "name": "其他产品", "type": "FINISHED_PRODUCT", "unit": "个"})
|
||
|
|
w["routings"].append({"id": 301, "productId": 200, "version": "V9", "isDefault": True})
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
ids = {key: [r["id"] for r in w[key]] for key in ("materials", "routings", "routingSteps", "boms", "bomItems", "equipment")}
|
||
|
|
step = w["routingSteps"][0]
|
||
|
|
apply_master_action(w, _ids(), "master.routing.upsert", {"stepId": step["id"], "runTimePerUnit": 19})
|
||
|
|
for key in ("flexMaterials", "flexRoutings", "flexBom", "flexEquipment"):
|
||
|
|
w[key] = copy.deepcopy(source[key])
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
assert {key: [r["id"] for r in w[key]] for key in ids} == ids
|
||
|
|
assert w["flexRoutings"][0]["stdTimePerUnit"] == 19
|
||
|
|
assert next(r for r in w["routings"] if r["id"] == 301)["version"] == "V9"
|
||
|
|
|
||
|
|
|
||
|
|
def test_partial_stock_import_does_not_reset_master_fields(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
w["flexMaterials"] = [{"code": "M1", "stock": 44}]
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
material = next(m for m in w["materials"] if m["code"] == "M1")
|
||
|
|
assert material["stock"] == 44
|
||
|
|
assert material["name"] == "钢板" and material["unit"] == "kg" and material["inTransit"] == 30
|
||
|
|
|
||
|
|
|
||
|
|
def test_equipment_targeted_disable_changes_flex_and_preserves_capability(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
eq = w["equipment"][0]
|
||
|
|
apply_master_action(w, _ids(), "master.equipment.upsert", {"id": eq["id"], "status": "INACTIVE"})
|
||
|
|
assert eq["status"] == w["flexEquipment"][0]["status"] == "INACTIVE"
|
||
|
|
assert w["flexEquipment"][0]["operations"] == ["CUT"]
|
||
|
|
w["flexEquipment"][0]["status"] = "RUNNING"
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
assert w["flexEquipment"][0]["status"] == "INACTIVE"
|
||
|
|
|
||
|
|
|
||
|
|
def test_bom_edit_preserves_key_loss_and_routes_the_consumed_quantity(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
item = w["bomItems"][0]
|
||
|
|
assert item["isKeyMaterial"] and item["lossRate"] == 0.02
|
||
|
|
apply_master_action(w, _ids(), "master.bom.upsert", {"itemId": item["id"], "quantity": 7})
|
||
|
|
assert w["flexBom"][0]["quantity"] == 7
|
||
|
|
assert w["flexBom"][0]["isKey"] and w["flexBom"][0]["lossRate"] == 0.02
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("kind,head_key,detail_key,action,payload_field,edited_field,value", [
|
||
|
|
("routing", "routings", "routingSteps", "master.routing.upsert", "stepId", "runTimePerUnit", 19),
|
||
|
|
("bom", "boms", "bomItems", "master.bom.upsert", "itemId", "quantity", 7),
|
||
|
|
])
|
||
|
|
def test_release_edit_rollback_preserves_history_and_restores_engine_projection(
|
||
|
|
imported_world, kind, head_key, detail_key, action, payload_field, edited_field, value,
|
||
|
|
):
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
old_head = copy.deepcopy(w[head_key][0])
|
||
|
|
old_detail = copy.deepcopy(w[detail_key][0])
|
||
|
|
w["flexScheduleVersions"] = [{"id": 9, "inputs": copy.deepcopy(w["flexRoutings"])}]
|
||
|
|
history = copy.deepcopy(w["flexScheduleVersions"])
|
||
|
|
nid = _ids()
|
||
|
|
apply_master_action(w, nid, f"master.{kind}.release", {f"{kind}Id": old_head["id"]})
|
||
|
|
snapshot = copy.deepcopy(w["masterdataVersions"][0])
|
||
|
|
apply_master_action(w, nid, action, {payload_field: old_detail["id"], edited_field: value})
|
||
|
|
apply_master_action(w, nid, f"master.{kind}.rollback", {"productId": old_head["productId"], "version": old_head["version"]})
|
||
|
|
active = next(h for h in w[head_key] if h["isDefault"])
|
||
|
|
parent_key = "routingId" if kind == "routing" else "bomId"
|
||
|
|
detail = next(d for d in w[detail_key] if d[parent_key] == active["id"])
|
||
|
|
assert detail[edited_field] == old_detail[edited_field]
|
||
|
|
assert any(d["id"] == old_detail["id"] for d in w[detail_key])
|
||
|
|
assert w["masterdataVersions"][0] == snapshot
|
||
|
|
assert w["flexScheduleVersions"] == history
|
||
|
|
assert (w["flexRoutings"][0]["stdTimePerUnit"] if kind == "routing" else w["flexBom"][0]["quantity"]) == old_detail[edited_field]
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
assert len({r["id"] for r in w[detail_key]}) == len(w[detail_key])
|
||
|
|
|
||
|
|
|
||
|
|
def test_weekend_template_materializes_both_days_and_machine_overrides(imported_world):
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
nid = _ids()
|
||
|
|
template = apply_master_action(w, nid, "master.calendar.template.create", {
|
||
|
|
"name": "周末生产", "lineId": w["lines"][0]["id"], "shifts": [{"shiftId": 1, "workdays": [5, 6]}]})
|
||
|
|
payload = {"templateId": template["id"], "startDate": "2026-09-12", "endDate": "2026-09-14"}
|
||
|
|
result = apply_master_action(w, nid, "master.calendar.week.copy", payload)
|
||
|
|
assert result["generatedCount"] == 2 and result["skippedWeekendCount"] == 0
|
||
|
|
overrides = {r["date"]: r for r in w["flexCalendarOverrides"]}
|
||
|
|
assert overrides["2026-09-12"]["shifts"][0]["start"] == "08:00"
|
||
|
|
assert overrides["2026-09-12"]["shifts"][0]["shiftCode"] == "D"
|
||
|
|
assert overrides["2026-09-13"]["shifts"][0]["end"] == "17:00"
|
||
|
|
assert overrides["2026-09-14"]["shifts"] == []
|
||
|
|
repeated = apply_master_action(w, nid, "master.calendar.week.copy", payload)
|
||
|
|
assert repeated["generatedCount"] == 0
|
||
|
|
assert len(w["flexCalendarOverrides"]) == 3
|
||
|
|
|
||
|
|
|
||
|
|
def test_real_imported_calendar_keeps_day_night_identity_for_personnel():
|
||
|
|
workbook = source_for_test()
|
||
|
|
from server.aps_domain.importers import apply_import_commit, preview_file
|
||
|
|
w = empty_world()
|
||
|
|
preview = preview_file(workbook.name, workbook.read_bytes(), w, soft=True)
|
||
|
|
apply_import_commit(w, _ids(), preview["batches"])
|
||
|
|
shifts = {s["code"]: s for s in w["shifts"]}
|
||
|
|
assert shifts[EXPECTED["shifts"]["dayCode"]]["enabled"] is True and shifts[EXPECTED["shifts"]["nightCode"]]["enabled"] is False
|
||
|
|
assert shifts[EXPECTED["shifts"]["dayCode"]]["breakPeriods"] == EXPECTED["shifts"]["dayBreaks"]
|
||
|
|
assert shifts[EXPECTED["shifts"]["dayCode"]]["sourceRef"]["sheet"] == EXPECTED["sheets"]["calendar"]
|
||
|
|
nid = _ids()
|
||
|
|
template = apply_master_action(w, nid, "master.calendar.template.create", {
|
||
|
|
"name": "批准周末白班", "lineId": w["lines"][0]["id"],
|
||
|
|
"shifts": [{"shiftId": shifts[EXPECTED["shifts"]["dayCode"]]["id"], "workdays": EXPECTED["weekend"]["workdays"]}]})
|
||
|
|
apply_master_action(w, nid, "master.calendar.week.copy", {
|
||
|
|
"templateId": template["id"], "startDate": EXPECTED["masterWeekend"]["startDate"], "endDate": EXPECTED["masterWeekend"]["endDate"]})
|
||
|
|
assert w["flexCalendarOverrides"]
|
||
|
|
assert all(s["shiftCode"] == EXPECTED["shifts"]["dayCode"] for row in w["flexCalendarOverrides"] for s in row["shifts"])
|
||
|
|
|
||
|
|
|
||
|
|
def test_real_ruiyang_row_values_survive_projection_and_edits():
|
||
|
|
workbook = source_for_test()
|
||
|
|
assert workbook.is_file(), f"必须使用指定客户输入：{workbook}"
|
||
|
|
wb = openpyxl.load_workbook(workbook, read_only=True, data_only=True)
|
||
|
|
try:
|
||
|
|
p = list(wb[EXPECTED["sheets"]["products"]].values)[1]
|
||
|
|
r = list(wb[EXPECTED["sheets"]["routing"]].values)[1]
|
||
|
|
b = list(wb[EXPECTED["sheets"]["bom"]].values)[1]
|
||
|
|
e = list(wb[EXPECTED["sheets"]["equipment"]].values)[1]
|
||
|
|
finally:
|
||
|
|
wb.close()
|
||
|
|
w = empty_world()
|
||
|
|
w["flexMaterials"] = [{"code": p[0], "name": p[1], "unit": p[3], "type": p[4], "stock": p[6]}]
|
||
|
|
w["flexRoutings"] = [{"productCode": r[0], "seq": r[2], "operationCode": r[3],
|
||
|
|
"operationName": r[4], "stdTimePerUnit": r[5], "stdTimeSource": r[6]}]
|
||
|
|
w["flexBom"] = [{"productCode": b[0], "materialCode": b[2], "quantity": b[4], "isKey": b[5] == "是", "lossRate": b[6]}]
|
||
|
|
w["flexEquipment"] = [{"code": e[0], "name": e[1], "operations": e[5].split(","), "zoneCode": e[6], "status": e[7]}]
|
||
|
|
reconcile_imported_masterdata(w, source=EXPECTED["profile"])
|
||
|
|
assert w["routingSteps"][0]["runTimePerUnit"] == r[5]
|
||
|
|
assert w["bomItems"][0]["lossRate"] == b[6]
|
||
|
|
apply_master_action(w, _ids(), "master.routing.upsert", {"stepId": w["routingSteps"][0]["id"], "runTimePerUnit": 11})
|
||
|
|
assert w["flexRoutings"][0]["stdTimePerUnit"] == 11
|
||
|
|
assert w["flexRoutings"][0]["stdTimeSource"] == r[6]
|
||
|
|
|
||
|
|
|
||
|
|
def test_database_roundtrip_persists_mapping_calendar_and_source_without_project_leaks(tmp_path, monkeypatch, imported_world):
|
||
|
|
from server.db.database import reset_engine
|
||
|
|
from server.db.sync import db_to_world, set_active_project, world_to_db
|
||
|
|
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "masters.db"))
|
||
|
|
reset_engine()
|
||
|
|
try:
|
||
|
|
w = imported_world
|
||
|
|
reconcile_imported_masterdata(w)
|
||
|
|
w["flexPersonnel"] = [{"id": "P1", "skills": ["CUT"]}]
|
||
|
|
w["planningContext"] = {"dataDate": "2026-08-04", "trialOnly": True}
|
||
|
|
w["intakeSources"] = [{"sha256": "verified-source"}]
|
||
|
|
w["flexCalendarOverrides"] = [{"equipmentCode": "EQ1", "date": "2026-09-12", "shifts": []}]
|
||
|
|
set_active_project("w2-site-a")
|
||
|
|
world_to_db(w)
|
||
|
|
shadow = {}
|
||
|
|
db_to_world(shadow)
|
||
|
|
for key in ("materials", "routingSteps", "flexPersonnel", "planningContext", "intakeSources", "flexCalendarOverrides"):
|
||
|
|
assert shadow[key] == w[key]
|
||
|
|
set_active_project("w2-site-b")
|
||
|
|
world_to_db({"materials": [{"id": 1, "code": "OTHER"}]})
|
||
|
|
db_to_world(shadow)
|
||
|
|
assert shadow["materials"] == [{"id": 1, "code": "OTHER"}]
|
||
|
|
assert shadow["routingSteps"] == shadow["flexPersonnel"] == shadow["flexCalendarOverrides"] == []
|
||
|
|
assert shadow["planningContext"] == {}
|
||
|
|
set_active_project("w2-site-a")
|
||
|
|
db_to_world(shadow)
|
||
|
|
assert shadow["materials"] == w["materials"] and shadow["planningContext"] == w["planningContext"]
|
||
|
|
finally:
|
||
|
|
reset_engine()
|
||
|
|
|
||
|
|
|
||
|
|
def test_real_equipment_zone_scope_survives_import_and_weekend_calendar_edit():
|
||
|
|
workbook = source_for_test()
|
||
|
|
from collections import Counter
|
||
|
|
|
||
|
|
from server.aps_domain.importers import apply_import_commit, preview_file
|
||
|
|
w = empty_world()
|
||
|
|
preview = preview_file(workbook.name, workbook.read_bytes(), w, soft=True)
|
||
|
|
apply_import_commit(w, _ids(), preview["batches"])
|
||
|
|
lines = {row["id"]: row for row in w["lines"]}
|
||
|
|
stations = {row["id"]: row for row in w["workstations"]}
|
||
|
|
actual = Counter(lines[stations[eq["workstationId"]]["lineId"]]["code"] for eq in w["equipment"])
|
||
|
|
assert actual == EXPECTED["zones"]["equipmentCounts"]
|
||
|
|
scope = next(row for row in w["lines"] if row["code"] == EXPECTED["zones"]["targetLine"])
|
||
|
|
target_devices = {row["code"] for row in w["flexEquipment"] if row["zone"] == EXPECTED["zones"]["targetZone"]}
|
||
|
|
nid = _ids()
|
||
|
|
template = apply_master_action(w, nid, "master.calendar.template.create", {
|
||
|
|
"name": "组装区周末白班", "lineId": scope["id"],
|
||
|
|
"shifts": [{"shiftId": next(s["id"] for s in w["shifts"] if s["code"] == EXPECTED["shifts"]["dayCode"]), "workdays": EXPECTED["weekend"]["workdays"]}]})
|
||
|
|
apply_master_action(w, nid, "master.calendar.week.copy", {
|
||
|
|
"templateId": template["id"], "startDate": EXPECTED["masterWeekend"]["startDate"], "endDate": EXPECTED["masterWeekend"]["endDate"]})
|
||
|
|
assert {r["equipmentCode"] for r in w["flexCalendarOverrides"]} == target_devices
|
||
|
|
# Repair old generated UNASSIGNED ownership, leaving historical IDs stable.
|
||
|
|
old_station = stations[w["equipment"][0]["workstationId"]]
|
||
|
|
old_station["lineId"] = scope["id"]
|
||
|
|
reconcile_imported_masterdata(w, refresh=False)
|
||
|
|
assert lines[old_station["lineId"]]["code"] == EXPECTED["zones"]["firstLine"]
|