refactor(phase1): 统一渲染入口 + 模板系统双读兼容 (#630)
CI/CD Pipeline / Unit Tests (push) Failing after 5m2s
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Failing after 1394h49m36s
CI/CD Pipeline / Build Production Worker Image (push) Failing after 1394h49m37s
CI/CD Pipeline / Build Production Web Image (push) Failing after 1394h49m37s
CI/CD Pipeline / Check if frontend-only change (push) Failing after 1394h49m39s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been skipped
CI/CD Pipeline / Integration Tests (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Has been skipped
CI/CD Pipeline / Build Staging Web Image (push) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Failing after 1395h21m45s
CI/CD Pipeline / Build Production API Image (push) Failing after 1395h21m46s
CI/CD Pipeline / ACR Image Cleanup (push) Has been skipped

Co-authored-by: xiaoxia <dev@xiaoxiajianji.com>
Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
This commit was merged in pull request #630.
This commit is contained in:
2026-07-20 12:39:43 +08:00
committed by auto-approve-bot
parent 8dc8de0c59
commit e5fffe40c1
3 changed files with 524 additions and 117 deletions
+210
View File
@@ -1151,3 +1151,213 @@ class TestRenderPlanWithBgmAsr:
assert "c_bad" in result.failed_clip_ids
assert len(result.rendered_clip_ids) == 2
assert len(result.failed_clip_ids) == 1
# ── render_from_memory 测试 ──────────────────────────────────────────────────
class TestRenderFromMemory:
"""render_from_memory 内存模式渲染测试。"""
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
def test_successful_render(self, mock_render_cls, mock_upload, tmp_path):
"""内存模式渲染成功。"""
mock_render = MagicMock()
mock_render.render.return_value = MagicMock(
output_path=tmp_path / "out.mp4",
duration=10.0,
file_size=1024,
width=1280,
height=720,
)
mock_render_cls.return_value = mock_render
mock_upload.return_value = "https://oss.example.com/out.mp4"
plan = FakePlan(id="mem_plan_001", config={"bgm": {"enabled": False}})
clips = [
_make_clip("c1", order=0, duration=5.0),
_make_clip("c2", order=1, duration=5.0),
]
asset_path_map = {
"asset_c1.mp4": tmp_path / "c1.mp4",
"asset_c2.mp4": tmp_path / "c2.mp4",
}
# 创建假文件
for p in asset_path_map.values():
p.write_bytes(b"fake")
adapter, _, _ = _make_adapter()
result = adapter.render_from_memory(
plan=plan,
clips=clips,
asset_path_map=asset_path_map,
plan_id="mem_plan_001",
work_dir=tmp_path / "work",
)
assert result.success
assert result.duration == 10.0
assert result.file_size == 1024
assert result.width == 1280
assert result.height == 720
assert result.clip_count == 2
assert len(result.rendered_clip_ids) == 2
assert len(result.failed_clip_ids) == 0
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
def test_empty_clips_returns_failure(self, mock_render_cls, mock_upload, tmp_path):
"""clips 为空时返回失败。"""
plan = FakePlan(id="mem_empty")
adapter, _, _ = _make_adapter()
result = adapter.render_from_memory(
plan=plan,
clips=[],
asset_path_map={},
plan_id="mem_empty",
work_dir=tmp_path / "work",
)
assert not result.success
assert "没有可渲染的片段" in result.error_message
mock_render_cls.assert_not_called()
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
def test_empty_asset_map_returns_failure(self, mock_render_cls, mock_upload, tmp_path):
"""asset_path_map 为空时返回失败。"""
plan = FakePlan(id="mem_no_assets")
clips = [_make_clip("c1", order=0, duration=5.0)]
adapter, _, _ = _make_adapter()
result = adapter.render_from_memory(
plan=plan,
clips=clips,
asset_path_map={},
plan_id="mem_no_assets",
work_dir=tmp_path / "work",
)
assert not result.success
assert "素材路径映射为空" in result.error_message
mock_render_cls.assert_not_called()
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
def test_bgm_prepared_for_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
"""内存模式下 BGM 配置也会被正确处理。"""
mock_render = MagicMock()
mock_render.render.return_value = MagicMock(
output_path=tmp_path / "out.mp4",
duration=5.0,
file_size=512,
width=1280,
height=720,
)
mock_render_cls.return_value = mock_render
mock_upload.return_value = "https://oss.example.com/out.mp4"
plan = FakePlan(
id="mem_bgm",
config={
"bgm": {
"enabled": True,
"preset_id": "preset_001",
}
},
)
clips = [_make_clip("c1", order=0, duration=5.0)]
asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
asset_path_map["asset_c1.mp4"].write_bytes(b"fake")
adapter, _, _ = _make_adapter()
# mock _prepare_bgm 返回一个假的 bgm 路径
fake_bgm_path = tmp_path / "bgm.mp3"
fake_bgm_path.write_bytes(b"fake bgm")
with patch.object(adapter, "_prepare_bgm", return_value=str(fake_bgm_path)):
result = adapter.render_from_memory(
plan=plan,
clips=clips,
asset_path_map=asset_path_map,
plan_id="mem_bgm",
work_dir=tmp_path / "work",
)
assert result.success
call_kwargs = mock_render_cls.call_args
assert call_kwargs.kwargs["bgm_path"] == str(fake_bgm_path)
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
def test_resolution_from_config_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
"""内存模式下从 plan.config.export.resolution 读取分辨率。"""
mock_render = MagicMock()
mock_render.render.return_value = MagicMock(
output_path=tmp_path / "out.mp4",
duration=5.0,
file_size=512,
width=720,
height=1280,
)
mock_render_cls.return_value = mock_render
mock_upload.return_value = "https://oss.example.com/out.mp4"
plan = FakePlan(
id="mem_res",
config={"export": {"resolution": "720x1280"}},
)
clips = [_make_clip("c1", order=0, duration=5.0)]
asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
asset_path_map["asset_c1.mp4"].write_bytes(b"fake")
adapter, _, _ = _make_adapter()
result = adapter.render_from_memory(
plan=plan,
clips=clips,
asset_path_map=asset_path_map,
plan_id="mem_res",
work_dir=tmp_path / "work",
)
assert result.success
call_kwargs = mock_render_cls.call_args
assert call_kwargs.kwargs["output_width"] == 720
assert call_kwargs.kwargs["output_height"] == 1280
@patch("video_processing.render_adapter.upload_to_oss")
@patch("video_processing.render_adapter.UnifiedRenderService")
def test_ffmpeg_error_captured_memory_mode(self, mock_render_cls, mock_upload, tmp_path):
"""内存模式下 FFmpeg 错误也会被正确捕获。"""
import subprocess
mock_render = MagicMock()
mock_render.render.side_effect = subprocess.CalledProcessError(
returncode=1,
cmd=["ffmpeg", "-i", "input.mp4", "output.mp4"],
stderr="test error output",
)
mock_render_cls.return_value = mock_render
plan = FakePlan(id="mem_err")
clips = [_make_clip("c1", order=0, duration=5.0)]
asset_path_map = {"asset_c1.mp4": tmp_path / "c1.mp4"}
asset_path_map["asset_c1.mp4"].write_bytes(b"fake")
adapter, _, _ = _make_adapter()
result = adapter.render_from_memory(
plan=plan,
clips=clips,
asset_path_map=asset_path_map,
plan_id="mem_err",
work_dir=tmp_path / "work",
)
assert not result.success
assert "FFmpeg渲染失败" in result.error_message
assert "test error output" in result.error_detail