chendelian
1c51139883
All checks were successful
Publish Python Package to Gitea / publish (push) Successful in 17s
Introduce SAS v1 emit/finish/journal API, sibling CLI bridge, RPA video session integration, and tests for the host Job Runner activity timeline. Co-authored-by: Cursor <cursoragent@cursor.com>
114 lines
4.0 KiB
Python
114 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""RpaVideoSession.add_step integrates with activity.emit."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from jiangchang_skill_core import activity, config
|
|
from jiangchang_skill_core.rpa.video_session import RpaVideoSession
|
|
from jiangchang_skill_core.runtime_env import get_runs_dir
|
|
|
|
|
|
class TestVideoSessionActivityIntegration(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
config.reset_cache()
|
|
self._saved_root = os.environ.get("JIANGCHANG_DATA_ROOT")
|
|
self._saved_job = os.environ.get("JIANGCHANG_JOB_ID")
|
|
self._saved_video = os.environ.get("OPENCLAW_RECORD_VIDEO")
|
|
self._tmps: list[str] = []
|
|
activity._reset_for_tests()
|
|
|
|
def tearDown(self) -> None:
|
|
activity._reset_for_tests()
|
|
if self._saved_root is None:
|
|
os.environ.pop("JIANGCHANG_DATA_ROOT", None)
|
|
else:
|
|
os.environ["JIANGCHANG_DATA_ROOT"] = self._saved_root
|
|
if self._saved_job is None:
|
|
os.environ.pop("JIANGCHANG_JOB_ID", None)
|
|
else:
|
|
os.environ["JIANGCHANG_JOB_ID"] = self._saved_job
|
|
if self._saved_video is None:
|
|
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
|
else:
|
|
os.environ["OPENCLAW_RECORD_VIDEO"] = self._saved_video
|
|
config.reset_cache()
|
|
for tmp in self._tmps:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
def _tmp_data_root(self) -> str:
|
|
tmp = tempfile.mkdtemp()
|
|
self._tmps.append(tmp)
|
|
return tmp
|
|
|
|
def test_add_step_emits_activity_when_disabled(self) -> None:
|
|
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
|
tmp = self._tmp_data_root()
|
|
os.environ["JIANGCHANG_DATA_ROOT"] = tmp
|
|
os.environ["JIANGCHANG_JOB_ID"] = "video-act-1"
|
|
activity._reset_for_tests()
|
|
config.reset_cache()
|
|
|
|
session = RpaVideoSession(
|
|
skill_slug="publish-toutiao",
|
|
skill_data_dir=tmp,
|
|
batch_id="batch1",
|
|
)
|
|
self.assertFalse(session.enabled)
|
|
session.add_step("打开浏览器")
|
|
|
|
journal = Path(get_runs_dir()) / "video-act-1.jsonl"
|
|
self.assertTrue(journal.is_file())
|
|
content = journal.read_text(encoding="utf-8")
|
|
self.assertIn("打开浏览器", content)
|
|
self.assertIn("rpa.video", content)
|
|
self.assertIn("publish-toutiao", content)
|
|
|
|
def test_emit_activity_false_skips_journal(self) -> None:
|
|
os.environ.pop("OPENCLAW_RECORD_VIDEO", None)
|
|
tmp = self._tmp_data_root()
|
|
os.environ["JIANGCHANG_DATA_ROOT"] = tmp
|
|
os.environ["JIANGCHANG_JOB_ID"] = "video-act-off"
|
|
activity._reset_for_tests()
|
|
|
|
session = RpaVideoSession(
|
|
skill_slug="publish-toutiao",
|
|
skill_data_dir=tmp,
|
|
batch_id="batch2",
|
|
emit_activity=False,
|
|
)
|
|
session.add_step("不应写入")
|
|
journal = Path(get_runs_dir()) / "video-act-off.jsonl"
|
|
self.assertFalse(journal.exists())
|
|
|
|
def test_add_step_writes_both_journal_and_subtitle_steps(self) -> None:
|
|
os.environ["OPENCLAW_RECORD_VIDEO"] = "1"
|
|
tmp = self._tmp_data_root()
|
|
os.environ["JIANGCHANG_DATA_ROOT"] = tmp
|
|
os.environ["JIANGCHANG_JOB_ID"] = "video-act-both"
|
|
activity._reset_for_tests()
|
|
config.reset_cache()
|
|
|
|
session = RpaVideoSession(
|
|
skill_slug="receive-order",
|
|
skill_data_dir=tmp,
|
|
batch_id="batch3",
|
|
)
|
|
session.add_step("录制步骤")
|
|
self.assertEqual(len(session._steps), 1)
|
|
|
|
journal = Path(get_runs_dir()) / "video-act-both.jsonl"
|
|
lines = [json.loads(ln) for ln in journal.read_text(encoding="utf-8").splitlines()]
|
|
activity_events = [e for e in lines if e.get("type") == "activity"]
|
|
self.assertEqual(len(activity_events), 1)
|
|
self.assertEqual(activity_events[0]["text"], "录制步骤")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|