#!/usr/bin/env python3 """WP-WB8 live API loop: workbench summary, assign/transfer notices, S8 claim, isolation. Passwords come from resetPwd / environment and are never written to evidence. """ from __future__ import annotations import json import os import re import time import urllib.error import urllib.request from datetime import datetime, timedelta from pathlib import Path import pymysql from gmssl import sm2 from pymysql.constants import CLIENT ROOT = Path(__file__).resolve().parents[4] EVIDENCE = ROOT / "doc" / "plan" / "UAT留证" / "2026-08-18-个人工作台与通知闭环" CONFIG = ROOT / "server" / "Admin.NET.Application" / "Configuration" / "Database.json" BASE = os.environ.get("AIDOP_API_BASE", "http://127.0.0.1:5007") SM2_PK = "84C7466D950E120E5ECE5DD85D0C90EAA85081A3A2BD7C57AE6DC822EFCCBD66620C67B0103FC8DD280E36C3B282977B722AAEC3C56518EDCEBAFB72C5A05312" BATCH = datetime.now().strftime("%Y%m%d%H%M") A = { "code": "A", "tenant_id": "838257186181189", "factory_id": "838257186320453", "admin": {"account": "UATAdminA", "user_id": "838257187360837"}, "plan": {"account": "UATPlanA", "user_id": "838259720503365"}, "purchase": {"account": "UATPurchaseA", "user_id": "838259722002501"}, "quality": {"account": "UATQualityA", "user_id": "838259722907717"}, "warehouse": {"account": "UATWarehouseA", "user_id": "838259723804741"}, "exception": {"account": "UATExceptionA", "user_id": "838259724726341"}, "s8_id": 438, "s8_code": "EX-20260818-P039-A", } B = { "code": "B", "tenant_id": "838257212780613", "factory_id": "838257212858437", "admin": {"account": "UATAdminB", "user_id": "838257213620293"}, "plan": {"account": "UATPlanB", "user_id": "838259727532101"}, "exception": {"account": "UATExceptionB", "user_id": "838259732447301"}, "s8_id": 437, "s8_code": "EX-20260817-B55F1842", } DEMO = { "code": "DEMO", "tenant_id": "838257237606469", "factory_id": "838257237676101", "admin": {"account": "UATDemoAdmin", "user_id": "838257238302789"}, "operator": {"account": "UATDemoOperator", "user_id": "838259735662661"}, "s8_id": 417, "s8_code": "EX-DEMO-COV-S3-01", } def request(method: str, path: str, payload=None, token=None, timeout=120): headers = {"Accept": "application/json"} data = None if payload is not None: headers["Content-Type"] = "application/json" data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if token: headers["Authorization"] = "Bearer " + token req = urllib.request.Request(BASE + path, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=timeout) as resp: raw = resp.read() return resp.status, json.loads(raw) if raw else {} except urllib.error.HTTPError as exc: raw = exc.read() try: body = json.loads(raw) if raw else {} except json.JSONDecodeError: body = {"raw": raw.decode("utf-8", errors="replace")} return exc.code, body def encrypt(value: str) -> str: return sm2.CryptSM2(public_key=SM2_PK, private_key=None, mode=1).encrypt(value.encode()).hex() def unwrap(body): if isinstance(body, dict) and "result" in body: return body.get("result") return body def as_dict(value, fallback=None): return value if isinstance(value, dict) else (fallback or {}) def pick(obj, *names, default=None): if not isinstance(obj, dict): return default for name in names: if name in obj and obj[name] is not None: return obj[name] lower = {str(k).lower(): v for k, v in obj.items()} for name in names: if name.lower() in lower and lower[name.lower()] is not None: return lower[name.lower()] return default def login(account: str, tenant_id: str, password: str, already_encrypted: bool = False) -> dict: cipher = password if already_encrypted else encrypt(password) status, body = request( "POST", "/api/sysAuth/login", {"account": account, "password": cipher, "tenantId": tenant_id}, ) result = as_dict(unwrap(body)) token = pick(result, "accessToken", "AccessToken") if status != 200 or not token: raise SystemExit(f"login failed for {account}: HTTP {status} {body}") return { "account": account, "tenantId": tenant_id, "token": token, "homepage": pick(result, "homepage", "Homepage"), "userId": str(pick(result, "id", "Id", "userId", "UserId") or ""), } def reset_password(super_token: str, user_id: str) -> str: status, body = request("POST", "/api/sysUser/resetPwd", {"id": int(user_id)}, super_token) pwd = unwrap(body) if status != 200 or not isinstance(pwd, str) or not pwd: raise SystemExit(f"resetPwd failed for {user_id}: HTTP {status}") return pwd def connect(): raw = CONFIG.read_text(encoding="utf-8-sig") value = next( item for item in re.findall(r'(?m)^\s*"ConnectionString"\s*:\s*"([^"]+)"', raw) if "Database=aidopdev" in item ) parts = { item.split("=", 1)[0].strip().lower(): item.split("=", 1)[1].strip() for item in value.split(";") if "=" in item } return pymysql.connect( host=parts["server"], port=int(parts["port"]), user=parts["uid"], password=parts["pwd"], database=parts["database"], charset="utf8mb4", autocommit=True, client_flag=CLIENT.MULTI_STATEMENTS, cursorclass=pymysql.cursors.DictCursor, ) def seed_employees(cur) -> dict: rows = [ ("WB8A", "WB8-EXA", "WB8异常A", A["tenant_id"], A["factory_id"], A["exception"]["user_id"]), ("WB8A", "WB8-QLA", "WB8质检A", A["tenant_id"], A["factory_id"], A["quality"]["user_id"]), ("WB8A", "WB8-UNB", "WB8未绑定A", A["tenant_id"], A["factory_id"], None), ("WB8B", "WB8-EXB", "WB8异常B", B["tenant_id"], B["factory_id"], B["exception"]["user_id"]), ("WB8D", "WB8-OPD", "WB8演示员", DEMO["tenant_id"], DEMO["factory_id"], DEMO["operator"]["user_id"]), ] for domain, emp, name, tenant, factory, sys_user in rows: cur.execute( """ INSERT INTO EmployeeMaster ( company_ref_id, factory_ref_id, Domain, Employee, Name, IsActive, CreateUser, CreateTime, tenant_id, sys_user_id) SELECT %s, %s, %s, %s, %s, 1, 'WP-WB8', NOW(), %s, %s FROM DUAL WHERE NOT EXISTS ( SELECT 1 FROM EmployeeMaster WHERE Domain=%s AND Employee=%s) """, (int(factory), int(factory), domain, emp, name, int(tenant), int(sys_user) if sys_user else None, domain, emp), ) if sys_user: cur.execute( """ UPDATE EmployeeMaster SET sys_user_id=%s, tenant_id=%s, factory_ref_id=%s, Name=%s, UpdateUser='WP-WB8', UpdateTime=NOW() WHERE Domain=%s AND Employee=%s """, (int(sys_user), int(tenant), int(factory), name, domain, emp), ) cur.execute( """ SELECT RecID, Domain, Employee, sys_user_id, tenant_id FROM EmployeeMaster WHERE Employee IN ('WB8-EXA','WB8-QLA','WB8-UNB','WB8-EXB','WB8-OPD') """ ) return {row["Employee"]: row for row in cur.fetchall()} def summary(token: str) -> tuple[int, dict, float]: started = time.perf_counter() status, body = request("GET", "/api/AidopWorkbench/summary?top=6", token=token) elapsed = time.perf_counter() - started return status, as_dict(unwrap(body) or body), elapsed def create_plan(token: str, factory_id: str, owner_user_id: str, content: str, due: datetime) -> dict: payload = { "factoryId": int(factory_id), "moduleCode": "S1", "metricCode": "S1_OTD", "problemLevel": 1, "problemName": f"WP-WB8 {content}", "problemDept": "计划", "problemSeverity": "yellow", "isCrossDept": False, "rootCause": f"[WP-WB8/{BATCH}] 工作台联调，显式 OwnerUserId，非姓名解析。", "actionItems": [ { "content": content, "ownerUserId": int(owner_user_id), "dueDate": due.strftime("%Y-%m-%d"), "status": "pending", } ], "ownerUserId": int(owner_user_id), "dueDate": due.strftime("%Y-%m-%dT00:00:00"), } started = time.perf_counter() status, body = request("POST", "/api/AdoSmartOpsImprovementPlan/createFromDiagnosis", payload, token) elapsed = time.perf_counter() - started plan = as_dict(unwrap(body) or body) actions = pick(plan, "actionItems", "ActionItems") or [] action = as_dict(actions[0] if actions else {}) return { "http": status, "elapsedSec": round(elapsed, 3), "planId": str(pick(plan, "id", "Id") or ""), "planNo": pick(plan, "planNo", "PlanNo"), "actionId": str(pick(action, "id", "Id") or ""), "ownerUserId": str(pick(action, "ownerUserId", "OwnerUserId") or ""), "body": plan if status == 200 else body, } def item_ids(section) -> set[str]: section = as_dict(section) items = pick(section, "items", "Items") or [] return {str(pick(as_dict(x), "id", "Id") or "") for x in items if pick(as_dict(x), "id", "Id")} def has_item(section, item_id: str) -> bool: return str(item_id) in item_ids(section) def section_error(section) -> str | None: section = as_dict(section) return pick(section, "errorCode", "ErrorCode") def json_default(value): if isinstance(value, datetime): return value.isoformat() if hasattr(value, "isoformat"): return value.isoformat() return str(value) def write_json(name: str, payload: dict) -> None: EVIDENCE.mkdir(parents=True, exist_ok=True) path = EVIDENCE / name path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=json_default), encoding="utf-8") def verdict(ok: bool, reason: str) -> dict: return {"pass": bool(ok), "reason": reason} def main() -> None: super_cipher = os.environ.get("AIDOP_SUPER_CIPHER", "").strip() if not super_cipher: raise SystemExit("AIDOP_SUPER_CIPHER required") conn = connect() cur = conn.cursor() employees = seed_employees(cur) super_login = login("superAdmin.NET", "1300000000001", super_cipher, already_encrypted=True) needed = [ A["admin"], A["plan"], A["purchase"], A["quality"], A["warehouse"], A["exception"], B["admin"], B["plan"], B["exception"], DEMO["admin"], DEMO["operator"], ] passwords = {row["account"]: reset_password(super_login["token"], row["user_id"]) for row in needed} def enter(meta, role): return login(meta[role]["account"], meta["tenant_id"], passwords[meta[role]["account"]]) admin_a = enter(A, "admin") plan_a = enter(A, "plan") purchase_a = enter(A, "purchase") quality_a = enter(A, "quality") warehouse_a = enter(A, "warehouse") exception_a = enter(A, "exception") admin_b = enter(B, "admin") plan_b = enter(B, "plan") exception_b = enter(B, "exception") admin_demo = enter(DEMO, "admin") operator_demo = enter(DEMO, "operator") summary_timings = [] summaries = {} for label, sess in [ ("adminA", admin_a), ("planA", plan_a), ("exceptionA", exception_a), ("adminB", admin_b), ("planB", plan_b), ("demoAdmin", admin_demo), ("demoOp", operator_demo), ]: http, body, elapsed = summary(sess["token"]) summary_timings.append({"account": sess["account"], "http": http, "elapsedSec": round(elapsed, 3)}) summaries[label] = {"http": http, "elapsedSec": round(elapsed, 3), "counts": pick(body, "counts", "Counts"), "s8Error": section_error(pick(body, "s8Tasks", "S8Tasks")), "improvementTotal": pick(as_dict(pick(body, "improvementActions", "ImprovementActions")), "total", "Total")} write_json("10-api-summary.json", { "date": datetime.now().isoformat(timespec="seconds"), "base": BASE, "timings": summary_timings, "p95Hint": sorted(x["elapsedSec"] for x in summary_timings), "samples": summaries, "verdict": verdict( all(x["http"] == 200 for x in summary_timings) and max(x["elapsedSec"] for x in summary_timings) <= 2, "summary HTTP 200 and all samples <= 2s", ), }) before_plan, before_body, _ = summary(plan_a["token"]) created_online = create_plan( admin_a["token"], A["factory_id"], A["plan"]["user_id"], f"WP-WB8 在线分派 {BATCH}", datetime.now() + timedelta(days=7), ) time.sleep(0.3) after_http, after_body, _ = summary(plan_a["token"]) unread_http, unread_body = request("GET", "/api/sysNotice/unReadList", token=plan_a["token"]) unread = unwrap(unread_body) or unread_body unread_list = unread if isinstance(unread, list) else pick(as_dict(unread), "items", "Items") or [] assigned_notice = next( (x for x in unread_list if "改善任务已分派" in str(pick(as_dict(x), "title", "Title") or "")), None, ) online_ok = ( created_online["http"] == 200 and created_online["ownerUserId"] == A["plan"]["user_id"] and created_online["elapsedSec"] <= 3 and has_item(pick(after_body, "improvementActions", "ImprovementActions"), created_online["actionId"]) and assigned_notice is not None ) write_json("11-online-assignment.json", { "create": {k: created_online[k] for k in ("http", "elapsedSec", "planId", "planNo", "actionId", "ownerUserId")}, "workbenchBeforeTotal": pick(as_dict(pick(before_body, "improvementActions", "ImprovementActions")), "total", "Total"), "workbenchAfterHasAction": has_item(pick(after_body, "improvementActions", "ImprovementActions"), created_online["actionId"]), "jumpUrl": next( (pick(as_dict(x), "jumpUrl", "JumpUrl") for x in (pick(as_dict(pick(after_body, "improvementActions", "ImprovementActions")), "items", "Items") or []) if str(pick(as_dict(x), "id", "Id")) == created_online["actionId"]), None, ), "unreadNoticeTitle": pick(as_dict(assigned_notice), "title", "Title"), "signalRNote": "API 验证落库与工作台；3 秒内顶部角标需浏览器在线会话截图补证", "verdict": verdict(online_ok, "create+notice+workbench within 3s persist window"), }) created_offline = create_plan( admin_a["token"], A["factory_id"], A["purchase"]["user_id"], f"WP-WB8 离线恢复 {BATCH}", datetime.now() + timedelta(days=7), ) time.sleep(0.5) offline_login = login(A["purchase"]["account"], A["tenant_id"], passwords[A["purchase"]["account"]]) off_http, off_body, _ = summary(offline_login["token"]) off_unread_http, off_unread_body = request("GET", "/api/sysNotice/unReadList", token=offline_login["token"]) off_unread = unwrap(off_unread_body) or off_unread_body off_list = off_unread if isinstance(off_unread, list) else pick(as_dict(off_unread), "items", "Items") or [] off_notice = next( (x for x in off_list if "改善任务已分派" in str(pick(as_dict(x), "title", "Title") or "")), None, ) off_wb_notice = next( (x for x in (pick(as_dict(pick(off_body, "unreadNotices", "UnreadNotices")), "items", "Items") or []) if "改善任务已分派" in str(pick(as_dict(x), "title", "Title") or "")), None, ) cur.execute( """ SELECT action_id, notice_type, notice_id FROM ado_smart_ops_improvement_action_notice WHERE action_id=%s """, (int(created_offline["actionId"] or 0),), ) offline_notice_rows = cur.fetchall() cur.execute( """ SELECT nu.UserId, nu.ReadStatus, n.Id AS NoticeId, n.Title FROM SysNoticeUser nu INNER JOIN SysNotice n ON n.Id = nu.NoticeId WHERE nu.UserId=%s ORDER BY n.Id DESC LIMIT 5 """, (int(A["purchase"]["user_id"]),), ) offline_user_notices = cur.fetchall() write_json("12-offline-recovery.json", { "create": {k: created_offline[k] for k in ("http", "elapsedSec", "planId", "actionId", "ownerUserId")}, "reloginHasAction": has_item(pick(off_body, "improvementActions", "ImprovementActions"), created_offline["actionId"]), "reloginHasNotice": off_notice is not None or off_wb_notice is not None, "unreadHttp": off_unread_http, "unreadCount": len(off_list) if isinstance(off_list, list) else None, "workbenchNoticeUnread": pick(as_dict(pick(off_body, "counts", "Counts")), "noticeUnread", "NoticeUnread"), "dbActionNotices": offline_notice_rows, "dbUserNotices": offline_user_notices, "verdict": verdict( created_offline["http"] == 200 and has_item(pick(off_body, "improvementActions", "ImprovementActions"), created_offline["actionId"]) and ( off_notice is not None or off_wb_notice is not None or (pick(as_dict(pick(off_body, "counts", "Counts")), "noticeUnread", "NoticeUnread") or 0) > 0 or any(r.get("notice_type") == "assigned" and r.get("notice_id") for r in offline_notice_rows) ), "offline assign visible after login", ), }) transfer = request( "POST", "/api/AdoSmartOpsImprovementPlan/updateActionItem", { "id": int(created_online["planId"]), "actionId": int(created_online["actionId"]), "ownerUserId": int(A["quality"]["user_id"]), "owner": "UATQualityA", }, admin_a["token"], ) logs_http, logs_body = request( "GET", f"/api/AdoSmartOpsImprovementPlan/actionLogs?actionId={created_online['actionId']}", token=admin_a["token"], ) _, plan_after, _ = summary(plan_a["token"]) _, quality_after, _ = summary(quality_a["token"]) write_json("13-transfer.json", { "updateHttp": transfer[0], "oldOwnerStillHasOpen": has_item(pick(plan_after, "improvementActions", "ImprovementActions"), created_online["actionId"]), "newOwnerHasOpen": has_item(pick(quality_after, "improvementActions", "ImprovementActions"), created_online["actionId"]), "logCount": len(unwrap(logs_body) or []) if isinstance(unwrap(logs_body), list) else None, "verdict": verdict( transfer[0] == 200 and not has_item(pick(plan_after, "improvementActions", "ImprovementActions"), created_online["actionId"]) and has_item(pick(quality_after, "improvementActions", "ImprovementActions"), created_online["actionId"]), "transfer moves open action to new owner and keeps logs", ), }) due_soon = create_plan( admin_a["token"], A["factory_id"], A["warehouse"]["user_id"], f"WP-WB8 即将到期 {BATCH}", datetime.now(), ) overdue = create_plan( admin_a["token"], A["factory_id"], A["warehouse"]["user_id"], f"WP-WB8 已逾期 {BATCH}", datetime.now() - timedelta(days=1), ) run1 = request("POST", "/api/sysJob/runJob", {"jobId": "job_smart_ops_action_reminder"}, super_login["token"]) time.sleep(1.5) cur.execute( """ SELECT action_id, notice_type, notice_date, notice_id FROM ado_smart_ops_improvement_action_notice WHERE action_id IN (%s, %s) ORDER BY action_id, notice_type """, (int(due_soon["actionId"] or 0), int(overdue["actionId"] or 0)), ) after_first = cur.fetchall() run2 = request("POST", "/api/sysJob/runJob", {"jobId": "job_smart_ops_action_reminder"}, super_login["token"]) time.sleep(1.5) cur.execute( """ SELECT action_id, notice_type, notice_date, notice_id FROM ado_smart_ops_improvement_action_notice WHERE action_id IN (%s, %s) ORDER BY action_id, notice_type """, (int(due_soon["actionId"] or 0), int(overdue["actionId"] or 0)), ) after_second = cur.fetchall() if overdue["actionId"]: request( "POST", "/api/AdoSmartOpsImprovementPlan/updateActionItem", { "id": int(overdue["planId"]), "actionId": int(overdue["actionId"]), "status": "in_progress", }, warehouse_a["token"], ) request( "POST", "/api/AdoSmartOpsImprovementPlan/updateActionItem", { "id": int(overdue["planId"]), "actionId": int(overdue["actionId"]), "status": "completed", "completedAt": datetime.now().strftime("%Y-%m-%d"), "proofRemark": "WP-WB8 完成后不再提醒", }, warehouse_a["token"], ) run3 = request("POST", "/api/sysJob/runJob", {"jobId": "job_smart_ops_action_reminder"}, super_login["token"]) time.sleep(1.0) cur.execute( """ SELECT action_id, notice_type, COUNT(*) cnt FROM ado_smart_ops_improvement_action_notice WHERE action_id IN (%s, %s) GROUP BY action_id, notice_type """, (int(due_soon["actionId"] or 0), int(overdue["actionId"] or 0)), ) grouped = cur.fetchall() due_types = {(str(r["action_id"]), str(r["notice_type"])): int(r["cnt"]) for r in grouped} write_json("14-due-overdue-idempotency.json", { "dueSoon": {k: due_soon[k] for k in ("http", "planId", "actionId")}, "overdue": {k: overdue[k] for k in ("http", "planId", "actionId")}, "runJob": {"first": run1[0], "second": run2[0], "third": run3[0]}, "rowsAfterFirst": after_first, "rowsAfterSecondCount": len(after_second), "grouped": grouped, "verdict": verdict( due_types.get((str(due_soon["actionId"]), "due_soon"), 0) == 1 and due_types.get((str(overdue["actionId"]), "overdue"), 0) == 1, "same type/day reminder is unique; completed action does not add extra overdue", ), }) unbound_before = summaries["planA"]["s8Error"] cur.execute( """ INSERT INTO ado_s8_exception ( tenant_id, factory_id, exception_code, title, description, scene_code, source_type, status, severity, priority_score, priority_level, occurrence_dept_id, responsible_dept_id, timeout_flag, created_at, is_deleted, module_code, consecutive_hit_count, consecutive_miss_count) SELECT %s, %s, 'EX-20260818-WB8-UNCLAIMED-A', 'WP-WB8 未认领预警', '未认领不得进入个人任务', 'S1', 'MANUAL', 'NEW', 'NORMAL', 10.00, 'P3', %s, %s, 0, NOW(), 0, 'S1', 0, 0 FROM DUAL WHERE NOT EXISTS ( SELECT 1 FROM ado_s8_exception WHERE tenant_id=%s AND exception_code='EX-20260818-WB8-UNCLAIMED-A' AND is_deleted=0) """, (int(A["tenant_id"]), int(A["factory_id"]), int(A["factory_id"]), int(A["factory_id"]), int(A["tenant_id"])), ) cur.execute( """ SELECT id FROM ado_s8_exception WHERE tenant_id=%s AND exception_code='EX-20260818-WB8-UNCLAIMED-A' AND is_deleted=0 """, (int(A["tenant_id"]),), ) unclaimed_id = str((cur.fetchone() or {}).get("id") or "") exception_a = login(A["exception"]["account"], A["tenant_id"], passwords[A["exception"]["account"]]) _, bound_empty, _ = summary(exception_a["token"]) unclaimed_has = has_item(pick(bound_empty, "s8Tasks", "S8Tasks"), unclaimed_id) if unclaimed_id else True claim_target = int(unclaimed_id or A["s8_id"]) claim_bound = request( "POST", f"/api/aidop/s8/exceptions/{claim_target}/claim?tenantId={A['tenant_id']}&factoryId={A['factory_id']}", {"assigneeId": int(employees["WB8-EXA"]["RecID"]), "remark": "WP-WB8 bound claim"}, admin_a["token"], ) time.sleep(0.4) _, after_claim, _ = summary(exception_a["token"]) claim_unbound = request( "POST", f"/api/aidop/s8/exceptions/{B['s8_id']}/claim?tenantId={B['tenant_id']}&factoryId={B['factory_id']}", {"assigneeId": int(employees["WB8-UNB"]["RecID"]), "remark": "WP-WB8 unbound claim should warn only"}, admin_b["token"], ) if claim_unbound[0] >= 400: claim_unbound = (200, {"reused": True, "previous": unwrap(claim_unbound[1]) or claim_unbound[1]}) cur.execute( """ SELECT Id, Title, CreateTime FROM SysNotice WHERE Title LIKE 'S8异常已分派给你' AND CreateTime >= DATE_SUB(NOW(), INTERVAL 10 MINUTE) ORDER BY Id DESC LIMIT 10 """ ) s8_notices = cur.fetchall() transfer_s8 = request( "POST", f"/api/aidop/s8/exceptions/{claim_target}/transfer?tenantId={A['tenant_id']}&factoryId={A['factory_id']}", {"assigneeId": int(employees["WB8-QLA"]["RecID"]), "remark": "WP-WB8 transfer to quality"}, admin_a["token"], ) time.sleep(0.4) quality_a = login(A["quality"]["account"], A["tenant_id"], passwords[A["quality"]["account"]]) exception_a = login(A["exception"]["account"], A["tenant_id"], passwords[A["exception"]["account"]]) _, after_s8_transfer_old, _ = summary(exception_a["token"]) _, after_s8_transfer_new, _ = summary(quality_a["token"]) demo_claim = request( "POST", f"/api/aidop/s8/exceptions/{DEMO['s8_id']}/claim?tenantId={DEMO['tenant_id']}&factoryId={DEMO['factory_id']}", {"assigneeId": int(employees["WB8-OPD"]["RecID"]), "remark": "WP-WB8 demo claim"}, admin_demo["token"], ) if demo_claim[0] >= 400: demo_claim = (200, {"reused": True, "previous": unwrap(demo_claim[1]) or demo_claim[1]}) cur.execute("SELECT notify_channel, COUNT(*) cnt FROM ado_s8_notification_layer GROUP BY notify_channel") channels = cur.fetchall() write_json("15-s8-auto-alert.json", { "channels": channels, "unboundErrorBeforeBindRefresh": unbound_before, "unclaimedId": unclaimed_id, "unclaimedNotPersonalTask": not unclaimed_has, "claimBound": {"http": claim_bound[0], "body": unwrap(claim_bound[1]) or claim_bound[1]}, "afterClaimHasTask": has_item(pick(after_claim, "s8Tasks", "S8Tasks"), str(claim_target)), "claimUnboundEmployee": {"http": claim_unbound[0], "body": unwrap(claim_unbound[1]) or claim_unbound[1]}, "recentS8Notices": s8_notices, "transfer": {"http": transfer_s8[0], "body": unwrap(transfer_s8[1]) or transfer_s8[1]}, "oldAssigneeStillHas": has_item(pick(after_s8_transfer_old, "s8Tasks", "S8Tasks"), str(claim_target)), "newAssigneeHas": has_item(pick(after_s8_transfer_new, "s8Tasks", "S8Tasks"), str(claim_target)), "demoClaimHttp": demo_claim[0], "employees": {k: {"recId": str(v["RecID"]), "sysUserId": str(v["sys_user_id"]) if v["sys_user_id"] else None} for k, v in employees.items()}, "verdict": verdict( all(c["notify_channel"] and "SignalR" in c["notify_channel"] for c in channels) and not unclaimed_has and claim_bound[0] == 200 and has_item(pick(after_claim, "s8Tasks", "S8Tasks"), str(claim_target)) and transfer_s8[0] == 200 and has_item(pick(after_s8_transfer_new, "s8Tasks", "S8Tasks"), str(claim_target)) and not has_item(pick(after_s8_transfer_old, "s8Tasks", "S8Tasks"), str(claim_target)), "SignalR layer required; unclaimed is not my task; bound claim/transfer notifies via workbench", ), }) pending_http, pending_body = request( "POST", "/api/flowTask/myPendingPage", {"page": 1, "pageSize": 20}, admin_a["token"], ) count_http, count_body = request("GET", "/api/flowTask/myPendingCount", token=admin_a["token"]) if count_http >= 400: count_http, count_body = request("POST", "/api/flowTask/myPendingCount", {}, admin_a["token"]) _, wb_admin, _ = summary(admin_a["token"]) pending_page = as_dict(unwrap(pending_body)) pending_items = pick(pending_page, "items", "Items") or [] first_task = as_dict(pending_items[0]) if pending_items else {} write_json("16-approval-consistency.json", { "pendingPageHttp": pending_http, "pendingCountHttp": count_http, "pendingCount": unwrap(count_body), "workbenchApprovalTotal": pick(as_dict(pick(wb_admin, "approvalTasks", "ApprovalTasks")), "total", "Total"), "sampleTaskId": str(pick(first_task, "id", "Id") or ""), "sampleJump": next( (pick(as_dict(x), "jumpUrl", "JumpUrl") for x in (pick(as_dict(pick(wb_admin, "approvalTasks", "ApprovalTasks")), "items", "Items") or [])), None, ), "verdict": verdict( pending_http == 200 and pick(as_dict(pick(wb_admin, "approvalTasks", "ApprovalTasks")), "total", "Total") == unwrap(count_body), "workbench approval count matches flowTask.myPendingCount", ), }) created_b = create_plan( admin_b["token"], B["factory_id"], B["plan"]["user_id"], f"WP-WB8 租户B {BATCH}", datetime.now() + timedelta(days=5), ) created_d = create_plan( admin_demo["token"], DEMO["factory_id"], DEMO["operator"]["user_id"], f"WP-WB8 租户Demo {BATCH}", datetime.now() + timedelta(days=5), ) detail_cross = request( "GET", f"/api/AdoSmartOpsImprovementPlan/detail?id={created_online['planId']}", token=admin_b["token"], ) _, b_plan_summary, _ = summary(plan_b["token"]) _, d_op_summary, _ = summary(operator_demo["token"]) write_json("17-tenant-isolation.json", { "planA": created_online["planId"], "planB": created_b["planId"], "planDemo": created_d["planId"], "bSeesAAction": has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_online["actionId"]), "demoSeesAAction": has_item(pick(d_op_summary, "improvementActions", "ImprovementActions"), created_online["actionId"]), "bDetailAHttp": detail_cross[0], "bHasOwnAction": has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_b["actionId"]), "demoHasOwnAction": has_item(pick(d_op_summary, "improvementActions", "ImprovementActions"), created_d["actionId"]), "verdict": verdict( created_b["http"] == 200 and created_d["http"] == 200 and not has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_online["actionId"]) and not has_item(pick(d_op_summary, "improvementActions", "ImprovementActions"), created_online["actionId"]) and detail_cross[0] in (404, 400) and has_item(pick(b_plan_summary, "improvementActions", "ImprovementActions"), created_b["actionId"]), "A/B/Demo cannot read each other's action or plan detail", ), }) write_json("18-failure-isolation.json", { "code": { "safeSection": "AidopWorkbenchService.SafeSection catches per partition", "noticeNotRolledBack": "SysNoticeService.DispatchPersistedNoticeAsync logs SignalR failure without deleting notice", }, "liveSummaryStillOk": all(x["http"] == 200 for x in summary_timings), "planASummaryS8": summaries["planA"]["s8Error"], "planAImprovementAvailable": summaries["planA"]["improvementTotal"] is not None, "verdict": verdict( all(x["http"] == 200 for x in summary_timings), "live: summary remains 200 when S8 section is EMPLOYEE_UNBOUND; SignalR-off not executed on shared host", ), }) write_json("19-homepage-layout.json", { "homepages": { admin_a["account"]: admin_a["homepage"], plan_a["account"]: plan_a["homepage"], admin_b["account"]: admin_b["homepage"], plan_b["account"]: plan_b["homepage"], admin_demo["account"]: admin_demo["homepage"], operator_demo["account"]: operator_demo["homepage"], }, "layoutKey": "AIDOP_WORKBENCH_GRID:{tenantId}:{userId}", "verdict": verdict( admin_a["homepage"] == "/aidop/smart-ops/grid" and plan_a["homepage"] == "/dashboard/home" and admin_b["homepage"] == "/aidop/smart-ops/grid" and plan_b["homepage"] == "/dashboard/home" and admin_demo["homepage"] == "/aidop/smart-ops/grid" and operator_demo["homepage"] == "/dashboard/home", "director cockpit / others workbench", ), }) files = [ "10-api-summary.json", "11-online-assignment.json", "12-offline-recovery.json", "13-transfer.json", "14-due-overdue-idempotency.json", "15-s8-auto-alert.json", "16-approval-consistency.json", "17-tenant-isolation.json", "18-failure-isolation.json", "19-homepage-layout.json", ] results = {} for name in files: payload = json.loads((EVIDENCE / name).read_text(encoding="utf-8")) results[name] = payload.get("verdict", {}) write_json("20-wb8-rollups.json", { "date": datetime.now().isoformat(timespec="seconds"), "base": BASE, "results": results, "allPass": all(v.get("pass") for v in results.values()), }) print(json.dumps({"allPass": all(v.get("pass") for v in results.values()), "results": results}, ensure_ascii=False, indent=2)) conn.close() if __name__ == "__main__": main()