test(knowledge): 生产E2E越权验证脚本入库 verify_knowledge_partition_e2e.py (entity-memory-partition)

2026-09-03 23:01:33 +08:00
parent e22e1702f8
commit 96844f0002
1 changed files with 137 additions and 0 deletions
@@ -0,0 +1,137 @@
+"""CMA 企业记忆分区——生产 E2E 越权验证（2026-09-03）
+
+流程：DB插标记文章(全局/E1专属/E2专属) → 登E1/E2互查 → 断言 → 清理回查
+标记: 【系统验证临时数据-验证后删除】
+"""
+import sys
+
+env = {}
+with open('/root/cma-management/backend/.env') as f:
+    for line in f:
+        line = line.strip()
+        if line and not line.startswith('#') and '=' in line:
+            k, v = line.split('=', 1)
+            env[k.strip()] = v.strip().strip('"').strip("'")
+
+import pymysql
+conn = pymysql.connect(
+    host=env.get('CMA_DB_HOST', '127.0.0.1'),
+    port=int(env.get('CMA_DB_PORT', '3306')),
+    user=env.get('CMA_DB_USER', 'cma_user'),
+    password=env.get('CMA_DB_PASS', 'cma_pass_2026'),
+    database=env.get('CMA_DB_NAME', 'cma'),
+    charset='utf8mb4',
+)
+cur = conn.cursor()
+
+MARK = "【系统验证临时数据-验证后删除】"
+ok = True
+def check(name, cond, extra=""):
+    global ok
+    print(("  ✅" if cond else "  ❌"), name, extra)
+    if not cond:
+        ok = False
+
+# 0. 三表列+索引已存在（DB 回查）
+for t in ['knowledge_events', 'knowledge_summaries', 'knowledge_articles']:
+    cur.execute(f"SHOW COLUMNS FROM {t} LIKE 'entity_id'")
+    has_col = cur.fetchone() is not None
+    cur.execute(f"SHOW INDEX FROM {t} WHERE Column_name='entity_id'")
+    has_idx = cur.fetchone() is not None
+    check(f"{t}.entity_id 列+索引", has_col and has_idx)
+# 存量全局文章（NULL）在加列后保持 38 篇
+cur.execute("SELECT COUNT(*) FROM knowledge_articles WHERE entity_id IS NULL")
+g = cur.fetchone()[0]
+print(f"  存量全局文章(NULL): {g} 篇")
+check("存量全局教学文章保持全局共享", g == 38, f"(期望38)")
+
+# 1. 插临时验证文章（标题带标记，source 专属 entity）
+cur.execute("DELETE FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",))
+conn.commit()
+cur.execute(
+    "INSERT INTO knowledge_articles (entity_id, title, summary, content, category) VALUES (%s,%s,%s,%s,%s)",
+    (1, f"{MARK}E1专属越权测试", "s", "c", "term"))
+cur.execute(
+    "INSERT INTO knowledge_articles (entity_id, title, summary, content, category) VALUES (%s,%s,%s,%s,%s)",
+    (2, f"{MARK}E2专属越权测试", "s", "c", "term"))
+cur.execute(
+    "INSERT INTO knowledge_articles (entity_id, title, summary, content, category) VALUES (%s,%s,%s,%s,%s)",
+    (None, f"{MARK}全局共享测试", "s", "c", "term"))
+conn.commit()
+cur.execute("SELECT id, entity_id, title FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",))
+rows = cur.fetchall()
+by_eid = {r[1]: (r[0], r[2]) for r in rows}
+ids = {r[1]: r[0] for r in rows}
+print(f"  插入临时文章: {rows}")
+
+# 2. 登录 E1/E2 调 API
+import urllib.request, json as _json
+
+def api(method, path, token=None, body=None):
+    url = f"http://127.0.0.1:8010{path}"
+    req = urllib.request.Request(url, method=method)
+    req.add_header("Content-Type", "application/json")
+    if token:
+        req.add_header("Authorization", f"Bearer {token}")
+    if body is not None:
+        req.data = _json.dumps(body).encode()
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            return resp.status, resp.read().decode()
+    except urllib.error.HTTPError as e:
+        return e.code, e.read().decode()
+
+def login(eid):
+    st, body = api("POST", "/api/cma/auth/login", body={"username": "admin", "password": "admin123", "entity_id": eid})
+    if st != 200:
+        print(f"  ⚠️ 登录 entity{eid} 失败 {st}: {body[:200]}")
+        return None
+    return _json.loads(body)["token"]
+
+t1 = login(1)
+t2 = login(2)
+if not t1 or not t2:
+    print("登录失败，跳过 API 断言")
+    sys.exit(1)
+
+# E1 视角
+st, body = api("GET", "/api/cma/knowledge-articles", token=t1)
+d1 = _json.loads(body)
+titles1 = {i["title"] for i in d1.get("data", [])}
+check("E1 见全局共享文章", f"{MARK}全局共享测试" in titles1)
+check("E1 见自己专属文章", f"{MARK}E1专属越权测试" in titles1)
+check("E1 不见 E2 专属文章", f"{MARK}E2专属越权测试" not in titles1, "(核心越权1)")
+# E2 视角
+st, body = api("GET", "/api/cma/knowledge-articles", token=t2)
+titles2 = {i["title"] for i in _json.loads(body).get("data", [])}
+check("E2 见全局共享文章", f"{MARK}全局共享测试" in titles2)
+check("E2 见自己专属文章", f"{MARK}E2专属越权测试" in titles2)
+check("E2 不见 E1 专属文章", f"{MARK}E1专属越权测试" not in titles2, "(核心越权2)")
+
+# E1 直取 E2 专属文章详情 → 404
+st, body = api("GET", f"/api/cma/knowledge-articles/{ids[2]}", token=t1)
+check("E1 直取 E2 专属文章 → 404", st == 404, f"(st={st})")
+# E2 直取 E1 专属文章详情 → 404
+st, body = api("GET", f"/api/cma/knowledge-articles/{ids[1]}", token=t2)
+check("E2 直取 E1 专属文章 → 404", st == 404, f"(st={st})")
+
+# 摘要接口（生产空表）两账套 200 不 500
+for eid, tok in ((1, t1), (2, t2)):
+    st, body = api("GET", "/api/cma/knowledge/summaries", token=tok)
+    d = _json.loads(body)
+    check(f"E{eid} /knowledge/summaries 200 + 空列表", st == 200 and d.get("total", -1) == 0, f"(st={st} total={d.get('total')})")
+    st, body = api("GET", "/api/cma/knowledge/events", token=tok)
+    check(f"E{eid} /knowledge/events 200", st == 200, f"(st={st})")
+
+# 3. 清理 + 回查
+cur.execute("DELETE FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",))
+conn.commit()
+cur.execute("SELECT COUNT(*) FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",))
+left = cur.fetchone()[0]
+check("临时数据清理回查 = 0", left == 0, f"(剩{left})")
+cur.execute("SELECT COUNT(*) FROM knowledge_articles")
+print(f"  knowledge_articles 最终行数: {cur.fetchone()[0]}")
+
+conn.close()
+print("\n" + ("全部通过 ✅" if ok else "存在失败 ❌"))
+sys.exit(0 if ok else 1)