jiangchang-platform-kit/tests/test_video_session.py

# -*- coding: utf-8 -*-
"""RpaVideoSession：ffmpeg 桌面录屏。"""
from __future__ import annotations

import asyncio
import os
import sys
import tempfile
import unittest
from unittest.mock import MagicMock, patch

from jiangchang_skill_core import config
from jiangchang_skill_core.rpa.video_session import RpaVideoSession


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"])
        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._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.shutil.which", return_value="ffmpeg")
    @patch("jiangchang_skill_core.rpa.video_session.subprocess.Popen")
    def test_enter_starts_ffmpeg_gdigrab(self, mock_popen: MagicMock, _which: MagicMock) -> None:
        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.assertIn("ffmpeg", cmd)
                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.shutil.which", return_value=None)
    def test_ffmpeg_not_found_warning(self, _which: 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()


if __name__ == "__main__":
    unittest.main()