File size: 8,552 Bytes
bc50ea0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | #!/usr/bin/env python3
"""Chapter-only manager output: mixed geometry, exact clocks and no writes."""
import importlib.util
import json
from datetime import datetime, timezone
from pathlib import Path
import tempfile
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location(
"chapter_output_helpers", ROOT / "tests/_checkpoint_local_output_unit_test.py")
h = importlib.util.module_from_spec(spec)
spec.loader.exec_module(h)
chain = h.chain
def main():
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
h.h.folder_paths.output_directory = temporary
run = root / "h3_chains" / "chapter_output"
metadata, lineage = [], []
for index in range(1, 5):
item, _ = h.h.write_revision(
run, index, str(index) * 32, index, active=True,
predecessor=metadata[-1] if metadata else None,
run_name=run.name, context_length=0, audio_context_length=0,
compatibility={"width":1344 if index < 3 else 960,
"height":768 if index < 3 else 544,
"audio_mode":"generated_audio"})
metadata.append(item)
lineage.append({"scene":index, "revision":str(index) * 32})
(run / "plan.json").write_text(json.dumps({
"run_name":run.name, "shots":[{"id":f"scene_{i}"} for i in range(1, 6)]}))
editorial = {
"scene_order":[{"scene":i, "scene_id":f"scene_{i}"} for i in range(1, 6)],
"chapters":[{"id":"one", "start_scene_id":"scene_1"},
{"id":"two", "start_scene_id":"scene_3"}],
"trims":[{"scene_id":"scene_1", "out_frame":294}],
}
(run / "editorial.json").write_text(json.dumps(editorial))
selection = {"run_name":run.name, "output_mode":"workflow_local",
"scope_start_scene":3, "scope_end_scene":5, "lineage":lineage}
h.rejected(lambda: chain._checkpoint_selection_manifest(selection), "compatibility")
selection["output_scope"] = "chapter"
before = h.snapshot(root)
with patch.object(chain, "_atomic_json", side_effect=AssertionError("project write")):
result = chain._checkpoint_selection_manifest(selection)
assert h.snapshot(root) == before
with patch.object(chain, "datetime") as clock:
clock.now.return_value = datetime(2000, 1, 1, tzinfo=timezone.utc)
first_read = chain._checkpoint_selection_manifest(selection)
clock.now.return_value = datetime(2040, 1, 1, tzinfo=timezone.utc)
assert chain._checkpoint_selection_manifest(selection) == first_read == result, \
"legacy sidecar reads must not change source hashes with wall-clock time"
assert [s["index"] for s in result["segments"]] == [3, 4]
assert result["format"] == chain.CHAPTER_MANIFEST_FORMAT
assert result["compatibility"]["width"] == 960
assert result["chapter"]["number"] == 2
assert result["chapter"]["complete"] is False
assert result["source_scene_count"] == 5
assert result["chapter"]["source_start_frame"] == 702
assert result["chapter"]["editorial_origin_frame"] == 634
assert result["editorial"]["placements"][0]["start_frame"] == 0
assert "chapter_manifest_path" not in result and not (run / "chapters").exists()
selected_metadata = root / metadata[3]["segment"]["revision_metadata"]
original_metadata = selected_metadata.read_text()
incompatible = json.loads(original_metadata)
incompatible["compatibility"]["width"] = 1024
selected_metadata.write_text(json.dumps(incompatible))
h.rejected(lambda: chain._checkpoint_selection_manifest(selection),
"compatible takes within this chapter")
selected_metadata.write_text(original_metadata)
# The reported Chapter 2 takes differ only in reference-catalog
# fingerprints. This is not a geometry or saved-lineage mismatch.
changed_catalog = json.loads(original_metadata)
changed_catalog["compatibility"].update({
"generation_fingerprint":"new-catalog",
"generation_fingerprint_lineage":{"current":"new-catalog"}})
selected_metadata.write_text(json.dumps(changed_catalog))
mixed = chain._checkpoint_selection_manifest(selection)
assert [s["generation_fingerprint"] for s in mixed["segments"]] == ["", "new-catalog"]
selected_metadata.write_text(original_metadata)
# Appending empty planned scenes after this take was generated must
# not prevent selecting it. Reference schedules retain the archived
# Plan's count, while the editorial chapter may extend beyond it.
plan_path = run / "plan.json"
plan_before = plan_path.read_text()
plan_path.write_text(json.dumps({
"run_name":run.name, "shots":[{"id":f"scene_{i}"} for i in range(1, 5)]}))
older = chain._checkpoint_selection_manifest(selection)
assert older["source_scene_count"] == 4
assert older["chapter"]["planned_end_scene"] == 5
plan_path.write_text(plan_before)
# UI scope ends at the current scene order; immutable archived Plans
# can include more empty scenes (real regression: UI 12, archive 13).
for planned_count, ui_end in ((6, 5), (5, 4), (5, 7)):
plan_path.write_text(json.dumps({
"run_name":run.name,
"shots":[{"id":f"scene_{i}"} for i in range(1, planned_count + 1)]}))
pinned = chain._checkpoint_selection_manifest({
**selection, "scope_end_scene":ui_end})
assert [s["revision"] for s in pinned["segments"]] == ["3" * 32, "4" * 32]
assert pinned["chapter"]["start_scene"] == 3
assert pinned["chapter"]["planned_end_scene"] == max(5, planned_count)
plan_path.write_text(plan_before)
# A real boundary crossing within the pinned takes is not an empty
# tail difference: this must still require an explicit new selection.
(run / "editorial.json").write_text(json.dumps({
**editorial, "chapters":[*editorial["chapters"],
{"id":"three", "start_scene_id":"scene_4"}]}))
h.rejected(lambda: chain._checkpoint_selection_manifest(selection), "boundaries")
(run / "editorial.json").write_text(json.dumps(editorial))
# A one-scene selection without explicit chapter markers is Chapter 1
# even when its archived Plan includes more ungenerated shots.
(run / "editorial.json").write_text("{}")
implicit = chain._checkpoint_selection_manifest({
**selection, "lineage":lineage[:1],
"scope_start_scene":1, "scope_end_scene":1})
assert [s["index"] for s in implicit["segments"]] == [1]
(run / "editorial.json").write_text(json.dumps(editorial))
# Earlier media is not an upscale input. Retain its immutable timing
# metadata, but remove the fixture's video and latent artifacts.
for item in metadata[:2]:
for key in ("segment", "checkpoint"):
(root / item["segment"][key]).unlink()
before = h.snapshot(root)
assert chain._checkpoint_selection_manifest(selection) == result
assert h.snapshot(root) == before
h.rejected(lambda: chain._checkpoint_selection_manifest({
**selection, "scope_start_scene":4}), "boundaries")
h.rejected(lambda: chain._checkpoint_selection_manifest({
**selection, "output_scope":"invalid"}), "unknown output scope")
# A wrong take within the selected chapter is still rejected.
other, _ = h.h.write_revision(run, 3, "a" * 32, 10, run_name=run.name,
compatibility=metadata[2]["compatibility"], context_length=0, audio_context_length=0)
h.rejected(lambda: chain._checkpoint_selection_manifest({
**selection, "lineage":[*lineage[:2], {"scene":3,"revision":"a" * 32}, lineage[3]]}),
"depends on")
(root / metadata[3]["segment"]["segment"]).write_bytes(b"corrupt")
h.rejected(lambda: chain._checkpoint_selection_manifest(selection), "integrity")
print("Checkpoint chapter output: mixed geometry, original indexes/clocks, unavailable prior media, read-only and fail-closed checks pass")
if __name__ == "__main__":
main()
|