AiDOPWarehouse/doc/db/mdp/contract_tests/smoke_api_ingest.py

"""今日可跑的 stg 级冒烟（不依赖 Phase 1、不涉 std）。

两段独立可用：
  A) Mock 契约冒烟（无需 DB、无需后端）：直连 Mock 校验响应形状
     （data.list 数组 + 每行含 bizKey），确认 API 契约与执行器约定一致。
  B) stg 落库冒烟（需 DB）：给定 source_table 与批次前缀，统计对应 stg 行数，
     确认 API 执行器入站后 stg 出数（在你已触发入站后运行）。

用法：
    # A：默认校验 Mock 的 /api/receipt
    python smoke_api_ingest.py --base-url http://127.0.0.1:8018 --path /api/receipt --token uat-mock-token
    # B：统计 stg（在触发入站之后）
    python smoke_api_ingest.py --stg mdp_stg_purchase_receipt --source-table /api/receipt

退出码：0=通过；1=失败。
"""
import argparse
import sys

import requests


def check_mock(base_url, path, token):
    url = base_url.rstrip("/") + path
    r = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=30)
    if r.status_code != 200:
        print(f"[FAIL] HTTP {r.status_code}: {r.text[:200]}", file=sys.stderr)
        return False
    body = r.json()
    rows = body.get("data", {}).get("list")
    if not isinstance(rows, list):
        print(f"[FAIL] data.list 不是数组：{body}", file=sys.stderr)
        return False
    missing = [i for i, row in enumerate(rows) if "bizKey" not in row]
    if missing:
        print(f"[FAIL] 第 {missing} 行缺 bizKey", file=sys.stderr)
        return False
    print(f"[PASS] Mock 契约 OK：{path} 返回 {len(rows)} 行，均含 bizKey。")
    return True


def check_stg(stg, source_table):
    from _db import query
    rows = query(
        f"SELECT COUNT(*) AS c FROM {stg} WHERE source_table=%s",
        (source_table,),
    )
    c = rows[0]["c"] if rows else 0
    if c > 0:
        print(f"[PASS] {stg} 中 source_table={source_table} 有 {c} 行。")
        return True
    print(f"[FAIL] {stg} 中 source_table={source_table} 无行（是否已触发入站？）", file=sys.stderr)
    return False


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--base-url", default="http://127.0.0.1:8018")
    ap.add_argument("--path", default="/api/receipt")
    ap.add_argument("--token", default="uat-mock-token")
    ap.add_argument("--stg", help="给出则做 stg 落库冒烟")
    ap.add_argument("--source-table", help="stg 冒烟时 source_table 过滤值（API 入站为 source_api_path）")
    args = ap.parse_args()

    if args.stg:
        return 0 if check_stg(args.stg, args.source_table or args.path) else 1
    return 0 if check_mock(args.base_url, args.path, args.token) else 1


if __name__ == "__main__":
    sys.exit(main())