dingtalk-workspace-cli/scripts/run_shortcut_real_write_matrix.py

#!/usr/bin/env python3
"""Run write/high-risk shortcuts against the real CLI with safe negative inputs.

This is intentionally not a mock and not a dry-run.  It launches the built CLI
for every write/high-risk shortcut and records the real stdout/stderr/exit code.

To avoid broad irreversible side effects, target identifiers and people are
rewritten to obviously-nonexistent test values.  Commands that have no
resource/user target and would necessarily mutate the current account state are
held and recorded rather than executed.
"""

from __future__ import annotations

import datetime as dt
import argparse
import json
import os
import shlex
import subprocess
import time
from pathlib import Path

from shortcut_real_result import (
    classify_failure,
    sanitize_result,
    summarize_failure_categories,
    classify_real_status,
    summarize_results,
)


ROOT = Path(__file__).resolve().parents[1]
MATRIX_PATH = Path("/private/tmp/dws-shortcut-matrix.json")
OUT_PATH = ROOT / "docs" / "shortcut-real-write-results.json"
BIN = os.environ.get("DWS_REAL_TEST_BIN", "/private/tmp/dws-real-test")

NO_SAFE_NEGATIVE_TARGET = {
    ("chat", "+conversation-clear-all-red-point"):
        "该命令没有目标 ID，会直接修改当前账号所有会话红点状态；需要逐项授权后单独跑。",
    ("chat", "+category-create"):
        "该命令只需要标题，真实执行会在当前账号创建会话分类；需要逐项授权后单独跑。",
}


def ensure_matrix() -> dict:
    if not MATRIX_PATH.exists():
        env = os.environ.copy()
        env.setdefault("GOCACHE", "/private/tmp/dws_gocache")
        raw = subprocess.check_output(
            ["go", "run", "./scripts/gen_shortcut_test_matrix.go"],
            cwd=ROOT,
            env=env,
            text=True,
        )
        MATRIX_PATH.write_text(raw, encoding="utf-8")
    return json.loads(MATRIX_PATH.read_text(encoding="utf-8"))


