432 lines
20 KiB
Python
432 lines
20 KiB
Python
"""Versioned contracts for scheduling workbooks and the profile-driven intake template."""
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
from io import BytesIO
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from openpyxl import load_workbook
|
||
|
||
from server.aps_domain.report_contracts import (
|
||
PLAN_REPORT_CONTRACT,
|
||
REPORT_CONTRACTS,
|
||
SCHEDULE_BLOCKED_CONTRACT,
|
||
SCHEDULE_EQUIPMENT_CONTRACT,
|
||
SCHEDULE_ORDER_CONTRACT,
|
||
build_contract_template,
|
||
contract_digest,
|
||
contract_summary,
|
||
validate_workbook_contract,
|
||
)
|
||
from server.aps_domain.reports import (
|
||
build_plan_report,
|
||
build_schedule_equipment_export,
|
||
build_schedule_order_export,
|
||
)
|
||
from server.engines import PoolEngine
|
||
from server.importers.template_workbook import (
|
||
build_intake_template,
|
||
intake_definition_rows,
|
||
template_contract,
|
||
validate_intake_template,
|
||
)
|
||
from server.importers.workbook_profiles import builtin_compatibility_profile
|
||
from server.state.seed import empty_world, seed_world
|
||
|
||
|
||
class _MemStore:
|
||
def __init__(self, data: dict, path: str = "<mem>"):
|
||
self.data = data
|
||
self.path = path
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
key = f"_c_{kind}"
|
||
self.data[key] = self.data.get(key, 100) + 1
|
||
return self.data[key]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def _world_with_version() -> dict:
|
||
world = empty_world()
|
||
world["flexScheduleVersions"] = [{
|
||
"id": 5, "versionNo": "V-CONTRACT-001", "sortMode": "BOTTLENECK", "solveStatus": "SOLVED",
|
||
"demandCount": 2, "admittedDemandCount": 2, "unscheduledDemandCount": 0,
|
||
}]
|
||
world["flexOrders"] = [
|
||
{"orderNo": "SO-C1", "productCode": "P-C1", "quantity": 10, "dueDate": "2031-02-10"},
|
||
{"orderNo": "SO-C2", "productCode": "P-C2", "quantity": 5, "dueDate": "2031-02-12"},
|
||
]
|
||
world["flexVirtualLines"] = [
|
||
{"versionId": 5, "orderNo": "SO-C1", "productCode": "P-C1", "quantity": 10,
|
||
"plannedStart": "2031-02-06 08:00", "plannedEnd": "2031-02-07 12:00"},
|
||
{"versionId": 5, "orderNo": "SO-C2", "productCode": "P-C2", "quantity": 5,
|
||
"plannedStart": "2031-02-06 13:00", "plannedEnd": "2031-02-06 18:00"},
|
||
]
|
||
world["flexWorkOrders"] = [{
|
||
"versionId": 5, "flexOrderNo": "SO-C1", "orderNo": "WO-C1-1", "productCode": "P-C1", "seq": 1,
|
||
"operationCode": "CUT", "operationName": "下料", "equipmentCode": "EQ-C1",
|
||
"equipmentName": "切割机", "zone": "Z1", "moldCode": "M1", "changeoverMin": 5, "moveMin": 2,
|
||
"runMin": 30, "plannedStartTime": "2031-02-06 08:00", "plannedEndTime": "2031-02-06 08:37",
|
||
"isBottleneck": True, "status": "PLANNED", "quantity": 10,
|
||
}]
|
||
return world
|
||
|
||
|
||
def _scheduled_world() -> dict:
|
||
world = seed_world()
|
||
counters: dict[str, int] = {}
|
||
|
||
def next_id(kind: str) -> int:
|
||
counters[kind] = counters.get(kind, 0) + 1
|
||
return counters[kind]
|
||
|
||
PoolEngine().solve(world, next_id, sort_mode="BOTTLENECK")
|
||
return world
|
||
|
||
|
||
def test_report_contracts_declare_every_sheet_and_column():
|
||
plan = PLAN_REPORT_CONTRACT.sheets["plan"]
|
||
assert plan.name == "工作计划"
|
||
assert (plan.title_row, plan.header_row, plan.data_start_row) == (1, 4, 5)
|
||
assert plan.freeze_panes == "A5" and plan.auto_filter
|
||
assert plan.headers == (
|
||
"订单号", "产品编码", "数量", "交期", "序", "工序编码", "工序名称",
|
||
"工位/设备", "设备编码", "区域", "模具", "换型(分)", "移栽(分)", "加工(分)",
|
||
"计划开始", "计划结束", "瓶颈", "状态",
|
||
)
|
||
assert PLAN_REPORT_CONTRACT.sheets["overview"].header_row == 12
|
||
assert PLAN_REPORT_CONTRACT.sheets["conflicts"].header_row == 3
|
||
assert SCHEDULE_ORDER_CONTRACT.sheets["orders"].headers == (
|
||
"订单号", "工单号", "产品编码", "工序编码", "工序名称", "资源编码", "资源名称",
|
||
"计划开始", "计划结束", "状态", "预计数量", "实际数量",
|
||
)
|
||
assert SCHEDULE_EQUIPMENT_CONTRACT.sheets["equipment"].headers == (
|
||
"设备编码", "设备名称", "工序编码", "工序名称", "占用开始", "占用结束", "负荷分钟", "利用率%",
|
||
)
|
||
assert SCHEDULE_BLOCKED_CONTRACT.sheets["blocked"].headers == (
|
||
"阻断类型", "对象", "严重度", "说明", "修复建议",
|
||
)
|
||
assert {contract.id for contract in REPORT_CONTRACTS.values()} == {
|
||
"schedule-plan.v1", "schedule-order.v1", "schedule-equipment.v1", "schedule-blocked.v1",
|
||
}
|
||
summary = {row["id"]: row for row in contract_summary()}
|
||
assert summary["schedule-plan.v1"]["schemaVersion"] == 1
|
||
assert len(contract_digest()) == 64
|
||
|
||
|
||
def test_generated_plan_report_matches_contract_and_is_deterministic():
|
||
report = build_plan_report(_world_with_version())
|
||
assert report["templateId"] == "schedule-plan.v1"
|
||
assert report["templateSchemaVersion"] == 1
|
||
assert validate_workbook_contract(report["xlsxBytes"], PLAN_REPORT_CONTRACT) == []
|
||
assert build_plan_report(_world_with_version())["xlsxBytes"] == report["xlsxBytes"]
|
||
workbook = load_workbook(BytesIO(report["xlsxBytes"]))
|
||
assert workbook.sheetnames == ["工作计划", "方案概览", "冲突"]
|
||
assert workbook["工作计划"].freeze_panes == "A5"
|
||
assert workbook["工作计划"].auto_filter.ref == "A4:R5"
|
||
workbook.close()
|
||
|
||
|
||
def test_plan_report_without_work_orders_uses_the_blocked_contract():
|
||
world = _world_with_version()
|
||
world["flexWorkOrders"] = []
|
||
report = build_plan_report(world)
|
||
assert report["blocked"] is True
|
||
assert report["templateId"] == "schedule-blocked.v1"
|
||
assert validate_workbook_contract(report["xlsxBytes"], SCHEDULE_BLOCKED_CONTRACT) == []
|
||
assert build_plan_report(world)["xlsxBytes"] == report["xlsxBytes"]
|
||
|
||
|
||
def test_plan_report_without_any_version_keeps_the_declared_empty_state():
|
||
report = build_plan_report(empty_world())
|
||
assert report["xlsxBytes"] is None and report["filename"] is None
|
||
assert report["contract"]["reason"] == "no-flexible-version"
|
||
|
||
|
||
def test_schedule_order_and_equipment_exports_match_their_contracts():
|
||
world = _scheduled_world()
|
||
before = copy.deepcopy(world)
|
||
order = build_schedule_order_export(world)
|
||
equipment = build_schedule_equipment_export(world)
|
||
assert order["templateId"] == "schedule-order.v1"
|
||
assert equipment["templateId"] == "schedule-equipment.v1"
|
||
assert validate_workbook_contract(order["xlsxBytes"], SCHEDULE_ORDER_CONTRACT) == []
|
||
assert validate_workbook_contract(equipment["xlsxBytes"], SCHEDULE_EQUIPMENT_CONTRACT) == []
|
||
assert world == before
|
||
|
||
|
||
def test_contract_validator_reports_missing_sheets_and_wrong_headers():
|
||
report = build_plan_report(_world_with_version())
|
||
workbook = load_workbook(BytesIO(report["xlsxBytes"]))
|
||
workbook.remove(workbook["方案概览"])
|
||
workbook["工作计划"].cell(4, 1, "订单号（被改动）")
|
||
buffer = BytesIO()
|
||
workbook.save(buffer)
|
||
workbook.close()
|
||
defects = validate_workbook_contract(buffer.getvalue(), PLAN_REPORT_CONTRACT)
|
||
assert any("方案概览" in defect for defect in defects)
|
||
assert any("订单号" in defect for defect in defects)
|
||
|
||
|
||
def test_intake_template_is_profile_driven_and_round_trips_through_preview():
|
||
profile = builtin_compatibility_profile()
|
||
template = build_intake_template()
|
||
assert template["profileId"] == profile["id"]
|
||
assert template["profileDigest"] == profile["profileDigest"]
|
||
assert template["intakeSheet"] == "数据采集说明"
|
||
assert template["sheetNames"] == [spec["name"] for spec in profile["sheets"].values()]
|
||
assert validate_intake_template(template["xlsxBytes"], profile) == []
|
||
workbook = load_workbook(BytesIO(template["xlsxBytes"]))
|
||
assert workbook.sheetnames[0] == "数据采集说明"
|
||
assert workbook.sheetnames[1:] == template["sheetNames"]
|
||
assert profile["id"] in str(workbook["数据采集说明"]["A2"].value)
|
||
for role, spec in profile["sheets"].items():
|
||
sheet = workbook[spec["name"]]
|
||
headers = [cell.value for cell in sheet[1]]
|
||
assert headers == list(spec["columns"].values()), role
|
||
assert sheet.freeze_panes == "A2"
|
||
workbook.close()
|
||
|
||
from server.aps_domain.importers import preview_file
|
||
|
||
preview = preview_file("blank-template.xlsx", template["xlsxBytes"], empty_world())
|
||
assert preview["totalOk"] == 0
|
||
assert [batch["sheet"] for batch in preview["batches"]] == template["sheetNames"]
|
||
assert all(batch["sourceProfile"] == profile["id"] for batch in preview["batches"])
|
||
assert all(batch["profileDigest"] == profile["profileDigest"] for batch in preview["batches"])
|
||
assert all(batch["fieldMap"] for batch in preview["batches"])
|
||
|
||
|
||
def test_intake_template_examples_come_from_the_supplied_world_and_stay_deterministic():
|
||
profile = builtin_compatibility_profile()
|
||
world = seed_world()
|
||
first = build_intake_template(include_examples=True, world=world)
|
||
second = build_intake_template(include_examples=True, world=world)
|
||
assert first["xlsxBytes"] == second["xlsxBytes"]
|
||
assert first["sha256"] == second["sha256"]
|
||
assert first["rows"]["equipment"] == min(3, len(world["flexEquipment"]))
|
||
workbook = load_workbook(BytesIO(first["xlsxBytes"]))
|
||
sheet = workbook[profile["sheets"]["equipment"]["name"]]
|
||
assert sheet["A2"].value == world["flexEquipment"][0]["code"]
|
||
workbook.close()
|
||
|
||
|
||
def test_intake_template_endpoint_requires_auth_and_serves_the_declared_profile(monkeypatch, tmp_path):
|
||
import server.gateway.app as gateway_module
|
||
import server.state.store as state_store
|
||
from tests.auth_provider import install_test_auth
|
||
install_test_auth(monkeypatch, "tenant-intake-template")
|
||
store = _MemStore(empty_world(), str(tmp_path / "world.json"))
|
||
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
||
monkeypatch.setattr(state_store, "get_store", lambda: store)
|
||
client = TestClient(gateway_module.create_app())
|
||
|
||
unauthenticated = client.get("/api/import/template")
|
||
assert unauthenticated.status_code == 401
|
||
assert "AUTH_REQUIRED" in unauthenticated.text
|
||
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
response = client.get("/api/import/template")
|
||
assert response.status_code == 200, response.text
|
||
profile = builtin_compatibility_profile()
|
||
assert response.headers["x-aps-profile-id"] == profile["id"]
|
||
assert response.headers["x-aps-profile-digest"] == profile["profileDigest"]
|
||
assert response.headers["x-aps-template-id"] == f"intake-{profile['id']}.v1"
|
||
assert "%E9%87%87%E9%9B%86%E6%A8%A1%E6%9D%BF" in response.headers["content-disposition"]
|
||
assert response.content[:2] == b"PK"
|
||
assert validate_intake_template(response.content, profile) == []
|
||
|
||
# 条件请求：同一份模板不重复下载，换了配置摘要才重新拉取。
|
||
cached = client.get("/api/import/template",
|
||
headers={"If-None-Match": response.headers["etag"]})
|
||
assert cached.status_code == 304
|
||
assert cached.content == b""
|
||
assert cached.headers["x-aps-profile-id"] == profile["id"]
|
||
assert client.get("/api/import/template",
|
||
headers={"If-None-Match": '"stale-digest"'}).status_code == 200
|
||
|
||
unknown = client.get("/api/import/template", params={"profileId": "not-registered"})
|
||
assert unknown.status_code == 404
|
||
assert "not-registered" in unknown.json()["detail"]
|
||
|
||
|
||
def _deployment_profile(profile_id: str = "customer-b-planning-v1") -> dict:
|
||
"""一份改变了格式ID、表名和列名的部署配置，用于验证模板跟随配置目录。"""
|
||
profile = copy.deepcopy(builtin_compatibility_profile())
|
||
profile.pop("profileDigest")
|
||
profile["id"] = profile_id
|
||
profile["label"] = "客户B排产"
|
||
for spec in profile["sheets"].values():
|
||
spec["name"] = "B-" + spec["name"]
|
||
spec["columns"] = {field: "B_" + header for field, header in spec["columns"].items()}
|
||
return profile
|
||
|
||
|
||
def _write_profile(directory, profile: dict, filename: str = "deployment.json"):
|
||
directory.mkdir(exist_ok=True)
|
||
(directory / filename).write_text(json.dumps(profile, ensure_ascii=False), encoding="utf-8")
|
||
|
||
|
||
def test_template_and_contract_follow_the_configured_profile_directory(tmp_path, monkeypatch):
|
||
profile = _deployment_profile()
|
||
_write_profile(tmp_path, profile)
|
||
monkeypatch.setenv("APS_WORKBOOK_PROFILE_DIR", str(tmp_path))
|
||
|
||
template = build_intake_template()
|
||
assert template["profileId"] == profile["id"]
|
||
assert template["templateId"] == f"intake-{profile['id']}.v1"
|
||
assert template["sheetNames"] == [spec["name"] for spec in profile["sheets"].values()]
|
||
assert validate_intake_template(template["xlsxBytes"]) == []
|
||
|
||
workbook = load_workbook(BytesIO(template["xlsxBytes"]))
|
||
equipment = profile["sheets"]["equipment"]
|
||
headers = [cell.value for cell in workbook[equipment["name"]][1]]
|
||
assert headers == list(equipment["columns"].values())
|
||
assert profile["id"] in str(workbook["数据采集说明"]["A2"].value)
|
||
workbook.close()
|
||
|
||
contract = template_contract()
|
||
assert contract["profileId"] == profile["id"]
|
||
assert [sheet["name"] for sheet in contract["sheets"]] == template["sheetNames"]
|
||
assert contract["sheets"][0]["requiredColumns"] == profile["sheets"]["sourceNotes"]["requiredColumns"]
|
||
|
||
|
||
def test_template_default_profile_fails_closed_when_the_directory_is_ambiguous(tmp_path, monkeypatch):
|
||
first = _deployment_profile("customer-b-planning-v1")
|
||
first["compatibilityDefault"] = False
|
||
_write_profile(tmp_path, first, "first.json")
|
||
monkeypatch.setenv("APS_WORKBOOK_PROFILE_DIR", str(tmp_path))
|
||
|
||
# 目录里只有一份配置：即使没有默认标记，也只有这一种格式可用。
|
||
assert build_intake_template()["profileId"] == first["id"]
|
||
|
||
second = _deployment_profile("customer-c-planning-v1")
|
||
second["compatibilityDefault"] = False
|
||
_write_profile(tmp_path, second, "second.json")
|
||
with pytest.raises(ValueError, match="compatibilityDefault"):
|
||
build_intake_template()
|
||
|
||
# 唯一标记为默认的配置胜出，不再依赖目录顺序。
|
||
second["compatibilityDefault"] = True
|
||
_write_profile(tmp_path, second, "second.json")
|
||
assert build_intake_template()["profileId"] == second["id"]
|
||
|
||
|
||
def test_builtin_profile_declares_required_columns_for_every_role():
|
||
profile = builtin_compatibility_profile()
|
||
for role, spec in profile["sheets"].items():
|
||
assert spec["requiredColumns"], role
|
||
assert set(spec["requiredColumns"]) <= set(spec["columns"]), role
|
||
|
||
def test_intake_template_contract_matches_the_authorized_workbook():
|
||
"""现场工作簿与模板必须逐表逐列同构，模板才代表真实导入格式。"""
|
||
from tests.workbook_acceptance import source_for_test
|
||
|
||
contract = template_contract()
|
||
workbook = load_workbook(source_for_test(), read_only=True)
|
||
try:
|
||
assert workbook.sheetnames == [sheet["name"] for sheet in contract["sheets"]]
|
||
for sheet in contract["sheets"]:
|
||
physical = [cell.value for cell in next(workbook[sheet["name"]].iter_rows())]
|
||
assert physical[:len(sheet["columns"])] == sheet["columns"], sheet["name"]
|
||
finally:
|
||
workbook.close()
|
||
|
||
|
||
def test_schedule_export_templates_are_blank_header_skeletons_of_every_contract():
|
||
"""排产导出模板必须是合同的表头骨架：无业务行，且生成后即通过合同复核。"""
|
||
for report_type, contract in (
|
||
("plan", PLAN_REPORT_CONTRACT),
|
||
("schedule-order", SCHEDULE_ORDER_CONTRACT),
|
||
("schedule-equipment", SCHEDULE_EQUIPMENT_CONTRACT),
|
||
("schedule-blocked.v1", SCHEDULE_BLOCKED_CONTRACT),
|
||
):
|
||
template = build_contract_template(report_type)
|
||
assert template["templateId"] == contract.id
|
||
assert template["contractDigest"] == contract_digest()
|
||
assert template["sheets"][0]["name"] == contract.primary_sheet
|
||
assert validate_workbook_contract(template["xlsxBytes"], contract) == []
|
||
workbook = load_workbook(BytesIO(template["xlsxBytes"]))
|
||
try:
|
||
for spec in contract.sheets.values():
|
||
sheet = workbook[spec.name]
|
||
if not spec.header_row:
|
||
continue
|
||
headers = [cell.value for cell in sheet[spec.header_row]]
|
||
assert headers[:len(spec.headers)] == list(spec.headers), spec.name
|
||
assert sheet.cell(spec.data_start_row, 1).value is None, spec.name
|
||
finally:
|
||
workbook.close()
|
||
|
||
with pytest.raises(ValueError, match="未注册"):
|
||
build_contract_template("schedule-nothing")
|
||
|
||
|
||
def test_schedule_template_endpoint_serves_the_contract_with_conditional_requests(monkeypatch, tmp_path):
|
||
import server.gateway.app as gateway_module
|
||
import server.state.store as state_store
|
||
from tests.auth_provider import install_test_auth
|
||
install_test_auth(monkeypatch, "tenant-schedule-template")
|
||
store = _MemStore(empty_world(), str(tmp_path / "world.json"))
|
||
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
||
monkeypatch.setattr(state_store, "get_store", lambda: store)
|
||
client = TestClient(gateway_module.create_app())
|
||
|
||
unauthenticated = client.get("/api/reports/schedule-template")
|
||
assert unauthenticated.status_code == 401
|
||
assert "AUTH_REQUIRED" in unauthenticated.text
|
||
|
||
login = client.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
response = client.get("/api/reports/schedule-template")
|
||
assert response.status_code == 200, response.text
|
||
assert response.headers["x-aps-template-id"] == "schedule-plan.v1"
|
||
assert response.headers["x-aps-contract-digest"] == contract_digest()
|
||
assert response.content[:2] == b"PK"
|
||
assert validate_workbook_contract(response.content, PLAN_REPORT_CONTRACT) == []
|
||
|
||
cached = client.get("/api/reports/schedule-template",
|
||
headers={"If-None-Match": response.headers["etag"]})
|
||
assert cached.status_code == 304
|
||
assert cached.content == b""
|
||
assert cached.headers["x-aps-template-id"] == "schedule-plan.v1"
|
||
|
||
orders = client.get("/api/reports/schedule-template", params={"reportType": "schedule-order"})
|
||
assert orders.status_code == 200, orders.text
|
||
assert orders.headers["x-aps-template-id"] == "schedule-order.v1"
|
||
assert validate_workbook_contract(orders.content, SCHEDULE_ORDER_CONTRACT) == []
|
||
|
||
unknown = client.get("/api/reports/schedule-template", params={"reportType": "not-registered"})
|
||
assert unknown.status_code == 404
|
||
assert "not-registered" in unknown.json()["detail"]
|
||
|
||
|
||
def test_intake_definition_declares_required_state_and_allowed_values():
|
||
"""采集说明必须把必填、单位/格式和允许取值讲清楚，且全部来自同一份配置。"""
|
||
profile = builtin_compatibility_profile()
|
||
rows = intake_definition_rows(profile)
|
||
by_field = {(row[0], row[4]): row for row in rows}
|
||
|
||
order_type = by_field[("orders", "orderType")]
|
||
assert order_type[5] == "必填"
|
||
assert order_type[7].startswith("订单分类：")
|
||
assert "正式订单" in order_type[7] and "SANDBOX" in order_type[7]
|
||
assert by_field[("equipment", "availabilityRate")][7] == "数字，不写单位"
|
||
assert by_field[("inventory", "expectedArrivalDate")][7].startswith("日期")
|
||
assert by_field[("products", "stock")][5] == "可选"
|
||
|
||
template = build_intake_template()
|
||
workbook = load_workbook(BytesIO(template["xlsxBytes"]))
|
||
try:
|
||
definition = workbook["数据采集说明"]
|
||
assert [cell.value for cell in definition[6]] == list(template_contract()["definitionHeaders"])
|
||
assert definition.cell(7, 8).value
|
||
assert definition.cell(definition.max_row, 8).value
|
||
finally:
|
||
workbook.close()
|