E2E 测试模式（Mock）实施方案

本文档自包含，可直接据此实施。系统提供测试模式，可在配置中开启（test_mode.enabled）：开启后端到端链路（剧本生成 → 分镜 → 图片/音频/视频生成 → 四宫格/多角度 → 人脸遮盖 → 时间轴合成）中除 LLM 外的所有外部付费生成 API 调用与外部下载都被挡板拦截，返回预设本地媒体资源。算力扣费、CDN 上传等副作用保留真实执行（靠独立隔离环境保证可控，见 §6）；测试数据采用幂等播种、不做清理（见 §14）。

关键约束：

web 接口与内部函数必须非阻塞；禁止在异步函数中调用同步阻塞 IO（如 requests）。
功能改动同步 docs/。
兼容 Windows / Linux / macOS（路径与编码）。
改表结构需同步迁移脚本与模型 SQL。

1. 背景与现状

1.1 现有测试模式（仅覆盖个别链路）

默认配置中已定义 test_mode.enabled 及若干 mock URL 配置项，但此前主链路的挡板基本未实现（仅个别链路有 mock 或日志输出）。本方案从零搭建统一拦截层。

1.2 覆盖范围

挡板需覆盖 E2E 链路中所有涉及外部生成调用的任务通道（视觉异步、视觉同步、TTS 音频、第三方服务异步、ComfyUI 工具直调、四宫格、多角度、人脸遮盖 pipeline 等）；时间轴合成与世界导入/导出两条链路靠上游本地路径/纯本地处理跑真实代码，无需 mock。

此外有若干副作用需隔离/可控：第三方服务槽位由 mock 短路天然不占用；算力扣费、LLM token 计费、CDN 上传在独立测试环境真实执行（不绕过，见 §6），靠环境隔离保证可控。

2. 总体架构

                         ┌──────────────────────────────────────────┐
                         │   mock 拦截器（纯逻辑层）                  │
                         │  · is_mock_enabled / mock_task_id 生成     │
                         │  · 按 category 解析 mock URL               │
                         │  · 构造各通道响应结构 / 命中计数             │
                         └───────────────┬──────────────────────────┘
                                         │ 被各生成任务的提交/轮询层调用
                                         │ （时间轴合成/世界导入导出无需 mock）
        ┌────────────┬────────────┬──────┴───────┬────────────┬──────────────┐
        ▼            ▼            ▼              ▼            ▼              ▼
   视觉异步      视觉同步        音频           第三方服务     ComfyUI工具/   人脸遮盖
   (调度层)      (子进程)       (TTS)           异步          四宫格/多角度
                                         │
                         ┌───────────────┴──────────────────────────┐
                         │  副作用策略（不绕过，靠隔离环境）            │
                         │  · 算力：扣费链路真实执行，测试前余额充足     │
                         │  · CDN：保留上传，由测试环境承载             │
                         │  · 槽位：mock 短路天然不占用                │
                         └──────────────────────────────────────────┘

2.1 三条铁律

写库留在各通道调度器内，复用现有成功处理函数；拦截器只负责"返回什么"，不直接写 DB。
mock 任务 ID 统一前缀 mock_task_；异步任务的外部任务 ID 同前缀。所有轮询/重试调度器据此识别并短路。
mock 资源一律本地 /upload/mock/... 真实文件（绝不使用远程 URL），避免触发外网下载与远程生成 API（CDN 上传在隔离环境中保留，见 §6）。

3. 核心模块：mock 拦截器（新建）

仅含纯逻辑，无 DB 写入。 核心代码如下：

"""
E2E 测试模式拦截器（纯逻辑层）。
当 test_mode.enabled=True 时，拦截所有外部媒体生成调用，返回预设本地媒体。
本模块只负责"判断与构造响应"，写库由各通道调度器复用现有成功处理函数完成。
"""
import logging
import threading
import uuid
from typing import Any, Dict, Optional

from config.config_util import get_dynamic_config_value
from config.unified_config import UnifiedConfigRegistry, TaskCategory

logger = logging.getLogger(__name__)

MOCK_PROJECT_PREFIX = "mock_task_"

# ============ 进程内命中计数（可观测性） ============
_counters_lock = threading.Lock()
_counters: Dict[str, int] = {}


def _bump(channel: str) -> None:
    with _counters_lock:
        _counters[channel] = _counters.get(channel, 0) + 1


def mock_hit_summary() -> Dict[str, int]:
    with _counters_lock:
        return dict(_counters)


# ============ 开关与 ID ============
def is_mock_enabled() -> bool:
    """挡板总开关。注意：动态配置缓存 TTL=30s，跨进程有传播延迟（见 §7）。"""
    return bool(get_dynamic_config_value("test_mode", "enabled", default=False))


def generate_mock_project_id() -> str:
    return f"{MOCK_PROJECT_PREFIX}{uuid.uuid4().hex[:16]}"


def is_mock_id(task_id: Optional[str]) -> bool:
    return isinstance(task_id, str) and task_id.startswith(MOCK_PROJECT_PREFIX)


# ============ 按 category 解析 mock URL ============
# TaskCategory 是字符串常量类（非 Enum），取值如 'text_to_image'/'image_edit'/
# 'image_to_video'/'text_to_video'/'digital_human'/'audio' 等。
def _img(key: str) -> Optional[str]:
    return get_dynamic_config_value("test_mode", "mock_images", key, default=None)


def _vid(key: str) -> Optional[str]:
    return get_dynamic_config_value("test_mode", "mock_videos", key, default=None)


def _aud(key: str) -> Optional[str]:
    return get_dynamic_config_value("test_mode", "mock_audio", key, default=None)


