From 02b904164e43ab20f4603f4fb5ccf6d8dcc5afa1 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Sat, 12 Sep 2026 09:05:52 +0800
Subject: [PATCH] =?UTF-8?q?test(mining):=20=E9=A1=B9=E7=9B=AE=E9=9A=94?=
 =?UTF-8?q?=E7=A6=BBE2E=E8=84=9A=E6=9C=AC=E2=80=94=E2=80=9411=E9=A1=B9?=
 =?UTF-8?q?=E6=96=AD=E8=A8=80(=E6=89=B9=E6=AC=A1=E8=90=BD=E9=A1=B9?=
 =?UTF-8?q?=E7=9B=AE/=E8=B7=A8=E9=A1=B9=E7=9B=AE=E6=89=B9=E6=AC=A1+?=
 =?UTF-8?q?=E7=B1=BB=E5=88=AB+=E6=98=8E=E7=BB=86=E4=B8=89=E5=B1=82?=
 =?UTF-8?q?=E6=8B=92=E7=BB=9D/B=E5=88=97=E8=A1=A8=E4=B8=8D=E5=90=ABA/?=
 =?UTF-8?q?=E5=AD=98=E9=87=8F=E5=BD=92=E5=B9=B3=E5=8F=B0=E7=BA=A7/?=
 =?UTF-8?q?=E6=97=A0=E9=A1=B9=E7=9B=AE=E4=BC=9A=E8=AF=9D=E5=8F=AA=E8=A7=81?=
 =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E7=BA=A7)=E5=85=A8=E8=BF=87?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 scripts/p1_proj_e2e.py | 91 ++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 91 insertions(+)
 create mode 100644 scripts/p1_proj_e2e.py

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