You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
153 lines
9.8 KiB
153 lines
9.8 KiB
# -*- coding: utf-8 -*-
|
|
"""票16 审计 · API 探针第2轮：权限面 / 状态机边界 / 动态 actionName / 卡重复绑定的落库形态"""
|
|
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):
|
|
data = urllib.parse.urlencode(body).encode() if body is not None else None
|
|
headers = {"Authorization": "Bearer " + tok} if tok else {}
|
|
if data: headers["Content-Type"] = "application/x-www-form-urlencoded;charset=UTF-8"
|
|
r = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(r, timeout=20) as resp:
|
|
return json.loads(resp.read().decode("utf-8", "replace"))
|
|
except urllib.error.HTTPError as e:
|
|
try: return json.loads(e.read().decode("utf-8", "replace"))
|
|
except Exception: return {"__status": e.code}
|
|
except Exception as e:
|
|
return {"__error": str(e)}
|
|
|
|
def P(case, cond, note=True):
|
|
print(("PASS " if cond else "!! ") + case + ((" | " + str(note)[:260]) if note is not True else ""))
|
|
return cond
|
|
|
|
TOKD, TOKS = token(DIRECTOR), token(SALES)
|
|
|
|
# ---------- A. 跨用户权限面（总监建，销售操作） ----------
|
|
nm = "A5AUD-权限-" + time.strftime("%H%M%S")
|
|
j = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)
|
|
pid = j["data"]
|
|
P("A1 总监建档成功", j.get("code") == 0, str(j)[:120])
|
|
|
|
j = _req("POST", "/api/project/filing/review", {"id": pid, "approved": "true"}, tok=TOKS)
|
|
P("A2 无关销售可报备审核（原型：仅报备专员+管理员）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("POST", "/api/project/filing/review", {"id": pid, "approved": "true"}, tok=TOKD)
|
|
P("A3 总监报备审核通过", j.get("code") == 0)
|
|
|
|
j = _req("POST", "/api/project/stage/advance", {"id": pid}, tok=TOKS)
|
|
P("A4 无关销售可推进他人项目（原型：确认通过限负责人/销售管理员）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
|
|
j = _req("GET", "/api/project/detail?id=" + pid, tok=TOKS)
|
|
P("A5 无关销售可读他人项目详情（数据权限）", j.get("code") == 0, f"code={j.get('code')}")
|
|
j = _req("GET", "/api/project/list?scope=manage¤t=1&size=50", tok=TOKS)
|
|
rows = (j.get("data") or {}).get("content") or []
|
|
P("A6 普通销售 manage 可见他人项目（DataScope 应限部门）", all(str(r.get("ownerUserId")) == SALES for r in rows), f"rows={len(rows)} leak={[r.get('ownerUserId') for r in rows][:5]}")
|
|
j = _req("GET", "/api/project/board?scope=manage", tok=TOKS)
|
|
cols = (j.get("data") or {}).get("columns") or []
|
|
leak = [c["id"] for c in (cols[1].get("cards") or []) if str(c.get("ownerUserId")) == str(DIRECTOR)] if len(cols) > 1 else []
|
|
P("A7 普通销售 manage 看板可见总监项目", not leak, f"stage1 leak={leak}")
|
|
|
|
j = _req("POST", "/api/project/close", {"id": pid, "closeType": "MANUAL"}, tok=TOKS)
|
|
P("A8 无关销售可关闭他人项目", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
|
|
# ---------- B. 状态机边界 ----------
|
|
nm = "A5AUD-边界-" + time.strftime("%H%M%S")
|
|
j = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)
|
|
pidB = j["data"]
|
|
_req("POST", "/api/project/filing/review", {"id": pidB, "approved": "true"}, tok=TOKD)
|
|
|
|
j = _req("POST", "/api/project/stage/advance", {"id": pidB, "skipBidding": "true"}, tok=TOKD)
|
|
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
|
|
P("B1 stage1 带 skipBidding 不应连跳（唯一连跳在 stage3）", d.get("projectStage") == 2, f"stage={d.get('projectStage')} code={j.get('code')}")
|
|
|
|
_req("POST", "/api/project/stage/advance", {"id": pidB}, tok=TOKD) # 2→3
|
|
j = _req("POST", "/api/project/stage/advance", {"id": pidB, "skipBidding": "false"}, tok=TOKD)
|
|
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
|
|
P("B2 stage3 skipBidding=false 走 3→4", d.get("projectStage") == 4, f"stage={d.get('projectStage')}")
|
|
|
|
j = _req("POST", "/api/project/stage/advance", {"id": pidB}, tok=TOKD) # 4→5
|
|
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
|
|
P("B3 4→5 推进", d.get("projectStage") == 5, f"stage={d.get('projectStage')} code={j.get('code')}")
|
|
|
|
j = _req("POST", "/api/project/stage/advance", {"id": pidB}, tok=TOKD) # stage5 未标记不允许推进→6
|
|
P("B4 stage5 未标记时推进被拒（不许线性 5→6）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
|
|
P("B5 stage5 未标记可回退（spec 一致性）", True, "")
|
|
j = _req("GET", "/api/project/stage/rollback-targets?id=" + pidB, tok=TOKD)
|
|
tg = sorted(t["stage"] for t in (j.get("data") or []))
|
|
P("B6 走满 1-4 后可选节点={1,2,3,4}", tg == [1, 2, 3, 4], str(tg))
|
|
|
|
_req("POST", "/api/project/bid-result/mark", {"id": pidB, "bidResult": "2"}, tok=TOKD) # 输单
|
|
j = _req("POST", "/api/project/filing/review", {"id": pidB, "approved": "true"}, tok=TOKD)
|
|
P("B7 结项后报备审核被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("POST", "/api/project/follow/add", {"projectId": pidB, "followContent": "终态后跟进", "followTime": "2026-09-10 11:00:00"}, tok=TOKD)
|
|
P("B8 结项终态后跟进新增（原型已关闭全部只读）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
|
|
# 关闭后再关闭 / 非法 closeType
|
|
nm = "A5AUD-关闭-" + time.strftime("%H%M%S")
|
|
pidC = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)["data"]
|
|
j = _req("POST", "/api/project/close", {"id": pidC, "closeType": "NOT_A_TYPE"}, tok=TOKD)
|
|
P("B9 非法 closeType 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
_req("POST", "/api/project/close", {"id": pidC, "closeType": "MANUAL", "closeReason": "r", "closeRemark": "m"}, tok=TOKD)
|
|
j = _req("POST", "/api/project/close", {"id": pidC, "closeType": "MANUAL"}, tok=TOKD)
|
|
P("B10 已关闭不可重复关闭", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("POST", "/api/project/stage/advance", {"id": pidC}, tok=TOKD)
|
|
P("B11 已关闭不可推进", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("POST", "/api/project/bid-result/mark", {"id": pidC, "bidResult": "1"}, tok=TOKD)
|
|
P("B12 已关闭不可标记赢单", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
|
|
# ---------- C. 参数健壮性 ----------
|
|
j = _req("GET", "/api/project/board", tok=TOKD)
|
|
P("C1 board 缺 scope 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("GET", "/api/project/board?scope=all", tok=TOKD)
|
|
P("C2 board 非法 scope 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("GET", "/api/project/detail", tok=TOKD)
|
|
P("C3 detail 缺 id 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("GET", "/api/project/detail?id=999", tok=TOKD)
|
|
P("C4 detail 不存在 id → 68002", j.get("code") == 68002, str(j.get("message")))
|
|
j = _req("GET", "/api/project/list?scope=manage¤t=0&size=0", tok=TOKD)
|
|
P("C5 list 分页参数退化不炸", j.get("code") == 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
j = _req("POST", "/api/project/follow/add", {"projectId": pidC, "followContent": "x", "followTime": "垃圾时间"}, tok=TOKD)
|
|
P("C6 跟进非法时间格式被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
|
|
|
|
# ---------- D. 动态 actionName 全集 + 跟进是否留痕 ----------
|
|
j = _req("GET", "/api/project/timeline/list?projectId=" + pidB + "¤t=1&size=50", tok=TOKD)
|
|
ev = [(t.get("action"), t.get("actionName"), t.get("fromStage"), t.get("toStage"), t.get("opDesc")) for t in (j.get("data") or {}).get("content") or []]
|
|
print("TIMELINE pidB:")
|
|
for e in ev: print(" ", e)
|
|
acts = [e[0] for e in ev]
|
|
P("D1 输单留痕 BID_LOSE", "BID_LOSE" in acts, str(acts))
|
|
j = _req("GET", "/api/project/timeline/list?projectId=" + pidC + "¤t=1&size=50", tok=TOKD)
|
|
evC = [(t.get("action"), t.get("actionName")) for t in (j.get("data") or {}).get("content") or []]
|
|
print("TIMELINE pidC:", evC)
|
|
|
|
# 跟进新增是否写动态（用一个新项目验证）
|
|
nm = "A5AUD-动态-" + time.strftime("%H%M%S")
|
|
pidD = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)["data"]
|
|
_req("POST", "/api/project/follow/add", {"projectId": pidD, "followContent": "留痕验证", "followTime": "2026-09-10 12:00:00"}, tok=TOKD)
|
|
j = _req("GET", "/api/project/timeline/list?projectId=" + pidD + "¤t=1&size=50", tok=TOKD)
|
|
actsD = [t.get("action") for t in (j.get("data") or {}).get("content") or []]
|
|
P("D2 跟进新增写项目动态", any("FOLLOW" in str(a) for a in actsD), str(actsD))
|
|
|
|
# ---------- E. 卡重复绑定后的卡归属形态（读 DB） ----------
|
|
try:
|
|
import pymysql
|
|
conn = pymysql.connect(host="8.129.84.155", port=3306, user="root", password="Itc@123456", database="itc_crm", charset="utf8mb4")
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT id, owner_type, owner_id, customer_id, delete_key FROM opportunity_scheme_card WHERE id=%s", (CARD,))
|
|
print("CARD ROW:", cur.fetchall())
|
|
cur.execute("SELECT table_schema FROM information_schema.schemata WHERE table_schema LIKE '%%crm%%' LIMIT 5")
|
|
print("SCHEMAS:", cur.fetchall())
|
|
conn.close()
|
|
except Exception as e:
|
|
print("DB CHECK SKIP:", e)
|
|
|