diff --git a/backend/app/services/oa_engine.py b/backend/app/services/oa_engine.py new file mode 100644 index 00000000..1b746230 --- /dev/null +++ b/backend/app/services/oa_engine.py @@ -0,0 +1,1215 @@ +"""OA 审批运行时引擎（钉钉式竖向树）""" +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple + +from sqlalchemy.orm import Session + +from app.core.utils import generate_uuid +from app.crud import oa_approval as oa_crud +from app.models.models import OaApp, OaAppVersion, OaInstance, OaRelation, OaRole, OaTask, SysUser +from app.services.oa_assignee import AssigneeResolveError, resolve_assignees +from app.services.oa_formula import FormulaError, apply_formulas_and_linkages, eval_formula + + +class OaEngineError(ValueError): + """引擎业务错误""" + + +def _loads(text: Optional[str], default: Any = None) -> Any: + if text is None or text == "": + return default + if not isinstance(text, str): + return text + try: + return json.loads(text) + except (TypeError, ValueError): + return default + + +def _dumps(data: Any) -> str: + return json.dumps(data if data is not None else {}, ensure_ascii=False) + + +def _node_type(node: Dict[str, Any]) -> str: + return (node.get("type") or "").strip() + + +def _flow_root(flow_tree: Any) -> Dict[str, Any]: + if not isinstance(flow_tree, dict): + raise OaEngineError("流程树无效") + if isinstance(flow_tree.get("root"), dict): + return flow_tree["root"] + return flow_tree + + +def _find_node(root: Dict[str, Any], node_key: str) -> Optional[Dict[str, Any]]: + if not root or not node_key: + return None + if (root.get("id") or "") == node_key: + return root + child = root.get("child") + if isinstance(child, dict): + found = _find_node(child, node_key) + if found: + return found + for g in root.get("conditionGroups") or []: + if isinstance(g, dict) and isinstance(g.get("child"), dict): + found = _find_node(g["child"], node_key) + if found: + return found + return None + + +def _flow_node_key_order(root: Dict[str, Any]) -> Dict[str, int]: + """按流程树先序遍历为 node.id 分配稳定序号（用于详情时间轴排序）。""" + order: Dict[str, int] = {} + counter = 0 + + def walk(node: Any) -> None: + nonlocal counter + if not isinstance(node, dict): + return + nid = (node.get("id") or "").strip() + if nid and nid not in order: + order[nid] = counter + counter += 1 + if _node_type(node) == "condition": + for g in node.get("conditionGroups") or []: + if isinstance(g, dict) and isinstance(g.get("child"), dict): + walk(g["child"]) + child = node.get("child") + if isinstance(child, dict): + walk(child) + + walk(root) + return order + + +def _next_task_order_num(db: Session, instance_id: str) -> int: + """同一实例内任务展示序号取 max+1（无任务从 1 起）。""" + rows = ( + db.query(OaTask.order_num) + .filter(OaTask.instance_id == instance_id, OaTask.del_flag == 0) + .all() + ) + vals = [int(r[0]) for r in rows if r[0] is not None] + return (max(vals) + 1) if vals else 1 + + +def _sort_tasks_for_detail( + tasks: List[OaTask], *, flow_root: Optional[Dict[str, Any]] +) -> List[OaTask]: + """详情时间轴：按流程树节点顺序，再 order_num / create_time（兼容旧数据同秒创建）。""" + node_order = _flow_node_key_order(flow_root) if isinstance(flow_root, dict) else {} + far = 10**9 + + def key(t: OaTask) -> Tuple[int, int, datetime, str]: + nk = (t.node_key or "").strip() + return ( + node_order.get(nk, far), + int(t.order_num) if t.order_num is not None else 0, + t.create_time or datetime.min, + t.id or "", + ) + + return sorted(tasks, key=key) + + +def _fmt_dt(dt: Any) -> str: + if not dt: + return "" + if hasattr(dt, "strftime"): + return dt.strftime("%Y-%m-%d %H:%M:%S") + return str(dt) + + +def _terminal_item(status: str, title: str, inst: Any) -> Dict[str, Any]: + return { + "kind": "terminal", + "roundNo": getattr(inst, "current_round", None) or 1, + "title": title, + "status": status, + "assigneeUserId": "", + "assigneeUserName": "", + "timestamp": _fmt_dt(getattr(inst, "update_time", None)), + "commentText": "", + "nodeKey": "", + "taskType": "", + } + + +def build_progress_items( + *, + inst: Any, + flow_root: Optional[Dict[str, Any]], + tasks: List[Any], + name_map: Dict[str, str], + events: Optional[List[Any]] = None, +) -> List[Dict[str, Any]]: + """投影审批进度。P0：submit + 按树序 tasks + pending_node + terminal。 + P1：插入 return/resubmit 事件并带 roundNo。""" + _ = events # P1 使用 + items: List[Dict[str, Any]] = [] + round_no = 1 + items.append({ + "kind": "submit", + "roundNo": round_no, + "title": "发起人", + "status": "submitted", + "assigneeUserId": inst.initiator_user_id, + "assigneeUserName": name_map.get(inst.initiator_user_id or "", ""), + "timestamp": _fmt_dt(getattr(inst, "create_time", None)), + "commentText": "", + "nodeKey": "", + "taskType": "", + }) + sorted_tasks = _sort_tasks_for_detail(list(tasks), flow_root=flow_root) + reached_keys = {(t.node_key or "").strip() for t in sorted_tasks if (t.node_key or "").strip()} + for t in sorted_tasks: + items.append({ + "kind": "task", + "roundNo": round_no, + "title": t.node_name or t.node_key or "-", + "status": t.status, + "assigneeUserId": t.assignee_user_id, + "assigneeUserName": name_map.get(t.assignee_user_id or "", ""), + "timestamp": _fmt_dt(t.update_time or t.create_time), + "commentText": t.comment_text or "", + "nodeKey": t.node_key or "", + "taskType": t.task_type or "", + "resolveSource": getattr(t, "resolve_source", None) or "", + "taskId": t.id, + }) + if inst.status == "running" and isinstance(flow_root, dict): + order = _flow_node_key_order(flow_root) + for nid, _idx in sorted(order.items(), key=lambda x: x[1]): + node = _find_node(flow_root, nid) + if not node: + continue + nt = _node_type(node) + if nt not in ("approval", "cc") or nid in reached_keys: + continue + items.append({ + "kind": "pending_node", + "roundNo": round_no, + "title": node.get("name") or ("抄送人" if nt == "cc" else "审批人"), + "status": "not_reached", + "assigneeUserId": "", + "assigneeUserName": "", + "timestamp": "", + "commentText": "", + "nodeKey": nid, + "taskType": nt, + }) + if inst.status == "completed": + items.append(_terminal_item("done", "流程结束", inst)) + elif inst.status == "rejected": + items.append(_terminal_item("rejected", "流程结束（已拒绝）", inst)) + elif inst.status == "cancelled": + items.append(_terminal_item("cancelled", "已撤销", inst)) + return items + + +def _advanced(ver: OaAppVersion) -> Dict[str, Any]: + data = _loads(ver.advanced_settings_json, {}) or {} + if not isinstance(data, dict): + return dict(oa_crud.DEFAULT_ADVANCED_SETTINGS) + merged = dict(oa_crud.DEFAULT_ADVANCED_SETTINGS) + merged.update(data) + return merged + + +def _check_initiator_scope(app: OaApp, user_id: str) -> None: + scope = (app.initiator_scope or "all").strip() + if scope == "all": + return + ids = _loads(app.initiator_user_ids, []) or [] + if user_id not in ids: + raise OaEngineError("您不在该审批的发起范围内") + + +def _instance_to_dict(inst: OaInstance, *, app_name: str = "") -> Dict[str, Any]: + return { + "id": inst.id, + "appId": inst.app_id, + "appVersionId": inst.app_version_id, + "appName": app_name, + "title": inst.title or "", + "initiatorUserId": inst.initiator_user_id, + "initiatorDeptId": inst.initiator_dept_id or "", + "formDataJson": _loads(inst.form_data_json, {}), + "status": inst.status, + "createTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "", + "updateTime": inst.update_time.strftime("%Y-%m-%d %H:%M:%S") if inst.update_time else "", + } + + +def _task_to_dict(t: OaTask, *, assignee_user_name: str = "") -> Dict[str, Any]: + return { + "id": t.id, + "instanceId": t.instance_id, + "nodeKey": t.node_key or "", + "nodeName": t.node_name or "", + "taskType": t.task_type, + "assigneeUserId": t.assignee_user_id, + "assigneeUserName": assignee_user_name or t.assignee_user_id or "", + "status": t.status, + "commentText": t.comment_text or "", + "resolveSource": t.resolve_source or "", + "signMode": t.sign_mode or "", + "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "", + "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "", + } + + +def _user_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]: + ids = [uid for uid in {str(u or "").strip() for u in user_ids} if uid] + if not ids: + return {} + rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all() + return {r.id: (r.nick_name or r.id) for r in rows} + + +def list_templates(db: Session, *, tenant_no: str, user_id: str) -> List[Dict[str, Any]]: + apps = ( + db.query(OaApp) + .filter(OaApp.tenant_no == tenant_no, OaApp.del_flag == 0, OaApp.status == "1") + .order_by(OaApp.order_num.asc(), OaApp.create_time.desc()) + .all() + ) + out: List[Dict[str, Any]] = [] + for app in apps: + pub = oa_crud._find_latest_published(db, app.id, tenant_no) + if not pub: + continue + try: + _check_initiator_scope(app, user_id) + except OaEngineError: + continue + out.append( + { + "id": app.id, + "appName": app.app_name, + "groupId": app.group_id or "", + "iconUrl": app.icon_url or "", + "description": app.description or "", + "versionId": pub.id, + "versionNo": pub.version_no, + } + ) + return out + + +def get_template_detail( + db: Session, *, tenant_no: str, app_id: str, user_id: str +) -> Dict[str, Any]: + app = oa_crud.get_app(db, app_id, tenant_no) + if not app or app.status != "1": + raise OaEngineError("应用不存在或已停用") + _check_initiator_scope(app, user_id) + pub = oa_crud._find_latest_published(db, app_id, tenant_no) + if not pub: + raise OaEngineError("应用尚未发布") + return { + "id": app.id, + "appName": app.app_name, + "groupId": app.group_id or "", + "iconUrl": app.icon_url or "", + "description": app.description or "", + "versionId": pub.id, + "versionNo": pub.version_no, + "formSchemaJson": _loads(pub.form_schema_json, {}), + "flowTreeJson": _loads(pub.flow_tree_json, {}), + "advancedSettingsJson": _advanced(pub), + } + + +_ASSIGNEE_TYPE_LABEL = { + "users": "指定成员", + "oa_roles": "工作流角色", + "dept_leader": "发起人部门负责人", + "direct_supervisor": "发起人的主管", + "initiator_select": "发起人自选", +} + + +def _role_name_map(db: Session, tenant_no: str, role_ids: List[str], codes: List[str]) -> List[str]: + names: List[str] = [] + ids = [str(x).strip() for x in (role_ids or []) if str(x).strip()] + code_list = [str(c).strip().upper() for c in (codes or []) if str(c).strip()] + if ids: + rows = ( + db.query(OaRole) + .filter(OaRole.tenant_no == tenant_no, OaRole.id.in_(ids), OaRole.del_flag == 0) + .all() + ) + names.extend([(r.role_name or r.role_code or r.id) for r in rows]) + if code_list: + rows = ( + db.query(OaRole) + .filter( + OaRole.tenant_no == tenant_no, + OaRole.role_code.in_(code_list), + OaRole.del_flag == 0, + ) + .all() + ) + for r in rows: + label = r.role_name or r.role_code or r.id + if label not in names: + names.append(label) + return names + + +def _assignee_meta(db: Session, tenant_no: str, cfg: Dict[str, Any]) -> str: + assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip() + base = _ASSIGNEE_TYPE_LABEL.get(assignee_type, assignee_type or "未配置审批人") + if assignee_type == "oa_roles": + names = _role_name_map( + db, + tenant_no, + cfg.get("oaRoleIds") or cfg.get("oa_role_ids") or [], + cfg.get("oaRoleCodes") or cfg.get("oa_role_codes") or [], + ) + if names: + return f"{base} · {'/'.join(names)}" + return base + + +def _preview_people(db: Session, user_ids: List[str]) -> List[Dict[str, str]]: + name_map = _user_name_map(db, user_ids) + out: List[Dict[str, str]] = [] + for uid in user_ids: + out.append({"id": uid, "nickName": name_map.get(uid) or uid}) + return out + + +def _walk_preview_nodes( + db: Session, + *, + node: Optional[Dict[str, Any]], + tenant_no: str, + initiator_user_id: str, + initiator_dept_id: Optional[str], + advanced: Dict[str, Any], + initiator_select_user_ids: Optional[List[str]], + steps: List[Dict[str, Any]], +) -> None: + cur = node + while cur and isinstance(cur, dict): + ntype = _node_type(cur) + if ntype == "start": + people = _preview_people(db, [initiator_user_id] if initiator_user_id else []) + steps.append( + { + "id": cur.get("id") or "", + "type": "start", + "name": cur.get("name") or "发起人", + "signMode": "", + "assigneeType": "", + "meta": "", + "people": people, + "resolveError": "", + } + ) + cur = cur.get("child") + continue + + if ntype == "end": + steps.append( + { + "id": cur.get("id") or "", + "type": "end", + "name": cur.get("name") or "结束", + "signMode": "", + "assigneeType": "", + "meta": "", + "people": [], + "resolveError": "", + } + ) + break + + if ntype == "condition": + groups = cur.get("conditionGroups") or [] + for g in groups: + if not isinstance(g, dict): + continue + steps.append( + { + "id": f"{cur.get('id') or 'cond'}_{(g.get('name') or '分支')}", + "type": "branch", + "name": g.get("name") or "条件分支", + "signMode": "", + "assigneeType": "", + "meta": "条件分支（实际路径以提交时表单为准）", + "people": [], + "resolveError": "", + } + ) + child = g.get("child") + if isinstance(child, dict): + _walk_preview_nodes( + db, + node=child, + tenant_no=tenant_no, + initiator_user_id=initiator_user_id, + initiator_dept_id=initiator_dept_id, + advanced=advanced, + initiator_select_user_ids=initiator_select_user_ids, + steps=steps, + ) + cur = cur.get("child") + continue + + if ntype in ("approval", "cc"): + cfg = cur.get("approverConfig") or {} + assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip() + sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "").strip() + if ntype == "cc": + sign_mode = "cc" + elif not sign_mode and assignee_type != "initiator_select": + sign_mode = "or" + meta = _assignee_meta(db, tenant_no, cfg) + people: List[Dict[str, str]] = [] + resolve_error = "" + if assignee_type == "initiator_select": + ids = initiator_select_user_ids or [] + people = _preview_people(db, [str(x) for x in ids if str(x).strip()]) + if not people: + meta = meta or "发起人自选" + else: + try: + assignees, _meta = resolve_assignees( + db, + tenant_no=tenant_no, + assignee_config=cfg, + initiator_user_id=initiator_user_id, + initiator_dept_id=initiator_dept_id, + advanced=advanced, + initiator_select_user_ids=initiator_select_user_ids, + ) + people = _preview_people(db, assignees) + except AssigneeResolveError as e: + resolve_error = str(e) + steps.append( + { + "id": cur.get("id") or "", + "type": ntype, + "name": cur.get("name") or ("抄送人" if ntype == "cc" else "审批人"), + "signMode": sign_mode, + "assigneeType": assignee_type, + "meta": meta, + "people": people, + "resolveError": resolve_error, + } + ) + cur = cur.get("child") + continue + + cur = cur.get("child") + + +def preview_start_flow( + db: Session, + *, + tenant_no: str, + app_id: str, + initiator_user_id: str, + initiator_dept_id: Optional[str] = None, + initiator_select_user_ids: Optional[List[str]] = None, +) -> Dict[str, Any]: + """发起前预览审批链（解析具体人员；条件分支展开展示）。""" + app = oa_crud.get_app(db, app_id, tenant_no) + if not app or app.status != "1": + raise OaEngineError("应用不存在或已停用") + _check_initiator_scope(app, initiator_user_id) + pub = oa_crud._find_latest_published(db, app_id, tenant_no) + if not pub: + raise OaEngineError("应用尚未发布") + + initiator = ( + db.query(SysUser) + .filter( + SysUser.id == initiator_user_id, + SysUser.tenant_no == tenant_no, + SysUser.del_flag == 0, + ) + .first() + ) + dept_id = (initiator_dept_id or "").strip() or ( + (initiator.dept_id or "") if initiator else "" + ) + + flow = _loads(pub.flow_tree_json, {}) or {} + root = flow.get("root") if isinstance(flow, dict) else None + if not isinstance(root, dict): + root = flow if isinstance(flow, dict) else {} + + steps: List[Dict[str, Any]] = [] + _walk_preview_nodes( + db, + node=root, + tenant_no=tenant_no, + initiator_user_id=initiator_user_id, + initiator_dept_id=dept_id or None, + advanced=_advanced(pub), + initiator_select_user_ids=initiator_select_user_ids, + steps=steps, + ) + return {"steps": steps} + + +def _create_tasks_for_node( + db: Session, + *, + inst: OaInstance, + node: Dict[str, Any], + ver: OaAppVersion, + user_id: Optional[str], + initiator_select_user_ids: Optional[List[str]] = None, + order_num_start: Optional[int] = None, +) -> List[OaTask]: + ntype = _node_type(node) + now = datetime.now() + created: List[OaTask] = [] + if ntype == "cc": + task_type = "cc" + elif ntype == "approval": + task_type = "approval" + else: + return created + + cfg = node.get("approverConfig") or {} + advanced = _advanced(ver) + try: + assignees, meta = resolve_assignees( + db, + tenant_no=inst.tenant_no, + assignee_config=cfg, + initiator_user_id=inst.initiator_user_id, + initiator_dept_id=inst.initiator_dept_id, + advanced=advanced, + initiator_select_user_ids=initiator_select_user_ids, + ) + except AssigneeResolveError as e: + raise OaEngineError(str(e)) from e + + sign_mode = (cfg.get("signMode") or cfg.get("sign_mode") or "or").strip() or "or" + resolve_source = meta.get("resolve_source") + status = "done" if task_type == "cc" else "pending" + seq = order_num_start if order_num_start is not None else _next_task_order_num(db, inst.id) + + for uid in assignees: + t = OaTask( + id=generate_uuid(), + instance_id=inst.id, + node_key=node.get("id") or "", + node_name=node.get("name") or "", + task_type=task_type, + assignee_user_id=uid, + status=status, + comment_text=None, + resolve_source=resolve_source, + sign_mode=sign_mode if task_type == "approval" else None, + tenant_no=inst.tenant_no, + order_num=seq, + remark=None, + create_by=user_id, + create_dept_id=inst.initiator_dept_id or "", + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + seq += 1 + db.add(t) + created.append(t) + return created + + +def _match_condition_group(groups: List[Any], form_data: Dict[str, Any]) -> Optional[Dict[str, Any]]: + default_g = None + for g in groups: + if not isinstance(g, dict): + continue + rules = g.get("rules") + when = g.get("when") + # 空规则 = 默认分支 + if (not rules and not when) or g.get("isDefault"): + default_g = g + continue + expr = when + if not expr and isinstance(rules, str): + expr = rules + if not expr and isinstance(rules, list) and rules: + # 简化：每条 {field, op, value} 用 AND + parts = [] + for r in rules: + if not isinstance(r, dict): + continue + field = r.get("field") + op = r.get("op") or "==" + val = r.get("value") + if field is None: + continue + if isinstance(val, str): + parts.append(f"{{{field}}} {op} '{val}'") + else: + parts.append(f"{{{field}}} {op} {val}") + expr = " and ".join(parts) if parts else None + if not expr: + default_g = g + continue + try: + if eval_formula(str(expr), form_data): + return g + except FormulaError: + continue + return default_g + + +def advance_from( + db: Session, + *, + inst: OaInstance, + ver: OaAppVersion, + from_node: Optional[Dict[str, Any]], + user_id: Optional[str], + initiator_select_user_ids: Optional[List[str]] = None, +) -> None: + """从某节点的下一跳开始推进；from_node 为 None 表示从 start 进入。""" + root = _flow_root(_loads(ver.flow_tree_json, {})) + form_data = _loads(inst.form_data_json, {}) or {} + + def next_node(node: Dict[str, Any]) -> Optional[Dict[str, Any]]: + ntype = _node_type(node) + if ntype == "condition": + groups = node.get("conditionGroups") or [] + g = _match_condition_group(groups, form_data) + if not g or not isinstance(g.get("child"), dict): + raise OaEngineError("条件分支无法匹配出线") + return g["child"] + child = node.get("child") + return child if isinstance(child, dict) else None + + # 起点 + if from_node is None: + cur = next_node(root) if _node_type(root) == "start" else root + else: + cur = next_node(from_node) + + # 同一次推进里可能连续落抄送+审批：共用递增序号，避免 create_time 同秒时乱序 + next_order = _next_task_order_num(db, inst.id) + safety = 0 + while cur and safety < 50: + safety += 1 + ntype = _node_type(cur) + if ntype == "end": + inst.status = "completed" + inst.update_time = datetime.now() + db.add(inst) + return + if ntype == "condition": + cur = next_node(cur) + continue + if ntype in ("approval", "cc"): + tasks = _create_tasks_for_node( + db, + inst=inst, + node=cur, + ver=ver, + user_id=user_id, + initiator_select_user_ids=initiator_select_user_ids, + order_num_start=next_order, + ) + next_order += max(len(tasks), 1) + if ntype == "cc": + # 抄送不阻断，继续往下 + cur = next_node(cur) + continue + if not tasks: + raise OaEngineError("审批节点未生成任务") + return + # 未知类型跳过子节点 + cur = next_node(cur) + + # 无结束也无待办：异常结束 + if inst.status == "running": + raise OaEngineError("流程推进异常：未到达结束且无待办任务") + + +def start_instance( + db: Session, + *, + tenant_no: str, + app_id: str, + initiator_user_id: str, + initiator_dept_id: Optional[str] = None, + form_data: Optional[Dict[str, Any]] = None, + title: Optional[str] = None, + initiator_select_user_ids: Optional[List[str]] = None, + related_instance_ids: Optional[List[str]] = None, +) -> Dict[str, Any]: + app = oa_crud.get_app(db, app_id, tenant_no) + if not app or app.status != "1": + raise OaEngineError("应用不存在或已停用") + _check_initiator_scope(app, initiator_user_id) + ver = oa_crud._find_latest_published(db, app_id, tenant_no) + if not ver: + raise OaEngineError("应用尚未发布，无法发起") + + schema = _loads(ver.form_schema_json, {}) + data = dict(form_data or {}) + try: + data = apply_formulas_and_linkages(schema, data) + except FormulaError as e: + raise OaEngineError(f"公式计算失败: {e}") from e + + user = ( + db.query(SysUser) + .filter(SysUser.id == initiator_user_id, SysUser.del_flag == 0) + .first() + ) + dept_id = (initiator_dept_id or (user.dept_id if user else "") or "").strip() or None + + now = datetime.now() + inst = OaInstance( + id=generate_uuid(), + app_id=app_id, + app_version_id=ver.id, + title=(title or app.app_name or "").strip()[:200], + initiator_user_id=initiator_user_id, + initiator_dept_id=dept_id, + form_data_json=_dumps(data), + status="running", + tenant_no=tenant_no, + order_num=0, + remark=None, + create_by=initiator_user_id, + create_dept_id=dept_id or "", + create_time=now, + update_by=initiator_user_id, + update_time=now, + del_flag=0, + ) + db.add(inst) + + for to_id in related_instance_ids or []: + to_id = (to_id or "").strip() + if not to_id: + continue + db.add( + OaRelation( + id=generate_uuid(), + from_instance_id=inst.id, + to_instance_id=to_id, + field_key=None, + tenant_no=tenant_no, + order_num=0, + remark=None, + create_by=initiator_user_id, + create_dept_id=dept_id or "", + create_time=now, + update_by=initiator_user_id, + update_time=now, + del_flag=0, + ) + ) + + advance_from( + db, + inst=inst, + ver=ver, + from_node=None, + user_id=initiator_user_id, + initiator_select_user_ids=initiator_select_user_ids, + ) + db.commit() + db.refresh(inst) + return _instance_to_dict(inst, app_name=app.app_name) + + +def _get_instance(db: Session, instance_id: str, tenant_no: str) -> OaInstance: + inst = ( + db.query(OaInstance) + .filter(OaInstance.id == instance_id, OaInstance.tenant_no == tenant_no, OaInstance.del_flag == 0) + .first() + ) + if not inst: + raise OaEngineError("实例不存在") + return inst + + +def get_instance_detail(db: Session, instance_id: str, *, tenant_no: str) -> Dict[str, Any]: + inst = _get_instance(db, instance_id, tenant_no) + app = oa_crud.get_app(db, inst.app_id, tenant_no) + ver = ( + db.query(OaAppVersion) + .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0) + .first() + ) + tasks = ( + db.query(OaTask) + .filter(OaTask.instance_id == instance_id, OaTask.del_flag == 0) + .all() + ) + flow_root = None + if ver: + try: + flow_root = _flow_root(_loads(ver.flow_tree_json, {})) + except OaEngineError: + flow_root = None + tasks = _sort_tasks_for_detail(tasks, flow_root=flow_root) + rels = ( + db.query(OaRelation) + .filter(OaRelation.from_instance_id == instance_id, OaRelation.del_flag == 0) + .all() + ) + name_map = _user_name_map( + db, + [t.assignee_user_id for t in tasks if t.assignee_user_id] + + ([inst.initiator_user_id] if inst.initiator_user_id else []), + ) + data = _instance_to_dict(inst, app_name=app.app_name if app else "") + data["initiatorUserName"] = name_map.get(inst.initiator_user_id or "", inst.initiator_user_id or "") + data["tasks"] = [ + _task_to_dict(t, assignee_user_name=name_map.get(t.assignee_user_id or "", "")) + for t in tasks + ] + data["relatedInstanceIds"] = [r.to_instance_id for r in rels] + data["formSchemaJson"] = _loads(ver.form_schema_json, {}) if ver else {} + data["advancedSettingsJson"] = _advanced(ver) if ver else dict(oa_crud.DEFAULT_ADVANCED_SETTINGS) + data["allowAddSign"] = bool((_advanced(ver) if ver else {}).get("allowAddSign", True)) + data["progress"] = build_progress_items( + inst=inst, + flow_root=flow_root, + tasks=tasks, + name_map=name_map, + events=[], + ) + return data + + +def list_tasks( + db: Session, + *, + tenant_no: str, + user_id: str, + kind: str, +) -> List[Dict[str, Any]]: + kind = (kind or "").strip() + if kind == "pending": + q = db.query(OaTask).filter( + OaTask.tenant_no == tenant_no, + OaTask.del_flag == 0, + OaTask.assignee_user_id == user_id, + OaTask.task_type == "approval", + OaTask.status == "pending", + ) + elif kind == "done": + q = db.query(OaTask).filter( + OaTask.tenant_no == tenant_no, + OaTask.del_flag == 0, + OaTask.assignee_user_id == user_id, + OaTask.task_type == "approval", + OaTask.status.in_(("approved", "rejected")), + ) + elif kind == "cc": + q = db.query(OaTask).filter( + OaTask.tenant_no == tenant_no, + OaTask.del_flag == 0, + OaTask.assignee_user_id == user_id, + OaTask.task_type == "cc", + ) + elif kind == "mine": + insts = ( + db.query(OaInstance) + .filter( + OaInstance.tenant_no == tenant_no, + OaInstance.del_flag == 0, + OaInstance.initiator_user_id == user_id, + ) + .order_by(OaInstance.create_time.desc()) + .all() + ) + return [_instance_to_dict(i) for i in insts] + else: + raise OaEngineError(f"不支持的列表类型: {kind}") + + rows = q.order_by(OaTask.create_time.desc()).all() + return [_task_to_dict(t) for t in rows] + + +def complete_task( + db: Session, + task_id: str, + *, + tenant_no: str, + user_id: str, + action: str, + comment: Optional[str] = None, +) -> Dict[str, Any]: + action = (action or "").strip() + if action not in ("approve", "reject"): + raise OaEngineError("不支持的操作") + + task = ( + db.query(OaTask) + .filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0) + .first() + ) + if not task: + raise OaEngineError("任务不存在") + if task.assignee_user_id != user_id: + raise OaEngineError("无权处理该任务") + if task.task_type != "approval" or task.status != "pending": + raise OaEngineError("任务状态不可处理") + + inst = _get_instance(db, task.instance_id, tenant_no) + if inst.status != "running": + raise OaEngineError("实例已结束") + + ver = ( + db.query(OaAppVersion) + .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0) + .first() + ) + if not ver: + raise OaEngineError("流程版本不存在") + + advanced = _advanced(ver) + if advanced.get("commentRequired") and not (comment or "").strip(): + raise OaEngineError("请填写审批意见") + + now = datetime.now() + task.comment_text = (comment or "").strip() or None + task.status = "approved" if action == "approve" else "rejected" + task.update_by = user_id + task.update_time = now + db.add(task) + # Session 默认 autoflush=False：不先 flush 时，后续按 status=pending 的查询仍会命中本任务， + # 或签「取消同节点其余待办」会把本人已同意/已拒绝覆盖成 done，导致「已处理」列表查不到。 + db.flush() + + if action == "reject": + # 同节点其他待办作废（排除当前任务） + siblings = ( + db.query(OaTask) + .filter( + OaTask.instance_id == inst.id, + OaTask.node_key == task.node_key, + OaTask.id != task.id, + OaTask.status == "pending", + OaTask.del_flag == 0, + ) + .all() + ) + for s in siblings: + s.status = "done" + s.update_by = user_id + s.update_time = now + db.add(s) + inst.status = "rejected" + inst.update_by = user_id + inst.update_time = now + db.add(inst) + db.commit() + return get_instance_detail(db, inst.id, tenant_no=tenant_no) + + # 或签：一人通过即可；会签：全部通过 + sign_mode = (task.sign_mode or "or").strip() or "or" + pending_same = ( + db.query(OaTask) + .filter( + OaTask.instance_id == inst.id, + OaTask.node_key == task.node_key, + OaTask.task_type == "approval", + OaTask.id != task.id, + OaTask.status == "pending", + OaTask.del_flag == 0, + ) + .count() + ) + if sign_mode == "and" and pending_same > 0: + db.commit() + return get_instance_detail(db, inst.id, tenant_no=tenant_no) + + # 或签：取消同节点其余待办（排除当前任务） + if sign_mode != "and": + for s in ( + db.query(OaTask) + .filter( + OaTask.instance_id == inst.id, + OaTask.node_key == task.node_key, + OaTask.id != task.id, + OaTask.status == "pending", + OaTask.del_flag == 0, + ) + .all() + ): + s.status = "done" + s.update_by = user_id + s.update_time = now + db.add(s) + + root = _flow_root(_loads(ver.flow_tree_json, {})) + node = _find_node(root, task.node_key or "") + if not node: + raise OaEngineError("找不到对应流程节点") + advance_from(db, inst=inst, ver=ver, from_node=node, user_id=user_id) + db.commit() + return get_instance_detail(db, inst.id, tenant_no=tenant_no) + + +def add_sign( + db: Session, + task_id: str, + *, + tenant_no: str, + user_id: str, + user_ids: List[str], + comment: Optional[str] = None, +) -> Dict[str, Any]: + """当前待办处理人加签：为同节点新增待办审批人。""" + task = ( + db.query(OaTask) + .filter(OaTask.id == task_id, OaTask.tenant_no == tenant_no, OaTask.del_flag == 0) + .first() + ) + if not task: + raise OaEngineError("任务不存在") + if task.assignee_user_id != user_id: + raise OaEngineError("无权加签该任务") + if task.task_type != "approval" or task.status != "pending": + raise OaEngineError("仅待处理的审批任务可加签") + + inst = _get_instance(db, task.instance_id, tenant_no) + if inst.status != "running": + raise OaEngineError("实例已结束") + + ver = ( + db.query(OaAppVersion) + .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0) + .first() + ) + if not ver: + raise OaEngineError("流程版本不存在") + advanced = _advanced(ver) + if not advanced.get("allowAddSign", True): + raise OaEngineError("该应用不允许加签") + + ids = [] + seen = set() + for uid in user_ids or []: + uid = (uid or "").strip() + if not uid or uid in seen: + continue + seen.add(uid) + ids.append(uid) + if not ids: + raise OaEngineError("请选择加签人") + + existing = { + r.assignee_user_id + for r in db.query(OaTask) + .filter( + OaTask.instance_id == inst.id, + OaTask.node_key == task.node_key, + OaTask.del_flag == 0, + OaTask.status == "pending", + ) + .all() + } + now = datetime.now() + created = 0 + next_order = _next_task_order_num(db, inst.id) + for uid in ids: + if uid in existing: + continue + # 校验用户有效 + u = ( + db.query(SysUser) + .filter( + SysUser.id == uid, + SysUser.tenant_no == tenant_no, + SysUser.del_flag == 0, + SysUser.status == "1", + ) + .first() + ) + if not u: + raise OaEngineError(f"加签用户无效：{uid}") + db.add( + OaTask( + id=generate_uuid(), + instance_id=inst.id, + node_key=task.node_key, + node_name=task.node_name, + task_type="approval", + assignee_user_id=uid, + status="pending", + comment_text=None, + resolve_source="add_sign", + sign_mode=task.sign_mode or "or", + tenant_no=tenant_no, + order_num=next_order, + remark=(comment or "").strip() or None, + create_by=user_id, + create_dept_id=inst.initiator_dept_id or "", + create_time=now, + update_by=user_id, + update_time=now, + del_flag=0, + ) + ) + next_order += 1 + created += 1 + if created == 0: + raise OaEngineError("所选用户已在当前节点待办中") + # 加签后默认按会签语义收口更稳妥：保留原 sign_mode；若原或签则一人通过仍可结束 + if comment: + task.comment_text = ((task.comment_text or "") + f"\n[加签说明] {comment}").strip() + task.update_by = user_id + task.update_time = now + db.add(task) + db.commit() + return get_instance_detail(db, inst.id, tenant_no=tenant_no) + + +def withdraw_instance( + db: Session, instance_id: str, *, tenant_no: str, user_id: str +) -> Dict[str, Any]: + inst = _get_instance(db, instance_id, tenant_no) + if inst.initiator_user_id != user_id: + raise OaEngineError("仅发起人可撤销") + if inst.status != "running": + raise OaEngineError("仅运行中的实例可撤销") + ver = ( + db.query(OaAppVersion) + .filter(OaAppVersion.id == inst.app_version_id, OaAppVersion.del_flag == 0) + .first() + ) + advanced = _advanced(ver) if ver else dict(oa_crud.DEFAULT_ADVANCED_SETTINGS) + if not advanced.get("allowWithdrawRunning", True): + raise OaEngineError("该应用不允许撤销进行中的审批") + + now = datetime.now() + inst.status = "cancelled" + inst.update_by = user_id + inst.update_time = now + db.add(inst) + for t in ( + db.query(OaTask) + .filter(OaTask.instance_id == inst.id, OaTask.status == "pending", OaTask.del_flag == 0) + .all() + ): + t.status = "done" + t.update_by = user_id + t.update_time = now + db.add(t) + db.commit() + return get_instance_detail(db, inst.id, tenant_no=tenant_no) diff --git a/backend/tests/test_oa_engine.py b/backend/tests/test_oa_engine.py new file mode 100644 index 00000000..4ce21721 --- /dev/null +++ b/backend/tests/test_oa_engine.py @@ -0,0 +1,528 @@ +"""OA 引擎单元测试（纯逻辑 + 轻量 mock）""" +import json +import unittest +from datetime import datetime +from unittest.mock import MagicMock, patch + +from app.services.oa_engine import ( + OaEngineError, + _find_node, + _flow_node_key_order, + _match_condition_group, + _sort_tasks_for_detail, + advance_from, +) + + +class TestOaEngineHelpers(unittest.TestCase): + def test_find_node(self): + root = { + "id": "s", + "type": "start", + "child": { + "id": "a1", + "type": "approval", + "child": {"id": "e1", "type": "end"}, + }, + } + self.assertEqual(_find_node(root, "a1")["id"], "a1") + self.assertEqual(_find_node(root, "e1")["id"], "e1") + self.assertIsNone(_find_node(root, "x")) + + def test_condition_match(self): + groups = [ + { + "name": "gt3", + "when": "{days} > 3", + "child": {"id": "n1", "type": "end"}, + }, + {"name": "default", "rules": [], "child": {"id": "n2", "type": "end"}}, + ] + g = _match_condition_group(groups, {"days": 5}) + self.assertEqual(g["name"], "gt3") + g2 = _match_condition_group(groups, {"days": 1}) + self.assertEqual(g2["name"], "default") + + def test_flow_node_key_order_cc_before_approval(self): + root = { + "id": "s", + "type": "start", + "child": { + "id": "c1", + "type": "cc", + "child": { + "id": "a1", + "type": "approval", + "child": {"id": "e1", "type": "end"}, + }, + }, + } + order = _flow_node_key_order(root) + self.assertLess(order["c1"], order["a1"]) + + def test_sort_tasks_by_flow_when_same_create_time(self): + """抄送先于审批设计时，即使 create_time / order_num 相同也应按树序展示。""" + root = { + "id": "s", + "type": "start", + "child": { + "id": "c1", + "type": "cc", + "child": { + "id": "a1", + "type": "approval", + "child": {"id": "e1", "type": "end"}, + }, + }, + } + ts = datetime(2026, 7, 15, 14, 5, 7) + approval = MagicMock(node_key="a1", order_num=0, create_time=ts, id="AAAA") + cc = MagicMock(node_key="c1", order_num=0, create_time=ts, id="BBBB") + # 故意把审批放在列表前面，模拟 create_time 同秒时 PK 乱序 + sorted_tasks = _sort_tasks_for_detail([approval, cc], flow_root=root) + self.assertEqual([t.node_key for t in sorted_tasks], ["c1", "a1"]) + + +class TestAdvanceFrom(unittest.TestCase): + def test_linear_creates_approval_then_stops(self): + flow = { + "id": "s", + "type": "start", + "child": { + "id": "a1", + "type": "approval", + "name": "负责人", + "approverConfig": {"assigneeType": "dept_leader"}, + "child": {"id": "e1", "type": "end"}, + }, + } + db = MagicMock() + inst = MagicMock( + id="inst1", + tenant_no="zkxw", + initiator_user_id="u0", + initiator_dept_id="d1", + form_data_json="{}", + status="running", + ) + ver = MagicMock( + flow_tree_json=json.dumps(flow), + advanced_settings_json=json.dumps({"supervisorFallbackToDeptLeader": True}), + ) + fake_task = MagicMock() + with patch( + "app.services.oa_engine._next_task_order_num", return_value=1 + ), patch( + "app.services.oa_engine._create_tasks_for_node", + return_value=[fake_task], + ) as mock_create: + advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0") + mock_create.assert_called_once() + self.assertEqual(inst.status, "running") + self.assertEqual(mock_create.call_args.kwargs.get("order_num_start"), 1) + + def test_complete_path_to_end_via_cc(self): + flow = { + "id": "s", + "type": "start", + "child": { + "id": "c1", + "type": "cc", + "name": "抄送", + "approverConfig": {"assigneeType": "users", "userIds": ["u9"]}, + "child": {"id": "e1", "type": "end"}, + }, + } + db = MagicMock() + inst = MagicMock( + id="inst1", + tenant_no="zkxw", + initiator_user_id="u0", + initiator_dept_id="d1", + form_data_json="{}", + status="running", + update_time=datetime.now(), + ) + ver = MagicMock( + flow_tree_json=json.dumps(flow), + advanced_settings_json="{}", + ) + with patch( + "app.services.oa_engine._next_task_order_num", return_value=1 + ), patch( + "app.services.oa_engine._create_tasks_for_node", + return_value=[MagicMock()], + ): + advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0") + self.assertEqual(inst.status, "completed") + + def test_empty_assignee_raises(self): + flow = { + "id": "s", + "type": "start", + "child": { + "id": "a1", + "type": "approval", + "approverConfig": {"assigneeType": "dept_leader"}, + "child": {"id": "e1", "type": "end"}, + }, + } + db = MagicMock() + inst = MagicMock( + id="inst1", + tenant_no="zkxw", + initiator_user_id="u0", + initiator_dept_id="d1", + form_data_json="{}", + status="running", + ) + ver = MagicMock(flow_tree_json=json.dumps(flow), advanced_settings_json="{}") + with patch( + "app.services.oa_engine._next_task_order_num", return_value=1 + ), patch("app.services.oa_engine._create_tasks_for_node", return_value=[]): + with self.assertRaises(OaEngineError): + advance_from(db, inst=inst, ver=ver, from_node=None, user_id="u0") + + +class TestAddSign(unittest.TestCase): + def test_add_sign_rejects_when_disabled(self): + from app.services.oa_engine import add_sign + + db = MagicMock() + task = MagicMock( + id="t1", + assignee_user_id="u1", + task_type="approval", + status="pending", + instance_id="i1", + node_key="a1", + node_name="审批", + sign_mode="or", + comment_text=None, + ) + inst = MagicMock(id="i1", status="running", initiator_dept_id="d1") + ver = MagicMock( + id="v1", + advanced_settings_json=json.dumps({"allowAddSign": False}), + del_flag=0, + ) + db.query.return_value.filter.return_value.first.side_effect = [task, inst, ver] + with self.assertRaises(OaEngineError) as ctx: + add_sign( + db, + "t1", + tenant_no="zkxw", + user_id="u1", + user_ids=["u2"], + ) + self.assertIn("不允许加签", str(ctx.exception)) + + def test_add_sign_requires_users(self): + from app.services.oa_engine import add_sign + + db = MagicMock() + task = MagicMock( + id="t1", + assignee_user_id="u1", + task_type="approval", + status="pending", + instance_id="i1", + ) + inst = MagicMock(id="i1", status="running") + ver = MagicMock(advanced_settings_json="{}", del_flag=0) + db.query.return_value.filter.return_value.first.side_effect = [task, inst, ver] + with self.assertRaises(OaEngineError) as ctx: + add_sign(db, "t1", tenant_no="zkxw", user_id="u1", user_ids=[]) + self.assertIn("加签人", str(ctx.exception)) + + +class TestCompleteTaskOrSign(unittest.TestCase): + """回归：Session autoflush=False 时，或签取消同节点待办不可覆盖本人已同意状态。""" + + def setUp(self): + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + from app.core.database import Base + from app.models.models import OaAppVersion, OaInstance, OaTask + + self.engine = create_engine("sqlite:///:memory:") + # 仅建引擎相关表 + OaAppVersion.__table__.create(self.engine) + OaInstance.__table__.create(self.engine) + OaTask.__table__.create(self.engine) + Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) + self.db = Session() + self.Base = Base + self.OaAppVersion = OaAppVersion + self.OaInstance = OaInstance + self.OaTask = OaTask + + flow = { + "id": "s", + "type": "start", + "child": { + "id": "a1", + "type": "approval", + "name": "审批人", + "approverConfig": {"signMode": "or"}, + "child": {"id": "e1", "type": "end"}, + }, + } + now = datetime.now() + self.db.add( + OaAppVersion( + id="v1", + app_id="app1", + version_no=1, + status="published", + is_latest_published="1", + form_schema_json="{}", + flow_tree_json=json.dumps(flow), + advanced_settings_json="{}", + tenant_no="zkxw", + create_dept_id="", + create_time=now, + update_time=now, + del_flag=0, + ) + ) + self.db.add( + OaInstance( + id="i1", + app_id="app1", + app_version_id="v1", + title="单人审批", + initiator_user_id="u0", + initiator_dept_id="d1", + form_data_json="{}", + status="running", + tenant_no="zkxw", + create_dept_id="", + create_time=now, + update_time=now, + del_flag=0, + ) + ) + self.db.add( + OaTask( + id="t1", + instance_id="i1", + node_key="a1", + node_name="审批人", + task_type="approval", + assignee_user_id="u1", + status="pending", + sign_mode="or", + tenant_no="zkxw", + create_dept_id="", + create_time=now, + update_time=now, + del_flag=0, + ) + ) + self.db.commit() + + def tearDown(self): + self.db.close() + self.engine.dispose() + + def test_approve_keeps_approved_not_done(self): + from app.services.oa_engine import complete_task + + with patch( + "app.services.oa_engine.get_instance_detail", + return_value={"id": "i1", "status": "completed"}, + ): + complete_task( + self.db, + "t1", + tenant_no="zkxw", + user_id="u1", + action="approve", + comment="同意", + ) + + task = self.db.query(self.OaTask).filter(self.OaTask.id == "t1").first() + self.assertEqual(task.status, "approved") + self.assertEqual(task.comment_text, "同意") + + # 已处理列表应能查到 + from app.services.oa_engine import list_tasks + + done = list_tasks(self.db, tenant_no="zkxw", user_id="u1", kind="done") + self.assertEqual(len(done), 1) + self.assertEqual(done[0]["status"], "approved") + + def test_or_sign_cancels_siblings_only(self): + from app.services.oa_engine import complete_task + + now = datetime.now() + self.db.add( + self.OaTask( + id="t2", + instance_id="i1", + node_key="a1", + node_name="审批人", + task_type="approval", + assignee_user_id="u2", + status="pending", + sign_mode="or", + tenant_no="zkxw", + create_dept_id="", + create_time=now, + update_time=now, + del_flag=0, + ) + ) + self.db.commit() + + with patch( + "app.services.oa_engine.get_instance_detail", + return_value={"id": "i1", "status": "completed"}, + ): + complete_task( + self.db, + "t1", + tenant_no="zkxw", + user_id="u1", + action="approve", + comment="同意", + ) + + t1 = self.db.query(self.OaTask).filter(self.OaTask.id == "t1").first() + t2 = self.db.query(self.OaTask).filter(self.OaTask.id == "t2").first() + self.assertEqual(t1.status, "approved") + self.assertEqual(t2.status, "done") + + +class TestBuildProgress(unittest.TestCase): + def test_cc_before_approval_and_pending_ahead(self): + from app.services.oa_engine import build_progress_items + + root = { + "id": "s", + "type": "start", + "child": { + "id": "c1", + "type": "cc", + "name": "抄送人", + "child": { + "id": "a1", + "type": "approval", + "name": "审批人", + "child": { + "id": "a2", + "type": "approval", + "name": "总经理", + "child": {"id": "e1", "type": "end"}, + }, + }, + }, + } + ts = datetime(2026, 7, 15, 14, 5, 7) + # 审批任务 create 同秒但列表故意审批在前 + t_approval = MagicMock( + id="tA", + node_key="a1", + node_name="审批人", + task_type="approval", + status="pending", + assignee_user_id="u2", + order_num=0, + create_time=ts, + update_time=ts, + comment_text=None, + resolve_source="users", + ) + t_cc = MagicMock( + id="tC", + node_key="c1", + node_name="抄送人", + task_type="cc", + status="done", + assignee_user_id="u9", + order_num=0, + create_time=ts, + update_time=ts, + comment_text=None, + resolve_source="users", + ) + inst = MagicMock( + status="running", + initiator_user_id="u0", + create_time=ts, + update_time=ts, + ) + items = build_progress_items( + inst=inst, + flow_root=root, + tasks=[t_approval, t_cc], + name_map={"u0": "发起人甲", "u2": "审批乙", "u9": "抄送丙"}, + events=[], # P0 可空 + ) + kinds = [x["kind"] for x in items] + self.assertEqual(kinds[0], "submit") + # 已产生：cc 再 approval + self.assertEqual( + [x["nodeKey"] for x in items if x["kind"] == "task"], + ["c1", "a1"], + ) + # running 时未到达 a2 + pending = [x for x in items if x["kind"] == "pending_node"] + self.assertTrue(any(p["nodeKey"] == "a2" for p in pending)) + + def test_terminal_status_excludes_pending_nodes(self): + from app.services.oa_engine import build_progress_items + + root = { + "id": "s", + "type": "start", + "child": { + "id": "a1", + "type": "approval", + "name": "审批人", + "child": { + "id": "a2", + "type": "approval", + "name": "总经理", + "child": {"id": "e1", "type": "end"}, + }, + }, + } + ts = datetime(2026, 7, 15, 14, 5, 7) + t_approval = MagicMock( + id="tA", + node_key="a1", + node_name="审批人", + task_type="approval", + status="rejected", + assignee_user_id="u2", + order_num=0, + create_time=ts, + update_time=ts, + comment_text="不同意", + resolve_source="users", + ) + for status in ("rejected", "completed", "cancelled"): + inst = MagicMock( + status=status, + initiator_user_id="u0", + create_time=ts, + update_time=ts, + ) + items = build_progress_items( + inst=inst, + flow_root=root, + tasks=[t_approval], + name_map={"u0": "发起人甲", "u2": "审批乙"}, + events=[], + ) + pending = [x for x in items if x["kind"] == "pending_node"] + self.assertEqual(pending, [], f"status={status} should not have pending_node") + terminal = [x for x in items if x["kind"] == "terminal"] + self.assertEqual(len(terminal), 1, f"status={status} should have one terminal") + + +if __name__ == "__main__": + unittest.main() +