def resolve_mock_url_for_visual(ai_tool_type: int) -> Optional[str]:
    """
    根据 AI 工具类型解析视觉任务的 mock URL。
    同时考虑主分类 category 与附加分类 categories（一个图编模型可能也支持文生图）。
    """
    try:
        cfg = UnifiedConfigRegistry.get_by_id(ai_tool_type)
        if not cfg:
            return None
        cats = [cfg.category] + list(cfg.categories or [])

        def pick(*candidates, getter):
            for c in candidates:
                if c in cats:
                    return getter(c)
            return None

        # 视频（数字人归视频桶）
        url = pick(TaskCategory.IMAGE_TO_VIDEO, TaskCategory.TEXT_TO_VIDEO,
                   TaskCategory.DIGITAL_HUMAN, getter=_vid)
        if url:
            return url
        # 图片：优先按主分类精确映射，再用附加分类兜底，
        # 否则图编模型（主分类=IMAGE_EDIT）会被误判为文生图。
        primary = cfg.category
        if primary == TaskCategory.IMAGE_EDIT:
            url = _img("image_edit")
        elif primary == TaskCategory.TEXT_TO_IMAGE:
            url = _img("text_to_image")
        else:
            url = pick(TaskCategory.TEXT_TO_IMAGE, TaskCategory.IMAGE_EDIT, getter=_img)
        if url:
            return url
        # 兜底
        return _vid("image_to_video") or _img("text_to_image")
    except Exception as e:
        logger.warning(f"[MOCK] resolve_mock_url_for_visual({ai_tool_type}) failed: {e}")
        return None


def mock_image(subkey: str = "text_to_image") -> Optional[str]:
    return _img(subkey)


def mock_video(subkey: str = "image_to_video") -> Optional[str]:
    return _vid(subkey)


def mock_audio(subkey: str = "tts") -> Optional[str]:
    return _aud(subkey)


# ============ 各通道响应结构构造 ============
def visual_async_submit_result(ai_tool_type: int) -> Dict[str, Any]:
    """视觉异步：submit 返回任务 ID，check_status 稍后返回 SUCCESS+url。"""
    _bump("visual_async_submit")
    pid = generate_mock_project_id()
    logger.info(f"[MOCK] channel=visual_async_submit type={ai_tool_type} project_id={pid}")
    return {"success": True, "project_id": pid}


def visual_async_status_result(ai_tool_type: int) -> Dict[str, Any]:
    """视觉异步：check_status 直接返回 SUCCESS。"""
    url = resolve_mock_url_for_visual(ai_tool_type)
    _bump("visual_async_poll")
    logger.info(f"[MOCK] channel=visual_async_poll type={ai_tool_type} url={url}")
    return {"status": "SUCCESS", "result_url": url}


def visual_sync_result(ai_tool_type: int) -> Dict[str, Any]:
    """视觉同步：返回同步模式结果。"""
    url = resolve_mock_url_for_visual(ai_tool_type)
    _bump("visual_sync_submit")
    logger.info(f"[MOCK] channel=visual_sync_submit type={ai_tool_type} url={url}")
    return {"success": True, "sync_mode": True, "result_url": url}


def comfyui_submit_result(ai_tool_type: int) -> Dict[str, Any]:
    """ComfyUI 工具直调 submit：兼容 {project_ids, status} 结构。"""
    _bump("comfyui_submit")
    pid = generate_mock_project_id()
    logger.info(f"[MOCK] channel=comfyui_submit type={ai_tool_type} project_id={pid}")
    return {"status": "submitted", "project_ids": [pid]}


def comfyui_status_success(file_url: str) -> Dict[str, Any]:
    """ComfyUI 轮询成功：返回任务元素结构（含 results[0].file_url）。
    注意：成功处理函数接收的是任务元素本身，不是外层 envelope。"""
    _bump("comfyui_poll")
    logger.info(f"[MOCK] channel=comfyui_poll file_url={file_url}")
    return {
        "status": "SUCCESS",
        "results": [{"file_url": file_url, "result_url": file_url,
                     "cdn_status": "skip", "task_cost_time": 0}],
    }


def async_submit_result(impl_label: str) -> Dict[str, Any]:
    """第三方服务异步（音频/人脸遮盖）：submit 返回 mock 任务 ID。"""
    _bump(f"async_submit:{impl_label}")
    pid = generate_mock_project_id()
    logger.info(f"[MOCK] channel=async_submit impl={impl_label} project_id={pid}")
    return {"success": True, "project_id": pid}

4. 配置项定义

4.1 扩展默认配置（test_mode 段）

