refactor: 删除legacy渲染引擎feature flag测试

tests/unit/test_generation_feature_flag.py

@@ -1,332 +0,0 @@
-"""generate_video 任务 Feature Flag 灰度引擎选择单元测试.
-
-覆盖：
-- _resolve_render_engine 正常返回 unified / legacy
-- Feature Flag 不可用时 fallback 到 unified
-- 白名单 / 百分比 / 全局开关各场景
-- _render_with_legacy_engine 命令构建与输出验证
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-from types import ModuleType
-from unittest.mock import MagicMock
-
-os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
-os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
-
-from pathlib import Path
-
-# ── Mock worker 模块以避免数据库连接 ──────────────────────────────────────────
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "worker"))
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
-
-_mock_db_mod = ModuleType("worker_app.db")
-_mock_db_mod.SessionLocal = MagicMock()
-sys.modules.setdefault("worker_app.db", _mock_db_mod)
-
-_mock_celery_mod = ModuleType("worker_app.celery_app")
-_mock_celery_app = MagicMock()
-_mock_celery_app.task = lambda **kwargs: lambda fn: fn
-_mock_celery_mod.celery_app = _mock_celery_app
-sys.modules.setdefault("worker_app.celery_app", _mock_celery_mod)
-
-# Mock worker_app.core.config 避免 settings 加载
-_mock_config_mod = ModuleType("worker_app.core.config")
-_mock_settings = MagicMock()
-_mock_settings.redis_url = None
-_mock_settings.render_engine = "unified"
-_mock_config_mod.get_settings = lambda: _mock_settings
-sys.modules.setdefault("worker_app.core", ModuleType("worker_app.core"))
-sys.modules.setdefault("worker_app.core.config", _mock_config_mod)
-
-
-# ── 测试用数据类 ──────────────────────────────────────────────────────────────
-
-
-class _TestClip:
-    def __init__(self, asset_id, duration=30.0, clip_type="main", config=None, order=0):
-        self.id = f"clip_{asset_id}"
-        self.plan_id = "test-plan"
-        self.clip_type = clip_type
-        self.order = order
-        self.asset_id = asset_id
-        self.duration = duration
-        self.config = config or {}
-        self.start_time = 0.0
-        self.transition_effect = "cut"
-
-
-# ── RenderEngineResolver 基础行为测试 ───────────────────────────────────────
-
-
-def test_resolver_unified_when_enabled_100_percent():
-    """flag 全局开启（percentage=100）时，返回 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "unified"
-
-
-def test_resolver_legacy_when_flag_disabled():
-    """flag 全局关闭时，返回默认引擎 legacy。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=100))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "legacy"
-
-
-def test_resolver_whitelist_overrides_percentage_0():
-    """白名单用户即使 percentage=0 也走 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(
-        FeatureFlagConfig(
-            name="render_engine",
-            enabled=True,
-            percentage=0,
-            whitelist={"user-vip"},
-        )
-    )
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    assert resolver.get_engine(user_id="user-vip") == "unified"
-    assert resolver.get_engine(user_id="user-other") == "legacy"
-
-
-def test_resolver_percentage_0_all_legacy():
-    """percentage=0 且无白名单时，全部走 legacy。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=0))
-    resolver = RenderEngineResolver(default_engine="legacy", store=store)
-
-    for i in range(50):
-        assert resolver.get_engine(user_id=f"user-{i}") == "legacy"
-
-
-def test_resolver_default_unified_when_flag_off():
-    """默认引擎设为 unified 且 flag 关闭时，返回 unified。"""
-    from video_processing.render_engine_resolver import RenderEngineResolver
-
-    from packages.adapters.redis.feature_flag_store import (
-        FeatureFlagConfig,
-        InMemoryFeatureFlagStore,
-    )
-
-    store = InMemoryFeatureFlagStore()
-    store.set(FeatureFlagConfig(name="render_engine", enabled=False, percentage=0))
-    resolver = RenderEngineResolver(default_engine="unified", store=store)
-
-    assert resolver.get_engine(user_id="user-123") == "unified"
-
-
-# ── _render_with_legacy_engine 集成测试 ──────────────────────────────────────
-
-
-def test_legacy_engine_single_clip_keeps_original_fps():
-    """单 clip 场景：输出保持原帧率（不做 fps 归一化），分辨率缩放正确。"""
-    import subprocess
-    import tempfile
-
-    from video_processing.ffmpeg_utils import probe_video_info
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input_path = tmp_path / "input.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        # 生成 1 秒 30fps 测试视频（带音频）
-        subprocess.run(
-            [
-                "ffmpeg",
-                "-y",
-                "-f",
-                "lavfi",
-                "-i",
-                "color=c=red:s=640x360:d=1:r=30",
-                "-f",
-                "lavfi",
-                "-i",
-                "anullsrc=r=44100:cl=stereo:d=1",
-                "-c:v",
-                "libx264",
-                "-pix_fmt",
-                "yuv420p",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(input_path),
-            ],
-            check=True,
-            capture_output=True,
-        )
-
-        clip = _TestClip(asset_id="asset-1", duration=1.0)
-        asset_path_map = {"asset-1": input_path}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert duration > 0
-
-        # 旧引擎保持原帧率（30fps），不做 fps 归一化
-        info = probe_video_info(str(output_path))
-        assert abs(info.get("fps", 0) - 30.0) < 0.5
-        assert info.get("width") == 1280
-        assert info.get("height") == 720
-
-
-def test_legacy_engine_two_clips_concat_duration():
-    """多 clip 场景：concat 后时长为两片段之和。"""
-    import subprocess
-    import tempfile
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input1 = tmp_path / "input1.mp4"
-        input2 = tmp_path / "input2.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        for idx, inp in enumerate([input1, input2]):
-            color = "red" if idx == 0 else "blue"
-            subprocess.run(
-                [
-                    "ffmpeg",
-                    "-y",
-                    "-f",
-                    "lavfi",
-                    "-i",
-                    f"color=c={color}:s=640x360:d=1:r=30",
-                    "-f",
-                    "lavfi",
-                    "-i",
-                    "anullsrc=r=44100:cl=stereo:d=1",
-                    "-c:v",
-                    "libx264",
-                    "-pix_fmt",
-                    "yuv420p",
-                    "-c:a",
-                    "aac",
-                    "-shortest",
-                    str(inp),
-                ],
-                check=True,
-                capture_output=True,
-            )
-
-        clip1 = _TestClip(asset_id="asset-1", duration=1.0, clip_type="main", order=0)
-        clip2 = _TestClip(asset_id="asset-2", duration=1.0, clip_type="main", order=1)
-        asset_path_map = {"asset-1": input1, "asset-2": input2}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip1, clip2],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert abs(duration - 2.0) < 0.2
-
-
-def test_legacy_engine_broll_mode_supported():
-    """b_roll 类型的 clip 也被正确识别为主图层并渲染。"""
-    import subprocess
-    import tempfile
-
-    from apps.worker.worker_app.tasks.generation import _render_with_legacy_engine
-
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmp_path = Path(tmpdir)
-        input_path = tmp_path / "input.mp4"
-        output_path = tmp_path / "output.mp4"
-
-        subprocess.run(
-            [
-                "ffmpeg",
-                "-y",
-                "-f",
-                "lavfi",
-                "-i",
-                "color=c=green:s=640x360:d=1:r=30",
-                "-f",
-                "lavfi",
-                "-i",
-                "anullsrc=r=44100:cl=stereo:d=1",
-                "-c:v",
-                "libx264",
-                "-pix_fmt",
-                "yuv420p",
-                "-c:a",
-                "aac",
-                "-shortest",
-                str(input_path),
-            ],
-            check=True,
-            capture_output=True,
-        )
-
-        clip = _TestClip(
-            asset_id="asset-1",
-            duration=1.0,
-            clip_type="main",
-            config={"role": "b_roll"},
-        )
-        asset_path_map = {"asset-1": input_path}
-
-        duration, file_size = _render_with_legacy_engine(
-            task_id="test-task",
-            virtual_clips=[clip],
-            asset_path_map=asset_path_map,
-            work_dir=tmp_path,
-            output_path=output_path,
-        )
-
-        assert output_path.exists()
-        assert file_size > 0
-        assert duration > 0