AiDOPWarehouse/doc/_verify_s1_s4_continue_e2e.py

#!/usr/bin/env python3
"""Continue unfinished S1-S4 tests: routing sync, reschedule, dispatch, S3/S4 smoke."""
from __future__ import annotations

import json
import sys
import urllib.error
import urllib.request
from datetime import date

import pymysql
from gmssl import sm2

BASE = "http://127.0.0.1:5005"
TENANT = 797403760988229
DOMAIN = str(TENANT)
BILL_NO = "SO202607120001"
WORK_ORD = "M000000005"
ORDER_ID = 825854621474885
PK = "84C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312"
CONN = dict(
    host="123.60.180.165",
    port=3306,
    user="aidopremote",
    password="1234567890aiDOP#",
    database="aidopdev",
    charset="utf8mb4",
    connect_timeout=20,
    cursorclass=pymysql.cursors.DictCursor,
)


def http(method, path, body=None, token=None, timeout=180):
    headers = {"Content-Type": "application/json", "Accept": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(f"{BASE}{path}", data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8")
            return resp.status, json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        try:
            return e.code, json.loads(raw)
        except json.JSONDecodeError:
            return e.code, {"raw": raw[:1500]}
    except Exception as e:
        return 0, {"error": str(e)}


def unwrap(body):
    if isinstance(body, dict) and body.get("code") == 200 and "result" in body:
        return body["result"]
    return body


def ok(status, body):
    if status != 200:
        return False
    if isinstance(body, dict) and body.get("error"):
        return False
    if isinstance(body, dict) and body.get("code") not in (None, 200):
        return False
    if isinstance(body, dict) and body.get("type") == "error":
        return False
    return True


def login():
    crypt = sm2.CryptSM2(public_key=PK, private_key=None, mode=1)
    pwd = crypt.encrypt(b"1234567890dop").hex()
    st, body = http("POST", "/api/sysAuth/login", {"account": "AIDOPDemo", "password": pwd, "tenantId": DOMAIN})
    token = (body.get("result") or {}).get("accessToken") or body.get("accessToken")
    if not token:
        raise RuntimeError(f"login failed: {body}")
    return token


def add(report, name, status, body, extra=None):
    item = {
        "step": name,
        "http": status,
        "ok": ok(status, body),
        "body": unwrap(body) if isinstance(body, dict) else body,
    }
    if isinstance(body, dict) and body.get("message"):
        item["message"] = body.get("message")
    if isinstance(body, dict) and body.get("raw"):
        item["raw"] = body.get("raw")[:300]
    if isinstance(body, dict) and body.get("error"):
        item["error"] = body.get("error")
    if extra:
        item.update(extra)
    # Truncate large bodies
    b = item.get("body")
    if isinstance(b, dict) and len(json.dumps(b, default=str)) > 2000:
        keep = {}
        for k in (
            "message",
            "Message",
            "total",
            "workOrderCount",
            "scheduleRowCount",
            "skippedWorkOrders",
            "shortageItemCount",
            "prCreatedCount",
            "prUpdatedCount",
            "createdCount",
            "candidateCount",
            "demandCount",
            "enabled",
            "batchId",
            "dwdCount",
            "shortageCount",
            "workOrdTempCount",
            "list",
            "actionCode",
            "warnings",
            "workOrders",
        ):
            if k in b:
                keep[k] = b[k] if k != "list" else (len(b[k]) if isinstance(b[k], list) else b[k])
        if not keep:
            keep = {"keys": list(b.keys())[:12]}
        item["body"] = keep
    report["steps"].append(item)
    return item


def q(conn, sql, params=None):
    with conn.cursor() as cur:
        cur.execute(sql, params or ())
        return cur.fetchall()


def q1(conn, sql, params=None):
    rows = q(conn, sql, params)
    return rows[0] if rows else {}


def main():
    report = {
        "title": "S1-S4 continue unfinished tests",
        "bill_no": BILL_NO,
        "work_ord": WORK_ORD,
        "steps": [],
        "db": {},
        "checks": [],
    }

    st, body = http("GET", "/api/sysAuth/captcha")
    add(report, "backend_health", st, body, {"ok": st == 200})
    if st != 200:
        report["passed"] = False
        print(json.dumps(report, ensure_ascii=False, indent=2, default=str))
        sys.exit(1)

    token = login()
    report["login"] = "ok"

    with pymysql.connect(**CONN) as conn:
        report["db"]["before"] = {
            "wo": q1(
                conn,
                "SELECT WorkOrd, ItemNum, Status, BusinessID FROM WorkOrdMaster WHERE tenant_id=%s AND WorkOrd=%s",
                (TENANT, WORK_ORD),
            ),
            "routing_cnt": q1(
                conn,
                "SELECT COUNT(*) AS c FROM WorkOrdRouting WHERE tenant_id=%s AND WorkOrd=%s",
                (TENANT, WORK_ORD),
            ).get("c", 0),
            "detail_cnt": q1(
                conn,
                "SELECT COUNT(*) AS c FROM WorkOrdDetail WHERE tenant_id=%s AND WorkOrd=%s",
                (TENANT, WORK_ORD),
            ).get("c", 0),
            "demand": q1(conn, "SELECT COUNT(*) AS c FROM ic_demandschedule WHERE tenant_id=%s", (TENANT,)).get("c", 0),
            "pr": q1(
                conn,
                "SELECT COUNT(*) AS c FROM srm_pr_main WHERE tenant_id=%s AND IFNULL(IsDeleted,0)=0",
                (TENANT,),
            ).get("c", 0),
            "workordtemp_has_tenant": q1(
                conn,
                """
                SELECT COUNT(*) AS c FROM information_schema.COLUMNS
                WHERE TABLE_SCHEMA='aidopdev' AND TABLE_NAME='WorkOrdTemp' AND COLUMN_NAME='tenant_id'
                """,
            ).get("c", 0),
        }

    # S2: sync routing for unfinished schedule path
    st, body = http(
        "POST",
        "/api/Production/scheduling/sync-routing",
        {"workOrd": WORK_ORD, "domain": DOMAIN},
        token,
    )
    add(report, "s2_sync_routing", st, body)

    st, body = http("GET", f"/api/Production/scheduling/routings?workOrd={WORK_ORD}&domain={DOMAIN}", token=token)
    add(report, "s2_routings_list", st, body)

    st, body = http("GET", f"/api/Production/scheduling/materials?workOrd={WORK_ORD}&domain={DOMAIN}", token=token)
    add(report, "s2_materials_list", st, body)

    st, body = http("POST", f"/api/Production/scheduling/generate?domain={DOMAIN}", None, token, timeout=300)
    add(report, "s2_schedule_generate_retry", st, body)

    st, body = http("GET", f"/api/Production/scheduling/list?page=1&pageSize=10&domain={DOMAIN}&keyword={WORK_ORD}", token=token)
    add(report, "s2_scheduling_list_wo", st, body)

    st, body = http("GET", f"/api/Production/work-order-progress/list?page=1&pageSize=10&keyword={WORK_ORD}", token=token)
    add(report, "s2_progress_list", st, body)

    # S2 dispatch / kitting
    st, body = http(
        "POST",
        "/api/WorkOrder/dispatch/kitting-check",
        {"workOrd": WORK_ORD, "tenantId": TENANT},
        token,
        timeout=180,
    )
    add(report, "s2_dispatch_kitting_check", st, body)

    st, body = http(
        "POST",
        "/api/WorkOrder/dispatch/work-order-kitting-check",
        {"workOrd": WORK_ORD, "tenantId": TENANT},
        token,
        timeout=180,
    )
    add(report, "s2_work_order_kitting_check", st, body)

    st, body = http(
        "POST",
        "/api/WorkOrder/dispatch/release",
        {
            "workOrd": WORK_ORD,
            "tenantId": TENANT,
            "ordDate": date.today().isoformat(),
            "lotSerial": "E2E-CONT-001",
        },
        token,
        timeout=180,
    )
    add(report, "s2_dispatch_release", st, body)

    # S3
    for name, path, method, body in [
        ("s3_demand_schedule_list", "/api/Supply/demand-schedule/list?page=1&pageSize=10", "GET", None),
        ("s3_purchase_request_list", "/api/Supply/purchase-request/list?page=1&pageSize=10", "GET", None),
        ("s3_purchase_order_list", "/api/Supply/purchase-order/list?page=1&pageSize=10", "GET", None),
        ("s3_delivery_schedule_list", "/api/Supply/delivery-schedule/list?page=1&pageSize=10", "GET", None),
        ("s3_material_readiness_list", "/api/Supply/work-order-material-readiness/list?page=1&pageSize=10", "GET", None),
        ("s3_kitting_refresh", "/api/Supply/work-order-material-readiness/refresh", "POST", None),
        ("s3_delivery_generate", "/api/Supply/delivery-schedule/generate", "POST", None),
        ("s3_process_outsource_list", "/api/Supply/process-outsource-order/list?page=1&pageSize=5", "GET", None),
        ("s3_outsource_order_list", "/api/Supply/outsource-order/list?page=1&pageSize=5", "GET", None),
    ]:
        st, body = http(method, path, body, token, timeout=300)
        add(report, name, st, body)

    # S4
    for name, path in [
        ("s4_supplier_shipment_list", "/api/ProcurementExecution/supplier-shipment/list?page=1&pageSize=10"),
        ("s4_receiving_list", "/api/ProcurementExecution/receiving/list?page=1&pageSize=5"),
        ("s4_iqc_list", "/api/ProcurementExecution/iqc/list?page=1&pageSize=5"),
        ("s4_shortage_board", "/api/ProcurementExecution/shortage-board/list?page=1&pageSize=5"),
    ]:
        st, body = http("GET", path, token=token)
        add(report, name, st, body)

    st, body = http("POST", f"/api/WorkOrder/material-requirement/generate?domain={DOMAIN}", None, token, timeout=300)
    add(report, "s4_mrp_generate_again", st, body)

    # S1 lists for sample order
    st, body = http("GET", f"/api/Order/seorder/list?billNo={BILL_NO}&page=1&pageSize=5", token=token)
    add(report, "s1_order_list", st, body)
    st, body = http("GET", f"/api/Order/delivery/list?billNo={BILL_NO}&page=1&pageSize=10", token=token)
    add(report, "s1_delivery_list", st, body)

    for mod in ("S1", "S2", "S3", "S4"):
        st, body = http("GET", f"/api/AidopKanban/module-detail?moduleCode={mod}", token=token)
        add(report, f"kanban_{mod}", st, body)

    with pymysql.connect(**CONN) as conn:
        report["db"]["after"] = {
            "wo": q1(
                conn,
                "SELECT WorkOrd, ItemNum, Status, BusinessID FROM WorkOrdMaster WHERE tenant_id=%s AND WorkOrd=%s",
                (TENANT, WORK_ORD),
            ),
            "routing_cnt": q1(
                conn,
                "SELECT COUNT(*) AS c FROM WorkOrdRouting WHERE tenant_id=%s AND WorkOrd=%s",
                (TENANT, WORK_ORD),
            ).get("c", 0),
            "detail_cnt": q1(
                conn,
                "SELECT COUNT(*) AS c FROM WorkOrdDetail WHERE tenant_id=%s AND WorkOrd=%s",
                (TENANT, WORK_ORD),
            ).get("c", 0),
            "demand": q1(conn, "SELECT COUNT(*) AS c FROM ic_demandschedule WHERE tenant_id=%s", (TENANT,)).get("c", 0),
            "pr": q1(
                conn,
                "SELECT COUNT(*) AS c FROM srm_pr_main WHERE tenant_id=%s AND IFNULL(IsDeleted,0)=0",
                (TENANT,),
            ).get("c", 0),
            "po": q1(
                conn,
                "SELECT COUNT(*) AS c FROM srm_po_main WHERE tenant_id=%s AND IFNULL(IsDeleted,0)=0",
                (TENANT,),
            ).get("c", 0),
            "dwd_readiness": q1(
                conn,
                "SELECT COUNT(*) AS c FROM dwd_material_readiness WHERE tenant_id=%s AND work_order=%s",
                (TENANT, WORK_ORD),
            ).get("c", 0),
            "entries": q(
                conn,
                """
                SELECT e.entry_seq, e.item_number, e.progress, w.WorkOrd, w.Status
                FROM crm_seorderentry e
                LEFT JOIN WorkOrdMaster w ON w.BusinessID=e.Id AND w.tenant_id=e.tenant_id
                WHERE e.seorder_id=%s AND e.IsDeleted=0
                ORDER BY e.entry_seq
                """,
                (ORDER_ID,),
            ),
        }

    # Checks
    steps = {s["step"]: s for s in report["steps"]}
    after = report["db"]["after"]
    checks = [
        {"name": "sync_routing_api", "ok": steps["s2_sync_routing"]["ok"]},
        {
            "name": "routing_rows_exist",
            "ok": (after.get("routing_cnt") or 0) >= 1,
            "detail": after.get("routing_cnt"),
        },
        {"name": "schedule_generate_api", "ok": steps["s2_schedule_generate_retry"]["ok"]},
        {
            "name": "schedule_has_rows_or_explained",
            "ok": steps["s2_schedule_generate_retry"]["ok"],
            "detail": steps["s2_schedule_generate_retry"].get("body"),
        },
        {"name": "s2_progress_list", "ok": steps["s2_progress_list"]["ok"]},
        {"name": "s3_purchase_request_list", "ok": steps["s3_purchase_request_list"]["ok"]},
        {"name": "s3_purchase_order_list", "ok": steps["s3_purchase_order_list"]["ok"]},
        {
            "name": "s3_material_readiness_list",
            "ok": steps["s3_material_readiness_list"]["ok"],
            "detail": steps["s3_material_readiness_list"].get("message")
            or steps["s3_material_readiness_list"].get("raw")
            or steps["s3_material_readiness_list"].get("body"),
        },
        {"name": "s3_kitting_refresh", "ok": steps["s3_kitting_refresh"]["ok"]},
        {"name": "s4_supplier_shipment_list", "ok": steps["s4_supplier_shipment_list"]["ok"]},
        {"name": "s4_mrp_generate", "ok": steps["s4_mrp_generate_again"]["ok"]},
        {"name": "s1_order_list", "ok": steps["s1_order_list"]["ok"]},
        {"name": "kanban_all", "ok": all(steps[f"kanban_{m}"]["ok"] for m in ("S1", "S2", "S3", "S4"))},
        {
            "name": "dispatch_release_or_blocked",
            "ok": steps["s2_dispatch_release"]["ok"]
            or (
                isinstance(steps["s2_dispatch_release"].get("body"), dict)
                and "物料" in str(steps["s2_dispatch_release"].get("body"))
            )
            or bool(steps["s2_dispatch_release"].get("message")),
            "detail": steps["s2_dispatch_release"].get("message") or steps["s2_dispatch_release"].get("body"),
            "note": "允许因缺料/工艺阻塞失败，但需有明确返回",
        },
    ]
    report["checks"] = checks
    report["passed"] = all(c["ok"] for c in checks)
    report["failed"] = [c["name"] for c in checks if not c["ok"]]
    report["step_failures"] = [s["step"] for s in report["steps"] if not s.get("ok")]

    out = r"d:\Projects\Ai-DOP\SourceCode\ZZYDOP\doc\_verify_s1_s4_continue_e2e_result.json"
    with open(out, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2, default=str)

    summary = {
        "passed": report["passed"],
        "failed": report["failed"],
        "step_failures": report["step_failures"],
        "db_before": report["db"]["before"],
        "db_after": {
            "wo": after.get("wo"),
            "routing_cnt": after.get("routing_cnt"),
            "detail_cnt": after.get("detail_cnt"),
            "demand": after.get("demand"),
            "pr": after.get("pr"),
            "po": after.get("po"),
            "dwd_readiness": after.get("dwd_readiness"),
            "entries": after.get("entries"),
        },
        "key_steps": {
            k: {
                "ok": steps[k]["ok"],
                "http": steps[k]["http"],
                "body": steps[k].get("body"),
                "message": steps[k].get("message"),
            }
            for k in (
                "s2_sync_routing",
                "s2_schedule_generate_retry",
                "s2_dispatch_release",
                "s3_material_readiness_list",
                "s3_kitting_refresh",
                "s3_delivery_generate",
                "s4_mrp_generate_again",
                "s4_supplier_shipment_list",
            )
            if k in steps
        },
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2, default=str))
    sys.exit(0 if report["passed"] else 5)


if __name__ == "__main__":
    main()