79 lines
3.5 KiB
Python
79 lines
3.5 KiB
Python
# ============================================================
|
||
# 工程目录 SQL：根据这个排产应入库并具备开排条件（不依赖 Excel 批次）
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
from server.aps_domain.folder_pack import analyze_work_dir
|
||
from server.aps_domain.readiness import check_readiness
|
||
from server.state.seed import seed_world
|
||
|
||
|
||
def test_sql_pack_in_workdir_covers_schedule(tmp_path: Path, monkeypatch):
|
||
"""最小 SQL 包（订单+工艺 craftl）写入目录后，analyze 应可开排且无 Excel batches。"""
|
||
sql = tmp_path / "mini.sql"
|
||
sql.write_text(
|
||
"""
|
||
CREATE TABLE `pl_order` (
|
||
`id` int, `code` varchar(64), `business_code` varchar(64),
|
||
`customer_business_code` varchar(64), `material_code` varchar(64),
|
||
`material_name` varchar(64), `parent_code` varchar(64), `craftl_code` varchar(64),
|
||
`machine_model` varchar(64), `dept` varchar(64), `people` varchar(64),
|
||
`drawing_code` varchar(64), `quantity` decimal(10,2), `man_quantity` decimal(10,2),
|
||
`all_quantity` decimal(10,2), `status` varchar(8), `stock_status` varchar(8),
|
||
`distribution_status` varchar(8), `type` varchar(8), `description` varchar(64),
|
||
`planned_end_time` datetime, `is_delete` varchar(8), `level` int
|
||
) ENGINE=InnoDB;
|
||
CREATE TABLE `r_production_craftl` (
|
||
`id` int, `craftl_code` varchar(64), `craftl_name` varchar(64),
|
||
`procedure_code` varchar(64), `procedure_name` varchar(64),
|
||
`output_material_code` varchar(64), `output_material_name` varchar(64),
|
||
`working_hours` decimal(10,2), `sort_no` int, `is_delete` varchar(8)
|
||
) ENGINE=InnoDB;
|
||
CREATE TABLE `md_equipment` (
|
||
`id` int, `code` varchar(64), `name` varchar(64), `is_delete` varchar(8)
|
||
) ENGINE=InnoDB;
|
||
CREATE TABLE `md_material` (
|
||
`id` int, `code` varchar(64), `name` varchar(64), `type` varchar(32),
|
||
`unit` varchar(8), `is_delete` varchar(8)
|
||
) ENGINE=InnoDB;
|
||
INSERT INTO `pl_order` VALUES
|
||
(1,'ORD1',NULL,'C1','FG001','成品A',NULL,'CRAFT1',NULL,NULL,NULL,NULL,10,NULL,NULL,'0','0','0','S',NULL,'2026-08-01 00:00:00','0',5);
|
||
INSERT INTO `r_production_craftl` VALUES
|
||
(1,'CRAFT1','成品A','OP10','粗车','WIP-X','半成品',0.5,10,'0'),
|
||
(2,'CRAFT1','成品A','OP20','精车','FG001','成品A',0.8,20,'0');
|
||
INSERT INTO `md_equipment` VALUES (1,'EQ1','车床','0');
|
||
INSERT INTO `md_material` VALUES (1,'FG001','成品A','成品','件','0');
|
||
""",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
class FakePS:
|
||
def snapshot(self, include_messages=False):
|
||
return {
|
||
"projects": [{"id": "p1", "name": "松岳测", "workDir": str(tmp_path)}],
|
||
"sessions": [{"id": "s1", "projectId": "p1"}],
|
||
"files": [],
|
||
}
|
||
|
||
monkeypatch.setattr("server.state.projects.get_project_store", lambda: FakePS())
|
||
world = seed_world()
|
||
for k in ("flexOrders", "flexMaterials", "flexRoutings", "flexEquipment", "flexBom"):
|
||
world[k] = []
|
||
|
||
report = analyze_work_dir(world, "s1", force_sql_replace=True)
|
||
assert report["sqlApplied"] is True
|
||
assert report["canSchedule"] is True
|
||
assert report["missing"] == []
|
||
assert (report.get("sqlStats") or {}).get("routedProducts", 0) >= 1
|
||
# 不依赖 Excel batches
|
||
assert not report.get("batches")
|
||
|
||
ready = check_readiness(world)
|
||
assert ready["summary"]["ready"] >= 1
|
||
assert ready["summary"]["blocked"] == 0
|
||
pcs = {s["productCode"] for s in world["flexRoutings"]}
|
||
assert "FG001" in pcs
|
||
assert "WIP-X" not in pcs
|