crm-backend-matt/.scratch/crm-project/audit/probe_api.py

# -*- coding: utf-8 -*-
"""票16 黑盒审计 · API 层探针（只读 + 自建数据操作，不改产品代码）
产出: probe-result.json 供 audit-defects.md 引用
"""
import io, json, sys, time, urllib.request, urllib.parse, urllib.error

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
BASE = "http://localhost:8080"
AUD = r"E:\code\crm-backend-matt\.scratch\crm-project\audit"
DIRECTOR, SALES = "99021002", "99021003"
CUSTOMER, CARD = "99021001", "99021008"

def token(uid):
    j = _req("GET", "/api/auth/debug/token?userId=" + uid)
    d = j["data"]
    return d if isinstance(d, str) else d["token"]

def _req(method, path, body=None, tok=None, ctype="application/x-www-form-urlencoded;charset=UTF-8"):
    url = BASE + path
    data = None
    headers = {}
    if body is not None:
        data = urllib.parse.urlencode(body).encode()
        headers["Content-Type"] = ctype
    if tok:
        headers["Authorization"] = "Bearer " + tok
    r = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(r, timeout=20) as resp:
            raw = resp.read().decode("utf-8", "replace")
            try:
                return json.loads(raw)
            except Exception:
                return {"__raw": raw[:300], "__status": resp.status}
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", "replace")
        try:
            return json.loads(raw)
        except Exception:
            return {"__raw": raw[:300], "__status": e.code}
    except Exception as e:
        return {"__error": str(e)}

results = []
def rec(case, got, expect, note=""):
    ok = got == expect if not isinstance(expect, (list, tuple, set)) else got in expect
    results.append({"case": case, "got": got, "expect": str(expect), "ok": bool(ok), "note": str(note)[:400]})
    print(("PASS " if ok else "!!   ") + case + " | got=" + str(got)[:160] + ((" | " + note) if note else ""))

TOKD = token(DIRECTOR)
TOKS = token(SALES)

# ---------- 0. 信封形态 ----------
j = _req("GET", "/api/project/board?scope=manage", tok=TOKD)
rec("envelope keys = code/success/message/data", sorted(j.keys()), ["code", "data", "message", "success"])
rec("board.conflictPendingCount 类型", type(j["data"]["conflictPendingCount"]).__name__, ["int", "str"], "原型为数字;若str则前端依赖宽松")

# ---------- 1. 建档校验 ----------
cases = [
    ("缺项目名称", {"customerId": CUSTOMER, "schemeCardId": CARD}),
    ("缺客户", {"projectName": "A5AUD-x", "schemeCardId": CARD}),
    ("缺方案卡", {"projectName": "A5AUD-x", "customerId": CUSTOMER}),
    ("卡客户不匹配", {"projectName": "A5AUD-x", "customerId": "99999001", "schemeCardId": CARD}),
    ("客户不存在", {"projectName": "A5AUD-x", "customerId": "88888888", "schemeCardId": CARD}),
    ("方案卡不存在", {"projectName": "A5AUD-x", "customerId": CUSTOMER, "schemeCardId": "77777777"}),
]
for name, body in cases:
    j = _req("POST", "/api/project/create", body, tok=TOKD)
    rec("create 校验:「" + name + "」被拒", (j.get("code"), j.get("success")), ("not0", False) if j.get("code") else j,
        f"code={j.get('code')} msg={j.get('message')}")

# 成功建档 → 落库形态
nm = "A5AUD-主流程-" + time.strftime("%H%M%S")
j = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD,
                                          "projectAmount": "1200000", "brand": "itc", "regionCode": "440113"}, tok=TOKD)
rec("create 合法入参成功", j.get("code"), 0, str(j)[:200])
PID = j.get("data")