def replace_flag_values(args: list[str], now: str, service: str = "", command: str = "") -> list[str]:
    future_start = (
        dt.datetime.now() + dt.timedelta(days=7)
    ).replace(hour=10, minute=0, second=0, microsecond=0).isoformat() + "+08:00"
    future_end = (
        dt.datetime.now() + dt.timedelta(days=7)
    ).replace(hour=10, minute=15, second=0, microsecond=0).isoformat() + "+08:00"

    no_user = "__DWS_SHORTCUT_REAL_TEST_NO_SUCH_USER__"
    no_id = "DWSREALTESTNOSUCHID0000000000000"
    no_conv = "cidDWSREALTESTNOSUCHCONV"
    huge_num = "999999999999"
    text = f"DWS shortcut 真实测试 {now}，安全负向输入，请忽略。"

    replacements = {
        # People / users.
        "to": no_user,
        "with": no_user,
        "owner": no_user,
        "target": no_user,
        "user": no_user,
        "users": no_user,
        "user-ids": no_user,
        "new-owner": no_user,
        "applicant": no_user,
        "inviter": no_user,
        "approver-user-id": no_user,
        "add-users": no_user,
        "remove-users": "",
        "add-extra-users": "",
        "remove-extra-users": "",
        "at-user-ids": no_user,
        "at-users": no_user,
        "at-open-dingtalk-ids": "",
        "open-dingtalk-ids": "",
        # Conversation / message / Ding targets.
        "group": no_conv,
        "id": no_id,
        "conversation-id": no_conv,
        "open-conversation-id": no_conv,
        "src-conversation-id": no_conv,
        "dest-conversation-id": no_conv,
        "message-id": no_id,
        "msg-id": no_id,
        "msg-ids": no_id,
        "src-msg-id": no_id,
        "src-thread-id": no_id,
        "keys": no_id,
        "biz-id": no_id,
        "category-ids": huge_num,
        "receiver": no_user,
        "robot-code": "__DWS_SHORTCUT_REAL_TEST_NO_SUCH_ROBOT__",
        "bot-id": no_id,
        "token": "__DWS_SHORTCUT_REAL_TEST_NO_SUCH_WEBHOOK_TOKEN__",
        # DingTalk document / app / aitable / wiki ids.
        "base-id": no_id,
        "base": no_id,
        "table-id": no_id,
        "table": no_id,
        "field-id": no_id,
        "record-id": huge_num,
        "record-ids": no_id,
        "view-id": no_id,
        "dashboard-id": no_id,
        "chart-id": no_id,
        "workflow-id": no_id,
        "role-id": no_id,
        "role-ids": no_id,
        "section-id": no_id,
        "parent-section-id": no_id,
        "new-parent-section-id": no_id,
        "node-id": no_id,
        "node": no_id,
        "doc": no_id,
        "folder": no_id,
        "workspace": no_id,
        "template-id": no_id,
        "comment-key": no_id,
        "block-id": no_id,
        "unified-app-id": no_id,
        "icon-media-id": no_id,
        "media-id": no_id,
        "session-id": no_id,
        "import-id": no_id,
        "space": "__DWS_SHORTCUT_REAL_TEST_NO_SUCH_SPACE__",
        # Numeric identifiers.
        "category-id": huge_num,
        "record-id-num": huge_num,
        "record-id-int": huge_num,
        "class-id": huge_num,
        "group-id": huge_num,
        "plan-id": huge_num,
        "result-id": huge_num,
        # Text-like values.
        "title": f"DWS shortcut 真实测试 {now}",
        "name": f"DWS shortcut 真实测试 {now}",
        "new-name": f"DWS shortcut 真实测试 {now}",
        "alias-title": f"DWS shortcut 真实测试 {now}",
        "nick": f"DWS测试{now}",
        "desc": "DWS shortcut 真实测试描述，可删除",
        "description": "DWS shortcut 真实测试描述，可删除",
        "field-description": "DWS shortcut 真实测试字段说明",
        "text": text,
        "content": text,
        "comment": text,
        "note": text,
        "remark": text,
        "selected-text": "DWS测试",
        "task": f"DWS shortcut 真实测试待办 {now}",
        "keyword": f"__DWS_SHORTCUT_REAL_TEST_NO_SUCH_APPROVAL_{now}__",
        "reason": "DWS shortcut 真实测试安全负向清理",
        "uuid": f"dws-shortcut-real-negative-{now}",
        # Times.
        "due": future_end,
        "at": future_start,
        "start": future_start,
        "end": future_end,
        "time": future_start,
        # Structured values.
        "config": "{}",
        "layout": "{}",
        "json": "{}",
        "records": "[]",
        "ai-config": "{}",
        "field-mapping": "{}",
        "class-vo": '{"sections":[{"times":[{"checkType":"OnDuty","checkTime":"09:00","across":0},{"checkType":"OffDuty","checkTime":"18:00","across":0}]}]}',
        "group-vo": '{"defaultClassId":999999999999,"workDayClassList":[999999999999,999999999999,999999999999,999999999999,999999999999,0,0]}',
        "visibility-rules": '[{"type":"dept","visible":["-1"]}]',
        "sub-roles": "[]",
        "skills": "[]",
        "scope-values": "[]",
        "event-codes": "[]",
        "schedules": '[{"userId":"202397","workDate":"2026-07-16 09:00:00","classId":999999999999,"isRest":"N"}]',
        "class-ids": "[999999999999]",
        "add-depts": "",
        "remove-depts": "",
        # URLs / app config.
        "homepage-url": "https://example.invalid/dws-shortcut-real-test",
        "pc-homepage-url": "https://example.invalid/dws-shortcut-real-test",
        "omp-url": "https://example.invalid/dws-shortcut-real-test",
        "outgoing-url": "https://example.invalid/dws-shortcut-real-test",
        "event-callback-url": "https://example.invalid/dws-shortcut-real-test",
        "redirect-urls": "https://example.invalid/dws-shortcut-real-test",
        "sso-urls": "https://example.invalid/dws-shortcut-real-test",
        "ip-whitelist": "127.0.0.1",
        "url": "https://example.invalid/dws-shortcut-real-test",
        # Misc.
        "type": "FIXED",
        "record-name-key": "task",
        "file-name": f"dws-shortcut-real-{now}.txt",
        "mime-type": "text/plain",
        "version": "0.0.1-dws-real-test",
        "version-id": no_id,
        "member-type": "USER",
        "emotion-id": no_id,
        "emotion-name": "DWS测试表情",
        "background-id": no_id,
        "emoji": "👍",
        "pair": "测试=>测试",
        "leave-code": no_id,
        "num": "1",
    }
    if service == "chat" and command == "+messages-send-card":
        replacements["receiver"] = "103262"
    if service == "chat" and command == "+chat-audit-join":
        replacements["applicant"] = "103262"
        replacements["inviter"] = "519019"
    if service == "doc" and command == "+comment-create-inline":
        replacements["start"] = "0"
        replacements["end"] = "1"
    if service == "attendance" and command == "+save-leave-balance":
        replacements["start"] = (dt.date.today() + dt.timedelta(days=1)).isoformat()
        replacements["end"] = (dt.date.today() + dt.timedelta(days=30)).isoformat()
    if service == "aitable" and command == "+record-update":
        replacements["records"] = '[{"recordId":"recDWSREALTEST","cells":{}}]'
    if service == "aitable" and command == "+view-update":
        replacements["desc"] = "{}"

    out = list(args)
    if "--format" not in out:
        out.extend(["--format", "json"])
    i = 0
    while i < len(out) - 1:
        if out[i].startswith("--"):
            key = out[i][2:]
            if key in replacements and not out[i + 1].startswith("--"):
                out[i + 1] = replacements[key]
                i += 2
                continue
        i += 1
    return out


