"""#1970 PR2 渲染服务微变换注入测试。 不做真实渲染，只验证 UnifiedRenderService 上微变换计划的开关、缓存、 滤镜注入与速度因子；纯参数生成在 test_1970_micro_transform_pure 覆盖。 """ from __future__ import annotations from pathlib import Path from unittest.mock import MagicMock import pytest def _make_service(plan_config: dict | None = None, clips=None): from video_processing.unified_render_service import UnifiedRenderService svc = object.__new__(UnifiedRenderService) svc.plan = MagicMock() svc.plan.config = plan_config or {} svc.plan.id = "plan-1" svc.plan.clips = clips or [] svc._micro_plan_cache = None svc._micro_plan_loaded = False return svc class TestDedupGate: def test_default_enabled_when_config_missing(self): svc = _make_service({}) assert svc._dedup_enabled() is True def test_explicit_true(self): svc = _make_service({"dedup_enabled": True}) assert svc._dedup_enabled() is True def test_explicit_false(self): svc = _make_service({"dedup_enabled": False}) assert svc._dedup_enabled() is False def test_plan_none_config_treated_enabled(self): svc = _make_service(None) svc.plan.config = None assert svc._dedup_enabled() is True class TestPlanBuild: def test_disabled_returns_none_and_cached(self): svc = _make_service({"dedup_enabled": False, "generation_task_id": "t1"}) assert svc._get_micro_transform_plan(5) is None # 第二次走缓存 svc._dedup_enabled = MagicMock(side_effect=AssertionError("不应再次计算")) assert svc._get_micro_transform_plan(5) is None def test_zero_clips_returns_none(self): svc = _make_service({"generation_task_id": "t1"}) assert svc._get_micro_transform_plan(0) is None def test_enabled_builds_reproducible_plan(self): cfg = {"generation_task_id": "task-abc", "video_index": 2, "bgm": {"enabled": True}} svc1 = _make_service(cfg) svc2 = _make_service(dict(cfg)) p1 = svc1._get_micro_transform_plan(6) p2 = svc2._get_micro_transform_plan(6) assert p1 is not None and p2 is not None assert [c.speed for c in p1.clips] == [c.speed for c in p2.clips] assert p1.seed == p2.seed assert len(p1.clips) == 6 def test_p1_conservative_no_hflip(self): svc = _make_service({"generation_task_id": "t1"}) plan = svc._get_micro_transform_plan(30) assert all(not c.hflip for c in plan.clips) def test_no_bgm_config_zero_offset(self): svc = _make_service({"generation_task_id": "t1"}) plan = svc._get_micro_transform_plan(3) assert plan.bgm_start_offset == 0.0 def test_bgm_enabled_offset_in_range(self): svc = _make_service({"generation_task_id": "t1", "bgm": {"enabled": True}}) plan = svc._get_micro_transform_plan(3) assert 2.0 <= plan.bgm_start_offset <= 8.0 class TestFilterInjection: def test_none_mt_noop(self): svc = _make_service({}) from video_processing.unified_render_service import UnifiedRenderService filters = ["scale=100:100"] UnifiedRenderService._apply_micro_transform_video(filters, None) UnifiedRenderService._apply_micro_hflip(filters, None) assert filters == ["scale=100:100"] def test_eq_injection(self): svc = _make_service({"generation_task_id": "t1"}) plan = svc._get_micro_transform_plan(1) mt = plan.clips[0] from video_processing.unified_render_service import UnifiedRenderService filters: list[str] = [] UnifiedRenderService._apply_micro_transform_video(filters, mt) assert filters and filters[0].startswith("eq=brightness=") assert "contrast=" in filters[0] and "saturation=" in filters[0] def test_hflip_skipped_p1(self): svc = _make_service({"generation_task_id": "t1"}) plan = svc._get_micro_transform_plan(10) from video_processing.unified_render_service import UnifiedRenderService for mt in plan.clips: filters: list[str] = [] UnifiedRenderService._apply_micro_hflip(filters, mt) assert filters == [] def test_speed_factor(self): from video_processing.micro_transform_pure import ClipMicroTransform from video_processing.unified_render_service import UnifiedRenderService assert UnifiedRenderService._micro_speed_factor(None) == 1.0 assert UnifiedRenderService._micro_speed_factor(ClipMicroTransform(0, speed=1.025)) == pytest.approx(1.025) assert UnifiedRenderService._micro_speed_factor(MagicMock(speed=0.97)) == pytest.approx(0.97) def test_bgm_offset_reader_respects_flag(self): svc_off = _make_service({"dedup_enabled": False}) assert svc_off._get_micro_bgm_offset() == 0.0 svc_on = _make_service({"generation_task_id": "t1", "bgm": {"enabled": True}}, clips=[MagicMock()]) off = svc_on._get_micro_bgm_offset() assert 2.0 <= off <= 8.0 def test_bgm_offset_zero_without_bgm(self): svc = _make_service({"generation_task_id": "t1"}, clips=[MagicMock()]) assert svc._get_micro_bgm_offset() == 0.0 class TestStreamCopyGate: """dedup 开启时微变换需要重编码，stream copy 必须被拒绝。""" def _build(self, config): from types import SimpleNamespace from unittest.mock import patch from video_processing.unified_render_service import ResolvedClip, UnifiedRenderService source = SimpleNamespace( id="c1", clip_type="main", ) svc = object.__new__(UnifiedRenderService) svc.output_width = 1280 svc.output_height = 720 svc.output_fps = 25 svc.plan = MagicMock() svc.plan.id = "plan-1" svc.plan.config = config svc.plan.clips = [source] svc.clips = [source] svc._micro_plan_cache = None svc._micro_plan_loaded = False resolved = ResolvedClip( clip_id="c1", asset_id="a1", local_path=Path("/tmp/a1.mp4"), clip_type="main", order=0, ) info = { "width": 1280, "height": 720, "fps": 25.0, "video_codec": "h264", "pix_fmt": "yuv420p", "duration": 5.0, "has_audio": True, "audio_codec": "aac", } return svc, resolved, info def test_dedup_enabled_blocks_stream_copy(self): from unittest.mock import patch svc, resolved, info = self._build({"dedup_enabled": True, "generation_task_id": "t1"}) with patch("video_processing.unified_render_service.probe_video_info", return_value=info): can_copy, reason = svc._can_use_stream_copy(resolved) assert can_copy is False assert "微变换" in reason def test_dedup_disabled_allows_stream_copy(self): from unittest.mock import patch svc, resolved, info = self._build({"dedup_enabled": False}) with patch("video_processing.unified_render_service.probe_video_info", return_value=info): can_copy, _ = svc._can_use_stream_copy(resolved) assert can_copy is True