# PROJECT 卡重复绑定：同卡同客户再建一个项目
j2 = _req("POST", "/api/project/create", {"projectName": "A5AUD-重复卡-" + time.strftime("%H%M%S"),
                                            "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)
rec("create 同方案卡重复绑定（一客户一卡铁律）", j2.get("code"), ["not0_or_0待复核"], f"code={j2.get('code')} msg={j2.get('message')}")
PID_DUP = j2.get("data") if j2.get("code") == 0 else None

# 落库形态
j = _req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD)
d = j.get("data") or {}
rec("建项目落库 stage=0", d.get("projectStage"), 0)
rec("建项目落库 status=1 进行中", d.get("projectStatus"), 1)
rec("建项目落库 filingStatus=1 审核中", d.get("filingStatus"), 1)
rec("建项目落库 bidResult 未定", d.get("bidResult"), [0, None])
rec("detail 含 ownerNameSnapshot", bool(d.get("ownerNameSnapshot")), True, str(d.get("ownerNameSnapshot")))
rec("detail 含 customerName", bool(d.get("customerName")), True, str(d.get("customerName")))
print("DETAIL KEYS:", sorted(d.keys()))

# ---------- 2. 无权限/未登录 ----------
j = _req("GET", "/api/project/detail?id=" + str(PID))
rec("未登录读 detail → 401 信封", (j.get("code"), j.get("success")), ("not0", False))
j = _req("GET", "/api/project/detail/" + str(PID), tok=TOKD)
rec("@PathVariable 形态 /project/detail/{id} 不存在", j.get("code"), ["not0", 404, "not_found_404"], str(j)[:120])
j = _req("PUT", "/api/project/detail?id=" + str(PID), {"projectName": "x"}, tok=TOKD)
rec("PUT 方法被拒（只 POST 写）", j.get("__status", j.get("code")), [405, "not0"], str(j)[:120])
j = _req("DELETE", "/api/project/follow/list?projectId=" + str(PID), tok=TOKD)
rec("DELETE 方法被拒", j.get("__status", j.get("code")), [405, "not0"], str(j)[:120])

# ---------- 3. 列表/分页/筛选 ----------
j = _req("GET", "/api/project/list?scope=manage&current=1&size=2", tok=TOKD)
p = j.get("data") or {}
rec("list 分页结构含 content/total", ("content" in p and "total" in p), True, str(sorted(p.keys())))
rec("list size=2 生效", len((p.get("content") or [])), 2)
for st in ["0", "1", "2", "3", "4", "5", "6"]:
    jj = _req("GET", f"/api/project/list?scope=manage&stage={st}&current=1&size=50", tok=TOKD)
    rows = (jj.get("data") or {}).get("content") or []
    bad = [r for r in rows if str(r.get("projectStage")) != st]
    rec(f"list 阶段筛 stage={st} 只回该阶段", len(bad), 0, f"rows={len(rows)}")
j = _req("GET", "/api/project/list?scope=manage&stage=7", tok=TOKD)
rec("list 非法阶段 stage=7", j.get("code"), ["not0", 0], f"code={j.get('code')} msg={j.get('message')}")
j = _req("GET", "/api/project/list?scope=manage&keyword=A5AUD&current=1&size=50", tok=TOKD)
rows = (j.get("data") or {}).get("content") or []
rec("list keyword 筛命中自建项目", any(r.get("projectName", "").startswith("A5AUD") for r in rows), True, f"rows={len(rows)}")
j = _req("GET", "/api/project/list?scope=manage&filingStatus=1&current=1&size=50", tok=TOKD)
rows = (j.get("data") or {}).get("content") or []
bad = [r for r in rows if r.get("filingStatus") != 1]
rec("list 报备独立筛 filing=1", len(bad), 0, f"rows={len(rows)}")
j = _req("GET", "/api/project/list?scope=mine&current=1&size=50", tok=TOKS)
rows = (j.get("data") or {}).get("content") or []
rec("普通销售 mine 空或只含本人", all((r.get("ownerUserId") in (None, int(SALES))) for r in rows), True, f"rows={len(rows)}")

# ---------- 4. 状态机主流程（在 PID 上） ----------
# 报备驳回 → filing=3 停 stage0
j = _req("POST", "/api/project/filing/review", {"id": PID, "approved": "false", "remark": "材料不齐"}, tok=TOKD)
rec("报备驳回成功", j.get("code"), 0, str(j)[:150])
d = (_req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD).get("data") or {})
rec("驳回后 filing=3 且 stage 仍 0", (d.get("filingStatus"), d.get("projectStage")), (3, 0))
# 再审通过 → filing=2 + 0→1
j = _req("POST", "/api/project/filing/review", {"id": PID, "approved": "true", "remark": "复审通过"}, tok=TOKD)
rec("驳回后可再审通过", j.get("code"), 0, str(j)[:150])
d = (_req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD).get("data") or {})
rec("通过后 filing=2 + stage=1 双写", (d.get("filingStatus"), d.get("projectStage")), (2, 1))
# stage0 时回退 targets 应为空
j = _req("GET", "/api/project/stage/rollback-targets?id=" + str(0) + "00", tok=TOKD)
rec("rollback-targets 非法 id 报错", j.get("code"), "not0", str(j)[:120])

