xiaoxia-saas/tests/unit/test_empty_plan_id_marks_failed.py

"""回归测试：source_edit_plan_id 为空时，generate_video 必须调用 mark_failed。

背景 (2026-08-24)：确认生成卡在 10% 永不结束。根因是 worker 在
source_edit_plan_id 为空时直接 return failed 字典，但没有调用
_update_task_status(task_id, "mark_failed")，导致 DB 状态永远停在
running，前端轮询永远等不到完成。
"""

from __future__ import annotations

import ast
from pathlib import Path

GENERATION_PY = Path(__file__).resolve().parents[2] / "apps" / "worker" / "worker_app" / "tasks" / "generation.py"


def _get_else_block_source() -> str:
    """解析 generate_video 函数，提取 if source_edit_plan_id: 的 else 分支源码。"""
    source = GENERATION_PY.read_text(encoding="utf-8")
    lines = source.splitlines(keepends=True)
    tree = ast.parse(source)

    for node in ast.walk(tree):
        if isinstance(node, ast.FunctionDef) and node.name == "generate_video":
            for child in ast.walk(node):
                if isinstance(child, ast.If):
                    test = child.test
                    if isinstance(test, ast.Name) and test.id == "source_edit_plan_id":
                        if not child.orelse:
                            raise AssertionError("if source_edit_plan_id 缺少 else 分支")
                        # 取 else 块第一行的行号和最后一行的行号
                        first_line = child.orelse[0].lineno  # 1-based
                        last_line = child.orelse[-1].end_lineno
                        return "".join(lines[first_line - 1 : last_line])
    raise AssertionError("未找到 if source_edit_plan_id: 分支")


class TestEmptyPlanIdMarksFailed:
    """验证 else 分支（source_edit_plan_id 为空）调用了 mark_failed。"""

    def test_else_branch_calls_mark_failed(self):
        """else 分支必须包含 _update_task_status + mark_failed 调用。"""
        else_source = _get_else_block_source()
        assert "mark_failed" in else_source, (
            "else 分支必须调用 _update_task_status(task_id, 'mark_failed', ...) "
            "以更新 DB 状态为 failed，否则任务永远卡在 running"
        )

    def test_mark_failed_before_return(self):
        """mark_failed 调用必须出现在 return 之前。"""
        else_source = _get_else_block_source()
        mark_failed_pos = else_source.find("mark_failed")
        return_pos = else_source.find("return")
        assert mark_failed_pos != -1, "else 分支中未找到 mark_failed"
        assert return_pos != -1, "else 分支中未找到 return"
        assert (
            mark_failed_pos < return_pos
        ), f"mark_failed (pos {mark_failed_pos}) 必须在 return (pos {return_pos}) 之前"