From 96844f0002ea5f4aae43e01cf255fe7ed6f1d8b6 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Thu, 3 Sep 2026 23:01:33 +0800 Subject: [PATCH] =?UTF-8?q?test(knowledge):=20=E7=94=9F=E4=BA=A7E2E?= =?UTF-8?q?=E8=B6=8A=E6=9D=83=E9=AA=8C=E8=AF=81=E8=84=9A=E6=9C=AC=E5=85=A5?= =?UTF-8?q?=E5=BA=93=20verify=5Fknowledge=5Fpartition=5Fe2e.py=20(entity-m?= =?UTF-8?q?emory-partition)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../scripts/verify_knowledge_partition_e2e.py | 137 ++++++++++++++++++ 1 file changed, 137 insertions(+) create mode 100644 backend/scripts/verify_knowledge_partition_e2e.py diff --git a/backend/scripts/verify_knowledge_partition_e2e.py b/backend/scripts/verify_knowledge_partition_e2e.py new file mode 100644 index 00000000..54a4e0bf --- /dev/null +++ b/backend/scripts/verify_knowledge_partition_e2e.py @@ -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)