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 re
from pathlib import Path


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


def _read_source() -> str:
    return GENERATION_PY.read_text(encoding="utf-8")


class TestEmptyPlanIdMarksFailed:
    """验证 generation.py 在 source_edit_plan_id 为空的 else 分支中调用了 mark_failed。"""

    def test_else_branch_has_mark_failed(self):
        """else 分支（source_edit_plan_id 为空）必须在 return 之前调用 mark_failed。"""
        source = _read_source()

        # 找到 else 分支: 包含 source_edit_plan_id 为空 的日志
        else_pattern = re.compile(
            r"else:\s*\n"
            r"\s+logger\.error\("
            r"[^)]*source_edit_plan_id",
            re.DOTALL,
        )
        assert else_pattern.search(source), "未找到 source_edit_plan_id 为空的 else 分支"

        else_match = else_pattern.search(source)
        else_start = else_match.start()
        else_block = source[else_start:else_start + 800]

        return_match = re.search(r'return\s*\{[^}]*"status"\s*:\s*"failed"', else_block, re.DOTALL)
        assert return_match, "else 分支中未找到 return failed"

        block_before_return = else_block[:return_match.start()]
        assert "mark_failed" in block_before_return, (
            "source_edit_plan_id 为空的 else 分支在 return 之前必须调用 mark_failed"
        )

    def test_mark_failed_before_return_in_else(self):
        """mark_failed 调用必须在 return 之前，而非之后。"""
        source = _read_source()

        else_idx = source.find("source_edit_plan_id 为空，无法渲染")
        assert else_idx != -1, "未找到 source_edit_plan_id 为空的日志"

        after_else = source[else_idx:]
        mark_failed_idx = after_else.find('"mark_failed"')
        return_failed_idx = after_else.find('"status": "failed"')

        assert mark_failed_idx != -1, "else 分支中没有 mark_failed 调用"
        assert return_failed_idx != -1, "else 分支中没有 return failed"
        assert mark_failed_idx < return_failed_idx, (
            f"mark_failed (pos {mark_failed_idx}) 必须在 return failed (pos {return_failed_idx}) 之前"
        )