From b1c341d488ade5c8b2edad01820e5f2c3fddbf58 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Sun, 19 Jul 2026 19:19:56 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=88=A0=E9=99=A4legacy=E6=B8=B2?= =?UTF-8?q?=E6=9F=93=E5=BC=95=E6=93=8Efeature=20flag=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_generation_feature_flag.py | 332 --------------------- 1 file changed, 332 deletions(-) delete mode 100644 tests/unit/test_generation_feature_flag.py diff --git a/tests/unit/test_generation_feature_flag.py b/tests/unit/test_generation_feature_flag.py deleted file mode 100644 index 5c743a598..000000000 --- a/tests/unit/test_generation_feature_flag.py +++ /dev/null @@ -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