410 lines
16 KiB
Python
410 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "verify_delivery.py"
|
|
SPEC = importlib.util.spec_from_file_location("verify_delivery", SCRIPT)
|
|
assert SPEC is not None and SPEC.loader is not None
|
|
delivery = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(delivery)
|
|
|
|
|
|
class VerifyDeliveryTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
self.source = self.root / "video.intermediate.mkv"
|
|
self.source.write_bytes(b"video")
|
|
self.subtitle = self.root / "video.source-srt.en.srt"
|
|
self.subtitle.write_text("1\n00:00:00,000 --> 00:00:01,000\nHello\n")
|
|
self.manifest = self.root / "download-manifest.json"
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "downloaded",
|
|
"output_directory": str(self.root),
|
|
"artifacts": {
|
|
"intermediate": {"path": self.source.name},
|
|
"subtitle": {
|
|
"source_srt": {"path": self.subtitle.name},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary.cleanup()
|
|
|
|
def test_subtitled_job_with_only_translation_inputs_is_incomplete(self) -> None:
|
|
inputs = self.root / "subtitles" / "translation-input"
|
|
inputs.mkdir(parents=True)
|
|
batch = inputs / "batch-0001.json"
|
|
batch.write_text("{}", encoding="utf-8")
|
|
outputs = self.root / "subtitles" / "translation-output"
|
|
outputs.mkdir()
|
|
(self.root / "subtitles" / "subtitle-manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"translation_batches": [{"path": str(batch)}],
|
|
"translation_output_dir": str(outputs),
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = delivery.assess_delivery(self.manifest)
|
|
|
|
self.assertFalse(result["complete"])
|
|
self.assertEqual(result["stage"], "translation_required")
|
|
self.assertIn("batch-0001.json", result["missing"])
|
|
|
|
def test_video_deliverable_completes_with_untranslated_subtitle(self) -> None:
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "video_complete",
|
|
"deliverable": "video",
|
|
"output_directory": str(self.root),
|
|
"artifacts": {
|
|
"intermediate": {"path": self.source.name},
|
|
"subtitle": {
|
|
"language": "en",
|
|
"source_srt": {"path": self.subtitle.name},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = delivery.assess_delivery(self.manifest)
|
|
|
|
self.assertTrue(result["complete"])
|
|
self.assertEqual(result["stage"], "video_complete")
|
|
|
|
def test_subs_deliverable_needs_no_video_file(self) -> None:
|
|
self.source.unlink()
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "subs_complete",
|
|
"deliverable": "subs",
|
|
"output_directory": str(self.root),
|
|
"artifacts": {
|
|
"intermediate": None,
|
|
"subtitle": {
|
|
"language": "en",
|
|
"source_srt": {"path": self.subtitle.name},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = delivery.assess_delivery(self.manifest)
|
|
|
|
self.assertTrue(result["complete"])
|
|
self.assertEqual(result["stage"], "subs_complete")
|
|
|
|
def test_bilingual_subs_deliverable_stops_at_render(self) -> None:
|
|
inputs = self.root / "subtitles" / "translation-input"
|
|
inputs.mkdir(parents=True)
|
|
batch = inputs / "batch-0001.json"
|
|
batch.write_text("{}", encoding="utf-8")
|
|
outputs = self.root / "subtitles" / "translation-output"
|
|
outputs.mkdir()
|
|
(outputs / "batch-0001.json").write_text("{}", encoding="utf-8")
|
|
(self.root / "subtitles" / "subtitle-manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"translation_batches": [{"path": str(batch)}],
|
|
"translation_output_dir": str(outputs),
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "bilingual_required",
|
|
"deliverable": "bilingual-subs",
|
|
"output_directory": str(self.root),
|
|
"artifacts": {
|
|
"intermediate": None,
|
|
"subtitle": {
|
|
"language": "en",
|
|
"source_srt": {"path": self.subtitle.name},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
pending = delivery.assess_delivery(self.manifest)
|
|
self.assertFalse(pending["complete"])
|
|
self.assertEqual(pending["stage"], "render_required")
|
|
|
|
rendered = self.root / "subtitles" / "rendered"
|
|
rendered.mkdir()
|
|
(rendered / "bilingual.ass").write_text("[Script Info]\n", encoding="utf-8")
|
|
(rendered / "validation.json").write_text("{}", encoding="utf-8")
|
|
|
|
result = delivery.assess_delivery(self.manifest)
|
|
self.assertTrue(result["complete"])
|
|
self.assertEqual(result["stage"], "bilingual_subs_complete")
|
|
|
|
def test_full_delivery_uses_manifest_bilingual_filename(self) -> None:
|
|
inputs = self.root / "subtitles" / "translation-input"
|
|
inputs.mkdir(parents=True)
|
|
batch = inputs / "batch-0001.json"
|
|
batch.write_text("{}", encoding="utf-8")
|
|
outputs = self.root / "subtitles" / "translation-output"
|
|
outputs.mkdir()
|
|
(outputs / "batch-0001.json").write_text("{}", encoding="utf-8")
|
|
(self.root / "subtitles" / "subtitle-manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"translation_batches": [{"path": str(batch)}],
|
|
"translation_output_dir": str(outputs),
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
rendered = self.root / "subtitles" / "rendered"
|
|
rendered.mkdir()
|
|
(rendered / "bilingual.ass").write_text("[Script Info]\n", encoding="utf-8")
|
|
(rendered / "validation.json").write_text("{}", encoding="utf-8")
|
|
expected = self.root / "双语字幕版「测试视频」.mp4"
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "bilingual_required",
|
|
"deliverable": "full",
|
|
"output_directory": str(self.root),
|
|
"delivery_names": {
|
|
"cover": "封面.jpg",
|
|
"bilingual_video": expected.name,
|
|
},
|
|
"artifacts": {
|
|
"intermediate": {"path": self.source.name},
|
|
"subtitle": {
|
|
"language": "en",
|
|
"source_srt": {"path": self.subtitle.name},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(self.root / "legacy.bilingual.mp4").write_bytes(b"legacy")
|
|
|
|
pending = delivery.assess_delivery(self.manifest)
|
|
self.assertFalse(pending["complete"])
|
|
self.assertEqual(pending["missing"], [str(expected.resolve())])
|
|
|
|
expected.write_bytes(b"burned")
|
|
result = delivery.assess_delivery(self.manifest)
|
|
self.assertTrue(result["complete"])
|
|
self.assertEqual(result["burned_video"], str(expected.resolve()))
|
|
|
|
def test_scientific_review_gate_is_required_and_checksum_bound(self) -> None:
|
|
inputs = self.root / "subtitles" / "translation-input"
|
|
inputs.mkdir(parents=True)
|
|
batch = inputs / "batch-0001.json"
|
|
batch.write_text("{}", encoding="utf-8")
|
|
outputs = self.root / "subtitles" / "translation-output"
|
|
outputs.mkdir()
|
|
(outputs / "batch-0001.json").write_text("{}", encoding="utf-8")
|
|
subtitle_manifest = self.root / "subtitles" / "subtitle-manifest.json"
|
|
subtitle_manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"translation_batches": [{"path": str(batch)}],
|
|
"translation_output_dir": str(outputs),
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"deliverable": "bilingual-subs",
|
|
"output_directory": str(self.root),
|
|
"execution": {"scientific_review_required": True},
|
|
"artifacts": {
|
|
"intermediate": None,
|
|
"subtitle": {"source_srt": {"path": self.subtitle.name}},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
pending = delivery.assess_delivery(self.manifest)
|
|
self.assertEqual(pending["stage"], "scientific_review_required")
|
|
|
|
review_dir = self.root / "subtitles" / "scientific-review"
|
|
reviewed = review_dir / "reviewed-translations"
|
|
reviewed.mkdir(parents=True)
|
|
(reviewed / "translations.json").write_text("{}", encoding="utf-8")
|
|
report = review_dir / "report.json"
|
|
report.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "complete",
|
|
"human_expert_reviewed": False,
|
|
"subtitle_manifest_sha256": hashlib.sha256(
|
|
subtitle_manifest.read_bytes()
|
|
).hexdigest(),
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
rendered = self.root / "subtitles" / "rendered"
|
|
rendered.mkdir()
|
|
(rendered / "bilingual.ass").write_text("[Script Info]\n", encoding="utf-8")
|
|
(rendered / "validation.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"translation_quality_reviewed": True,
|
|
"scientific_review": {
|
|
"report_sha256": hashlib.sha256(report.read_bytes()).hexdigest()
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
result = delivery.assess_delivery(self.manifest)
|
|
self.assertTrue(result["complete"])
|
|
|
|
report.write_text("{}", encoding="utf-8")
|
|
with self.assertRaisesRegex(delivery.DeliveryError, "incomplete"):
|
|
delivery.assess_delivery(self.manifest)
|
|
|
|
def test_declared_citation_requires_matching_burn_receipt(self) -> None:
|
|
inputs = self.root / "subtitles" / "translation-input"
|
|
inputs.mkdir(parents=True)
|
|
batch = inputs / "batch-0001.json"
|
|
batch.write_text("{}", encoding="utf-8")
|
|
outputs = self.root / "subtitles" / "translation-output"
|
|
outputs.mkdir()
|
|
(outputs / "batch-0001.json").write_text("{}", encoding="utf-8")
|
|
(self.root / "subtitles" / "subtitle-manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"translation_batches": [{"path": str(batch)}],
|
|
"translation_output_dir": str(outputs),
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
rendered = self.root / "subtitles" / "rendered"
|
|
rendered.mkdir()
|
|
(rendered / "bilingual.ass").write_text("[Script Info]\n", encoding="utf-8")
|
|
(rendered / "validation.json").write_text("{}", encoding="utf-8")
|
|
expected = self.root / "双语字幕版「测试视频」.mp4"
|
|
expected.write_bytes(b"burned")
|
|
citation_file = self.root / "citation-watermark.txt"
|
|
citation_file.write_text("Formal citation.\n", encoding="utf-8")
|
|
citation_hash = hashlib.sha256(citation_file.read_bytes()).hexdigest()
|
|
manifest_value = {
|
|
"deliverable": "full",
|
|
"output_directory": str(self.root),
|
|
"delivery_names": {"bilingual_video": expected.name},
|
|
"citation_watermark": {
|
|
"enabled": True,
|
|
"citation_file": citation_file.name,
|
|
"citation_sha256": citation_hash,
|
|
"position": "top-left",
|
|
},
|
|
"artifacts": {
|
|
"intermediate": {"path": self.source.name},
|
|
"subtitle": {"source_srt": {"path": self.subtitle.name}},
|
|
},
|
|
}
|
|
self.manifest.write_text(json.dumps(manifest_value), encoding="utf-8")
|
|
|
|
with self.assertRaisesRegex(delivery.DeliveryError, "burn receipt is missing"):
|
|
delivery.assess_delivery(self.manifest)
|
|
|
|
receipt = expected.with_suffix(expected.suffix + ".citation.json")
|
|
receipt.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"output_file": expected.name,
|
|
"output_sha256": hashlib.sha256(expected.read_bytes()).hexdigest(),
|
|
"citation_sha256": citation_hash,
|
|
"position": "top-left",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
result = delivery.assess_delivery(self.manifest)
|
|
self.assertTrue(result["complete"])
|
|
|
|
expected.write_bytes(b"changed")
|
|
with self.assertRaisesRegex(delivery.DeliveryError, "output_sha256"):
|
|
delivery.assess_delivery(self.manifest)
|
|
|
|
def test_no_dialogue_subtitle_counts_as_video_only(self) -> None:
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "video_only_complete",
|
|
"output_directory": str(self.root),
|
|
"artifacts": {
|
|
"intermediate": {"path": self.source.name},
|
|
"subtitle": {
|
|
"language": "en",
|
|
"dialogue": False,
|
|
"source_srt": {"path": self.subtitle.name},
|
|
},
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = delivery.assess_delivery(self.manifest)
|
|
|
|
self.assertTrue(result["complete"])
|
|
self.assertEqual(result["stage"], "video_only_complete")
|
|
|
|
def test_video_only_job_requires_the_video_file_on_disk(self) -> None:
|
|
self.manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "downloaded",
|
|
"output_directory": str(self.root),
|
|
"artifacts": {
|
|
"intermediate": {"path": self.source.name},
|
|
"subtitle": None,
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = delivery.assess_delivery(self.manifest)
|
|
self.assertTrue(result["complete"])
|
|
self.assertEqual(result["stage"], "video_only_complete")
|
|
|
|
self.source.unlink()
|
|
with self.assertRaisesRegex(delivery.DeliveryError, "video artifact"):
|
|
delivery.assess_delivery(self.manifest)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|