83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
import json
|
|
from unittest.mock import MagicMock
|
|
|
|
from apps.admin_console.routers import media
|
|
from apps.admin_console.services import evidence_service as evidence
|
|
from apps.admin_console.services import report_export as exports
|
|
|
|
|
|
def fixture_files(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(evidence, "TRACES_PATH", tmp_path)
|
|
root = tmp_path / "task" / "evidence" / "recordings"
|
|
root.mkdir(parents=True)
|
|
video = root / "clip.mp4"
|
|
video.write_bytes(b"fixture video")
|
|
(root / "recordings.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 1,
|
|
"sources": [
|
|
{
|
|
"id": "pc",
|
|
"kind": "computer",
|
|
"target_ids": ["window"],
|
|
"status": "ready",
|
|
"segments": [
|
|
{
|
|
"file": "evidence/recordings/clip.mp4",
|
|
"offset_ms": 5000,
|
|
"duration": 2,
|
|
}
|
|
],
|
|
}
|
|
],
|
|
}
|
|
),
|
|
"utf-8",
|
|
)
|
|
return video
|
|
|
|
|
|
def test_plugin_media_preserves_session_offsets_and_exports_relative_files(tmp_path, monkeypatch):
|
|
fixture_files(tmp_path, monkeypatch)
|
|
monkeypatch.setattr(exports, "session_screenshots", lambda _: [])
|
|
monkeypatch.setattr(exports, "session_artifacts", lambda _: [])
|
|
source = evidence.plugin_sources("task")[0]
|
|
assert source["segments"][0]["start"] == 5
|
|
index = evidence.export_index(
|
|
"task", {"name": "case", "status": "completed", "verdict": "未验证"}, []
|
|
)
|
|
clip = index["recordings"][0]["segments"][0]
|
|
assert clip["file"] == "任务录屏/电脑/clip.mp4" and "url" not in clip
|
|
assert clip["file"] in {item["path"] for item in index["files"]}
|
|
|
|
|
|
def test_legacy_url_only_does_not_enable_strict_timeline(monkeypatch):
|
|
monkeypatch.setattr(
|
|
media,
|
|
"_get_legacy_session_video_sync",
|
|
lambda _: {"status": "ready", "video_url": "/videos/old.mp4", "video_segments": []},
|
|
)
|
|
monkeypatch.setattr(evidence, "plugin_sources", lambda _: [])
|
|
monkeypatch.setattr(media, "session_repo", MagicMock(get_session_by_id=lambda _: {}))
|
|
response = media._get_session_video_sync("task")
|
|
assert response["session_timeline"] is False
|
|
assert response["video_url"] == "/videos/old.mp4"
|
|
|
|
|
|
def test_computer_only_recording_can_be_downloaded(tmp_path, monkeypatch):
|
|
import zipfile
|
|
|
|
fixture_files(tmp_path, monkeypatch)
|
|
monkeypatch.setattr(exports, "EXPORT_ROOT", tmp_path / "exports")
|
|
monkeypatch.setattr(
|
|
exports,
|
|
"session_report",
|
|
lambda _: {"id": "task", "name": "case", "status": "completed", "verdict": "未验证"},
|
|
)
|
|
monkeypatch.setattr(exports, "session_videos", lambda _: [])
|
|
with zipfile.ZipFile(exports.export_session_media("task", "video")) as archive:
|
|
index = json.loads(archive.read("文件清单.json"))
|
|
assert index["videos"] == ["任务录屏/电脑/clip.mp4"]
|
|
assert archive.getinfo(index["videos"][0]).compress_type == zipfile.ZIP_STORED
|