feat(plugins): add source-aware evidence and recording timelines

This commit is contained in:
chenmingxuan committed 2026-10-16 10:20:11 +08:00
1 parent 72ff342605
commit ef7af9988e
61 files changed
+3847 -108

No files matched your search

+2 -1
View File
@@ -583,7 +583,8 @@ async def test_fail_fast_includes_skipped_reports(isolated, monkeypatch):
assert [j["status"] for j in isolated.jobs(batch["id"])] == ["failed", "skipped"]
archive = reports.export_batch(batch["id"])
with zipfile.ZipFile(archive) as z:
assert len(z.namelist()) == 6 # two summary files + two files per task
assert len(z.namelist()) == 8 # summaries + two reports and evidence index per task
assert len([name for name in z.namelist() if name.endswith("证据索引.json")]) == 2
assert sum(n.endswith(".docx") for n in z.namelist()) == 3
@@ -0,0 +1,82 @@
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