在现有 test_mode 段后追加（保留原有 mock URL 项）：

    # ---- test_mode 行为开关 ----
    {'key': 'test_mode.inject_failure_rate',     'value_type': 'int',    'description': '失败注入概率(0-100)，0=不注入', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.inject_failure_scenario', 'value_type': 'string', 'description': '失败注入场景，如 submit/poll/timeout', 'editable': True, 'is_sensitive': False},

    # ---- mock_images ----
    {'key': 'test_mode.mock_images.comfyui_text_to_image', 'value_type': 'string', 'description': 'ComfyUI 文生图预设URL', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.mock_images.comfyui_image_edit',    'value_type': 'string', 'description': 'ComfyUI 图片编辑预设URL', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.mock_images.grid_image',            'value_type': 'string', 'description': '四宫格整图预设URL(2x2)', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.mock_images.multi_angle_front',     'value_type': 'string', 'description': '多角度-正面图', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.mock_images.multi_angle_side',      'value_type': 'string', 'description': '多角度-侧面图', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.mock_images.multi_angle_back',      'value_type': 'string', 'description': '多角度-背面图', 'editable': True, 'is_sensitive': False},

    # ---- mock_videos ----
    {'key': 'test_mode.mock_videos.digital_human', 'value_type': 'string', 'description': '数字人预设视频URL', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.mock_videos.face_mask',     'value_type': 'string', 'description': '人脸遮盖预设视频URL(必须本地真实mp4)', 'editable': True, 'is_sensitive': False},

    # ---- mock_audio ----
    {'key': 'test_mode.mock_audio.tts',            'value_type': 'string', 'description': 'TTS 预设音频URL', 'editable': True, 'is_sensitive': False},
    {'key': 'test_mode.mock_audio.character_audio','value_type': 'string', 'description': '角色音频预设URL', 'editable': True, 'is_sensitive': False},

说明：get_dynamic_config_value 支持多级 key 拼接（如 ("test_mode", "mock_images", "text_to_image") 拼成 test_mode.mock_images.text_to_image）。原有的 test_mode.mock_videos.image_to_video/text_to_video、mock_images.image_edit/text_to_image 等保留不动，继续被 resolve_mock_url_for_visual 使用。

4.2 预设资源约定（必须本地落盘）

key	文件	要求
`mock_images.text_to_image`	`upload/mock/e2e_text_to_image.png`	9:16，≤1MB
`mock_images.image_edit`	`upload/mock/e2e_image_edit.png`	同上
`mock_images.comfyui_text_to_image`	`upload/mock/e2e_comfyui_tti.png`	同上
`mock_images.comfyui_image_edit`	`upload/mock/e2e_comfyui_ie.png`	同上
`mock_images.grid_image`	`upload/mock/e2e_grid_2x2.png`	真实 2x2 拼图，1024×1024
`mock_images.multi_angle_front/side/back`	`upload/mock/e2e_ma_*.png`	场景图
`mock_videos.image_to_video`	`upload/mock/e2e_i2v.mp4`	9:16，≤5MB，≤5s，真实可播
`mock_videos.text_to_video`	`upload/mock/e2e_t2v.mp4`	同上
`mock_videos.digital_human`	`upload/mock/e2e_dh.mp4`	同上
`mock_videos.face_mask`	`upload/mock/e2e_face_mask.mp4`	真实 mp4（下游有文件存在性校验，见 §5.8）
`mock_audio.tts`	`upload/mock/e2e_tts.mp3`	≤10s
`mock_audio.character_audio`	`upload/mock/e2e_char.mp3`	≤10s
（测试资产）世界导出包	`upload/mock/world_export_sample.zip`	由 §14 播种世界导出生成，结构见 §5.10，供导入测试用

资源准备脚本：见 §12。所有 mock 文件放 upload/mock/（该目录已挂载为静态资源，/upload/mock/x.png 可直接 HTTP 访问）。

5. 分链路实施要点

挡板统一嵌入各生成任务的提交层与轮询层。以下按链路给出关键要点，具体插入点以对应模块代码中的"创建驱动/发起外部调用"位置为准（统一原则：在真正发起外部调用之前短路）。

5.1 视觉异步

提交：mock 短路必须插在提交入口最开头，且在参数准备（param_prepare）检查、模型服务记录、同步分流之前，否则：带人脸遮盖前置步骤的任务类型会死锁；同步实现进了子进程后拦不住。
轮询：拿到外部任务 ID 后、创建驱动之前，识别 mock 前缀直接构造成功数据（无 URL 时按失败处理）。
本地路径短路（必须）：下载缓存工具不识别 /upload/ 本地路径（对相对路径会抛 InvalidURL），异步成功路径必须自行判断 is_local_path（/upload/ 开头则跳过下载直接采用）。同步分支已有此判断，异步分支需补齐保持一致。

    ai_tool_type = ai_tool.type
    task_id = ai_tool.id

    # ===== 测试模式短路（必须在 param_prepare / 模型服务记录 / sync 分流之前）=====
    from task.mock_interceptor import is_mock_enabled, visual_async_submit_result
    if is_mock_enabled():
        mock = visual_async_submit_result(ai_tool_type)
        project_id = mock["project_id"]
        # 写回视觉任务与任务表状态为 PROCESSING（复用现有更新函数）
        return True
    # =====================================================================

5.2 视觉同步（子进程池）

同步任务在子进程池内执行，不继承主进程状态；但 mock 判定来自动态配置，子进程首次读取即可命中（注意 §7 缓存延迟）。
推荐：mock 判定放在主进程的同步分流之前——命中则不提交子进程，直接在主进程写 mock 结果，彻底规避子进程缓存问题（见 §7 应对方案 1）。
命中后返回"同步成功 + 本地 result_url"，后续走现有结果处理；result_url 以 /upload/ 开头时跳过下载缓存。

5.3 TTS 音频

拦截点必须在音频生成调用之前，直接写 mock 结果 URL（不要放在音频生成工具内部——工具内部的 URL 拼接逻辑会覆盖返回值，导致 URL 指向不存在的文件）。
音频通道当前不扣费（生成接口只建记录），因此音频侧无需任何扣费处理。

    task_id = ai_audio.id

    # ===== 测试模式短路 =====
    from task.mock_interceptor import is_mock_enabled, mock_audio
    if is_mock_enabled():
        url = mock_audio("tts")
        if url:
            # 写回音频记录为 COMPLETED 并写 result_url（复用现有更新函数）
            return True
    # =========================

5.4 第三方服务异步（角色音频 + 人脸遮盖，统一拦截点）

音频与人脸遮盖两条异步链路都汇聚到异步驱动基类的"带槽位管理的提交"方法，是唯一统一拦截点：

提交：在创建异步任务记录之后、获取槽位之前短路，直接写 mock 外部任务 ID 并返回。由于在槽位获取之前 return，mock 任务天然不占槽位，无需额外处理。
轮询：异步任务轮询调度器会取到 QUEUED 状态的 mock 任务，是 mock 完成的唯一主路径。识别 mock 外部任务 ID 后：

不要调用原成功处理器——其内部下载函数不识别 /upload/mock/ 本地路径会失败；
直接写完成状态与 result_url，并自行补关键副作用（如角色音频需回写角色默认音色）；
人脸遮盖的 result_url 经 pipeline 结果应用写回 AI 工具的视频路径，供下游视频生成使用（格式与生产一致：/upload/mock/... 相对路径）。

另一条"可重试任务"轮询流程按 status=QUEUED AND next_retry_at IS NOT NULL 过滤，而 mock 分支创建记录时不设重试时间、且提前 return 不会触发重试调度，普通 mock 任务不会进入该流程，无需在此拦截；仅在任务内做防御：若外部任务 ID 是 mock 前缀（理论上不出现），跳过真实提交直接标记完成。

5.5 ComfyUI 工具直调（Agent 工具）

两个方法（文生图 / 图片编辑）直接 HTTP 调用本机 ComfyUI，完全绕开驱动，需分别短路。
关键：只替换"获取 project_ids"那一步，保留后续创建任务记录的逻辑。 否则轮询扫描任务表时无记录，参考图永远不会更新，Agent 工具调用无法完成。

        # ===== 测试模式短路：仅替换 project_ids 的获取，保留后续创建逻辑 =====
        from task.mock_interceptor import is_mock_enabled, generate_mock_project_id
        if is_mock_enabled():
            project_id = generate_mock_project_id()
            result_data = {'project_ids': [project_id]}
        else:
            response = httpx.post(api_url, data=request_data, timeout=30, verify=False)
            response.raise_for_status()
            result_data = response.json()
        # =================================================================
        # 后续创建任务记录逻辑原样保留

死代码澄清：某定时任务管理器中的状态轮询是死代码（轮询已迁移到调度器进程），不要在它上面加拦截。

5.6 四宫格

轮询：识别 mock 任务 ID 后直接构造成功数据，复用现有成功处理函数（成功处理读取 results[0].file_url，传入的是任务元素本身）。
失败重试：重试函数开头命中 mock 直接返回 mock 任务 ID。
四宫格特殊性：特定条目类型会触发 2x2 拆分，要求下载开关开启且本地文件是真实 2x2 图（mock 轮询分支仅读取 test_mode.mock_images.grid_image 并做 2x2 校验，无"直接给拆分图"分支）。E2E 素材须准备真实 2x2 拼图（非宫格条目则回退读取 mock_images.comfyui_text_to_image）。

5.7 多角度

重要差异：本链路不调状态轮询接口，完成判定走数据库（读对应视觉任务的完成状态与结果 URL）。
提交：mock 模式下每个调度 tick 推进一个角度（与真实"一次一个角度"状态机一致）：

def _apply_mock_angle(task_key: str, comfyui_base_url: str):
    """测试模式推进一个角度：选图→落盘→写场景引用→递增索引，末角度置完成。"""
    task = get_latest_multi_angle_task(task_key)   # 读取最新任务状态
    angles = task.get_angles_list()
    idx = task.current_angle_index or 0
    generated = task.get_generated_images_list() or []
    if idx >= len(angles):
        update_status(task_key, COMPLETED, generated_images=generated)
        return

    angle = angles[idx].get('angle', 0)
    angle_key = angles[idx].get('angleKey', 'unknown')
    label = angles[idx].get('label', f'{angle}°')

    # 角度→mock 图映射（front≈0°，back≈180°，其余 side）
    if angle >= 337.5 or angle < 22.5:
        file_url = _img('multi_angle_front') or '/upload/mock/e2e_ma_front.png'
    elif 157.5 <= angle < 202.5:
        file_url = _img('multi_angle_back') or '/upload/mock/e2e_ma_back.png'
    else:
        file_url = _img('multi_angle_side') or '/upload/mock/e2e_ma_side.png'

    # 本地 mock 文件直接复制落盘（改动 C 的本地路径短路，不外网下载）
    local_image_url, local_file_path = _download_and_store_image(file_url, comfyui_base_url)

    new_image = {'angle': angle_key, 'label': label,
                 'url': local_image_url, 'local_file_path': local_file_path}
    generated.append(new_image)
    _update_reference_images_to_staging(task, [new_image])   # 写入场景 JSON reference_images

    next_idx = idx + 1
    if next_idx >= len(angles):
        update_status(task_key, COMPLETED, generated_images=generated)
    else:
        update_status(task_key, PROCESSING,
                      current_angle_index=next_idx, generated_images=generated)

下载：图片下载函数在发起 HTTP 之前加本地路径短路（/upload/mock/ 开头直接本地复制）。

5.8 人脸遮盖 pipeline（param_prepare 产物）

关键约束：人脸遮盖的 result_url 经 pipeline 结果应用写回 AI 工具视频路径，作为下游视频生成的输入；下游对应驱动均做文件存在性校验，文件不存在则回退/失败。
硬性要求：test_mode.mock_videos.face_mask 必须指向本地真实存在的可播 mp4（如 /upload/mock/e2e_face_mask.mp4 预置）。
推荐策略：跳过遮罩合成（overlay），直接写完成（§5.4 已采用）。原因：原成功处理器内部对 result_url 调下载缓存，而下载缓存不识别本地路径 → 处理器返回空 → 被判失败。正确做法是 mock 轮询短路时直接写完成状态 + mock 视频路径，供下游使用。
备选（不推荐）：若必须验证 overlay 合成本身，需先给成功处理器的下载调用补本地短路，且 mock mp4 需与原视频尺寸/编码兼容。工作量大，建议用单独的集成测试覆盖 overlay，E2E 主链路用跳过策略。

5.9 时间轴合成（E2E 终点）—— 无需 mock

结论：本通道不做任何 mock 拦截。 该接口对每个片段先做本地上传文件解析：只要视频 URL 以 /upload/ 开头且文件存在，就走本地拷贝分支，天然跳过远程下载与 CDN 签名刷新，只剩本地 ffprobe/ffmpeg（本地二进制，非外网调用）。
由于上游所有通道产出的结果 URL 都是 /upload/mock/... 本地路径，整个导出链路用真实代码跑在本地 mock 素材上，无需任何改动。这是覆盖真实代码（草稿生成、多轨拼接、打 zip）最多的环节，应保留而非绕过。
环境前提（非 mock 关注点）：测试环境需安装 ffmpeg/ffprobe；/upload/mock/ 下的音视频样本必须真实可播（ffprobe 能读出时长/轨道）；导出阶段只读本地素材，无需关闭 CDN 上传开关。

5.10 世界导入 / 导出 —— 无需 mock

导出（GET /api/export-world）：把世界目录（角色/场景/道具/剧本/世界的 JSON + 图片 + 音频）打成 zip，经存储上传图床，返回下载 URL。唯一外部依赖是图床上传，按 §6.2 保留真实执行（测试环境需配置与生产隔离的 bucket）。
导入（POST /api/import-world）：零外部依赖——纯本地 zip 解压 + 文件复制 + JSON 写入，只需提供一份合法的测试 zip 资产。

测试资产：预置世界导出 zip，结构须与导出产出一致：

world_export_sample.zip
├── metadata.json              # {export_version, world_id, user_id, image_count, audio_count, subdirs:[...]}
├── image_mapping.json         # {filename: "/upload/{type}/pic/{filename}"}   （有图片时）
├── audio_mapping.json         # {filename: "/upload/character/voice/{filename}"} （有音频时）
├── characters/<name>.json     # 角色 JSON（reference_image 用 images/<file> 相对路径）
├── locations/<name>.json
├── props/<name>.json
├── scripts/<name>.json
├── worlds/<name>.json
├── images/<file>.png          # 与 image_mapping 对应的真实图片（可复用 §4.2 的 mock 图）
└── audios/<file>.mp3          # 与 audio_mapping 对应（可复用 mock 音频）

生成方式（二选一，推荐 1）：

先按 §14 幂等播种一个已知世界（含若干角色/场景/道具 + mock 图片），调一次真实导出接口得到 zip，固化为测试资产。格式天然与代码一致，最稳。
手工按上面结构打包（适合最小用例）。

导入测试建议指向专用测试 world_id，避免覆盖其它测试数据：导入会覆盖目标世界目录下的同名 JSON；图片在目标已存在时跳过（天然幂等）。

E2E 用例要点：

导出：导出已播种世界 → 拿下载 URL → 下载 zip → 校验含 metadata.json 且图片数与播种一致。
导入：上传 world_export_sample.zip → 校验返回的各类计数 → 校验目标世界目录下 JSON 与图片落盘。

6. 副作用处理策略（不绕过，靠隔离环境）

决策：算力扣减与 CDN 上传均不绕过——它们是必须覆盖的重要链路，绕过会导致大量测试不完整。E2E 在独立隔离环境运行，副作用影响可控：

副作用	策略	说明
媒体算力扣费/退款	保留真实执行	扣费/退款链路正常跑，覆盖计费链路
LLM token 计费	保留真实执行	token 扣减正常执行，覆盖 Agent 计费
测试账号算力	测试前保证充足	隔离环境内独立维护测试账户余额，保证运行期间不被扣穿
CDN/对象存储上传	保留	上传由测试环境的隔离 bucket 承载，不碰生产
第三方服务槽位	天然不占	mock 短路在获取槽位之前 return，mock 任务不占槽位

6.1 算力扣费/退款：保留 + 测试前余额充足

扣费链路保持原样，不拦截：媒体扣费/退款经计费服务（与 token 计费分属两套独立后端）正常执行，幂等键为交易 ID。

测试前需保证测试账户余额充足（隔离环境内独立维护，每次运行前对齐到固定高值），"验收"不再是"算力不变"，而是"扣费/退款链路被实际执行（日志可见扣减/增加）+ 测试前余额已对齐 + 不影响生产账户"。

6.2 CDN / 对象存储上传：保留，不改动

自动上传开关保持不动（由测试环境配置文件决定，建议测试环境配置独立 bucket 或本地存储）。各真实上传链路全部保留——CDN 上传也被 E2E 覆盖，且无需任何代码改动，降低复杂度与线上风险。

mock 产出的 /upload/mock/ 结果若被上传到测试 bucket 是预期行为；测试环境与生产 CDN 隔离即可。无需给上传入口加测试模式短路。

6.3 第三方服务槽位：天然不占

§5.4 的 mock 短路在获取槽位之前 return，mock 任务根本不获取槽位，槽位数据中无 mock 记录；轮询调度器 mock 分支的释放槽位调用仅作保险。无需任何 bypass 开关。

7. 配置缓存一致性（critical）

事实：动态配置缓存是进程级内存字典，TTL 30 秒；配置失效/重载接口只清当前进程缓存。

受影响：

同步任务进程池的 worker（独立进程，独立缓存）。
任何已派生的 worker 在 30s 内可能读到旧的 test_mode.enabled。

应对（必须执行其一）：

主进程预判（推荐）：视觉同步任务的 mock 判定放在主进程的同步分流之前。若 mock，则不提交子进程，直接在主进程写 mock 结果。这样子进程缓存不影响 mock 判定。
fixture 操作时序：开 test_mode 后调用配置重载接口，并重启同步任务进程池（shutdown + start，让 worker 重新派生），或等待 >30s 再提交任务。
批量配置接口注意：批量更新接口更新后不自动刷缓存（单条更新会刷）。用批量更新后必须再调重载接口。

主进程的后台调度器是线程非进程，与主进程共享缓存，30s TTL 对其可接受；问题主要在同步任务子进程。故方案 1 优先。

8. 任务清理豁免（防止 mock 任务被误杀）

任务清理逻辑不区分 mock 任务：

逻辑	对 mock 的影响
过期清理（默认 7 天）	历史残留 mock 任务可能过期失败
最大重试（默认 30 次）	失败注入会累加尝试次数，30 次后强制失败
孤儿重置（外部任务 ID 为空）	mock 同步任务子进程异常会被重置回待处理，丢尝试次数一致性

应对：

E2E fixture 创建任务时刷新创建时间为当前时间，避免误过期。
在过期检查与最大重试检查开头加 mock 任务 ID 豁免：

from task.mock_interceptor import is_mock_id
if is_mock_id(getattr(task, 'project_id', None)) or is_mock_id(getattr(ai_tool, 'project_id', None)):
    return False

失败注入路径建议不累加尝试次数（mock 失败是受控注入，不应消耗重试额度）。

9. PARAM_PREPARE 死锁与 pipeline 时序（critical）

事实：提交入口最开头就检查是否存在待处理的参数准备步骤（param_prepare），有则推进"等待参数准备"状态并返回——mock 短路必须在此之前。

因此 §5.1 的短路必须在 param_prepare 检查之前。

额外要求：当任务类型带人脸遮盖（param_prepare）时，mock 主流程短路后直接成功，不进 pipeline，因此人脸遮盖步骤不会创建。这是期望行为（E2E 不需要真实遮罩）。若 E2E 需测人脸遮盖链路本身，则该任务类型的 mock 应不在参数准备之前短路，而是让人脸遮盖步骤走 §5.4/§5.8 的 mock，再让主任务 mock——此时需确保主任务在结果应用后回到待处理时仍命中 §5.1 短路（基于 test_mode 判断，与模型服务无关，满足）。

10. 重试机制 / 重试尝试记录

链路：重试 pipeline 驱动切换模型服务并设回待处理重新提交。

要求：

mock 短路必须基于 test_mode 开关判断，不基于模型服务名——这样重试切换模型服务后第二次提交仍命中 §5.1 短路。
重试尝试记录在提交/同步成功/过期/最大重试/收尾等多处读写。mock 任务可创建尝试记录但不应影响判定。建议 mock 成功时正常标记当前尝试为完成，保持数据一致性；无需冻结。

11. 前端 / 后端开关同步

前端 TEST_MODE（URL ?test=1）与后端 test_mode.enabled 是两套独立机制，且 mock 任务 ID 前缀不同（前端 mock_project_、后端 mock_task_）。

E2E fixture 要求：同时开启——

后端：test_mode.enabled=true（通过管理配置接口）。
前端：访问 URL 带 ?test=1。
明确职责：后端为唯一真相源；前端 TEST_MODE 仅做 UI 乐观渲染，不应在 ?test=1 下向后端提交"真实"任务（否则前后端 mock 体系错位）。

12. 预设资源落地脚本

新建 scripts/prepare_mock_assets.py（或手动放置）：

"""
准备 E2E mock 预设资源到 upload/mock/。
需人工提供真实样本文件（视频/音频/2x2图），脚本仅做拷贝与校验。
"""
import os, shutil, sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DST = os.path.join(ROOT, "upload", "mock")

# (源文件, 目标相对路径) —— 源文件由实施者准备
ASSETS = [
    ("auto_test/samples/e2e_text_to_image.png", "e2e_text_to_image.png"),
    ("auto_test/samples/e2e_image_edit.png",    "e2e_image_edit.png"),
    ("auto_test/samples/e2e_grid_2x2.png",      "e2e_grid_2x2.png"),
    ("auto_test/samples/e2e_i2v.mp4",           "e2e_i2v.mp4"),
    ("auto_test/samples/e2e_t2v.mp4",           "e2e_t2v.mp4"),
    ("auto_test/samples/e2e_dh.mp4",            "e2e_dh.mp4"),
    ("auto_test/samples/e2e_face_mask.mp4",     "e2e_face_mask.mp4"),
    ("auto_test/samples/e2e_tts.mp3",           "e2e_tts.mp3"),
    ("auto_test/samples/e2e_char.mp3",          "e2e_char.mp3"),
    ("auto_test/samples/e2e_ma_front.png",      "e2e_ma_front.png"),
    ("auto_test/samples/e2e_ma_side.png",       "e2e_ma_side.png"),
    ("auto_test/samples/e2e_ma_back.png",       "e2e_ma_back.png"),
    # 世界导入测试资产（由 §5.10 方式 1 从已播种世界导出生成后放入 auto_test/samples/）
    ("auto_test/samples/world_export_sample.zip", "world_export_sample.zip"),
]

def main():
    os.makedirs(DST, exist_ok=True)
    missing = []
    for src, name in ASSETS:
        s = os.path.join(ROOT, src)
        d = os.path.join(DST, name)
        if not os.path.exists(s):
            missing.append(src); continue
        shutil.copy2(s, d)
        print(f"OK  {d}")
    if missing:
        print(f"\n缺失样本（请准备后重跑）: {missing}", file=sys.stderr); sys.exit(1)

if __name__ == "__main__":
    main()

配置写入脚本 scripts/enable_test_mode.py：直接调 config.config_util.set_dynamic_config_value（用配置模型 upsert，可自动新建配置项；而单条更新接口在配置不存在时报 404）。

import os
from config.config_util import set_dynamic_config_value, invalidate_dynamic_cache

# ⚠️ bool 值必须传 Python bool（True/False），不能传字符串 "false"——
# 内部按"非空即真"序列化，非空字符串 "false" 会被当 True 写成 'true'。
KV = {
    ("test_mode", "enabled"): (True, "bool"),
    ("test_mode", "mock_images", "text_to_image"): ("/upload/mock/e2e_text_to_image.png", "string"),
    ("test_mode", "mock_images", "image_edit"):    ("/upload/mock/e2e_image_edit.png", "string"),
    ("test_mode", "mock_images", "comfyui_text_to_image"): ("/upload/mock/e2e_comfyui_tti.png", "string"),
    ("test_mode", "mock_images", "comfyui_image_edit"):    ("/upload/mock/e2e_comfyui_ie.png", "string"),
    ("test_mode", "mock_images", "grid_image"):    ("/upload/mock/e2e_grid_2x2.png", "string"),
    ("test_mode", "mock_videos", "image_to_video"):("/upload/mock/e2e_i2v.mp4", "string"),
    ("test_mode", "mock_videos", "text_to_video"): ("/upload/mock/e2e_t2v.mp4", "string"),
    ("test_mode", "mock_videos", "digital_human"): ("/upload/mock/e2e_dh.mp4", "string"),
    ("test_mode", "mock_videos", "face_mask"):     ("/upload/mock/e2e_face_mask.mp4", "string"),
    ("test_mode", "mock_audio", "tts"):            ("/upload/mock/e2e_tts.mp3", "string"),
    ("test_mode", "mock_audio", "character_audio"):("/upload/mock/e2e_char.mp3", "string"),
    # multi_angle_* 按需补充
}

for keys, (val, vtype) in KV.items():
    set_dynamic_config_value(*keys, value=val, value_type=vtype)

invalidate_dynamic_cache()  # 清当前进程缓存；跨进程见 §7

# 测试账户余额在隔离环境内独立维护（见 §6.1），此处不做重置操作
print("test_mode enabled & mock URLs written.")

13. E2E fixture 模板（conftest.py）

import os
import pytest
from config.config_util import (
    set_dynamic_config_value,
    get_dynamic_config_value,
    invalidate_dynamic_cache,   # 签名 invalidate_dynamic_cache(config_key: str = None)
)

# ⚠️ bool 必须传 Python True/False，不能传 "false" 字符串（见 §12）。
# ⚠️ 算力与 CDN 均不绕过（见 §6）：不关自动上传开关，不设扣费旁路。
TEST_USER_ID = int(os.environ.get("E2E_TEST_USER_ID", "0"))


def _invalidate(k):
    """动态配置用点分字符串做缓存 key。"""
    invalidate_dynamic_cache(".".join(k))


@pytest.fixture(scope="session", autouse=True)
def e2e_mock_setup():
    # 仅保存/恢复 test_mode.enabled（mock URL 留着无害）
    saved_enabled = get_dynamic_config_value("test_mode", "enabled", default=False)

    # 1) 开启挡板（传 Python bool）
    set_dynamic_config_value("test_mode", "enabled", value=True, value_type="bool")
    _invalidate(("test_mode", "enabled"))

    # 2) 测试账户余额在隔离环境内独立维护（§6.1），保证运行期间不被扣穿

    # 3) 重启同步任务进程池，使子进程重新派生并读到新 test_mode（§7）
    from task.sync_task_executor import get_sync_task_executor
    exe = get_sync_task_executor()
    if exe.is_running():
        exe.shutdown(wait=False)
        exe.start()

    yield

    # teardown：仅恢复 test_mode.enabled（不清理数据，见 §14）
    set_dynamic_config_value("test_mode", "enabled",
                             value=bool(saved_enabled), value_type="bool")
    _invalidate(("test_mode", "enabled"))

说明：

算力扣费、token 计费、CDN 上传均保留真实执行（§6），fixture 不再关闭上传开关、不设旁路开关。
测试账户余额在隔离环境内独立维护，保证运行期间不被扣穿。
不做数据清理（§14）：基础数据采用幂等播种，删除会破坏关联性。
fixture 直接在测试进程内调配置写入（不经 HTTP，无需管理员权限）。

14. 测试数据：幂等播种，不做清理

决策：不做删除清理——担心破坏外键/业务关联性。基础数据改为幂等播种：播种前先查存在性，存在则跳过、不存在才新建。E2E 在隔离环境运行，累积的测试数据可接受。

原则：

所有"准备基础数据"的步骤（测试用户、world、character、location、workflow、session 等）都必须先查后建（create-if-not-exists）。
业务任务数据（视觉任务/异步任务/四宫格任务等，均为相关数据表）由被测流程自然产生，无需预置、也不清理。
若需识别某次运行的产物用于排查，创建时在附加配置/消息字段打 is_e2e_test=true 标记（仅用于查询定位，不用于删除）。

幂等播种示例（伪代码，按各模型实际查询/创建方法实现）：

def ensure_world(user_id, world_name):
    world = get_world_by_name(user_id, world_name)
    if world:
        return world.id                 # 已存在，直接复用
    return create_world(user_id=user_id, name=world_name, ...)

def ensure_character(world_id, name, **fields):
    char = get_character_by_name(world_id, name)
    if char:
        return char.id
    return create_character(world_id=world_id, name=name, **fields)

若未来确需清理某次失败的脏数据，应针对该次运行标记的记录定点删除，并严格遵守依赖顺序（先删子表再删父表），且仅在隔离环境执行。本期默认不提供批量清理脚本。

15. 可观测性

拦截器已内置进程内计数（mock_hit_summary()）。E2E 结束时输出：

from task.mock_interceptor import mock_hit_summary
print("[MOCK SUMMARY]", mock_hit_summary())

预期每个被测通道至少命中 1 次；若某通道计数为 0，说明该路径漏到真实外部服务。channel 维度：visual_async_submit/poll、visual_sync_submit、comfyui_submit/poll、async_submit:<impl>。

16. 验收清单（逐条可验证）

Phase 1：不真实调用外部 API

[ ] 各同步视觉驱动（全部同步模式对应驱动）在测试模式下不请求第三方图像生成服务。

[ ] Agent 工具不真实 POST ComfyUI（文生图 / 图片编辑）。

[ ] 四宫格链路不真实 POST/GET ComfyUI（含失败重试）。

[ ] 多角度链路不真实 POST ComfyUI、不远程下载图片。

[ ] 第三方服务异步链路不真实上传文件、不提交/轮询第三方服务。

[ ] 时间轴导出不远程下载片段、不刷新 CDN 签名。

[ ] 音频任务最终结果 URL 是 mock 本地路径（非生成工具虚构的拼接路径）。

[ ] 日志中所有 mock 命中均有 [MOCK]；mock_hit_summary() 覆盖所有被测通道。

[ ] 世界导出返回下载 URL，下载 zip 含 metadata.json 且计数正确（真实代码，图床上传保留）。

[ ] 世界导入用预置 zip 成功，返回计数与落盘文件一致（真实代码，零外部依赖）。

[ ] 时间轴导出在本地 mock 素材上成功生成草稿 zip（真实代码）。

Phase 2：副作用可控（不绕过，靠隔离环境）

[ ] 媒体算力扣费/退款链路被真实执行（日志可见扣减/增加）。

[ ] LLM token 计费被真实执行（日志可见 token 扣减）。

[ ] 测试前测试账户余额充足，运行期间不被扣穿。

[ ] CDN 上传在测试环境正常运行，不触及生产 bucket（环境隔离）。

[ ] 槽位数据无 mock 占用残留（mock 短路天然不获取）。

[ ] teardown 后 test_mode.enabled 恢复原值；基础数据幂等播种，无删除。

Phase 3（可选）：错误注入

[ ] inject_failure_rate>0 时能稳定复现 submit/poll/timeout 失败与重试成功，且失败路径正确退款（扣减→增加，净额为零）。

17. 实施分阶段

Phase 1（核心，防止真实外部调用）

新建 mock 拦截器模块（§3）
扩展默认配置（§4.1）
视觉异步链路：提交/轮询/成功路径三处短路（§5.1）
视觉同步链路短路（§5.2）
音频链路短路（§5.3）
第三方服务异步统一拦截点（§5.4）
ComfyUI 工具直调短路（§5.5）
四宫格轮询/重试短路（§5.6）
多角度提交/下载短路 + mock 角度推进（§5.7）
时间轴导出——无需改动（§5.9，上游产出本地路径即可）
资源准备脚本 + 落盘样本 + 预置世界导出 zip（§5.10 方式 1）

Phase 2（副作用可控：不绕过，靠隔离环境）

测试模式开启脚本（§6.1 + §12）
CDN/算力/token 不改动，保留真实链路（§6.1/§6.2）
任务清理豁免（§8，防 mock 任务被误杀）
缓存一致性处理（§7：主进程预判 + 重启进程池）
fixture 模板（§13）+ 基础数据幂等播种（§14，无删除）

Phase 3（错误注入，可选）

inject_failure_rate/scenario 状态机
重试链路 E2E 用例

18. 附录：关键事实速查

TaskCategory 是字符串常量类非 Enum；多分类判断用"主分类 + 附加分类"列表。
同步模式驱动共 13 个（多个第三方图像生成模型对应驱动）。
下载缓存工具不识别本地路径，调用方须自行判断本地短路。
多角度任务轮询走数据库，非状态轮询接口。
已迁移走调度器进程的历史轮询代码是死代码，勿在其上加拦截。
音频通道不扣费；扣费在计费服务（媒体）与 token 任务（LLM token）两套独立后端。
动态配置缓存 TTL 30 秒，跨进程独立（§7）。
异步任务外部任务 ID 字段长度 100，mock 任务 ID（前缀 + 16 位 hex）远小于上限。
管理配置接口中：批量更新不自动刷缓存；单条更新会刷；另有配置重载接口。
配置写入用 upsert 可新建配置项；单条更新接口在配置不存在时报 404。
bool 配置必须传 Python True/False：内部按"非空即真"序列化，非空字符串 "false" 会被当真值写成 'true'。
世界导出/导入无需 mock：导出仅图床上传（§6.2 保留），导入纯本地 zip 处理；测试需预置合法世界导出 zip（结构见 §5.10）。
算力/CDN/token 均不绕过：靠独立隔离环境 + 测试前对齐测试账户余额；不做数据删除，基础数据幂等播种（§6/§14）。

实施完成后，运行 §16 验收清单逐条勾选；任一未通过即对应链路拦截未生效。