File size: 6,436 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 | #!/usr/bin/env python3
"""Local manager output uses exact saved takes without promoting any pointer."""
import importlib.util
import json
import math
from pathlib import Path
import tempfile
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location(
"local_output_helpers", ROOT / "tests/_checkpoint_revision_unit_test.py")
h = importlib.util.module_from_spec(spec)
spec.loader.exec_module(h)
chain = h.chain
def snapshot(root):
return {str(path.relative_to(root)): path.read_bytes()
for path in root.rglob("*") if path.is_file()}
def rejected(call, text):
try:
call()
except (OSError, ValueError) as exc:
assert text in str(exc), str(exc)
else:
raise AssertionError("Expected failure: " + text)
def main():
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
h.folder_paths.output_directory = temporary
run = root / "h3_chains" / "local_test"
old1, _ = h.write_revision(run, 1, "1" * 32, 11, run_name=run.name)
old2, _ = h.write_revision(run, 2, "2" * 32, 12, predecessor=old1, run_name=run.name)
new1, _ = h.write_revision(run, 1, "3" * 32, 21, active=True, run_name=run.name)
new2, _ = h.write_revision(run, 2, "4" * 32, 22, active=True, predecessor=new1, run_name=run.name)
(run / "plan.json").write_text(json.dumps({
"run_name":run.name, "shots":[{"id":f"scene_{i}"} for i in range(1, 4)],
}))
(run / "manifest.json").write_text('{"keep": "project final"}')
(run / "editorial.json").write_text(json.dumps({
"scene_order":[{"scene":i, "scene_id":f"scene_{i}"} for i in range(1, 4)],
}))
# Workflow ownership is deliberately nightly-only. The stable port
# still guarantees no changes to any project-wide active pointers.
selection = {"run_name":run.name, "output_mode":"workflow_local",
"scope_start_scene":1, "scope_end_scene":3,
"lineage":[{"scene":1, "revision":"1" * 32},
{"scene":2, "revision":"2" * 32}]}
node = chain.MiniMaxH3ChainCheckpointManager()
assert math.isnan(node.IS_CHANGED())
assert node.passthrough() == (None,)
before = snapshot(root)
with patch.object(chain, "_atomic_json", side_effect=AssertionError("local output wrote JSON")):
result = node.passthrough(json.dumps(selection))[0]
assert [row["revision"] for row in result["segments"]] == ["1" * 32, "2" * 32]
assert result["selection_complete"] is False and result["planned_clip_count"] == 3
assert snapshot(root) == before, "local selection modified project files"
assert node.passthrough({k:v for k, v in selection.items() if k != "output_mode"})[0] == result, \
"pinning alone does not change upscale source hashes"
# A second workflow's exact output and the project's active state remain
# independent. Changing shared pointers after the pin does not move it.
second = {**selection, "lineage":[{"scene":1,"revision":"3" * 32},
{"scene":2,"revision":"4" * 32}]}
assert node.passthrough(second)[0]["segments"][1]["revision"] == "4" * 32
for i, metadata in enumerate((old1, old2), start=1):
(run / "checkpoints" / f"clip_{i:04d}.json").write_text(json.dumps(metadata))
changed = snapshot(root)
assert node.passthrough(second)[0]["segments"][1]["revision"] == "4" * 32
assert node.passthrough(selection)[0] == result
assert snapshot(root) == changed
rejected(lambda: node.passthrough({**selection, "output_mode":"global"}), "unknown output")
rejected(lambda: node.passthrough({**selection, "lineage":[
{"scene":1,"revision":"3" * 32}, {"scene":2,"revision":"2" * 32}]}), "depends on")
rejected(lambda: node.passthrough({**selection, "lineage":[
{"scene":1,"revision":"../escape"}]}), "32-character")
rejected(lambda: node.passthrough({**selection, "lineage":[selection["lineage"][1]]}), "contiguous")
# Legacy reference caches can be read in place, but a local output must
# not adopt/copy them into this run as a side effect.
descriptor = {"format":"h3_reference_cache_v2", "signature":"cache-test",
"reference_fingerprint":"refs", "metadata":"cache/ref.json",
"tensors":"cache/ref.safetensors", "tensors_sha256":"0" * 64}
metadata_path = root / old2["segment"]["revision_metadata"]
old2["segment"]["reference_cache"] = descriptor
metadata_path.write_text(json.dumps(old2))
changed = snapshot(root)
with patch.object(chain, "_run_local_reference_cache", return_value=None), \
patch.object(chain, "_load_reference_cache_descriptor", return_value=descriptor), \
patch.object(chain, "_adopt_reference_cache_for_run", side_effect=AssertionError("local cache migration")):
assert node.passthrough(selection)[0]["segments"][1]["reference_cache"] == descriptor
assert snapshot(root) == changed
# Canonical pointers still exist. They must not serve as fallback when
# an immutable selection is removed or fails integrity verification.
prefix_path = root / old1["segment"]["revision_metadata"]
prefix_data = prefix_path.read_bytes()
prefix_path.unlink()
changed = snapshot(root)
rejected(lambda: node.passthrough(selection), "no longer available")
assert snapshot(root) == changed, "local output recreated a missing prefix"
prefix_path.write_bytes(prefix_data)
metadata_path.unlink()
changed = snapshot(root)
rejected(lambda: node.passthrough(selection), "no longer available")
assert snapshot(root) == changed
metadata_path.write_text(json.dumps(old2))
video = root / old2["segment"]["segment"]
video.write_bytes(b"corrupt")
changed = snapshot(root)
rejected(lambda: node.passthrough(selection), "integrity")
assert snapshot(root) == changed
print("Checkpoint local output: read-only, exact revisions, stable source hash, legacy cache and fail-closed checks pass")
if __name__ == "__main__":
main()
|