diff --git a/scripts/p1_proj_e2e.py b/scripts/p1_proj_e2e.py new file mode 100644 index 0000000..cc137a8 --- /dev/null +++ b/scripts/p1_proj_e2e.py @@ -0,0 +1,91 @@ +# -*- coding:utf-8 -*- +"""P1 项目隔离 E2E:批次落项目归属 + 跨项目不可见 + 平台级本机构可见 + 存量归平台级。""" +import asyncio, json, os, sys, time +W="/d/pipeline/pipeline-app"; os.chdir(W); sys.path.insert(0,W) +from appPublic.folderUtils import ProgramPath +from appPublic.jsonConfig import getConfig +from appPublic.event_dispatcher import EventDispatcher +from sqlor.dbpools import DBPools +from ahserver.serverenv import ServerEnv +from ahserver.globalEnv import initEnv +p=ProgramPath(); c=getConfig(W,{"workdir":W,"ProgramPath":p}); DBPools(c.databases) +se=ServerEnv(); se.event_dispatcher=EventDispatcher(); se.get_module_dbname=lambda m:"pipeline" +initEnv() +PID_A=os.environ.get("P1_PROJECT_ID","") +FAIL=[] +def check(n,cond,d=""): + print("[%s] %s %s"%("OK " if cond else "FAIL",n,str(d)[:180])) + if not cond: FAIL.append(n) + +async def m(): + from pipeline_opportunity import opp_ability + ab=opp_ability.register_opp_ability() + # 找同机构的第二个商机项目做跨项目测试 + async with DBPools().sqlorContext("pipeline") as sor: + r=await sor.sqlExe("SELECT id FROM sd_projects WHERE pipeline_id='opportunity_general' AND id<>${a}$ LIMIT 1",{"a":PID_A}) + await sor.sqlExe("COMMIT",{}) + PID_B=r[0].id if r else "" + print("项目A:",PID_A,"项目B:",PID_B or "(无第二项目,跳过跨项目实测)") + ctxA={"project_id":PID_A,"user_id":"user-01","org_id":"0","pipeline_id":"opportunity_general", + "space":"opportunity_general","session_id":"e2e-proj","workspace_dir":"/tmp"} + async def call(t,p_,ctx): + async with DBPools().sqlorContext("pipeline") as sor: + return await ab.handlers[t](sor,p_,ctx) + + # 1. 项目A启动 targeted 批次(小数据快) + out=await call("opp_start_mining",{"scope":"targeted","keyword":"公众号","days":3650},ctxA) + check("A项目启动",out.startswith("OK:"),out[:100]) + bid=out.split("batch_id=")[1].split("(")[0] + t0=time.time() + while time.time()-t0<300: + d=json.loads(await call("opp_mining_status",{"batch_id":bid},ctxA)) + if d.get("status") in ("done","failed"): break + await asyncio.sleep(10) + check("批次done",d.get("status")=="done",d.get("status")+" "+str(d.get("error_msg"))[:80]) + + # 2. 落库 project_id 验证 + async with DBPools().sqlorContext("pipeline") as sor: + r=await sor.sqlExe("SELECT project_id FROM opp_mining_batches WHERE id=${b}$",{"b":bid}) + await sor.sqlExe("COMMIT",{}) + check("批次落项目A",str(r[0].project_id or "")==PID_A,r[0].project_id) + + # 3. 项目B查A的批次 → 拒绝 + if PID_B: + ctxB=dict(ctxA); ctxB["project_id"]=PID_B + s=await call("opp_mining_status",{"batch_id":bid},ctxB) + check("跨项目查批次被拒",s.startswith("ERROR:") and "其他项目" in s,s[:120]) + s2=await call("opp_list_clusters",{"batch_id":bid},ctxB) + check("跨项目查类别被拒",s2.startswith("ERROR:"),s2[:120]) + # B 的列表不含 A 的批次 + lst=json.loads(await call("opp_mining_status",{},ctxB)) + ids=[x["id"] for x in (lst if isinstance(lst,list) else [])] + check("B列表不含A批次",bid not in ids,ids[:5]) + + # 4. A 的列表含本批次 + 存量平台级批次(project_id='') + lst=json.loads(await call("opp_mining_status",{},ctxA)) + ids=[x["id"] for x in (lst if isinstance(lst,list) else [])] + check("A列表含本批次",bid in ids,ids[:5]) + plat=[x for x in (lst if isinstance(lst,list) else []) if not str(x.get("project_id") or "")] + check("存量批次归平台级可见",len(plat)>=1,"平台级批次数=%d"%len(plat)) + + # 5. 无项目会话只见平台级(不见A的批次) + ctx0=dict(ctxA); ctx0["project_id"]="" + lst0=json.loads(await call("opp_mining_status",{},ctx0)) + ids0=[x["id"] for x in (lst0 if isinstance(lst0,list) else [])] + check("无项目会话不见A批次",bid not in ids0 and all(not str(x.get("project_id") or "") for x in (lst0 if isinstance(lst0,list) else [])),ids0[:5]) + s=await call("opp_mining_status",{"batch_id":bid},ctx0) + check("无项目会话查A批次被拒",s.startswith("ERROR:"),s[:100]) + + # 6. cluster_detail 经批次可见性(项目B查A批次里的类别) + if PID_B: + cl=json.loads(await call("opp_list_clusters",{"batch_id":bid},ctxA)) + rows=cl.get("排名") or [] + if rows: + ctxB=dict(ctxA); ctxB["project_id"]=PID_B + s=await call("opp_cluster_detail",{"cluster_id":rows[0]["cluster_id"]},ctxB) + check("跨项目查类内明细被拒",s.startswith("ERROR:"),s[:120]) + + print("\n%s (%d fail)"%("PROJ-ISOLATION PASS" if not FAIL else "FAIL: "+",".join(FAIL),len(FAIL))) + sys.exit(1 if FAIL else 0) + +asyncio.run(m())