aps-agent/tests/golden/test_sql_pack.py

# ============================================================
# SQL 数据包解析（松岳 MES dump 口径）
# ============================================================
from __future__ import annotations

from pathlib import Path

from server.importers.sql_pack import (
    _iter_insert_tuples,
    _map_orders,
    apply_sql_pack_to_world,
    sql_pack_to_flex,
)
from server.state.seed import seed_world
from tests.external_data import external_dir


def test_parse_insert_tuple_with_quotes():
    line = (
        "INSERT INTO `pl_order` VALUES "
        "(1, 'SYS001', 'B1', 'C1', '030129001', '支重轮', NULL, NULL, 'PC800', "
        "NULL, NULL, 'D1', 117.00, NULL, NULL, 'G', '0', '0', 'S', NULL, "
        "'2023-09-24 00:00:00', '2023-10-01 00:00:00', NULL, NULL, NULL, NULL, NULL, NULL, "
        "NULL, '0', NULL, 5, 'u1');\n"
    )
    rows = list(_iter_insert_tuples(line))
    assert len(rows) == 1
    assert rows[0][1] == "SYS001"
    assert rows[0][4] == "030129001"


def test_map_orders_and_apply():
    rows = [{
        "code": "SYS202309210001", "material_code": "030129001", "material_name": "支重轮",
        "quantity": 117, "planned_end_time": "2023-10-01 00:00:00",
        "customer_business_code": "CUST-A", "is_delete": "0", "level": 5,
    }]
    orders = _map_orders(rows)
    assert len(orders) == 1
    assert orders[0]["orderNo"] == "SYS202309210001"
    assert orders[0]["productCode"] == "030129001"
    assert orders[0]["dueDate"].startswith("2023-10-01")

    world = seed_world()
    for k in ("flexOrders", "flexMaterials", "flexRoutings", "flexEquipment", "flexBom"):
        world[k] = []
    flex = {
        "orders": orders,
        "materials": [{
            "code": "030129001", "name": "支重轮", "type": "FINISHED_PRODUCT",
            "unit": "件", "stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0,
        }],
        "equipment": [{
            "code": "JY01001", "name": "数控车床", "capabilities": ["GENERAL"],
            "opStdTime": {"GENERAL": 1.0}, "movable": False, "moveTimeMin": 0,
            "zone": "ZONE-A", "adaptableMolds": [], "availabilityRate": 0.95, "status": "RUNNING",
        }],
        "routing": [{
            "productCode": "030129001", "productName": "支重轮", "seq": 10,
            "operationCode": "NZ0027", "operationName": "精车",
            "requireMold": False, "stdTimePerUnit": 30.0,
        }],
        "bom": [],
    }
    summary = apply_sql_pack_to_world(world, flex, replace=True)
    assert summary["orders"] == 1
    assert any(o["orderNo"] == "SYS202309210001" for o in world["flexOrders"])


def test_routing_binds_via_craftl_not_output_material():
    """工艺挂成品：用订单 craftl→料号，不能用中间件 output_material_code。"""
    flex = sql_pack_to_flex({
        "rowsByTable": {
            "pl_order": [{
                "code": "ORD1", "material_code": "FG001", "material_name": "成品A",
                "quantity": 10, "planned_end_time": "2026-08-01", "is_delete": "0",
                "craftl_code": "CRAFT-9", "level": 5,
            }],
            "md_craftl": [{"code": "CRAFT-9", "product_code": "FG001", "name": "成品A工艺"}],
            "r_production_craftl": [
                {
                    "craftl_code": "CRAFT-9", "procedure_code": "OP10", "procedure_name": "粗车",
                    "output_material_code": "WIP-X", "sort_no": 10, "working_hours": 0.5,
                    "is_delete": "0",
                },
                {
                    "craftl_code": "CRAFT-9", "procedure_code": "OP20", "procedure_name": "精车",
                    "output_material_code": "FG001", "sort_no": 20, "working_hours": 0.8,
                    "is_delete": "0",
                },
            ],
            "md_material": [],
            "md_equipment": [{
                "code": "EQ1", "name": "车床", "is_delete": "0",
            }],
        },
        "counts": {},
    })
    assert flex["stats"]["orders"] == 1
    pcs = {s["productCode"] for s in flex["routing"]}
    assert "FG001" in pcs
    assert "WIP-X" not in pcs
    assert flex["stats"]["routedProducts"] == 1
    assert flex["stats"]["missingRouting"] == 0


def test_songyue_sql_if_present():
    root = external_dir("SONGYUE_SQL_DIR", "songyue")
    sql = root / "mesdb_pro.sql"
    if not sql.exists():
        return
    from server.importers.sql_pack import scan_sql_pack
    scan = scan_sql_pack(
        str(sql),
        limits={
            "orders": 80, "materials": 200, "routing": 2000,
            "routing_prefer": 15000, "equipment": 50, "bom": 200, "craft": 500,
        },
    )
    assert "pl_order" in (scan.get("counts") or {})
    flex = sql_pack_to_flex(scan)
    assert flex["stats"]["orders"] >= 1
    assert flex["stats"]["materials"] >= 1
    # 现场库有完整工艺时，订单产品应大部分挂上工艺
    if flex["stats"].get("orderProducts", 0) >= 10:
        ratio = flex["stats"]["routedProducts"] / max(flex["stats"]["orderProducts"], 1)
        assert ratio >= 0.7, flex["stats"]