xiaoxia-saas/tests/unit/test_temp_dir_cleanup_race.py

"""回归测试：渲染产物临时目录不在 render_plan 中提前清理。

根因：render_plan 的 finally 块在返回前清理了临时目录，
但 generation.py 还需要访问其中的文件进行 OSS 上传。
修复：将清理责任交给调用方（generation.py），render_plan 只在失败时清理。
"""

import ast


def test_render_adapter_result_has_temp_dir_field():
    """RenderAdapterResult 包含 temp_dir 字段"""
    with open("apps/worker/video_processing/render_adapter.py") as f:
        source = f.read()
    tree = ast.parse(source)
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef) and node.name == "RenderAdapterResult":
            for item in node.body:
                if isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name):
                    if item.target.id == "temp_dir":
                        return
    raise AssertionError("RenderAdapterResult 缺少 temp_dir 字段")


def test_render_plan_does_not_cleanup_on_success():
    """render_plan 成功时不在 finally 中清理临时目录（通过将 temp_dir 置为 None）"""
    with open("apps/worker/video_processing/render_adapter.py") as f:
        source = f.read()

    # 成功路径必须将 temp_dir 置为 None，以阻止 finally 清理
    assert "temp_dir = None" in source, "render_plan 成功时应将 temp_dir 置为 None 以阻止 finally 清理"


def test_render_plan_passes_temp_dir_to_result():
    """render_plan 将 temp_dir 传递给返回结果"""
    with open("apps/worker/video_processing/render_adapter.py") as f:
        source = f.read()

    assert "result.temp_dir = temp_dir" in source, "render_plan 应将 temp_dir 设置到 result 上"


def test_generation_cleans_up_temp_dir():
    """generation.py 在上传完成后清理临时目录"""
    with open("apps/worker/worker_app/tasks/generation.py") as f:
        source = f.read()

    # 验证 _render_from_edit_plan 返回 temp_dir
    assert "render_temp_dir" in source, "generation.py 应接收 render_temp_dir"

    # 验证有清理逻辑（shutil.rmtree(render_temp_dir...）
    assert "rmtree(render_temp_dir" in source, "generation.py 应清理 render_temp_dir"

    # 验证清理发生在上传之后（通过查找顺序）
    # #1743：_upload_and_record 拆分为 _upload_rendered_video（仅 OSS 上传）
    upload_pos = source.find("_upload_rendered_video")
    cleanup_pos = source.find("rmtree(render_temp_dir")
    assert upload_pos > 0 and cleanup_pos > upload_pos, "清理临时目录应在 _upload_rendered_video 之后执行"