# -*- coding: utf-8 -*- """RpaVideoSession 模板示范与 result_summary / CLI 输出测试。""" from __future__ import annotations import io import json import os import re import unittest from contextlib import redirect_stderr, redirect_stdout from unittest.mock import AsyncMock, MagicMock, patch from _support import IsolatedDataRoot, get_skill_root from jiangchang_skill_core import config def _active_env_value(text: str, key: str) -> str | None: """Return the active (non-comment) KEY=value line from .env-style text.""" prefix = f"{key}=" for line in text.splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#"): continue active = stripped.split("#", 1)[0].strip() if active.startswith(prefix): return active return None def _combined_service_source(skill_root: str | None = None) -> str: """Merge scripts/service/*.py source (excludes __pycache__).""" root = skill_root or get_skill_root() service_dir = os.path.join(root, "scripts", "service") parts: list[str] = [] for dirpath, dirnames, filenames in os.walk(service_dir): dirnames[:] = [d for d in dirnames if d != "__pycache__"] for name in sorted(filenames): if not name.endswith(".py"): continue path = os.path.join(dirpath, name) with open(path, encoding="utf-8") as f: parts.append(f.read()) return "\n".join(parts) def _missing_video_integration_tokens(service_text: str) -> list[str]: missing: list[str] = [] if "RpaVideoSession" not in service_text: missing.append("RpaVideoSession") if "video.add_step" not in service_text and ".add_step(" not in service_text: missing.append("video.add_step or .add_step(") if "merge_video_into_result_summary" not in service_text: missing.append("merge_video_into_result_summary") return missing class TestEnvExampleVideoDefaults(unittest.TestCase): def test_env_example_contains_required_keys(self) -> None: path = os.path.join(get_skill_root(), ".env.example") with open(path, encoding="utf-8") as f: text = f.read() for key in ( "OPENCLAW_" + "TEST_TARGET=mock", "OPENCLAW_RECORD_VIDEO=0", "OPENCLAW_ARTIFACTS_ON_FAILURE=1", "OPENCLAW_BROWSER_HEADLESS=0", "OPENCLAW_PLAYWRIGHT_STEALTH=1", ): self.assertIn(key, text, msg=f"missing {key!r} in .env.example") def test_env_example_default_off_but_template_keeps_video_integration(self) -> None: path = os.path.join(get_skill_root(), ".env.example") with open(path, encoding="utf-8") as f: text = f.read() self.assertEqual( _active_env_value(text, "OPENCLAW_RECORD_VIDEO"), "OPENCLAW_RECORD_VIDEO=0", ) service_text = _combined_service_source() missing = _missing_video_integration_tokens(service_text) self.assertEqual( missing, [], msg=( "default OPENCLAW_RECORD_VIDEO=0 does not allow removing video integration; " f"scripts/service/*.py missing: {', '.join(missing)}" ), ) def test_env_example_allows_record_video_one_in_comments_only(self) -> None: sample = "\n".join( [ "# example: OPENCLAW_RECORD_VIDEO=1 for production", "OPENCLAW_RECORD_VIDEO=0", ] ) self.assertEqual( _active_env_value(sample, "OPENCLAW_RECORD_VIDEO"), "OPENCLAW_RECORD_VIDEO=0", ) class TestPrintVideoSummary(unittest.TestCase): def test_cli_prints_video_path_when_enabled(self) -> None: from service.task_run_support import _print_video_summary buf = io.StringIO() with redirect_stdout(buf), redirect_stderr(io.StringIO()): _print_video_summary( { "enabled": True, "path": r"D:\data\videos\your-skill-slug_demo.mp4", "record_log_path": r"D:\data\logs\ffmpeg-record.log", "warnings": [], } ) out = buf.getvalue() self.assertIn("录屏路径", out) self.assertIn("your-skill-slug_demo.mp4", out) self.assertIn("录屏日志", out) def test_disabled_video_prints_nothing(self) -> None: from service.task_run_support import _print_video_summary buf = io.StringIO() with redirect_stdout(buf), redirect_stderr(io.StringIO()): _print_video_summary({"enabled": False, "path": r"D:\tmp\x.mp4"}) self.assertEqual(buf.getvalue(), "") class TestMergeVideoIntoResultSummary(unittest.TestCase): def test_result_summary_includes_video_fields(self) -> None: from service.task_run_support import merge_video_into_result_summary video_info = { "enabled": True, "path": r"D:\data\videos\demo.mp4", "capture_path": r"D:\data\capture.mp4", "record_log_path": r"D:\data\logs\ffmpeg-record.log", "warnings": ["test_warning"], "voiceover_path": r"D:\data\voiceover.wav", "music_path": r"D:\data\music.mp3", "audio_warnings": ["tts_skipped"], } payload = merge_video_into_result_summary({"demo": True}, video_info) self.assertIn("video", payload) self.assertEqual(payload["video_path"], video_info["path"]) self.assertEqual(payload["raw_video"], video_info["capture_path"]) self.assertEqual(payload["video_log"], video_info["record_log_path"]) self.assertEqual(payload["video_warnings"], ["test_warning"]) self.assertEqual(payload["voiceover_path"], video_info["voiceover_path"]) self.assertEqual(payload["music_path"], video_info["music_path"]) self.assertEqual(payload["audio_warnings"], ["tts_skipped"]) class TestTemplateRunVideoSession(unittest.TestCase): def _fake_video_session(self, mock_video: MagicMock): class _FakeVideoSession: def __init__(self, **kwargs): self.kwargs = kwargs async def __aenter__(self): return mock_video async def __aexit__(self, *args): return None return _FakeVideoSession def test_cmd_run_creates_video_session_with_chinese_titles(self) -> None: with IsolatedDataRoot(user_id="_video_run"): os.environ["OPENCLAW_RECORD_VIDEO"] = "0" config.reset_cache() from service import task_service mock_video = MagicMock() mock_video.add_step = MagicMock() mock_video.summary.return_value = { "enabled": False, "path": None, "capture_path": None, "record_log_path": None, "warnings": [], "voiceover_path": None, "music_path": None, "audio_warnings": [], } created: list[dict] = [] class _CapturingFakeSession: def __init__(self, **kwargs): created.append(kwargs) async def __aenter__(self): return mock_video async def __aexit__(self, *args): return None with patch.object(task_service, "RpaVideoSession", _CapturingFakeSession): buf = io.StringIO() with redirect_stdout(buf), redirect_stderr(io.StringIO()): rc = task_service.cmd_run(target="demo-target", input_id="42") self.assertEqual(rc, 1) self.assertTrue(created, "RpaVideoSession should be constructed") kwargs = created[0] self.assertEqual(kwargs.get("title"), "开始执行示例任务") self.assertEqual(kwargs.get("closing_title"), "示例任务执行完成") self.assertTrue(re.search(r"[\u4e00-\u9fff]", kwargs.get("title", ""))) self.assertTrue(re.search(r"[\u4e00-\u9fff]", kwargs.get("closing_title", ""))) mock_video.add_step.assert_any_call("准备执行示例任务") mock_video.add_step.assert_any_call("示例任务执行完成") mock_video.summary.assert_called() out = buf.getvalue() self.assertIn("模板", out) from db import task_logs_repository as tlr rows = tlr.list_task_logs(1) self.assertTrue(rows) summary = json.loads(rows[0][7]) self.assertIn("video", summary) self.assertIn("video_path", summary) class TestCmdRunTaskLogCoverage(unittest.TestCase): def test_cmd_run_entitlement_failure_writes_task_log(self) -> None: with IsolatedDataRoot(user_id="_entitlement_fail"): from service import task_service with patch.object(task_service, "check_entitlement", return_value=(False, "mock denied")): buf = io.StringIO() with redirect_stdout(buf), redirect_stderr(io.StringIO()): rc = task_service.cmd_run(target="t1", input_id="99") self.assertEqual(rc, 1) from db import task_logs_repository as tlr rows = tlr.list_task_logs(1) self.assertTrue(rows) _rid, _ttype, tid, iid, _ititle, status, err, rsum, _cat, _uat = rows[0] self.assertEqual(status, "failed") self.assertEqual(tid, "t1") self.assertEqual(iid, "99") self.assertIn("mock denied", err or "") summary = json.loads(rsum or "{}") self.assertEqual(summary.get("stage"), "auth") def test_cmd_run_exception_writes_task_log(self) -> None: with IsolatedDataRoot(user_id="_exception_run"): os.environ["OPENCLAW_RECORD_VIDEO"] = "0" config.reset_cache() from service import task_service async def _boom(*_args, **_kwargs): raise RuntimeError("boom") with patch.object(task_service, "check_entitlement", return_value=(True, "")): with patch.object(task_service, "_run_template_demo", side_effect=_boom): buf = io.StringIO() with redirect_stdout(buf), redirect_stderr(io.StringIO()): rc = task_service.cmd_run(target="t2", input_id="88") self.assertEqual(rc, 1) from db import task_logs_repository as tlr rows = tlr.list_task_logs(1) self.assertTrue(rows) _rid, _ttype, tid, iid, _ititle, status, err, rsum, _cat, _uat = rows[0] self.assertEqual(status, "failed") self.assertEqual(tid, "t2") self.assertEqual(iid, "88") self.assertIn("任务执行异常", err or "") summary = json.loads(rsum or "{}") self.assertEqual(summary.get("stage"), "run") self.assertEqual(summary.get("error"), "unexpected_exception") def test_get_task_logger_fallback_when_not_initialized(self) -> None: from service import task_service mock_logger = MagicMock() responses = [RuntimeError("logging not initialized"), mock_logger] def _fake_get_skill_logger(): item = responses.pop(0) if isinstance(item, Exception): raise item return item with patch.object(task_service, "get_skill_logger", side_effect=_fake_get_skill_logger): with patch.object(task_service, "setup_skill_logging") as mock_setup: logger = task_service._get_task_logger() mock_setup.assert_called_once_with(task_service.SKILL_SLUG, task_service.LOG_LOGGER_NAME) self.assertIs(logger, mock_logger) if __name__ == "__main__": unittest.main()