286 lines
15 KiB
Python
286 lines
15 KiB
Python
"""Exact customer workbook acceptance. Missing the specified input is a failure."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
|
|
import openpyxl
|
|
import pytest
|
|
|
|
from server.aps_domain.folder_pack import analyze_work_dir
|
|
from server.aps_domain.importers import apply_import_commit, preview_file
|
|
from server.state.seed import empty_world
|
|
from tests.workbook_acceptance import load_expectations, source_for_test
|
|
|
|
EXPECTED = load_expectations()
|
|
DIGEST = EXPECTED["sourceSha256"]
|
|
PROFILE = EXPECTED["profile"]
|
|
SHEETS = EXPECTED["sheets"]
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def source_bytes():
|
|
source = source_for_test()
|
|
raw = source.read_bytes()
|
|
assert hashlib.sha256(raw).hexdigest() == DIGEST, "验收输入已变更，需重新核对基准"
|
|
yield raw
|
|
assert hashlib.sha256(source.read_bytes()).hexdigest() == DIGEST
|
|
|
|
|
|
def _preview(raw):
|
|
return preview_file("uploaded-planning-data.xlsx", raw, empty_world(), soft=True)
|
|
|
|
|
|
def _apply(raw):
|
|
world = empty_world()
|
|
result = _preview(raw)
|
|
apply_import_commit(world, lambda kind: 1, result["batches"])
|
|
return world, result
|
|
|
|
|
|
def _mutate(raw, operation):
|
|
workbook = openpyxl.load_workbook(io.BytesIO(raw))
|
|
operation(workbook)
|
|
stream = io.BytesIO()
|
|
workbook.save(stream)
|
|
workbook.close()
|
|
return stream.getvalue()
|
|
|
|
|
|
def test_exact_workbook_all_sheets_and_business_counts(source_bytes):
|
|
result = _preview(source_bytes)
|
|
assert result["profile"] == PROFILE
|
|
assert result["canCommit"] is True
|
|
assert result["canSchedule"] is False # Missing required skills and unresolved WIP.
|
|
assert result["totalErrors"] == 0
|
|
assert {b["physicalSheet"] for b in result["batches"]} == set(SHEETS.values())
|
|
assert len(result["sheetSummary"]) == len(SHEETS)
|
|
assert result["entityCounts"] == EXPECTED["entityCounts"]
|
|
assert not any(d["severity"] == "ignored" for d in result["diagnostics"])
|
|
|
|
|
|
def test_stock_bom_time_and_source_semantics_survive_import(source_bytes):
|
|
world, _ = _apply(source_bytes)
|
|
materials = world["flexMaterials"]
|
|
assert len(materials) == EXPECTED["entityCounts"]["materials"]
|
|
assert sum(m.get("inTransit", 0) for m in materials) == EXPECTED["inventory"]["inTransit"]
|
|
arrivals = [m for m in materials if m.get("inTransit", 0) > 0]
|
|
assert len(arrivals) == EXPECTED["inventory"]["arrivalRows"]
|
|
assert {m["expectedArrivalDate"] for m in arrivals} == {EXPECTED["inventory"]["expectedArrivalDate"]}
|
|
raw_materials = [m for m in materials if m["type"] == "RAW_MATERIAL"]
|
|
assert len(raw_materials) == EXPECTED["inventory"]["rawMaterials"]
|
|
assert {m["unit"] for m in raw_materials} == {EXPECTED["inventory"]["unit"]}
|
|
assert all(m["spec"] for m in raw_materials)
|
|
assert raw_materials[0]["sourceRef"]["sheet"] == SHEETS["materials"]
|
|
assert raw_materials[0]["inventorySourceRef"]["sheet"] == SHEETS["inventory"]
|
|
assert len(world["flexBom"]) == EXPECTED["entityCounts"]["bom"]
|
|
assert sum(r["isKey"] for r in world["flexBom"]) == EXPECTED["inventory"]["keyBomRows"]
|
|
assert sum(r["lossRate"] > 0 for r in world["flexBom"]) == EXPECTED["inventory"]["lossBomRows"]
|
|
assert {r["stdTimeSource"] for r in world["flexRoutings"]} == {EXPECTED["timeSource"]}
|
|
assert all(not r["timeConfirmed"] for r in world["flexRoutings"])
|
|
assert all(r["sourceRef"]["sha256"] == DIGEST for r in world["flexRoutings"])
|
|
|
|
|
|
def test_sandbox_and_order_dates_preserved(source_bytes):
|
|
world, _ = _apply(source_bytes)
|
|
assert {o["orderNo"] for o in world["flexOrders"]} == set(EXPECTED["formalOrderNos"])
|
|
assert {o["orderNo"] for o in world["salesOrders"]} == set(EXPECTED["formalOrderNos"])
|
|
assert [o["orderNo"] for o in world["flexSandboxOrders"]] == EXPECTED["sandboxOrderNos"]
|
|
assert world["flexSandboxOrders"][0]["status"] == "PENDING_EVALUATION"
|
|
assert world["flexSandboxOrders"][0]["dueDate"] == EXPECTED["sandboxDueDate"]
|
|
assert world["flexOrders"][0]["dueDate"] == EXPECTED["firstFormalDueDate"]
|
|
assert len(world["flexScenarios"]) == EXPECTED["sourceCounts"]["scenarios"]
|
|
|
|
|
|
def test_people_calendar_wip_parameters_are_structured_without_guesses(source_bytes):
|
|
world, preview = _apply(source_bytes)
|
|
assert len(world["flexFactoryResources"]) == EXPECTED["sourceCounts"]["factoryResources"]
|
|
assert len(world["flexPartners"]) == EXPECTED["sourceCounts"]["partners"]
|
|
assert len(world["flexPersonnel"]) == EXPECTED["entityCounts"]["personnel"]
|
|
assert sum(t["memberCount"] for t in world["flexTeams"]) == EXPECTED["entityCounts"]["personnel"]
|
|
assert len({c for t in world["flexTeams"] for c in t["personCodes"]}) == EXPECTED["entityCounts"]["personnel"]
|
|
assert len(world["flexWip"]) == EXPECTED["entityCounts"]["wip"]
|
|
running = next(w for w in world["flexWip"] if w["taskNo"] == EXPECTED["runningWip"]["taskNo"])
|
|
assert running["status"] == "RUNNING" and running["completedQuantity"] == EXPECTED["runningWip"]["completedQuantity"]
|
|
assert running["operationCode"] == EXPECTED["runningWip"]["operationCode"] and running["seq"] == EXPECTED["runningWip"]["seq"]
|
|
assert running["expectedEnd"] == EXPECTED["runningWip"]["expectedEnd"]
|
|
assert "actualStart" not in running
|
|
assert running["sourceValues"][EXPECTED["runningWip"]["sourceColumn"]] == EXPECTED["runningWip"]["sourceValue"]
|
|
unresolved = next(w for w in world["flexWip"] if w["taskNo"] == EXPECTED["unknownWip"]["taskNo"])
|
|
assert unresolved["equipmentCode"] == EXPECTED["unknownWip"]["equipmentCode"]
|
|
assert not any(e["code"] == EXPECTED["unknownWip"]["equipmentCode"] for e in world["flexEquipment"])
|
|
shifts = {s["shiftCode"]: s for s in world["flexCalendar"]}
|
|
assert len(shifts) == EXPECTED["entityCounts"]["calendar"]
|
|
assert shifts[EXPECTED["shifts"]["dayCode"]]["enabled"] is True and shifts[EXPECTED["shifts"]["nightCode"]]["enabled"] is False
|
|
assert shifts[EXPECTED["shifts"]["dayCode"]]["breaks"] == EXPECTED["shifts"]["dayBreaks"]
|
|
assert shifts[EXPECTED["shifts"]["dayCode"]]["workdays"] == EXPECTED["shifts"]["workdays"]
|
|
assert world["flexMaintenance"][0]["equipmentCode"] == EXPECTED["maintenance"]["equipmentCode"]
|
|
assert world["flexMaintenance"][0]["start"] == EXPECTED["maintenance"]["start"]
|
|
assert world["flexMaintenance"][0]["end"] == EXPECTED["maintenance"]["end"]
|
|
context = world["planningContext"]
|
|
assert context["planStart"] == EXPECTED["planningContext"]["planStart"]
|
|
assert context["horizonDays"] == EXPECTED["planningContext"]["horizonDays"] and context["freezeHours"] == EXPECTED["planningContext"]["freezeHours"]
|
|
assert context["nightShiftEnabled"] is False and context["trialOnly"] is True
|
|
assert context["sourceSha256"] == DIGEST
|
|
codes = {d["code"] for d in preview["diagnostics"]}
|
|
assert {"NO_PERSONNEL_SKILL", "UNKNOWN_WIP_EQUIPMENT", "UNCONFIRMED_DEMO_VALUES"} <= codes
|
|
evidence = world["intakeSources"][0]
|
|
assert len(evidence["sourceValidation"]) == EXPECTED["sourceCounts"]["validation"] and len(evidence["parameters"]) == EXPECTED["sourceCounts"]["parameters"]
|
|
assert all(r["verification"] == "source_reported_only" for r in evidence["sourceValidation"])
|
|
|
|
|
|
def test_identical_reimport_is_idempotent_and_preserves_planner_edits(source_bytes):
|
|
world, preview = _apply(source_bytes)
|
|
world["flexMaterials"][0]["stock"] = 12345
|
|
world["flexRoutings"][0]["stdTimePerUnit"] = 17.5
|
|
world["flexEquipment"][0]["status"] = "DOWN"
|
|
before = copy.deepcopy(world)
|
|
result = apply_import_commit(world, lambda kind: 999, copy.deepcopy(preview["batches"]))
|
|
assert result["unchanged"] is True and result["total"] == 0
|
|
assert world == before
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", [
|
|
lambda wb: wb.remove(wb[SHEETS["personnel"]]),
|
|
lambda wb: setattr(wb[SHEETS["bom"]]["G1"], "value", "未知字段"),
|
|
lambda wb: setattr(wb[SHEETS["bom"]]["E2"], "value", -1),
|
|
lambda wb: setattr(wb[SHEETS["scenario"]]["B5"], "value", "999"),
|
|
lambda wb: setattr(wb[SHEETS["orders"]]["A7"], "value", "正式订单"),
|
|
lambda wb: setattr(wb[SHEETS["routing"]]["A2"], "value", "UNKNOWN-PRODUCT"),
|
|
lambda wb: setattr(wb[SHEETS["equipment"]]["A3"], "value", EXPECTED["duplicateEquipmentCode"]),
|
|
])
|
|
def test_invalid_profile_fails_closed_and_cannot_partially_write(source_bytes, mutation):
|
|
preview = _preview(_mutate(source_bytes, mutation))
|
|
assert preview["profile"] == PROFILE
|
|
assert preview["canCommit"] is False and preview["totalErrors"] > 0
|
|
world = empty_world()
|
|
before = copy.deepcopy(world)
|
|
with pytest.raises(ValueError, match="未通过"):
|
|
apply_import_commit(world, lambda kind: 1, preview["batches"])
|
|
assert world == before
|
|
|
|
|
|
def test_partial_profile_and_preexisting_formal_sandbox_are_not_silently_adopted(source_bytes):
|
|
preview = _preview(source_bytes)
|
|
world = empty_world()
|
|
with pytest.raises(ValueError, match="完整"):
|
|
apply_import_commit(world, lambda kind: 1, preview["batches"][:5])
|
|
world["flexOrders"] = [{"id": 7, "orderNo": EXPECTED["sandboxOrderNos"][0], "status": "RELEASED"}]
|
|
before = copy.deepcopy(world)
|
|
with pytest.raises(ValueError, match="已存在正式订单"):
|
|
apply_import_commit(world, lambda kind: 1, preview["batches"])
|
|
assert world == before
|
|
|
|
|
|
def test_folder_report_keeps_complete_frozen_contract_and_unique_counts(source_bytes, tmp_path, monkeypatch):
|
|
(tmp_path / "uploaded.xlsx").write_bytes(source_bytes)
|
|
monkeypatch.setattr("server.aps_domain.folder_pack._project_work_dir",
|
|
lambda sid: ({"id": "ruiyang", "name": "真实资料验收"}, str(tmp_path)))
|
|
world = empty_world()
|
|
before = copy.deepcopy(world)
|
|
report = analyze_work_dir(world, "test")
|
|
assert all(world[k] == value for k, value in before.items())
|
|
assert all(value == [] for k, value in world.items() if k not in before)
|
|
assert report["kindCounts"]["materials"] == EXPECTED["entityCounts"]["materials"]
|
|
assert report["kindCounts"]["orders"] == EXPECTED["entityCounts"]["orders"]
|
|
assert report["skippedRows"] == 0
|
|
assert report["totalErrors"] == 0
|
|
assert report["canSchedule"] is False
|
|
assert len(report["batches"]) == len(SHEETS)
|
|
assert report["files"][0]["profile"] == PROFILE
|
|
assert report["files"][0]["canCommit"] is True
|
|
applied = apply_import_commit(world, lambda kind: 1, report["batches"])
|
|
assert applied["summary"]["materials"] == EXPECTED["entityCounts"]["materials"]
|
|
assert world["planningContext"]["sourceSha256"] == DIGEST
|
|
|
|
|
|
@pytest.mark.parametrize("revision", ["removed-routing", "added-equipment"])
|
|
def test_revised_snapshot_requires_separate_project_without_mutating_adopted_world(source_bytes, revision):
|
|
def change(workbook):
|
|
if revision == "removed-routing":
|
|
workbook[SHEETS["routing"]].delete_rows(6)
|
|
else:
|
|
equipment = workbook[SHEETS["equipment"]]
|
|
row = [cell.value for cell in equipment[2]]
|
|
row[0] = "REVIEW-ADDED-EQUIPMENT"
|
|
equipment.append(row)
|
|
|
|
revised = _preview(_mutate(source_bytes, change))
|
|
assert revised["canCommit"] is True
|
|
assert revised["source"]["sha256"] != DIGEST
|
|
assert revised["entityCounts"]["routing"] == (EXPECTED["entityCounts"]["routing"] - 1 if revision == "removed-routing" else EXPECTED["entityCounts"]["routing"])
|
|
assert revised["entityCounts"]["equipment"] == (EXPECTED["entityCounts"]["equipment"] + 1 if revision == "added-equipment" else EXPECTED["entityCounts"]["equipment"])
|
|
|
|
world, _ = _apply(source_bytes)
|
|
world["flexMaterials"][0]["stock"] = 12345
|
|
world["flexRoutings"][0]["stdTimePerUnit"] = 17.5
|
|
world["flexEquipment"][0]["status"] = "DOWN"
|
|
before = copy.deepcopy(world)
|
|
with pytest.raises(ValueError, match="独立项目.*差异审核"):
|
|
apply_import_commit(world, lambda kind: 1, revised["batches"])
|
|
assert world == before
|
|
|
|
# The revised snapshot is valid in its own empty project, with no old
|
|
# source-deleted rows silently left behind and no edits from another world.
|
|
separate = empty_world()
|
|
apply_import_commit(separate, lambda kind: 1, revised["batches"])
|
|
assert len(separate["flexRoutings"]) == revised["entityCounts"]["routing"]
|
|
assert len(separate["routingSteps"]) == revised["entityCounts"]["routing"]
|
|
assert len(separate["flexEquipment"]) == revised["entityCounts"]["equipment"]
|
|
|
|
|
|
def test_planning_context_identity_also_protects_revision_and_identical_reimport(source_bytes):
|
|
world, original = _apply(source_bytes)
|
|
world["intakeSources"] = []
|
|
world["flexRoutings"][0]["stdTimePerUnit"] = 17.5
|
|
before = copy.deepcopy(world)
|
|
identical = apply_import_commit(world, lambda kind: 1, original["batches"])
|
|
assert identical["unchanged"] is True
|
|
assert world == before
|
|
revised = _preview(_mutate(source_bytes, lambda wb: wb[SHEETS["routing"]].delete_rows(6)))
|
|
with pytest.raises(ValueError, match="独立项目"):
|
|
apply_import_commit(world, lambda kind: 1, revised["batches"])
|
|
assert world == before
|
|
|
|
|
|
def test_other_import_formats_still_accept_explicit_updates_after_profile_adoption(source_bytes):
|
|
world, _ = _apply(source_bytes)
|
|
csv = "编码,名称,类型,单位,库存\nEXTERNAL-MATERIAL,外部新增材料,RAW_MATERIAL,KG,123\n".encode("utf-8-sig")
|
|
generic = preview_file("普通物料.csv", csv, world)
|
|
assert generic.get("profile") is None
|
|
result = apply_import_commit(world, lambda kind: 999, generic["batches"])
|
|
assert result["total"] > 0
|
|
material = next(row for row in world["flexMaterials"] if row["code"] == "EXTERNAL-MATERIAL")
|
|
assert material["stock"] == 123 and material["unit"] == "KG"
|
|
|
|
|
|
def test_acceptance_source_is_optional_only_when_not_explicitly_required(monkeypatch, tmp_path):
|
|
from tests.workbook_acceptance import resolve_source
|
|
|
|
monkeypatch.delenv("ROUND87_SOURCE", raising=False)
|
|
monkeypatch.delenv("ROUND87_REQUIRED", raising=False)
|
|
assert resolve_source() is None
|
|
monkeypatch.setenv("ROUND87_REQUIRED", "1")
|
|
with pytest.raises(ValueError, match="ROUND87_SOURCE"):
|
|
resolve_source()
|
|
monkeypatch.setenv("ROUND87_SOURCE", str(tmp_path / "missing.xlsx"))
|
|
with pytest.raises(FileNotFoundError):
|
|
resolve_source()
|
|
|
|
|
|
def test_acceptance_manifest_can_relocate_and_tampered_source_always_fails(monkeypatch, tmp_path):
|
|
import json
|
|
|
|
from tests.workbook_acceptance import load_expectations, resolve_source
|
|
|
|
manifest = tmp_path / "expectations.json"
|
|
manifest.write_text(json.dumps(EXPECTED), encoding="utf-8")
|
|
monkeypatch.setenv("ROUND87_EXPECTATIONS", str(manifest))
|
|
assert load_expectations() == EXPECTED
|
|
assert load_expectations(manifest) == EXPECTED
|
|
changed = tmp_path / "changed.xlsx"
|
|
changed.write_bytes(b"intentionally invalid workbook contents")
|
|
with pytest.raises(ValueError, match="SHA-256"):
|
|
resolve_source(changed, required=True)
|