def shell_join(cmd: list[str]) -> str:
    return " ".join(shlex.quote(x) for x in cmd)


def drop_flag(args: list[str], flag: str) -> list[str]:
    out: list[str] = []
    i = 0
    while i < len(args):
        if args[i] == flag:
            i += 2 if i + 1 < len(args) and not args[i + 1].startswith("--") else 1
            continue
        out.append(args[i])
        i += 1
    return out


def adjust_command_args(args: list[str], service: str, command: str) -> list[str]:
    if service == "chat" and command == "+messages-send-card":
        # --group and --receiver are mutually exclusive; exercise the direct-message path.
        return drop_flag(args, "--group")
    return args


def setup_prerequisite(service: str, command: str, now: str) -> dict | None:
    if service == "todo" and command == "+todo-done":
        title = f"DWS shortcut 真实测试待办 {now}"
        due = (
            dt.datetime.now() + dt.timedelta(days=7)
        ).replace(hour=10, minute=15, second=0, microsecond=0).isoformat() + "+08:00"
        cmd = [
            BIN,
            "todo",
            "+remind",
            "--task",
            title,
            "--at",
            due,
            "--yes",
            "--format",
            "json",
        ]
        started = time.time()
        p = subprocess.run(cmd, text=True, capture_output=True, timeout=25)
        return {
            "purpose": "为 todo +todo-done 创建当前账号自己的临时待办 fixture",
            "input": shell_join(cmd),
            "stdout": p.stdout.strip(),
            "stderr": p.stderr.strip(),
            "exit_code": p.returncode,
            "duration_ms": int((time.time() - started) * 1000),
        }
    return None


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--service", action="append", help="Only run shortcuts from this service; may repeat")
    parser.add_argument("--command", action="append", help="Only run this shortcut command; may repeat")
    parser.add_argument("--missing-only", action="store_true", help="Skip shortcuts already present in the output report")
    parser.add_argument("--failed-only", action="store_true", help="Only rerun shortcuts currently marked non-success in the output report")
    ns = parser.parse_args()
    services = set(ns.service or [])
    commands = set(ns.command or [])

    matrix = ensure_matrix()
    rows = [r for r in matrix["results"] if r.get("risk") != "read"]
    if services:
        rows = [r for r in rows if r["service"] in services]
    if commands:
        rows = [r for r in rows if r["command"] in commands]
    if ns.failed_only and OUT_PATH.exists():
        existing = json.loads(OUT_PATH.read_text(encoding="utf-8"))
        failed = {
            (r.get("service"), r.get("command"))
            for r in existing.get("results", [])
            if r.get("status") != "real-ok"
        }
        rows = [r for r in rows if (r["service"], r["command"]) in failed]
    if ns.missing_only and OUT_PATH.exists():
        existing = json.loads(OUT_PATH.read_text(encoding="utf-8"))
        done = {(r.get("service"), r.get("command")) for r in existing.get("results", [])}
        rows = [r for r in rows if (r["service"], r["command"]) not in done]
    now = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
    results = []
    batch_summary = {"total": len(rows), "ok": 0, "error": 0, "timeout": 0, "held": 0}

    for idx, r in enumerate(rows, 1):
        key = (r["service"], r["command"])
        if key in NO_SAFE_NEGATIVE_TARGET:
            batch_summary["held"] += 1
            item = {
                "service": r["service"],
                "command": r["command"],
                "risk": r["risk"],
                "method": "held; no safe negative target",
                "status": "held",
                "input": "",
                "args": [],
                "stdout": "",
                "stderr": NO_SAFE_NEGATIVE_TARGET[key],
                "exit_code": None,
                "duration_ms": 0,
            }
            item = sanitize_result(item)
            category, fixability, note = classify_failure(item)
            item["failure_category"] = category
            item["fixability"] = fixability
            item["diagnosis"] = note
            results.append(item)
            continue

        setup = None
        try:
            setup = setup_prerequisite(r["service"], r["command"], now)
        except subprocess.TimeoutExpired as e:
            setup = {
                "purpose": "创建前置真实资源超时",
                "input": "",
                "stdout": e.stdout.strip() if isinstance(e.stdout, str) else "",
                "stderr": e.stderr.strip() if isinstance(e.stderr, str) else "",
                "exit_code": None,
                "duration_ms": 25000,
            }

        args = adjust_command_args(replace_flag_values(r["args"], now, r["service"], r["command"]), r["service"], r["command"])
        cmd = [BIN] + args
        started = time.time()
        status = "real-error"
        stdout = ""
        stderr = ""
        exit_code = None
        try:
            p = subprocess.run(cmd, text=True, capture_output=True, timeout=25)
            stdout = p.stdout.strip()
            stderr = p.stderr.strip()
            exit_code = p.returncode
            status = classify_real_status(exit_code, stdout)
            if status == "real-ok":
                status = "real-ok"
                batch_summary["ok"] += 1
            else:
                batch_summary["error"] += 1
        except subprocess.TimeoutExpired as e:
            status = "timeout"
            batch_summary["timeout"] += 1
            if isinstance(e.stdout, str):
                stdout = e.stdout.strip()
            if isinstance(e.stderr, str):
                stderr = e.stderr.strip()
        duration_ms = int((time.time() - started) * 1000)
        item = {
            "service": r["service"],
            "command": r["command"],
            "risk": r["risk"],
            "method": "real-backend-safe-negative-write; no --mock; no --dry-run",
            "status": status,
            "input": shell_join(cmd),
            "args": args,
            "stdout": stdout,
            "stderr": stderr,
            "exit_code": exit_code,
            "duration_ms": duration_ms,
        }
        if setup is not None:
            item["setup"] = setup
        item = sanitize_result(item)
        category, fixability, note = classify_failure(item)
        if category != "passed":
            item["failure_category"] = category
            item["fixability"] = fixability
            item["diagnosis"] = note
        results.append(item)
        if idx % 10 == 0 or idx == len(rows):
            print(
                f"progress {idx}/{len(rows)} ok={batch_summary['ok']} "
                f"error={batch_summary['error']} timeout={batch_summary['timeout']} held={batch_summary['held']}",
                flush=True,
            )
    if OUT_PATH.exists():
        existing = json.loads(OUT_PATH.read_text(encoding="utf-8"))
    else:
        existing = {"results": []}
    replace_keys = {(r["service"], r["command"]) for r in results}
    merged = [
        r for r in existing.get("results", [])
        if (r.get("service"), r.get("command")) not in replace_keys
    ]
    merged.extend(results)
    summary = summarize_results(merged, include_held=True)
    failure_categories = summarize_failure_categories(merged)

    OUT_PATH.write_text(json.dumps({
        "generated_at": dt.datetime.now().isoformat(),
        "summary": summary,
        "failure_categories": failure_categories,
        "results": merged,
    }, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"saved {OUT_PATH} batch={batch_summary} merged={summary}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())