chendelian
857e7e3d73
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 20s
Co-authored-by: Cursor <cursoragent@cursor.com>
576 lines
22 KiB
Python
576 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""RpaVideoSession：ffmpeg 桌面录屏、背景音乐循环淡出、旁白混音。"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
from jiangchang_skill_core import config
|
||
from jiangchang_skill_core.rpa import video_session as vs
|
||
from jiangchang_skill_core.rpa.video_session import (
|
||
RpaVideoSession,
|
||
_StepEntry,
|
||
_INTRO_BUFFER_SECONDS,
|
||
_INTRO_STABILIZE_SECONDS,
|
||
_OUTRO_BUFFER_SECONDS,
|
||
_compose_capture_mp4,
|
||
_music_fade_params,
|
||
_select_voiceover_clips,
|
||
_voiceover_text_for_step,
|
||
)
|
||
|
||
|
||
class TestRpaVideoSession(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
config.reset_cache()
|
||
|
||
def test_disabled_is_noop(self) -> None:
|
||
old = os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
||
try:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="receive-order",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_test",
|
||
)
|
||
self.assertFalse(session.enabled)
|
||
summ = session.summary()
|
||
self.assertFalse(summ["enabled"])
|
||
self.assertIsNone(summ["path"])
|
||
self.assertIsNone(summ["capture_path"])
|
||
self.assertIsNone(summ["record_log_path"])
|
||
self.assertIsNone(summ["voiceover_path"])
|
||
self.assertIsNone(summ["music_path"])
|
||
finally:
|
||
if old is not None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = old
|
||
config.reset_cache()
|
||
|
||
def test_enabled_paths_use_ffmpeg_capture(self) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
try:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="receive-order",
|
||
skill_data_dir=tmp,
|
||
batch_id="exmail_20260602_120000",
|
||
title="接收订单",
|
||
)
|
||
self.assertTrue(session.enabled)
|
||
self.assertTrue(
|
||
session._capture_path.replace("\\", "/").endswith(
|
||
"/rpa-artifacts/exmail_20260602_120000/capture.mp4"
|
||
)
|
||
)
|
||
self.assertTrue(
|
||
session._ffmpeg_record_log_path.replace("\\", "/").endswith(
|
||
"/rpa-artifacts/exmail_20260602_120000/logs/ffmpeg-record.log"
|
||
)
|
||
)
|
||
self.assertTrue(
|
||
session._voiceover_path.replace("\\", "/").endswith(
|
||
"/rpa-artifacts/exmail_20260602_120000/voiceover/voiceover.wav"
|
||
)
|
||
)
|
||
self.assertTrue(
|
||
session._planned_output.replace("\\", "/").startswith(
|
||
tmp.replace("\\", "/") + "/videos/"
|
||
)
|
||
)
|
||
finally:
|
||
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
||
config.reset_cache()
|
||
|
||
@unittest.skipUnless(sys.platform == "win32", "gdigrab desktop capture")
|
||
@patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
|
||
@patch("jiangchang_skill_core.rpa.video_session.subprocess.Popen")
|
||
def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, mock_resolve: MagicMock) -> None:
|
||
mock_resolve.return_value = Path(r"C:\media\bin\win32-x64\ffmpeg.exe")
|
||
proc = MagicMock()
|
||
proc.poll.return_value = None
|
||
proc.stdin = MagicMock()
|
||
mock_popen.return_value = proc
|
||
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
try:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
|
||
async def _run() -> None:
|
||
async with RpaVideoSession(
|
||
skill_slug="receive-order",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_ff",
|
||
) as video:
|
||
video.add_step("测试步骤")
|
||
|
||
with patch.object(RpaVideoSession, "finalize", return_value=None):
|
||
asyncio.run(_run())
|
||
|
||
self.assertTrue(mock_popen.called)
|
||
cmd = mock_popen.call_args[0][0]
|
||
self.assertTrue(str(cmd[0]).endswith("ffmpeg.exe"))
|
||
self.assertIn("gdigrab", cmd)
|
||
self.assertIn("desktop", cmd)
|
||
kwargs = mock_popen.call_args[1]
|
||
self.assertIsNot(kwargs.get("stderr"), os.devnull)
|
||
finally:
|
||
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
||
config.reset_cache()
|
||
|
||
@patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe", return_value=None)
|
||
def test_ffmpeg_not_found_warning(self, _resolve: MagicMock) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
try:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="receive-order",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_no_ff",
|
||
)
|
||
session._start_ffmpeg_capture()
|
||
self.assertIn("ffmpeg_not_found", session.warnings)
|
||
finally:
|
||
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
||
config.reset_cache()
|
||
|
||
def test_finalize_missing_capture_warning(self) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
try:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="receive-order",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_warn",
|
||
)
|
||
asyncio.run(session.finalize())
|
||
self.assertTrue(
|
||
any("ffmpeg_capture_missing" in w for w in session.warnings)
|
||
)
|
||
self.assertIn("record_log=", "".join(session.warnings))
|
||
self.assertIsNone(session.output_video_path)
|
||
finally:
|
||
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
||
config.reset_cache()
|
||
|
||
|
||
class TestMusicFadeParams(unittest.TestCase):
|
||
def test_normal_duration(self) -> None:
|
||
start, dur = _music_fade_params(20.0)
|
||
self.assertAlmostEqual(start, 17.0)
|
||
self.assertAlmostEqual(dur, 3.0)
|
||
|
||
def test_short_video(self) -> None:
|
||
start, dur = _music_fade_params(4.0)
|
||
self.assertAlmostEqual(dur, 1.0)
|
||
self.assertAlmostEqual(start, 3.0)
|
||
|
||
def test_invalid_duration(self) -> None:
|
||
self.assertEqual(_music_fade_params(None), (None, None))
|
||
self.assertEqual(_music_fade_params(0), (None, None))
|
||
self.assertEqual(_music_fade_params(-1), (None, None))
|
||
|
||
|
||
class TestVoiceoverTextFilter(unittest.TestCase):
|
||
def test_keeps_meaningful_steps(self) -> None:
|
||
for text in (
|
||
"开始采集",
|
||
"获取1688账号",
|
||
"启动浏览器",
|
||
"检查登录状态",
|
||
"搜索关键词：螺丝",
|
||
"解析第 1 页店铺列表",
|
||
"打开店铺联系方式页",
|
||
"提取联系人信息",
|
||
"采集完成",
|
||
):
|
||
self.assertIsNotNone(_voiceover_text_for_step(text), msg=text)
|
||
|
||
def test_skips_noise(self) -> None:
|
||
for text in (
|
||
"等待 1-5s",
|
||
"跳过重复店铺",
|
||
"写入联系人库",
|
||
"探针模式",
|
||
"操作异常",
|
||
"登录失败",
|
||
"2024-01-01 12:00:00 INFO: debug",
|
||
"",
|
||
" ",
|
||
):
|
||
self.assertIsNone(_voiceover_text_for_step(text), msg=text)
|
||
|
||
def test_truncates_long_text(self) -> None:
|
||
long_text = "开" * 80
|
||
result = _voiceover_text_for_step(long_text)
|
||
self.assertIsNotNone(result)
|
||
assert result is not None
|
||
self.assertEqual(len(result), 60)
|
||
|
||
def test_dedup_within_ten_seconds(self) -> None:
|
||
entries = [
|
||
_StepEntry(0.0, "启动浏览器"),
|
||
_StepEntry(5.0, "启动浏览器"),
|
||
_StepEntry(12.0, "启动浏览器"),
|
||
]
|
||
clips = _select_voiceover_clips(entries)
|
||
texts = [t for _, t in clips]
|
||
self.assertEqual(texts.count("启动浏览器"), 2)
|
||
|
||
def test_min_gap_two_seconds(self) -> None:
|
||
entries = [
|
||
_StepEntry(0.0, "获取1688账号"),
|
||
_StepEntry(1.0, "连接CDP"),
|
||
_StepEntry(3.0, "检查登录状态"),
|
||
]
|
||
clips = _select_voiceover_clips(entries)
|
||
self.assertEqual(len(clips), 2)
|
||
self.assertEqual(clips[0][1], "获取1688账号")
|
||
self.assertEqual(clips[1][1], "检查登录状态")
|
||
|
||
def test_important_bypasses_min_gap(self) -> None:
|
||
entries = [
|
||
_StepEntry(0.0, "定位搜索框"),
|
||
_StepEntry(0.5, "输入关键词：宁德电池"),
|
||
_StepEntry(1.0, "点击搜索"),
|
||
_StepEntry(1.5, "等待搜索结果"),
|
||
]
|
||
clips = _select_voiceover_clips(entries)
|
||
texts = [t for _, t in clips]
|
||
self.assertEqual(len(clips), 4)
|
||
self.assertIn("输入关键词：宁德电池", texts)
|
||
self.assertIn("点击搜索", texts)
|
||
|
||
def test_noise_still_skipped_with_important_nearby(self) -> None:
|
||
entries = [
|
||
_StepEntry(0.0, "点击搜索"),
|
||
_StepEntry(0.5, "跳过重复店铺"),
|
||
_StepEntry(1.0, "写入联系人库"),
|
||
]
|
||
clips = _select_voiceover_clips(entries)
|
||
texts = [t for _, t in clips]
|
||
self.assertEqual(texts, ["点击搜索"])
|
||
|
||
|
||
class TestComposeCaptureMp4(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
self.ffmpeg = Path(r"C:\ffmpeg\ffmpeg.exe")
|
||
self.capture = Path(r"C:\tmp\capture.mp4")
|
||
self.srt = Path(r"C:\tmp\sub.srt")
|
||
self.output = Path(r"C:\tmp\out.mp4")
|
||
self.music = Path(r"C:\tmp\music.mp3")
|
||
self.voice = Path(r"C:\tmp\voiceover.wav")
|
||
|
||
def _run_compose(self, **kwargs) -> list[str]:
|
||
captured: list[str] = []
|
||
|
||
def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
|
||
captured.extend(cmd_tail)
|
||
return True, ""
|
||
|
||
with (
|
||
patch.object(vs, "_run_ffmpeg", side_effect=fake_run),
|
||
patch.object(vs, "_resolve_ffprobe_exe", return_value=Path(r"C:\ffmpeg\ffprobe.exe")),
|
||
patch.object(vs, "_probe_media_duration_seconds", return_value=kwargs.pop("duration", 20.0)),
|
||
patch.object(Path, "is_file", return_value=True),
|
||
):
|
||
_compose_capture_mp4(
|
||
self.ffmpeg,
|
||
self.capture,
|
||
self.srt,
|
||
self.output,
|
||
music_file=kwargs.get("music_file"),
|
||
voiceover_file=kwargs.get("voiceover_file"),
|
||
warnings=kwargs.get("warnings"),
|
||
)
|
||
return captured
|
||
|
||
def test_music_loop_params(self) -> None:
|
||
cmd = self._run_compose(music_file=self.music)
|
||
cmd_str = " ".join(cmd)
|
||
loop_idx = cmd.index("-stream_loop")
|
||
minus_one_idx = cmd.index("-1", loop_idx)
|
||
music_i_idx = cmd.index("-i", minus_one_idx)
|
||
self.assertEqual(cmd[music_i_idx + 1], str(self.music))
|
||
self.assertIn("-shortest", cmd)
|
||
filter_idx = cmd.index("-filter_complex")
|
||
filt = cmd[filter_idx + 1]
|
||
self.assertNotIn("apad", filt)
|
||
|
||
def test_music_fade_filter_long_video(self) -> None:
|
||
cmd = self._run_compose(music_file=self.music, duration=20.0)
|
||
filt = cmd[cmd.index("-filter_complex") + 1]
|
||
self.assertIn("afade=t=out:st=17", filt)
|
||
self.assertIn("d=3", filt)
|
||
self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt)
|
||
|
||
def test_music_fade_filter_short_video(self) -> None:
|
||
cmd = self._run_compose(music_file=self.music, duration=4.0)
|
||
filt = cmd[cmd.index("-filter_complex") + 1]
|
||
self.assertIn("afade=t=out:st=3", filt)
|
||
self.assertIn("d=1", filt)
|
||
|
||
def test_duration_probe_failed_no_afade(self) -> None:
|
||
warnings: list[str] = []
|
||
captured: list[str] = []
|
||
|
||
def fake_run(_exe: Path, cmd_tail: list[str]) -> tuple[bool, str]:
|
||
captured.extend(cmd_tail)
|
||
return True, ""
|
||
|
||
with (
|
||
patch.object(vs, "_run_ffmpeg", side_effect=fake_run),
|
||
patch.object(vs, "_resolve_ffprobe_exe", return_value=None),
|
||
patch.object(vs, "_probe_media_duration_seconds", return_value=None),
|
||
patch.object(Path, "is_file", return_value=True),
|
||
):
|
||
ok, _ = _compose_capture_mp4(
|
||
self.ffmpeg,
|
||
self.capture,
|
||
self.srt,
|
||
self.output,
|
||
music_file=self.music,
|
||
warnings=warnings,
|
||
)
|
||
self.assertTrue(ok)
|
||
self.assertIn("video_duration_probe_failed_for_music_fade", warnings)
|
||
filt = captured[captured.index("-filter_complex") + 1]
|
||
self.assertNotIn("afade", filt)
|
||
self.assertIn("-stream_loop", captured)
|
||
|
||
def test_music_and_voice_amix(self) -> None:
|
||
cmd = self._run_compose(music_file=self.music, voiceover_file=self.voice)
|
||
filt = cmd[cmd.index("-filter_complex") + 1]
|
||
self.assertIn("amix", filt)
|
||
self.assertIn(f"volume={vs._MUSIC_VOLUME}", filt)
|
||
self.assertIn(f"volume={vs._VOICE_VOLUME}", filt)
|
||
self.assertLess(vs._MUSIC_VOLUME, vs._VOICE_VOLUME)
|
||
self.assertIn("afade", filt)
|
||
self.assertNotIn("apad", filt)
|
||
self.assertIn("-stream_loop", cmd)
|
||
self.assertIn("-shortest", cmd)
|
||
|
||
def test_voice_only_no_amix(self) -> None:
|
||
cmd = self._run_compose(voiceover_file=self.voice)
|
||
filt = cmd[cmd.index("-filter_complex") + 1]
|
||
self.assertNotIn("amix", filt)
|
||
self.assertIn(f"volume={vs._VOICE_VOLUME}", filt)
|
||
self.assertIn("apad", filt)
|
||
self.assertIn("-shortest", cmd)
|
||
self.assertNotIn("-stream_loop", cmd)
|
||
|
||
|
||
class TestVideoSessionLifecycle(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
config.reset_cache()
|
||
|
||
def tearDown(self) -> None:
|
||
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
||
config.reset_cache()
|
||
|
||
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
|
||
@patch.object(RpaVideoSession, "_start_ffmpeg_capture")
|
||
@patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
|
||
@patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
|
||
def test_enter_starts_capture_before_title(
|
||
self,
|
||
mock_finalize: AsyncMock,
|
||
mock_stop: MagicMock,
|
||
mock_start: MagicMock,
|
||
mock_sleep: AsyncMock,
|
||
) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
call_order: list[str] = []
|
||
mock_start.side_effect = lambda: call_order.append("start_capture")
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="test",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_intro",
|
||
title="开始执行测试",
|
||
)
|
||
original_add = session.add_step
|
||
|
||
def _track_add(text: str, *, duration: float = 4.0) -> None:
|
||
call_order.append(f"add_step:{text}")
|
||
original_add(text, duration=duration)
|
||
|
||
with patch.object(session, "add_step", side_effect=_track_add):
|
||
asyncio.run(session.__aenter__())
|
||
|
||
self.assertEqual(call_order[0], "start_capture")
|
||
self.assertIn("add_step:开始执行测试", call_order)
|
||
mock_sleep.assert_awaited()
|
||
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
|
||
self.assertAlmostEqual(total_sleep, _INTRO_BUFFER_SECONDS)
|
||
|
||
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
|
||
@patch.object(RpaVideoSession, "_start_ffmpeg_capture")
|
||
def test_enter_no_title_still_buffers(
|
||
self,
|
||
mock_start: MagicMock,
|
||
mock_sleep: AsyncMock,
|
||
) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="test",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_no_title",
|
||
)
|
||
with patch.object(session, "add_step") as mock_add:
|
||
asyncio.run(session.__aenter__())
|
||
mock_start.assert_called_once()
|
||
mock_add.assert_not_called()
|
||
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
|
||
self.assertAlmostEqual(total_sleep, _INTRO_BUFFER_SECONDS)
|
||
|
||
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
|
||
@patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
|
||
@patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
|
||
def test_exit_closing_title_before_stop(
|
||
self,
|
||
mock_finalize: AsyncMock,
|
||
mock_stop: MagicMock,
|
||
mock_sleep: AsyncMock,
|
||
) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
call_order: list[str] = []
|
||
mock_stop.side_effect = lambda: call_order.append("stop_capture")
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="test",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_outro",
|
||
closing_title="任务已完成",
|
||
)
|
||
original_add = session.add_step
|
||
|
||
def _track_add(text: str, *, duration: float = 4.0) -> None:
|
||
call_order.append(f"add_step:{text}")
|
||
original_add(text, duration=duration)
|
||
|
||
with patch.object(session, "add_step", side_effect=_track_add):
|
||
asyncio.run(session.__aexit__(None, None, None))
|
||
|
||
self.assertIn("add_step:任务已完成", call_order)
|
||
self.assertEqual(call_order[-1], "stop_capture")
|
||
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
|
||
self.assertAlmostEqual(total_sleep, _OUTRO_BUFFER_SECONDS)
|
||
|
||
@patch("jiangchang_skill_core.rpa.video_session.asyncio.sleep", new_callable=AsyncMock)
|
||
@patch.object(RpaVideoSession, "_stop_ffmpeg_capture")
|
||
@patch.object(RpaVideoSession, "finalize", new_callable=AsyncMock)
|
||
def test_exit_no_closing_title_still_outro_buffer(
|
||
self,
|
||
mock_finalize: AsyncMock,
|
||
mock_stop: MagicMock,
|
||
mock_sleep: AsyncMock,
|
||
) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
config.reset_cache()
|
||
session = RpaVideoSession(
|
||
skill_slug="test",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch_outro_empty",
|
||
)
|
||
with patch.object(session, "add_step") as mock_add:
|
||
asyncio.run(session.__aexit__(None, None, None))
|
||
mock_add.assert_not_called()
|
||
mock_stop.assert_called_once()
|
||
total_sleep = sum(c.args[0] for c in mock_sleep.await_args_list if c.args)
|
||
self.assertAlmostEqual(total_sleep, _OUTRO_BUFFER_SECONDS)
|
||
|
||
|
||
class TestFinalizeDegradation(unittest.TestCase):
|
||
def setUp(self) -> None:
|
||
config.reset_cache()
|
||
|
||
def tearDown(self) -> None:
|
||
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
||
config.reset_cache()
|
||
|
||
@patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
|
||
@patch("jiangchang_skill_core.rpa.video_session._resolve_music_file")
|
||
@patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover")
|
||
@patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4")
|
||
def test_tts_failure_still_composes(
|
||
self,
|
||
mock_compose: MagicMock,
|
||
mock_tts: MagicMock,
|
||
mock_music: MagicMock,
|
||
mock_ffmpeg: MagicMock,
|
||
) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe")
|
||
mock_music.return_value = Path(r"C:\tmp\music.mp3")
|
||
mock_tts.return_value = None
|
||
mock_compose.return_value = (True, "")
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4"
|
||
capture.parent.mkdir(parents=True)
|
||
capture.write_bytes(b"fake")
|
||
|
||
session = RpaVideoSession(
|
||
skill_slug="test",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch",
|
||
)
|
||
session.add_step("启动浏览器")
|
||
asyncio.run(session.finalize())
|
||
|
||
self.assertTrue(mock_compose.called)
|
||
self.assertIsNotNone(session.output_video_path)
|
||
|
||
@patch("jiangchang_skill_core.rpa.video_session._resolve_ffmpeg_exe")
|
||
@patch("jiangchang_skill_core.rpa.video_session._resolve_music_file")
|
||
@patch("jiangchang_skill_core.rpa.video_session._synthesize_step_voiceover")
|
||
@patch("jiangchang_skill_core.rpa.video_session._compose_capture_mp4")
|
||
def test_compose_fallback_without_music(
|
||
self,
|
||
mock_compose: MagicMock,
|
||
mock_tts: MagicMock,
|
||
mock_music: MagicMock,
|
||
mock_ffmpeg: MagicMock,
|
||
) -> None:
|
||
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
||
mock_ffmpeg.return_value = Path(r"C:\ffmpeg\ffmpeg.exe")
|
||
mock_music.return_value = Path(r"C:\tmp\music.mp3")
|
||
mock_tts.return_value = None
|
||
mock_compose.side_effect = [(False, "music err"), (True, "")]
|
||
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
capture = Path(tmp) / "rpa-artifacts" / "batch" / "capture.mp4"
|
||
capture.parent.mkdir(parents=True)
|
||
capture.write_bytes(b"fake")
|
||
|
||
session = RpaVideoSession(
|
||
skill_slug="test",
|
||
skill_data_dir=tmp,
|
||
batch_id="batch",
|
||
)
|
||
asyncio.run(session.finalize())
|
||
|
||
self.assertEqual(mock_compose.call_count, 2)
|
||
self.assertTrue(any("ffmpeg_with_music_failed" in w for w in session.warnings))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|