92 lines
4.7 KiB
Python
92 lines
4.7 KiB
Python
# -*- 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())
|