ee2a4d9e22
All checks were successful
技能自动化发布 / release (push) Successful in 15s
基于 skill-template 脚手架开发，包含： - 5 张业务表（教职工、请假政策、考勤日历、请假记录、假期余额） - 请假核算引擎（扣薪计算、审批层级判定、年假抵扣） - CLI 入口（import-employees / calculate-leave / render-html） - HTML 审批与扣薪报告生成 - 示例数据与完整测试套件（295 tests passed）
231 lines
9.1 KiB
Python
231 lines
9.1 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""录屏辅助与请假核算 CLI 冒烟（本技能无外站 RPA）。"""
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import os
|
||
import unittest
|
||
from contextlib import redirect_stderr, redirect_stdout
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from _support import IsolatedDataRoot, get_skill_root
|
||
|
||
from jiangchang_skill_core import config
|
||
|
||
_RPA_MARKERS = ("playwright", "uiautomation", "@rpa_step", "rpa_step(")
|
||
|
||
|
||
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=real_" + "rpa",
|
||
"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_and_rpa_skills_keep_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()
|
||
if not any(marker in service_text for marker in _RPA_MARKERS):
|
||
return
|
||
missing = _missing_video_integration_tokens(service_text)
|
||
self.assertEqual(
|
||
missing,
|
||
[],
|
||
msg=(
|
||
"RPA service must retain video integration even when OPENCLAW_RECORD_VIDEO=0; "
|
||
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\process-leave-attendance-default_demo.mp4",
|
||
"record_log_path": r"D:\data\logs\ffmpeg-record.log",
|
||
"warnings": [],
|
||
}
|
||
)
|
||
out = buf.getvalue()
|
||
self.assertIn("录屏路径", out)
|
||
self.assertIn("process-leave-attendance-default_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_finish_payload_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 TestCalculateLeaveCmd(unittest.TestCase):
|
||
def test_cmd_calculate_leave_without_employee_returns_validation_error(self) -> None:
|
||
old_auth = os.environ.pop("JIANGCHANG_AUTH_BASE_URL", None)
|
||
try:
|
||
with IsolatedDataRoot(user_id="_calc_empty"):
|
||
config.reset_cache()
|
||
from service import task_service
|
||
|
||
buf = io.StringIO()
|
||
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
|
||
rc = task_service.cmd_calculate_leave(employee_id=None)
|
||
self.assertEqual(rc, 1)
|
||
self.assertIn("employee_id", buf.getvalue())
|
||
finally:
|
||
if old_auth is not None:
|
||
os.environ["JIANGCHANG_AUTH_BASE_URL"] = old_auth
|
||
|
||
|
||
class TestCmdRunFinishCoverage(unittest.TestCase):
|
||
def test_cmd_calculate_leave_entitlement_failure_calls_finish(self) -> None:
|
||
with IsolatedDataRoot(user_id="_entitlement_fail"):
|
||
from service import task_service
|
||
|
||
finish_kwargs: list[dict] = []
|
||
original_finish = task_service.finish
|
||
|
||
def _capture_finish(*args, **kwargs):
|
||
finish_kwargs.append(kwargs)
|
||
return original_finish(*args, **kwargs)
|
||
|
||
with patch.object(task_service, "check_entitlement", return_value=(False, "mock denied")):
|
||
with patch.object(task_service, "finish", side_effect=_capture_finish):
|
||
buf = io.StringIO()
|
||
with redirect_stdout(buf), redirect_stderr(io.StringIO()):
|
||
rc = task_service.cmd_calculate_leave(
|
||
employee_id="1",
|
||
leave_type="事假",
|
||
start_date="2026-01-01",
|
||
end_date="2026-01-02",
|
||
)
|
||
|
||
self.assertEqual(rc, 1)
|
||
self.assertTrue(finish_kwargs)
|
||
payload = finish_kwargs[-1]
|
||
self.assertEqual(payload.get("status"), "failed")
|
||
self.assertEqual(payload.get("error_code"), "ENTITLEMENT_DENIED")
|
||
self.assertEqual(payload.get("stage"), "auth")
|
||
self.assertIn("mock denied", payload.get("message") or "")
|
||
|
||
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()
|