AiDOPWarehouse/tools/mock/inbound/push_demo.py

#!/usr/bin/env python3
"""API_INBOUND 联调用例。签名按方案 §5.2：METHOD&path&key&ts&nonce&sha256&idem。

密钥只从 --secret 或环境变量 AIDOP_INBOUND_SECRET 读取，不写死生产 secret。
联调种子 TESTKPARTNER 的公开占位口令可用 --use-seed-secret（仅开发库）。
"""
from __future__ import annotations

import argparse
import hashlib
import hmac
import json
import os
import sys
import time
import urllib.error
import urllib.request
import uuid
from base64 import b64encode
from datetime import datetime, timezone, timedelta

# 公开联调种子，不是生产密钥
_SEED_SECRET = "TESTKPARTNER_SECRET_DO_NOT_USE_IN_PROD"


def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sign(secret: str, message: str) -> str:
    digest = hmac.new(secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).digest()
    return b64encode(digest).decode("ascii")


def now_iso() -> str:
    return datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%dT%H:%M:%S+08:00")


def sales_row(**extra) -> dict:
    row = {
        "bill_no": extra.pop("bill_no", "SO-INB-DEMO-001"),
        "entry_seq": extra.pop("entry_seq", 1),
        "seorder_id": extra.pop("seorder_id", 900001),
        "qty": extra.pop("qty", 10),
        "item_number": "ITEM-INB-001",
        "sourceUpdatedAt": extra.pop("sourceUpdatedAt", now_iso()),
    }
    row.update(extra)
    return row


def envelope(rows: list, snapshot_id: str | None = None, seq: int | None = None) -> dict:
    data: dict = {"list": rows}
    if snapshot_id:
        data["snapshotId"] = snapshot_id
    if seq is not None:
        data["seq"] = seq
    return {"data": data}


class InboundClient:
    def __init__(self, base_url: str, access_key: str, secret: str):
        self.base = base_url.rstrip("/")
        self.key = access_key
        self.secret = secret

    def request(
        self,
        method: str,
        path: str,
        body: bytes | None = None,
        *,
        idem: str | None = None,
        ts: int | None = None,
        nonce: str | None = None,
        skip_idem: bool = False,
        tamper_body: bool = False,
        extra_headers: dict | None = None,
    ) -> tuple[int, dict | str]:
        raw = body if body is not None else b""
        digest = sha256_hex(raw)
        if tamper_body:
            raw = raw + b"x"
        ts_s = str(ts if ts is not None else int(time.time()))
        nonce = nonce or uuid.uuid4().hex
        idem = idem or f"IDEM-{uuid.uuid4().hex}"
        message = f"{method.upper()}&{path}&{self.key}&{ts_s}&{nonce}&{digest}&{idem}"
        headers = {
            "Content-Type": "application/json",
            "X-Access-Key": self.key,
            "X-Timestamp": ts_s,
            "X-Nonce": nonce,
            "X-Content-SHA256": digest,
            "X-Signature": sign(self.secret, message),
        }
        if not skip_idem:
            headers["Idempotency-Key"] = idem
        if extra_headers:
            headers.update(extra_headers)
        req = urllib.request.Request(self.base + path, data=raw if method != "GET" else None, method=method, headers=headers)
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                text = resp.read().decode("utf-8")
                return resp.status, _parse(text)
        except urllib.error.HTTPError as ex:
            text = ex.read().decode("utf-8")
            return ex.code, _parse(text)
        except urllib.error.URLError as ex:
            return 0, {"message": str(ex.reason)}


def _parse(text: str):
    try:
        return json.loads(text) if text else {}
    except json.JSONDecodeError:
        return text


def report(name: str, ok: bool, detail) -> bool:
    mark = "PASS" if ok else "FAIL"
    print(f"[{mark}] {name}: {detail}")
    return ok