# 推进 1→2→3
for st in (1, 2):
    j = _req("POST", "/api/project/stage/advance", {"id": PID}, tok=TOKD)
    rec(f"推进 {st}→{st+1}", j.get("code"), 0, str(j)[:150])
d = (_req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD).get("data") or {})
rec("推进至 stage3", d.get("projectStage"), 3)

# stage3 推进须带 skipBidding 或普通推进 3→4
j = _req("POST", "/api/project/stage/advance", {"id": PID, "skipBidding": "true"}, tok=TOKD)
rec("3→5 连跳（无需招投标）", j.get("code"), 0, str(j)[:150])
d = (_req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD).get("data") or {})
rec("连跳后 stage=5", d.get("projectStage"), 5)
# 连跳后回退可选节点 = {1,2,3}（stage4 不在列）
j = _req("GET", "/api/project/stage/rollback-targets?id=" + str(PID), tok=TOKD)
targets = [t.get("stage") for t in (j.get("data") or [])]
rec("连跳后可选节点={1,2,3} 无4", targets, [1, 2, 3], str(targets))

# 回退到 2 → 再推回 5
j = _req("POST", "/api/project/stage/rollback", {"id": PID, "targetStage": "2"}, tok=TOKD)
rec("多步回退 5→2（总监）", j.get("code"), 0, str(j)[:150])
d = (_req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD).get("data") or {})
rec("回退后 stage=2", d.get("projectStage"), 2)
_req("POST", "/api/project/stage/advance", {"id": PID}, tok=TOKD)      # 2→3
_req("POST", "/api/project/stage/advance", {"id": PID, "skipBidding": "true"}, tok=TOKD)  # 3→5

# 回退越界：targetStage=0
j = _req("POST", "/api/project/stage/rollback", {"id": PID, "targetStage": "0"}, tok=TOKD)
rec("回退下限 stage1（target 0 被拒）", j.get("code"), "not0", f"code={j.get('code')} msg={j.get('message')}")

# 权限：普通销售回退 68005
j = _req("POST", "/api/project/stage/rollback", {"id": PID, "targetStage": "2"}, tok=TOKS)
rec("普通销售回退被拒 68005", j.get("code"), 68005, str(j.get("message")))

# stage5 未标记时先试 mark 非法值
j = _req("POST", "/api/project/bid-result/mark", {"id": PID, "bidResult": "9"}, tok=TOKD)
rec("bidResult 非法值被拒", j.get("code"), "not0", f"code={j.get('code')} msg={j.get('message')}")

# 标记赢单 → stage5 闭环 status 仍 1
j = _req("POST", "/api/project/bid-result/mark", {"id": PID, "bidResult": "1"}, tok=TOKD)
rec("标记赢单成功", j.get("code"), 0, str(j)[:150])
d = (_req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD).get("data") or {})
rec("赢单: stage 仍 5 + status 仍 1 + bidResult=1", (d.get("projectStage"), d.get("projectStatus"), d.get("bidResult")), (5, 1, 1))
j = _req("POST", "/api/project/stage/rollback", {"id": PID, "targetStage": "2"}, tok=TOKD)
rec("赢单锁死回退 68003", j.get("code"), 68003, str(j.get("message")))
j = _req("POST", "/api/project/bid-result/mark", {"id": PID, "bidResult": "2"}, tok=TOKD)
rec("赢单后不可改输单（锁）", j.get("code"), "not0", f"code={j.get('code')} msg={j.get('message')}")
# 赢单项目可否关闭？
j = _req("POST", "/api/project/close", {"id": PID, "closeType": "MANUAL", "closeReason": "audit"}, tok=TOKD)
rec("赢单项目关闭（spec 未禁）", j.get("code"), ["not0", 0], f"code={j.get('code')} msg={j.get('message')}")
d = (_req("GET", "/api/project/detail?id=" + str(PID), tok=TOKD).get("data") or {})
rec("关闭后 status=2", d.get("projectStatus"), 2)

