refactor: 简化回归测试为单函数文本检查

tests/unit/test_empty_plan_id_marks_failed.py

@@ -1,58 +1,32 @@
-"""回归测试：source_edit_plan_id 为空时，generate_video 必须调用 mark_failed。
+"""回归测试：source_edit_plan_id 为空时任务必须被标记为 failed。
 
-背景 (2026-08-24)：确认生成卡在 10% 永不结束。根因是 worker 在
-source_edit_plan_id 为空时直接 return failed 字典，但没有调用
-_update_task_status(task_id, "mark_failed")，导致 DB 状态永远停在
-running，前端轮询永远等不到完成。
+背景 (2026-08-24)：确认生成卡在 10%。根因是 worker 在
+source_edit_plan_id 为空时直接 return failed，但没有调用 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 分支源码。"""
+def test_mark_failed_in_else_branch():
+    """generation.py 中 source_edit_plan_id 为空的 else 分支必须调用 mark_failed。"""
     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: 分支")
+    # 定位 else 分支：紧跟在 'source_edit_plan_id 为空' 日志之后的 else 块
+    marker = "source_edit_plan_id 为空"
+    idx = source.find(marker)
+    assert idx != -1, f"generation.py 中未找到 '{marker}'"
 
+    # 从 marker 位置向后搜索到下一个 return 语句
+    after_marker = source[idx:]
+    return_idx = after_marker.find("return {")
+    assert return_idx != -1, "else 分支中未找到 return 语句"
 
-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}) 之前"
+    # 关键断言：marker 和 return 之间必须包含 mark_failed
+    block = source[idx:idx + return_idx]
+    assert "mark_failed" in block, (
+        "else 分支在 return 之前必须调用 _update_task_status(task_id, "
+        "'mark_failed', ...) 以更新 DB 状态，否则任务永远卡在 running"
+    )