test(mining): 项目隔离E2E脚本——11项断言(批次落项目/跨项目批次+类别+明细三层拒绝/B列表不含A/存量归平台级/无项目会话只见平台级)全过

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())