# ---------- 5. 输单分叉（用 PID_DUP 或新建） ----------
if PID_DUP:
    pid2 = PID_DUP
    _req("POST", "/api/project/filing/review", {"id": pid2, "approved": "true"}, tok=TOKD)
    for _ in range(2):
        _req("POST", "/api/project/stage/advance", {"id": pid2}, tok=TOKD)
    _req("POST", "/api/project/stage/advance", {"id": pid2, "skipBidding": "true"}, tok=TOKD)
    j = _req("POST", "/api/project/bid-result/mark", {"id": pid2, "bidResult": "2"}, tok=TOKD)
    rec("标记输单 → stage6 结项", j.get("code"), 0, str(j)[:150])
    d = (_req("GET", "/api/project/detail?id=" + str(pid2), tok=TOKD).get("data") or {})
    rec("输单: stage=6 status=3 bidResult=2", (d.get("projectStage"), d.get("projectStatus"), d.get("bidResult")), (6, 3, 2))
    j = _req("POST", "/api/project/stage/rollback", {"id": pid2, "targetStage": "1"}, tok=TOKD)
    rec("结项终态回退被拒", j.get("code"), "not0", f"code={j.get('code')} msg={j.get('message')}")
    j = _req("POST", "/api/project/close", {"id": pid2, "closeType": "MANUAL"}, tok=TOKD)
    rec("已结项不可再关闭", j.get("code"), "not0", f"code={j.get('code')} msg={j.get('message')}")

# ---------- 6. 动态事件流 ----------
j = _req("GET", f"/api/project/timeline/list?projectId={PID}&current=1&size=50", tok=TOKD)
events = [ (t.get("action") or t.get("actionName")) for t in ((j.get("data") or {}).get("content") or []) ]
print("TIMELINE:", events)
for want in ["创建", "报备"]:
    rec("动态含「" + want + "」事件", any(want in str(e) for e in events), True, str(events)[:200])

# ---------- 7. 页签读写 ----------
j = _req("POST", "/api/project/follow/add", {"projectId": PID, "followContent": "审计跟进", "followTime": "2026-09-10 10:00:00"}, tok=TOKD)
rec("跟进新增", j.get("code"), 0, str(j)[:150])
j = _req("POST", "/api/project/follow/add", {"projectId": PID, "followContent": ""}, tok=TOKD)
rec("跟进缺内容被拒", j.get("code"), "not0", f"code={j.get('code')} msg={j.get('message')}")
j = _req("POST", "/api/project/attachment/add", {"projectId": PID, "fileId": "1", "bizType": "PROJECT"}, tok=TOKD)
rec("附件挂载(fileId=1)", j.get("code"), ["not0", 0], f"code={j.get('code')} msg={j.get('message')}")
aid = (j.get("data") or {}).get("id") if isinstance(j.get("data"), dict) else j.get("data")
j = _req("GET", f"/api/project/attachment/list?projectId={PID}", tok=TOKD)
rec("附件列表可读", j.get("code"), 0, str(j)[:120])
for sub in ["scheme", "quote", "bidding", "bid-open"]:
    j = _req("GET", f"/api/project/{sub}?projectId={PID}", tok=TOKD)
    rec(f"GET /project/{sub} 可读", j.get("code"), 0, str(j)[:120])
j = _req("GET", f"/api/project/customer/list?projectId={PID}", tok=TOKD)
cl = j.get("data") or []
rec("客户页签=单客户(一项目一客户)", (j.get("code"), len(cl)), (0, 1), str(cl)[:200])

out = {"results": results, "PID": PID, "PID_DUP": PID_DUP}
open(AUD + r"\probe-result.json", "w", encoding="utf-8").write(json.dumps(out, ensure_ascii=False, indent=1))
fails = [r for r in results if not r["ok"]]
print(f"\n==== {len(results)-len(fails)}/{len(results)} PASS, {len(fails)} 偏差 ====")