| """Chapter recovery pins and edited-vs-source timeline regression tests.""" |
| import copy |
| import asyncio |
| from contextlib import contextmanager |
| from importlib import import_module |
| import json |
| from pathlib import Path |
| import tempfile |
| import unittest |
| from unittest.mock import patch |
|
|
| from _checkpoint_revision_unit_test import chain, folder_paths, write_revision |
|
|
| retirement = import_module(chain.__package__ + ".chapter_snapshot_retirement") |
|
|
|
|
| class ChapterRecoverySafetyTests(unittest.TestCase): |
| def setUp(self): |
| self.temp = tempfile.TemporaryDirectory() |
| self.addCleanup(self.temp.cleanup) |
| folder_paths.output_directory = self.temp.name |
| self.run = Path(self.temp.name) / "h3_chains" / "revision_test" |
| self.old = "a" * 32 |
| self.metadata, self.files = write_revision( |
| self.run, 1, self.old, 1, active=True, audio_context_length=0) |
| segment = self.metadata["segment"] |
| self.manifest = { |
| "format": "h3_chain_manifest_v3", "run_name": "revision_test", |
| "plan_hash": "test", "compatibility": {"continuation_mode": "guide"}, |
| "clip_count": 1, "segments": [segment], |
| "total_delivered_frames": segment["delivered_frames"], |
| } |
| write_revision(self.run, 1, "b" * 32, 2, active=True, audio_context_length=0) |
| self.manager = chain.CheckpointGraphManager(self.temp.name) |
|
|
| def test_sealing_invalidates_delete_preview_and_preserves_recovery(self): |
| before = self.manager.deletion_preview("revision_test", 1, self.old) |
| self.assertTrue(before["allowed"]) |
| snapshot, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| pinned = self.manager.deletion_preview("revision_test", 1, self.old) |
| self.assertFalse(pinned["allowed"]) |
| self.assertEqual(pinned["chapter_references"][0]["snapshot"], snapshot["chapter_manifest_id"]) |
| self.assertIn("Sealed Chapter 1", pinned["blockers"][0]) |
| with self.assertRaises(ValueError): |
| self.manager.delete("revision_test", 1, self.old, before["snapshot"]) |
| recovered, _ = chain._load_chapter_manifest("revision_test", 1, snapshot["chapter_manifest_id"]) |
| self.assertEqual(recovered["segments"][0]["revision"], self.old) |
| self.assertTrue(Path(path).is_file()) |
| self.assertTrue(self.manager.deletion_preview("revision_test", 1, "b" * 32)["allowed"]) |
|
|
| def test_selected_alternate_and_archive_paths_are_pinned(self): |
| snapshot, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| candidate = "c" * 32 |
| metadata, _ = write_revision(self.run, 1, candidate, 3, audio_context_length=0) |
| for references in [ |
| {"editorial": {"replacements": [{"scene": 1, "alternate_revision": candidate, "base_revision": self.old}]}}, |
| |
| {"archives": {"checkpoint": metadata["segment"]["checkpoint"]}}, |
| ]: |
| with self.subTest(references=references): |
| modified = copy.deepcopy(snapshot) |
| modified.update(references) |
| Path(path).write_text(json.dumps(modified)) |
| preview = self.manager.deletion_preview("revision_test", 1, candidate) |
| self.assertFalse(preview["allowed"]) |
| self.assertTrue(preview["chapter_references"]) |
|
|
| def test_unreadable_snapshot_fails_closed(self): |
| _, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| Path(path).write_text("{incomplete") |
| preview = self.manager.deletion_preview("revision_test", 1, "b" * 32) |
| self.assertFalse(preview["allowed"]) |
| self.assertIn("Cannot verify sealed chapter", preview["blockers"][0]) |
|
|
| def test_supersedes_history_is_not_a_recovery_dependency(self): |
| current = json.loads((self.run / "checkpoints/clip_0001.json").read_text())["segment"] |
| current["supersedes"] = self.metadata["segment"]["revision_metadata"] |
| manifest = dict(self.manifest, segments=[current]) |
| snapshot, _ = chain._chapter_manifest_from_manifest(manifest, 1) |
| self.assertEqual(snapshot["segments"][0]["supersedes"], current["supersedes"]) |
| preview = self.manager.deletion_preview("revision_test", 1, self.old) |
| self.assertTrue(preview["allowed"]) |
| self.manager.delete("revision_test", 1, self.old, preview["snapshot"]) |
| recovered, _ = chain._load_chapter_manifest("revision_test", 1, snapshot["chapter_manifest_id"]) |
| self.assertEqual(recovered["segments"][0]["revision"], "b" * 32) |
|
|
| def test_retirement_archives_only_snapshot_and_releases_only_its_pins(self): |
| snapshot, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| original_bytes = Path(path).read_bytes() |
| manager = retirement.ChapterSnapshotManager(self.temp.name) |
| address = snapshot["chapter_manifest_path"] |
| preview = manager.retirement_preview("revision_test", address) |
| self.assertFalse(preview["scenes"][0]["active"]) |
| before = {p: p.read_bytes() for p in self.run.rglob("*") if p.is_file()} |
| result = manager.retire("revision_test", address, preview["snapshot"]) |
| archive = Path(self.temp.name) / result["retired_path"] |
| self.assertEqual(archive.read_bytes(), original_bytes) |
| self.assertFalse(Path(path).exists()) |
| for p, content in before.items(): |
| if p != Path(path): |
| self.assertEqual(p.read_bytes(), content) |
| self.assertTrue(self.manager.deletion_preview("revision_test", 1, self.old)["allowed"]) |
| with self.assertRaisesRegex(FileNotFoundError, "No sealed"): |
| chain._load_chapter_manifest("revision_test", 1, snapshot["chapter_manifest_id"]) |
| |
| with self.assertRaisesRegex(ValueError, "retired"): |
| chain._persist_chapter_manifest(snapshot) |
| self.assertFalse(Path(path).exists()) |
| |
| archive.rename(path) |
| recovered, _ = chain._load_chapter_manifest("revision_test", 1, snapshot["chapter_manifest_id"]) |
| self.assertEqual(recovered["segments"][0]["revision"], self.old) |
|
|
| def test_other_snapshot_and_shared_branch_dependencies_still_block(self): |
| first, _ = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| other, _ = chain._chapter_manifest_from_manifest(dict(self.manifest, plan_hash="other"), 1) |
| child, child_files = write_revision(self.run, 2, "c" * 32, 3, |
| predecessor=self.metadata) |
| manager = retirement.ChapterSnapshotManager(self.temp.name) |
| for snapshot in (first, other): |
| address = snapshot["chapter_manifest_path"] |
| preview = manager.retirement_preview("revision_test", address) |
| manager.retire("revision_test", address, preview["snapshot"]) |
| blocked = self.manager.deletion_preview("revision_test", 1, self.old) |
| self.assertFalse(blocked["allowed"]) |
| self.assertEqual(blocked["dependents"][0]["revision"], child["segment"]["revision"]) |
| self.assertEqual(len(blocked["chapter_references"]), 1 if snapshot is first else 0) |
| self.assertTrue(all(path.exists() for path in child_files)) |
| |
| preview = self.manager.deletion_preview("revision_test", 2, "c" * 32) |
| self.manager.delete("revision_test", 2, "c" * 32, preview["snapshot"]) |
| self.assertTrue(self.manager.deletion_preview("revision_test", 1, self.old)["allowed"]) |
|
|
| def test_retirement_rejects_missing_and_stale_confirmations(self): |
| snapshot, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| manager = retirement.ChapterSnapshotManager(self.temp.name) |
| address = snapshot["chapter_manifest_path"] |
| preview = manager.retirement_preview("revision_test", address) |
| for token in ("", "wrong"): |
| with self.assertRaises(retirement.CheckpointDeleteBlocked): |
| manager.retire("revision_test", address, token) |
| snapshot["sealed_at"] = "changed after preview" |
| Path(path).write_text(json.dumps(snapshot)) |
| with self.assertRaises(retirement.CheckpointDeleteBlocked): |
| manager.retire("revision_test", address, preview["snapshot"]) |
| preview = manager.retirement_preview("revision_test", address) |
| write_revision(self.run, 1, "d" * 32, 4, active=True, audio_context_length=0) |
| with self.assertRaises(retirement.CheckpointDeleteBlocked): |
| manager.retire("revision_test", address, preview["snapshot"]) |
| self.assertTrue(Path(path).is_file()) |
|
|
| def test_retirement_refuses_invalid_paths_identity_and_symlinks(self): |
| snapshot, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| manager = retirement.ChapterSnapshotManager(self.temp.name) |
| address = snapshot["chapter_manifest_path"] |
| for target in ("../escape.json", str(path), address.replace("revision_test", "foreign"), |
| address.replace("/manifests/", "/./manifests/"), "", None): |
| with self.subTest(target=target), self.assertRaises(ValueError): |
| manager.retirement_preview("revision_test", target) |
| Path(path).write_text(json.dumps({**snapshot, "run_name": "foreign"})) |
| with self.assertRaisesRegex(ValueError, "identity"): |
| manager.retirement_preview("revision_test", address) |
| Path(path).write_text(json.dumps({**snapshot, "clip_count": 999})) |
| with self.assertRaisesRegex(ValueError, "identity"): |
| manager.retirement_preview("revision_test", address) |
| Path(path).write_text(json.dumps(snapshot)) |
| archive = Path(path).parent.parent / "retired_manifests" |
| archive.symlink_to(Path(path).parent, target_is_directory=True) |
| with self.assertRaisesRegex(ValueError, "symlinks"): |
| manager.retirement_preview("revision_test", address) |
| archive.unlink() |
| real = Path(path).with_suffix(".backup") |
| Path(path).rename(real) |
| Path(path).symlink_to(real) |
| with self.assertRaisesRegex(ValueError, "symlinks"): |
| manager.retirement_preview("revision_test", address) |
|
|
| def test_retirement_does_not_overwrite_archive_and_failed_move_keeps_pin(self): |
| snapshot, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| manager = retirement.ChapterSnapshotManager(self.temp.name) |
| address = snapshot["chapter_manifest_path"] |
| preview = manager.retirement_preview("revision_test", address) |
| with patch.object(retirement.os, "rename", side_effect=OSError("test failure")): |
| with self.assertRaises(OSError): |
| manager.retire("revision_test", address, preview["snapshot"]) |
| self.assertFalse(self.manager.deletion_preview("revision_test", 1, self.old)["allowed"]) |
| destination = Path(self.temp.name) / preview["retired_path"] |
| destination.write_bytes(b"existing archive") |
| with self.assertRaisesRegex(ValueError, "not be overwritten"): |
| manager.retire("revision_test", address, preview["snapshot"]) |
| self.assertEqual(destination.read_bytes(), b"existing archive") |
| self.assertTrue(Path(path).is_file()) |
|
|
| def test_retirement_route_requires_lock_and_preview_confirmation(self): |
| snapshot, path = chain._chapter_manifest_from_manifest(self.manifest, 1) |
| guards = [] |
|
|
| @contextmanager |
| def guard(root, run): |
| guards.append(run) |
| with retirement.checkpoint_run_lock(root, run): |
| yield |
|
|
| class Request: |
| path = "/chapter-snapshots/retire-preview" |
| body = {"run_name": "revision_test", "path": snapshot["chapter_manifest_path"]} |
|
|
| async def json(self): |
| return self.body |
|
|
| request = Request() |
| call = lambda: asyncio.run(chain._chapter_snapshot_retirement(request)) |
| with patch.object(chain, "checkpoint_run_lock", side_effect=guard): |
| response = call() |
| self.assertEqual(response.status, 200) |
| preview = json.loads(response.body) |
| self.assertEqual(guards, []) |
| request.path = "/chapter-snapshots/retire" |
| self.assertEqual(call().status, 409) |
| self.assertTrue(Path(path).exists()) |
| request.body["snapshot"] = preview["snapshot"] |
| self.assertEqual(call().status, 200) |
| self.assertIn("revision_test", guards) |
| self.assertFalse(Path(path).exists()) |
| self.assertEqual(call().status, 404) |
| request.body = [] |
| self.assertEqual(call().status, 400) |
|
|
| def test_deleted_input_cannot_be_sealed_from_stale_selection(self): |
| preview = self.manager.deletion_preview("revision_test", 1, self.old) |
| self.manager.delete("revision_test", 1, self.old, preview["snapshot"]) |
| scoped = dict(self.manifest, format=chain.CHAPTER_MANIFEST_FORMAT, |
| chapter={"number": 1, "id": "one", "title": "One"}) |
| with self.assertRaises((ValueError, FileNotFoundError)): |
| chain._persist_chapter_manifest(scoped) |
| self.assertEqual(list(self.run.glob("chapters/*/manifests/*.json")), []) |
|
|
| def test_chapter_origin_uses_full_edited_timeline_not_raw_audio_offset(self): |
| segments = [] |
| for index in range(1, 9): |
| metadata, _ = write_revision(self.run, index, "%032x" % index, index, |
| audio_context_length=0) |
| segment = metadata["segment"] |
| segment["raw_frames"] = segment["delivered_frames"] = 13 |
| segments.append(segment) |
| editorial = { |
| "format": "h3_chain_editorial_v1", "run_name": "revision_test", |
| "scene_order": [{"scene": i, "scene_id": s["id"]} for i, s in enumerate(segments, 1)], |
| "chapters": [{"id": "one", "title": "One", "start_scene": 1, "start_scene_id": segments[0]["id"]}, |
| {"id": "two", "title": "Two", "start_scene": 5, "start_scene_id": segments[4]["id"]}], |
| "trims": [{"scene": 1, "scene_id": segments[0]["id"], "out_frame": 9}], |
| "placements": [], |
| "subtitles": {"mode": "preview_srt", "asset_id": "song", "offset_seconds": 0}, |
| } |
| manifest = dict(self.manifest, segments=segments, clip_count=8, |
| total_delivered_frames=104, editorial=editorial) |
| for gap in (False, True): |
| if gap: |
| |
| editorial["placements"] = [{"scene": i, "scene_id": s["id"], "start_frame": 24 + 13 * (i - 1)} |
| for i, s in enumerate(segments, 1)] |
| _, records, _ = chain._editorial_timeline_records("revision_test", segments, editorial) |
| expected = next(r["start_frame"] for r in records if r.get("scene") == 5) |
| chapter, _ = chain._chapter_manifest_from_manifest(manifest, 2) |
| self.assertEqual(chapter["chapter"]["editorial_origin_frame"], expected) |
| self.assertEqual(chapter["chapter"]["source_start_frame"], 52) |
| if not gap: |
| self.assertEqual(expected, 48) |
| self.assertEqual(chapter["editorial"]["placements"][0]["start_frame"], 0) |
| with patch.object(chain, "ProjectAssetStore") as store: |
| store.return_value.load.return_value = {"assets": [{"id": "song", "kind": "audio", |
| "lyrics": "[00:02.50]Line one\n[00:03.50]Line two"}]} |
| cues = chain._editorial_subtitle_cues("revision_test", chapter["editorial"], 52, |
| timeline_origin_frames=expected) |
| if not gap: |
| self.assertAlmostEqual(cues[0]["start"], 0.5) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
