144 lines
7.7 KiB
Python
144 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
import pytest
|
|
|
|
from server.aps_domain.importers import apply_import_commit, preview_file
|
|
from server.aps_domain.intake_recovery import (
|
|
apply_reconciled_intake,
|
|
review_intake_recovery,
|
|
)
|
|
from server.aps_domain.intake_views import planning_data_views
|
|
from server.state.seed import empty_world
|
|
from tests.intake_recovery_fixture import seed_legacy_intake
|
|
from tests.workbook_acceptance import load_expectations, source_for_test
|
|
|
|
|
|
@pytest.fixture
|
|
def legacy():
|
|
source = source_for_test()
|
|
world = empty_world()
|
|
preview = preview_file(source.name, source.read_bytes(), world)
|
|
counters = {}
|
|
|
|
def next_id(key):
|
|
counters[key] = counters.get(key, 1000) + 1
|
|
return counters[key]
|
|
|
|
seed_legacy_intake(world, next_id, preview)
|
|
return world, preview, next_id, load_expectations()
|
|
|
|
|
|
def test_real_legacy_collision_requires_explicit_recovery_and_preserves_history(legacy):
|
|
world, preview, next_id, expected = legacy
|
|
assert len(world["flexOrders"]) == expected["entityCounts"]["orders"] + expected["entityCounts"]["sandboxOrders"]
|
|
before = copy.deepcopy(world)
|
|
with pytest.raises(ValueError, match="正式订单"):
|
|
apply_import_commit(world, next_id, preview["batches"])
|
|
assert world == before
|
|
world["flexScheduleVersions"] = [{"id": 7, "versionNo": "prior-draft", "status": "DRAFT"}]
|
|
history = copy.deepcopy(world["flexScheduleVersions"])
|
|
recovery = review_intake_recovery(world, preview["batches"])
|
|
assert [item["orderNo"] for item in recovery["conflicts"]] == expected["sandboxOrderNos"]
|
|
assert all(item["canRecover"] for item in recovery["conflicts"])
|
|
result = apply_reconciled_intake(world, next_id, preview["batches"], recovery)
|
|
assert result["recoveredOrderNos"] == expected["sandboxOrderNos"]
|
|
assert {row["orderNo"] for row in world["flexOrders"]} == set(expected["formalOrderNos"])
|
|
assert {row["orderNo"] for row in world["salesOrders"] if row["status"] != "CANCELLED"} == set(expected["formalOrderNos"])
|
|
archived_sale = next(row for row in world["salesOrders"] if row["orderNo"] == expected["sandboxOrderNos"][0])
|
|
original_sale = next(row for row in before["salesOrders"] if row["orderNo"] == archived_sale["orderNo"])
|
|
assert archived_sale["id"] == original_sale["id"] and archived_sale["reclassifiedAs"] == "sandbox"
|
|
assert world["flexScheduleVersions"] == history
|
|
assert world["intakeSources"][0]["legacyReconciliation"]["archivedRecords"]["flexOrders"][0]["orderNo"] == expected["sandboxOrderNos"][0]
|
|
|
|
|
|
@pytest.mark.parametrize("protection", ["published", "progress", "mes", "manual", "changed-order", "purchase", "outsource", "reviewed", "approved-version", "released-version", "derived", "frozen-work", "rush", "priority"])
|
|
def test_recovery_does_not_reclassify_committed_or_different_orders(legacy, protection):
|
|
world, preview, next_id, expected = legacy
|
|
number = expected["sandboxOrderNos"][0]
|
|
order = next(row for row in world["flexOrders"] if row["orderNo"] == number)
|
|
if protection in {"published", "progress", "mes", "approved-version", "released-version", "frozen-work"}:
|
|
world["flexWorkOrders"] = [{"id": 777, "flexOrderNo": number, "versionId": 7, "status": "PLANNED"}]
|
|
version_status = {"published": "PUBLISHED", "approved-version": "APPROVED", "released-version": "RELEASED"}.get(protection, "DRAFT")
|
|
world["flexScheduleVersions"] = [{"id": 7, "status": version_status}]
|
|
if protection == "progress":
|
|
world["flexWorkOrders"][0]["progressPct"] = 10
|
|
if protection == "mes":
|
|
world["mesLinks"] = [{"track": "flex", "woId": 777, "externalWoId": "external"}]
|
|
if protection == "frozen-work":
|
|
world["flexWorkOrders"][0]["frozen"] = True
|
|
elif protection == "manual":
|
|
next(row for row in world["salesOrders"] if row["orderNo"] == number)["source"] = "MANUAL"
|
|
elif protection in {"purchase", "outsource"}:
|
|
sales = next(row for row in world["salesOrders"] if row["orderNo"] == number)
|
|
world["purchaseOrders" if protection == "purchase" else "outsourceOrders"] = [{"id": 900, "salesOrderId": sales["id"], "status": "RELEASED"}]
|
|
elif protection == "reviewed":
|
|
next(row for row in world["salesOrders"] if row["orderNo"] == number)["changes"] = [{"reason": "manually revised"}]
|
|
elif protection == "derived":
|
|
world["flexOrders"].append({"id": 901, "orderNo": "derived-order", "source": "CLOSED_LOOP", "salesOrderNo": number})
|
|
elif protection == "rush":
|
|
next(row for row in world["salesOrders"] if row["orderNo"] == number).update(isRush=True, rushStrategy="STRATEGY_SHIFT")
|
|
elif protection == "priority":
|
|
order["priority"] = 999
|
|
else:
|
|
order["quantity"] += 1
|
|
before = copy.deepcopy(world)
|
|
recovery = review_intake_recovery(world, preview["batches"])
|
|
assert not recovery["conflicts"][0]["canRecover"]
|
|
with pytest.raises(ValueError, match="状态或整理范围"):
|
|
apply_reconciled_intake(world, next_id, preview["batches"], recovery)
|
|
assert world == before
|
|
|
|
|
|
def test_changed_legacy_row_after_review_is_not_overwritten(legacy):
|
|
world, preview, next_id, expected = legacy
|
|
recovery = review_intake_recovery(world, preview["batches"])
|
|
next(row for row in world["flexOrders"] if row["orderNo"] == expected["sandboxOrderNos"][0])["customerName"] = "updated by planner"
|
|
before = copy.deepcopy(world)
|
|
with pytest.raises(ValueError):
|
|
apply_reconciled_intake(world, next_id, preview["batches"], recovery)
|
|
assert world == before
|
|
|
|
|
|
def test_four_detail_snapshots_contain_complete_inspectable_data(legacy):
|
|
world, preview, next_id, expected = legacy
|
|
views = planning_data_views(world, preview["batches"], adopted=False)
|
|
for key in ("orders", "materials", "routing", "equipment"):
|
|
assert len(views[key]["rows"]) == expected["entityCounts"][key]
|
|
assert views[key]["columns"] and all("sourceLocation" in row for row in views[key]["rows"])
|
|
assert sum(row["inTransit"] or 0 for row in views["materials"]["rows"]) == expected["inventory"]["inTransit"]
|
|
assert expected["sandboxOrderNos"][0] not in {row["orderNo"] for row in views["orders"]["rows"]}
|
|
source_view = copy.deepcopy(views)
|
|
apply_reconciled_intake(world, next_id, preview["batches"], review_intake_recovery(world, preview["batches"]))
|
|
world["flexMaterials"][0]["stock"] = 987
|
|
current = planning_data_views(world, preview["batches"], adopted=True)
|
|
assert current["materials"]["rows"][0]["stock"] == 987
|
|
assert views == source_view
|
|
|
|
|
|
def test_recovery_binds_all_data_that_the_complete_workbook_will_update(legacy):
|
|
world, preview, next_id, expected = legacy
|
|
recovery = review_intake_recovery(world, preview["batches"])
|
|
order = next(row for row in world["flexOrders"] if row["orderNo"] == expected["formalOrderNos"][0])
|
|
order.update(quantity=999, source="SAP")
|
|
before = copy.deepcopy(world)
|
|
with pytest.raises(ValueError):
|
|
apply_reconciled_intake(world, next_id, preview["batches"], recovery)
|
|
assert world == before
|
|
assert any(not issue["canRecover"] for issue in review_intake_recovery(world, preview["batches"])["conflicts"])
|
|
|
|
|
|
def test_adoption_without_classification_conflicts_still_binds_current_data(legacy):
|
|
world, preview, next_id, expected = legacy
|
|
numbers = set(expected["sandboxOrderNos"])
|
|
for table in ("flexOrders", "salesOrders"):
|
|
world[table] = [row for row in world[table] if row["orderNo"] not in numbers]
|
|
review = review_intake_recovery(world, preview["batches"])
|
|
assert review["conflicts"] == []
|
|
world["flexOrders"][0]["quantity"] += 1
|
|
before = copy.deepcopy(world)
|
|
with pytest.raises(ValueError):
|
|
apply_reconciled_intake(world, next_id, preview["batches"], review)
|
|
assert world == before
|