def case_happy(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8")
    status, data = c.request("POST", path, body)
    return report("happy", status == 202 and isinstance(data, dict) and data.get("code") in (0, 2), (status, data))


def case_missing_idem(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8")
    status, data = c.request("POST", path, body, skip_idem=True)
    msg = data.get("message") if isinstance(data, dict) else data
    return report("missing Idempotency-Key", status == 400 and "Idempotency-Key" in str(msg), (status, data))


def case_tamper_body(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8")
    status, data = c.request("POST", path, body, tamper_body=True)
    msg = data.get("message") if isinstance(data, dict) else data
    return report("body+1 byte", status == 400 and "digest" in str(msg).lower(), (status, data))


def case_expired_ts(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    body = json.dumps(envelope([sales_row()]), ensure_ascii=False).encode("utf-8")
    status, data = c.request("POST", path, body, ts=int(time.time()) - 3600)
    return report("expired timestamp", status == 401, (status, data))


def case_nonce_replay(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    body = json.dumps(envelope([sales_row(bill_no="SO-NONCE-1")]), ensure_ascii=False).encode("utf-8")
    nonce = uuid.uuid4().hex
    status1, _ = c.request("POST", path, body, nonce=nonce)
    status2, data2 = c.request("POST", path, body, nonce=nonce, idem=f"IDEM-{uuid.uuid4().hex}")
    return report("nonce replay", status1 == 202 and status2 == 401, (status1, status2, data2))


def case_same_key_replay(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    body = json.dumps(envelope([sales_row(bill_no="SO-REPLAY-1")]), ensure_ascii=False).encode("utf-8")
    idem = f"IDEM-REPLAY-{uuid.uuid4().hex}"
    status1, data1 = c.request("POST", path, body, idem=idem)
    status2, data2 = c.request("POST", path, body, idem=idem)
    replay = isinstance(data2, dict) and (data2.get("data") or {}).get("idempotentReplay") is True
    return report("same-key replay", status1 == 202 and status2 == 202 and replay, (status1, status2, data2))


def case_same_key_diff_body(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    idem = f"IDEM-DIFF-{uuid.uuid4().hex}"
    b1 = json.dumps(envelope([sales_row(bill_no="SO-DIFF-1")]), ensure_ascii=False).encode("utf-8")
    b2 = json.dumps(envelope([sales_row(bill_no="SO-DIFF-2")]), ensure_ascii=False).encode("utf-8")
    status1, _ = c.request("POST", path, b1, idem=idem)
    status2, data2 = c.request("POST", path, b2, idem=idem)
    return report("same-key different body", status1 == 202 and status2 == 409, (status1, status2, data2))


def case_partial(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    rows = [sales_row(bill_no="SO-PART-OK"), {"qty": 1, "sourceUpdatedAt": now_iso()}]
    body = json.dumps(envelope(rows), ensure_ascii=False).encode("utf-8")
    status, data = c.request("POST", path, body)
    code = data.get("code") if isinstance(data, dict) else None
    return report("partial", status == 202 and code == 2, (status, data))


def case_stale(c: InboundClient, entity: str) -> bool:
    path = f"/api/mdp/inbound/{entity}"
    newer = "2026-09-11T12:00:00+08:00"
    older = "2026-09-01T12:00:00+08:00"
    b1 = json.dumps(envelope([sales_row(bill_no="SO-STALE-1", sourceUpdatedAt=newer)]), ensure_ascii=False).encode("utf-8")
    b2 = json.dumps(envelope([sales_row(bill_no="SO-STALE-1", sourceUpdatedAt=older)]), ensure_ascii=False).encode("utf-8")
    status1, _ = c.request("POST", path, b1)
    status2, data2 = c.request("POST", path, b2)
    rejected = ((data2.get("data") or {}).get("staleRejected") if isinstance(data2, dict) else 0) or 0
    return report("stale", status1 == 202 and status2 == 202 and rejected >= 1, (status1, status2, data2))


def case_snapshot(c: InboundClient, entity: str) -> bool:
    open_path = f"/api/mdp/inbound/{entity}/snapshots"
    status, data = c.request("POST", open_path, b"")
    snap = (data.get("data") or {}).get("snapshotId") if isinstance(data, dict) else None
    if status not in (200, 201) or not snap:
        return report("snapshot open", False, (status, data))
    push = f"/api/mdp/inbound/{entity}"
    b1 = json.dumps(envelope([sales_row(bill_no="SO-SNAP-1")], snap, 1), ensure_ascii=False).encode("utf-8")
    b2 = json.dumps(envelope([sales_row(bill_no="SO-SNAP-2")], snap, 2), ensure_ascii=False).encode("utf-8")
    s1, _ = c.request("POST", push, b1)
    s2, _ = c.request("POST", push, b2)
    commit_path = f"/api/mdp/inbound/{entity}/snapshots/{snap}/commit"
    sc, dc = c.request("POST", commit_path, b"")
    committed = isinstance(dc, dict) and ((dc.get("data") or {}).get("status") == "COMMITTED" or dc.get("code") == 0)
    return report("snapshot open→2 batches→commit", s1 == 202 and s2 == 202 and sc in (200, 202) and committed, (s1, s2, sc, dc))


CASES = {
    "happy": case_happy,
    "missing-idem": case_missing_idem,
    "tamper-body": case_tamper_body,
    "expired-ts": case_expired_ts,
    "nonce-replay": case_nonce_replay,
    "same-key-replay": case_same_key_replay,
    "same-key-diff-body": case_same_key_diff_body,
    "partial": case_partial,
    "stale": case_stale,
    "snapshot": case_snapshot,
}


def main() -> int:
    p = argparse.ArgumentParser(description="API_INBOUND mock push")
    p.add_argument("--base-url", default="http://127.0.0.1:5005")
    p.add_argument("--access-key", default="TESTKPARTNER")
    p.add_argument("--secret", default=os.environ.get("AIDOP_INBOUND_SECRET", ""))
    p.add_argument("--use-seed-secret", action="store_true", help="使用公开联调种子口令（非生产）")
    p.add_argument("--entity", default="S1_SALES_ORDER_ENTRY")
    p.add_argument("--case", default="all", choices=["all", *CASES])
    args = p.parse_args()

    secret = args.secret
    if args.use_seed_secret:
        secret = _SEED_SECRET
    if not secret:
        print("missing --secret or AIDOP_INBOUND_SECRET（开发库可加 --use-seed-secret）", file=sys.stderr)
        return 2

    client = InboundClient(args.base_url, args.access_key, secret)
    names = list(CASES) if args.case == "all" else [args.case]
    ok = True
    for name in names:
        ok = CASES[name](client, args.entity) and ok
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(main())