Bundle 54 reviewed terrain and boss HD remasters
This commit is contained in:
raptorreplica committed
1 parent
ca438faf45
commit
456a6c8c0c
59 files changed
+2639
-198
No files matched your search
@@ -31,6 +31,8 @@ def asset_list(value, label):
|
||||
def standalone_selection(candidate, job, rows):
|
||||
"""Standalone means a complete one-frame definition, never an animation slice."""
|
||||
if candidate.get("complete_standalone_selection") is not True:
|
||||
if candidate.get("carried_registered_assets") or candidate.get("carried_registered_identity"):
|
||||
raise ValueError("carried artwork requires an explicit standalone Terrain selection")
|
||||
return None
|
||||
if candidate.get("complete_family") is not False:
|
||||
raise ValueError("standalone selection must not claim whole-family approval")
|
||||
@@ -42,8 +44,21 @@ def standalone_selection(candidate, job, rows):
|
||||
full = asset_list(candidate.get("full_inventory"), "full_inventory")
|
||||
selected = asset_list(candidate.get("selected_assets"), "selected_assets")
|
||||
fallback = asset_list(candidate.get("original_fallback_assets"), "original_fallback_assets")
|
||||
if full != inventory or selected != set(rows) or not selected or selected & fallback or selected | fallback != full:
|
||||
carried = asset_list(candidate.get("carried_registered_assets", []), "carried_registered_assets")
|
||||
if (full != inventory or selected != set(rows) or not selected or
|
||||
selected & fallback or selected & carried or fallback & carried or
|
||||
selected | fallback | carried != full):
|
||||
raise ValueError("standalone selected/fallback partition differs from complete job inventory")
|
||||
if carried:
|
||||
# This is the reviewed additive contract for one original sixteen-tile
|
||||
# inventory, not a general permission to slice other content families.
|
||||
if (job.get("job_id") != "terrain-2059d-205ac" or
|
||||
selected != {"205a3", "205a4"} or len(full) != 16 or
|
||||
len(carried) != 14 or fallback or
|
||||
not isinstance(candidate.get("carried_registered_identity"), dict)):
|
||||
raise ValueError("carried selection differs from the reviewed master74 two-new/fourteen-carried scope")
|
||||
elif candidate.get("carried_registered_identity"):
|
||||
raise ValueError("carried identity has no registered siblings")
|
||||
if candidate.get("targets"):
|
||||
asset_list([target["asset"] for target in candidate["targets"]], "candidate targets")
|
||||
inventory_job = candidate.get("inventory_job_id", job["job_id"])
|
||||
@@ -55,8 +70,106 @@ def standalone_selection(candidate, job, rows):
|
||||
len(sequence["frames"]) != 1 or type(sequence["frames"][0]["frame"]) is not int or
|
||||
sequence["frames"][0]["frame"] != 0):
|
||||
raise ValueError("standalone selection contains an animation or partial sequence")
|
||||
return {key: sorted(value) for key, value in [
|
||||
selection = {key: sorted(value) for key, value in [
|
||||
("full_inventory", full), ("selected_assets", selected), ("original_fallback_assets", fallback)]}
|
||||
if carried:
|
||||
selection["carried_registered_assets"] = sorted(carried)
|
||||
return selection
|
||||
|
||||
|
||||
def same_json(left, right):
|
||||
"""Identity includes JSON types: False is not the integer frame zero."""
|
||||
return json.dumps(left, sort_keys=True, separators=(",", ":")) == json.dumps(right, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def require_standalone_membership(proposal, assets):
|
||||
"""A retained frame zero cannot be hidden inside a partial animation."""
|
||||
for sequence in proposal["sequences"]:
|
||||
base = sequence["base_asset"]
|
||||
base = int(base, 16) if isinstance(base, str) else base
|
||||
if type(sequence["frame_count"]) is not int:
|
||||
raise ValueError("carried context sequence count must be an integer")
|
||||
members = {f"{base + offset:05x}" for offset in range(sequence["frame_count"])}
|
||||
if (assets is None or members & assets) and (
|
||||
type(sequence["frame_count"]) is not int or sequence["frame_count"] != 1 or
|
||||
len(sequence["frames"]) != 1 or type(sequence["frames"][0]["frame"]) is not int or
|
||||
sequence["frames"][0]["frame"] != 0):
|
||||
raise ValueError("carried/new Terrain identity contains an animation or partial sequence")
|
||||
|
||||
|
||||
def validate_carried_registered_identity(root, candidate, job, rows, review):
|
||||
"""Authenticate already accepted siblings; never approve or insert them again."""
|
||||
carried = set(candidate.get("carried_registered_assets", []))
|
||||
if not carried:
|
||||
return
|
||||
identity = candidate["carried_registered_identity"]
|
||||
evidence = review.get("standalone_selection", {})
|
||||
if not same_json(identity, evidence.get("carried_registered_identity")):
|
||||
raise ValueError("review carried identity differs from candidate")
|
||||
baseline = identity.get("baseline", {})
|
||||
if baseline.get("path") != "hd.json":
|
||||
raise ValueError("carried baseline must identify the current HD manifest")
|
||||
manifest = receipt_json(root, baseline)
|
||||
active = proposal_frames(manifest)
|
||||
require_standalone_membership(manifest, carried)
|
||||
if set(rows) & set(active) or not carried <= set(active):
|
||||
raise ValueError("carried baseline omits siblings or already includes new selected IDs")
|
||||
records = identity.get("frames", {})
|
||||
if not isinstance(records, dict) or set(records) != carried:
|
||||
raise ValueError("carried identity omits or adds registered IDs")
|
||||
targets = {target["asset"]: target for target in job["targets"]}
|
||||
provenance = identity.get("provenance", {})
|
||||
if provenance.get("path") != "provenance.json":
|
||||
raise ValueError("carried provenance must identify the current registration evidence")
|
||||
additions = receipt_json(root, provenance).get("reviewed_sequence_additions", [])
|
||||
index = identity.get("registration_record_index")
|
||||
if type(index) is not int or not 0 <= index < len(additions):
|
||||
raise ValueError("unsafe carried registration record index")
|
||||
registered = additions[index]
|
||||
if (registered.get("job_id") != job["job_id"] or
|
||||
asset_list(registered.get("assets"), "registered carried assets") != carried or
|
||||
type(registered.get("frames")) is not int or registered["frames"] != 14 or
|
||||
registered.get("frame_bindings_sha256") != identity.get("frame_bindings_sha256")):
|
||||
raise ValueError("carried registration belongs to another job or frame map")
|
||||
accepted_receipt = identity.get("accepted_runtime_qa", {})
|
||||
if accepted_receipt.get("path") != registered.get("runtime_qa"):
|
||||
raise ValueError("carried evidence is not the previously registered runtime receipt")
|
||||
accepted = receipt_json(root, accepted_receipt)
|
||||
accepted_frames = {frame["asset"]: frame for frame in accepted.get("frames", [])}
|
||||
if (accepted.get("diagnostics") or accepted.get("frame_bindings_sha256") != identity.get("frame_bindings_sha256") or
|
||||
len(accepted_frames) != len(accepted.get("frames", [])) or set(accepted_frames) != carried):
|
||||
raise ValueError("previously registered carried runtime evidence differs")
|
||||
bindings = {}
|
||||
for asset in sorted(carried):
|
||||
record = records[asset]
|
||||
frame = active[asset]
|
||||
if not isinstance(record, dict) or not same_json(record.get("frame_binding"), frame):
|
||||
raise ValueError(f"carried complete registered frame changed: {asset}")
|
||||
target = targets[asset]
|
||||
if (type(frame.get("frame")) is not int or frame["frame"] != 0 or
|
||||
type(frame.get("logical_width")) is not int or type(frame.get("logical_height")) is not int or
|
||||
[frame["logical_width"], frame["logical_height"]] != target["logical_canvas"] or
|
||||
frame.get("origin") != target["origin"] or
|
||||
frame.get("source_sha256") != target["source_sha256"] or
|
||||
record.get("source_sha256") != target["source_sha256"]):
|
||||
raise ValueError(f"carried native source/canvas/origin differs: {asset}")
|
||||
if set(frame) - {"frame", "file", "crop", "logical_width", "logical_height", "origin", "source_sha256"}:
|
||||
raise ValueError(f"carried master74 frame has an unreviewed optional policy: {asset}")
|
||||
raw = contained(root, frame["file"])
|
||||
measurement = accepted_frames[asset]
|
||||
masks = measurement.get("hd_masks", [])
|
||||
if (measurement.get("source_sha256") != frame["source_sha256"] or
|
||||
measurement.get("logical_canvas") != target["logical_canvas"] or len(masks) != 4 or
|
||||
any(type(mask["scale"]) is not int for mask in masks) or
|
||||
{mask["scale"] for mask in masks} != {1, 2, 3, 4}):
|
||||
raise ValueError(f"previously registered carried source/scales differ: {asset}")
|
||||
accepted_raw = accepted.get("artwork_sha256", {}).get(frame["file"])
|
||||
if (record.get("raw_sha256") != accepted_raw or
|
||||
hashlib.sha256(raw.read_bytes()).hexdigest() != accepted_raw):
|
||||
raise ValueError(f"carried raw artwork changed: {asset}")
|
||||
bindings[asset] = frame
|
||||
if identity.get("frame_bindings_sha256") != binding_digest(bindings):
|
||||
raise ValueError("carried complete frame digest differs")
|
||||
|
||||
|
||||
def receipt_json(root, receipt):
|
||||
@@ -179,8 +292,17 @@ def approved_standalone_evidence(root, selection, review, rows, qa):
|
||||
if any(sequence["frame_count"] != 1 or len(sequence["frames"]) != 1 or sequence["frames"][0]["frame"] != 0
|
||||
for sequence in proposal["sequences"]):
|
||||
raise ValueError("combined context candidate includes animation slices")
|
||||
for key in ["baseline", "future_projection"]:
|
||||
receipt_json(root, authority.get(key))
|
||||
baseline = receipt_json(root, authority.get("baseline"))
|
||||
future = receipt_json(root, authority.get("future_projection"))
|
||||
if selection.get("carried_registered_assets"):
|
||||
identity = evidence["carried_registered_identity"]
|
||||
active = proposal_frames(baseline)
|
||||
projected = proposal_frames(future)
|
||||
if (authority["baseline"]["sha256"] != identity["baseline"]["sha256"] or
|
||||
set(active) & set(bindings) or not same_json(projected, {**active, **bindings})):
|
||||
raise ValueError("carried context is not the exact current baseline plus new definitions")
|
||||
require_standalone_membership(proposal, None)
|
||||
require_standalone_membership(future, set(rows) | set(selection["carried_registered_assets"]))
|
||||
closure = receipt_json(root, authority.get("changed_case_closure"))
|
||||
if (closure.get("status") != "CLOSED_QUALIFIED_CONTEXT" or
|
||||
closure.get("candidate_sha256") != authority["candidate"]["sha256"] or
|
||||
@@ -268,6 +390,7 @@ def approved_job(root, identity, project):
|
||||
raise ValueError(f"raw artwork changed since review: {asset}")
|
||||
if selection is not None:
|
||||
pictures, payloads = authentic_terrain_inventory(root, job)
|
||||
validate_carried_registered_identity(root, candidate, job, rows, review)
|
||||
for asset, frame in rows.items():
|
||||
validate_hd_frame(root, frame, payloads[asset], pictures[asset])
|
||||
approved_standalone_evidence(root, selection, review, rows, qa)
|
||||
@@ -310,7 +433,8 @@ def main():
|
||||
raise ValueError("original fallback already has registered HD artwork")
|
||||
overlaps = set(rows) & set(active)
|
||||
if overlaps:
|
||||
if overlaps == set(rows) and all(rows[asset] == active[asset] for asset in rows):
|
||||
if (not candidate.get("carried_registered_assets") and overlaps == set(rows) and
|
||||
all(rows[asset] == active[asset] for asset in rows)):
|
||||
continue
|
||||
raise ValueError(f"duplicate registered ID; review a replacement explicitly: {sorted(overlaps)}")
|
||||
manifest["sequences"].extend(candidate["sequences"])
|
||||
@@ -336,6 +460,10 @@ def main():
|
||||
additions[-1]["approval_scope"] = "complete_standalone_terrain_selection"
|
||||
additions[-1]["original_fallback_assets"] = candidate["original_fallback_assets"]
|
||||
additions[-1]["full_inventory"] = candidate["full_inventory"]
|
||||
if candidate.get("carried_registered_assets"):
|
||||
additions[-1]["approval_scope"] = "complete_standalone_terrain_selection_with_unchanged_registered_siblings"
|
||||
additions[-1]["carried_registered_assets"] = candidate["carried_registered_assets"]
|
||||
additions[-1]["carried_registered_identity"] = candidate["carried_registered_identity"]
|
||||
if additions and not args.dry_run:
|
||||
# Do not overwrite another registration completed during validation.
|
||||
if manifest_path.read_bytes() != original_manifest or provenance_path.read_bytes() != original_provenance:
|
||||
|
||||
@@ -24,6 +24,7 @@ class StandaloneTerrainRegistrationTest(unittest.TestCase):
|
||||
["10530", "10532", "10533", "10534", "10537", "10538", "10539", "1053a", "1053b", "1053c", "1053d", "1053f"],
|
||||
["10540", "10541", "10542", "10543", "10545", "10546", "10547", "10548", "10549", "1054a", "1054b", "1054d", "1054e"],
|
||||
]
|
||||
START_ASSET = 0x10530
|
||||
|
||||
def setUp(self):
|
||||
temporary = tempfile.TemporaryDirectory()
|
||||
@@ -43,12 +44,12 @@ class StandaloneTerrainRegistrationTest(unittest.TestCase):
|
||||
palette = bytes(value // 4 for value in range(256) for _ in range(3))
|
||||
self.palette = self.data / "misc/PALETTE_DAT"
|
||||
self.palette.write_bytes(palette)
|
||||
self.entries = ["1052f 0 tiles/STARTG1TILES"]
|
||||
self.entries = [f"{self.START_ASSET-1:05x} 0 tiles/STARTG1TILES"]
|
||||
self.jobs = []
|
||||
self.pictures = []
|
||||
all_bindings = {}
|
||||
for number, selected in enumerate(self.SELECTED):
|
||||
start = 0x10530 + number * 16
|
||||
start = self.START_ASSET + number * 16
|
||||
identity = f"terrain-{start:05x}-{start+15:05x}"
|
||||
directory = self.hd / "jobs" / identity
|
||||
directory.mkdir(parents=True)
|
||||
@@ -115,7 +116,8 @@ class StandaloneTerrainRegistrationTest(unittest.TestCase):
|
||||
"standalone_selection": {**selection, "material_approvals": approvals}}
|
||||
self.jobs.append({"id": identity, "directory": directory, "candidate": candidate,
|
||||
"job": job, "review": review, "qa": qa})
|
||||
self.entries.append("10550 0 tiles/ENDG1TILES")
|
||||
end = self.START_ASSET + len(self.SELECTED) * 16
|
||||
self.entries.append(f"{end:05x} 0 tiles/ENDG1TILES")
|
||||
(self.data / "INDEX").write_text("\n".join(self.entries) + "\n")
|
||||
self.context_candidate_path = self.hd / "context-candidate.json"
|
||||
self.write(self.context_candidate_path, {"sequences": sum([row["candidate"]["sequences"] for row in self.jobs], [])})
|
||||
@@ -140,7 +142,7 @@ class StandaloneTerrainRegistrationTest(unittest.TestCase):
|
||||
"receipt": self.ref(self.context_path), "selected_assets": row["candidate"]["selected_assets"],
|
||||
"selected_frame_bindings_sha256": binding_digest(proposal_frames(row["candidate"]))}
|
||||
self.write(row["directory"] / "coordinator-review.json", row["review"])
|
||||
self.audit = {"assets": self.pictures, "sequences": {"Terrain": [{"name": "Terrain set 1", "frames": list(range(0x10530, 0x10550))}]},
|
||||
self.audit = {"assets": self.pictures, "sequences": {"Terrain": [{"name": "Terrain set 1", "frames": list(range(self.START_ASSET, end))}]},
|
||||
"terrain": {"runtime_hd_composition_supported": True}}
|
||||
self.write(self.hd / "coverage-audit.json", self.audit)
|
||||
self.write(self.hd / "runtime-content-audit.json", {})
|
||||
@@ -486,5 +488,354 @@ class StandaloneTerrainRegistrationTest(unittest.TestCase):
|
||||
with self.assertRaisesRegex(ValueError, "omits source IDs"): self.approve()
|
||||
|
||||
|
||||
class CarriedRegisteredTerrainRegistrationTest(unittest.TestCase):
|
||||
"""Master74 adds two definitions while its fourteen active siblings stay exact."""
|
||||
START_ASSET = 0x2059d
|
||||
SELECTED = [["205a3", "205a4"]]
|
||||
sha = staticmethod(StandaloneTerrainRegistrationTest.sha)
|
||||
write = staticmethod(StandaloneTerrainRegistrationTest.write)
|
||||
ref = StandaloneTerrainRegistrationTest.ref
|
||||
approve = StandaloneTerrainRegistrationTest.approve
|
||||
save = StandaloneTerrainRegistrationTest.save
|
||||
|
||||
def setUp(self):
|
||||
StandaloneTerrainRegistrationTest.setUp(self)
|
||||
row = self.jobs[0]
|
||||
peer = json.loads((row["directory"] / "peer.json").read_text())
|
||||
carried = [record for record in peer["records"] if record["asset"] not in self.SELECTED[0]]
|
||||
self.carried = {record["asset"]: record["selected_frame_binding"] for record in carried}
|
||||
self.baseline = {"format_version": 1, "sequences": [
|
||||
{"base_asset": int(asset, 16), "frame_count": 1, "frames": [frame]}
|
||||
for asset, frame in self.carried.items()]}
|
||||
self.write(self.hd / "hd.json", self.baseline)
|
||||
accepted = copy.deepcopy(row["qa"])
|
||||
accepted["frame_bindings_sha256"] = binding_digest(self.carried)
|
||||
accepted["frames"] = [{"asset": asset, "source_sha256": frame["source_sha256"], "logical_canvas": [4, 4],
|
||||
"hd_masks": [{"scale": n} for n in range(1, 5)]}
|
||||
for asset, frame in self.carried.items()]
|
||||
accepted["artwork_sha256"] = {frame["file"]: self.sha((self.hd / frame["file"]).read_bytes())
|
||||
for frame in self.carried.values()}
|
||||
self.accepted_path = self.hd / "accepted-carried-runtime-qa.json"
|
||||
self.write(self.accepted_path, accepted)
|
||||
self.write(self.hd / "provenance.json", {"registered_frame_count": 14, "reviewed_sequence_additions": [{
|
||||
"job_id": row["id"], "assets": sorted(self.carried), "frames": 14,
|
||||
"frame_bindings_sha256": binding_digest(self.carried), "runtime_qa": self.ref(self.accepted_path)["path"]}]})
|
||||
candidate = copy.deepcopy(row["candidate"])
|
||||
candidate["original_fallback_assets"] = []
|
||||
candidate["carried_registered_assets"] = sorted(self.carried)
|
||||
candidate["carried_registered_identity"] = {
|
||||
"baseline": self.ref(self.hd / "hd.json"), "provenance": self.ref(self.hd / "provenance.json"),
|
||||
"registration_record_index": 0, "accepted_runtime_qa": self.ref(self.accepted_path),
|
||||
"frame_bindings_sha256": binding_digest(self.carried),
|
||||
"frames": {record["asset"]: {"frame_binding": record["selected_frame_binding"],
|
||||
"source_sha256": record["source_payload_sha256"], "raw_sha256": record["full_untouched_raw_sha256"]}
|
||||
for record in carried}}
|
||||
self.resign(candidate)
|
||||
self.write(self.hd / "context-future_projection.json", {
|
||||
"format_version": 1, "sequences": self.baseline["sequences"] + candidate["sequences"]})
|
||||
self.reseal_context(baseline=self.ref(self.hd / "hd.json"),
|
||||
future=self.ref(self.hd / "context-future_projection.json"))
|
||||
|
||||
def resign(self, candidate):
|
||||
"""Reseal hypothetical claims so tests reach the actual semantic guards."""
|
||||
digest = binding_digest(proposal_frames(candidate))
|
||||
for key in ["qa", "review"]:
|
||||
self.jobs[0][key]["frame_bindings_sha256"] = digest
|
||||
self.resign_identity(candidate)
|
||||
|
||||
def resign_identity(self, candidate):
|
||||
row = self.jobs[0]
|
||||
row["candidate"] = candidate
|
||||
self.save("candidate", candidate)
|
||||
candidate_sha = self.ref(row["directory"] / "candidate.json")["sha256"]
|
||||
for key in ["qa", "review"]:
|
||||
row[key]["candidate_sha256"] = candidate_sha
|
||||
evidence = row["review"]["standalone_selection"]
|
||||
for key in ["full_inventory", "selected_assets", "original_fallback_assets", "carried_registered_assets", "carried_registered_identity"]:
|
||||
if key in candidate:
|
||||
evidence[key] = copy.deepcopy(candidate[key])
|
||||
else:
|
||||
evidence.pop(key, None)
|
||||
self.save("qa", row["qa"]); self.save("review", row["review"])
|
||||
|
||||
def reseal_context(self, baseline=None, future=None, proposal=None):
|
||||
context = json.loads(self.context_path.read_text())
|
||||
authority = context["standalone_selection_context"]
|
||||
if baseline is not None: authority["baseline"] = baseline
|
||||
if future is not None: authority["future_projection"] = future
|
||||
if proposal is not None:
|
||||
self.write(self.context_candidate_path, proposal)
|
||||
authority["candidate"] = self.ref(self.context_candidate_path)
|
||||
bindings = proposal_frames(proposal)
|
||||
authority["selected_assets"] = sorted(bindings)
|
||||
authority["frame_bindings_sha256"] = binding_digest(bindings)
|
||||
closure_path = self.hd / "context-changed_case_closure.json"
|
||||
closure = json.loads(closure_path.read_text())
|
||||
for field, value in [("candidate_sha256", authority["candidate"]["sha256"]),
|
||||
("frame_bindings_sha256", authority["frame_bindings_sha256"]),
|
||||
("baseline_sha256", authority["baseline"]["sha256"]),
|
||||
("future_projection_sha256", authority["future_projection"]["sha256"])]:
|
||||
closure[field] = value
|
||||
self.write(closure_path, closure)
|
||||
authority["changed_case_closure"] = self.ref(closure_path)
|
||||
self.write(self.context_path, context)
|
||||
self.jobs[0]["review"]["standalone_selection"]["contextual_approval"]["receipt"] = self.ref(self.context_path)
|
||||
self.save("review", self.jobs[0]["review"])
|
||||
|
||||
@contextlib.contextmanager
|
||||
def original_fixture(self):
|
||||
files = {path: (path.read_bytes(), path.stat().st_mtime_ns) for path in self.hd.rglob("*") if path.is_file()}
|
||||
jobs = copy.deepcopy(self.jobs)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
for path in self.hd.rglob("*"):
|
||||
if path.is_file() and path not in files: path.unlink()
|
||||
for path, (payload, stamp) in files.items():
|
||||
path.write_bytes(payload); os.utime(path, ns=(stamp, stamp))
|
||||
self.jobs = jobs
|
||||
|
||||
def test_exact_two_new_fourteen_carried_contract_passes_without_approving_siblings_again(self):
|
||||
candidate, review, rows = self.approve()
|
||||
self.assertEqual(set(rows), {"205a3", "205a4"})
|
||||
self.assertEqual(len(candidate["full_inventory"]), 16)
|
||||
self.assertEqual(len(candidate["carried_registered_assets"]), 14)
|
||||
self.assertEqual(candidate["original_fallback_assets"], [])
|
||||
self.assertEqual(set(review["standalone_selection"]["material_approvals"]), set(rows))
|
||||
|
||||
def test_real_registration_adds_only_two_and_preserves_all_fourteen_complete_frames(self):
|
||||
before = proposal_frames(json.loads((self.hd / "hd.json").read_text()))
|
||||
argv = ["registrar", "--hd-root", str(self.hd), "--job", self.jobs[0]["id"]]
|
||||
with patch.object(registrar, "__file__", str(self.project / "tools/hd_register_reviewed.py")), patch("sys.argv", argv), contextlib.redirect_stdout(io.StringIO()):
|
||||
registrar.main() # Temporary synthetic pack only; no production invocation.
|
||||
active = proposal_frames(json.loads((self.hd / "hd.json").read_text()))
|
||||
self.assertEqual(len(active), 16)
|
||||
self.assertEqual({asset: active[asset] for asset in before}, before)
|
||||
additions = json.loads((self.hd / "provenance.json").read_text())["reviewed_sequence_additions"]
|
||||
self.assertEqual([row["frames"] for row in additions], [14, 2])
|
||||
self.assertEqual(additions[-1]["assets"], ["205a3", "205a4"])
|
||||
self.assertEqual(additions[-1]["approval_scope"], "complete_standalone_terrain_selection_with_unchanged_registered_siblings")
|
||||
self.assertEqual(additions[-1]["carried_registered_assets"], sorted(before))
|
||||
|
||||
def test_duplicate_new_request_is_rejected_without_pack_write(self):
|
||||
before = {name: (self.hd / name).read_bytes() for name in ["hd.json", "provenance.json"]}
|
||||
argv = ["registrar", "--hd-root", str(self.hd), "--job", self.jobs[0]["id"], "--job", self.jobs[0]["id"]]
|
||||
with patch.object(registrar, "__file__", str(self.project / "tools/hd_register_reviewed.py")), patch("sys.argv", argv):
|
||||
with self.assertRaisesRegex(ValueError, "duplicate registered ID"): registrar.main()
|
||||
self.assertEqual(before, {name: (self.hd / name).read_bytes() for name in before})
|
||||
|
||||
def test_carried_partition_category_cross_job_and_animation_mismatches_reject(self):
|
||||
for mutation in ["missing_carried", "duplicate_carried", "new_overlap", "fallback_overlap", "fallback_instead_of_carry",
|
||||
"foreign_carried", "short_inventory", "cross_job", "category", "kind", "whole_family",
|
||||
"no_explicit_selection", "animation", "partial_animation", "boolean_count", "boolean_frame",
|
||||
"duplicate_sequence", "duplicate_target"]:
|
||||
with self.subTest(mutation=mutation), self.original_fixture():
|
||||
candidate = copy.deepcopy(self.jobs[0]["candidate"])
|
||||
if mutation == "missing_carried": candidate["carried_registered_assets"].pop()
|
||||
elif mutation == "duplicate_carried": candidate["carried_registered_assets"].append("2059d")
|
||||
elif mutation == "new_overlap": candidate["carried_registered_assets"].append("205a3")
|
||||
elif mutation == "fallback_overlap": candidate["original_fallback_assets"] = ["2059d"]
|
||||
elif mutation == "fallback_instead_of_carry":
|
||||
candidate["original_fallback_assets"] = candidate.pop("carried_registered_assets"); candidate.pop("carried_registered_identity")
|
||||
elif mutation == "foreign_carried": candidate["carried_registered_assets"][0] = "10530"
|
||||
elif mutation == "short_inventory": candidate["full_inventory"].pop()
|
||||
elif mutation == "cross_job":
|
||||
self.jobs[0]["job"]["job_id"] = "terrain-10530-1053f"
|
||||
self.save("job", self.jobs[0]["job"])
|
||||
elif mutation == "category": candidate["category"] = "Enemies"
|
||||
elif mutation == "kind": candidate["kind"] = "animation_sequence"
|
||||
elif mutation == "whole_family": candidate["complete_family"] = True
|
||||
elif mutation == "no_explicit_selection": candidate.pop("complete_standalone_selection")
|
||||
elif mutation == "animation":
|
||||
candidate["sequences"][0]["frame_count"] = 2
|
||||
candidate["sequences"][0]["frames"].append({**candidate["sequences"][0]["frames"][0], "frame": 1})
|
||||
elif mutation == "partial_animation": candidate["sequences"][0]["frame_count"] = 2
|
||||
elif mutation == "boolean_count": candidate["sequences"][0]["frame_count"] = True
|
||||
elif mutation == "boolean_frame": candidate["sequences"][0]["frames"][0]["frame"] = False
|
||||
elif mutation == "duplicate_sequence": candidate["sequences"].append(candidate["sequences"][0])
|
||||
else:
|
||||
self.jobs[0]["job"]["targets"].append(self.jobs[0]["job"]["targets"][0]); self.save("job", self.jobs[0]["job"])
|
||||
self.resign_identity(candidate)
|
||||
if mutation == "fallback_instead_of_carry":
|
||||
# The legacy original-fallback schema still parses, but
|
||||
# cannot register over already active siblings.
|
||||
argv = ["registrar", "--hd-root", str(self.hd), "--job", self.jobs[0]["id"]]
|
||||
with patch.object(registrar, "__file__", str(self.project / "tools/hd_register_reviewed.py")), patch("sys.argv", argv):
|
||||
with self.assertRaisesRegex(ValueError, "original fallback already"): registrar.main()
|
||||
else:
|
||||
with self.assertRaises(ValueError): self.approve()
|
||||
|
||||
def test_changed_carried_complete_binding_rejects_even_with_rehashed_candidate(self):
|
||||
for mutation in ["raw_path", "crop", "canvas", "origin", "source", "optional_policy", "frame_index", "omitted_record", "extra_record", "digest"]:
|
||||
with self.subTest(mutation=mutation), self.original_fixture():
|
||||
candidate = copy.deepcopy(self.jobs[0]["candidate"])
|
||||
identity = candidate["carried_registered_identity"]
|
||||
record = identity["frames"]["2059d"]
|
||||
frame = record["frame_binding"]
|
||||
if mutation == "raw_path": frame["file"] = identity["frames"]["2059e"]["frame_binding"]["file"]
|
||||
elif mutation == "crop": frame["crop"] = [0, 0, 4, 4]
|
||||
elif mutation == "canvas": frame["logical_width"] = 8
|
||||
elif mutation == "origin": frame["origin"] = [1, 0]
|
||||
elif mutation == "source": record["source_sha256"] = "0" * 64
|
||||
elif mutation == "optional_policy": frame["native_nearest_cues"] = [[0, 0]]
|
||||
elif mutation == "frame_index": frame["frame"] = 1
|
||||
elif mutation == "omitted_record": identity["frames"].pop("2059d")
|
||||
elif mutation == "extra_record": identity["frames"]["205a3"] = record
|
||||
else: identity["frame_bindings_sha256"] = "0" * 64
|
||||
self.resign(candidate)
|
||||
with self.assertRaises(ValueError): self.approve()
|
||||
|
||||
def test_actual_carried_raw_change_rejects_even_with_forged_new_raw_hash_and_preserved_mtime(self):
|
||||
candidate = copy.deepcopy(self.jobs[0]["candidate"])
|
||||
record = candidate["carried_registered_identity"]["frames"]["2059d"]
|
||||
raw = self.hd / record["frame_binding"]["file"]
|
||||
stamp = raw.stat().st_mtime_ns
|
||||
raw.write_bytes(raw.read_bytes() + b"changed carried art")
|
||||
os.utime(raw, ns=(stamp, stamp))
|
||||
record["raw_sha256"] = self.sha(raw.read_bytes())
|
||||
self.resign(candidate)
|
||||
with self.assertRaisesRegex(ValueError, "carried raw artwork changed"): self.approve()
|
||||
|
||||
def test_matching_carried_raw_symlink_cannot_escape_hd_root(self):
|
||||
raw = self.hd / self.carried["2059d"]["file"]
|
||||
outside = self.hd.parent / "outside-carried.png"
|
||||
outside.write_bytes(raw.read_bytes())
|
||||
raw.unlink(); raw.symlink_to(outside)
|
||||
with self.assertRaisesRegex(ValueError, "escapes HD root"): self.approve()
|
||||
|
||||
def test_candidate_and_review_carried_identities_must_agree(self):
|
||||
review = copy.deepcopy(self.jobs[0]["review"])
|
||||
review["standalone_selection"]["carried_registered_identity"]["frame_bindings_sha256"] = "0" * 64
|
||||
self.save("review", review)
|
||||
with self.assertRaisesRegex(ValueError, "review carried identity differs"): self.approve()
|
||||
|
||||
def test_rehashed_actual_baseline_changes_cannot_replace_accepted_carry_identity(self):
|
||||
for mutation in ["raw_path", "crop", "canvas", "origin", "source", "optional_policy"]:
|
||||
with self.subTest(mutation=mutation), self.original_fixture():
|
||||
baseline = copy.deepcopy(self.baseline)
|
||||
frame = baseline["sequences"][0]["frames"][0]
|
||||
if mutation == "raw_path": frame["file"] = baseline["sequences"][1]["frames"][0]["file"]
|
||||
elif mutation == "crop": frame["crop"] = [0, 0, 4, 4]
|
||||
elif mutation == "canvas": frame["logical_width"] = 8
|
||||
elif mutation == "origin": frame["origin"] = [1, 0]
|
||||
elif mutation == "source": frame["source_sha256"] = "0" * 64
|
||||
else: frame["preserve_reserved_coverage"] = True
|
||||
self.write(self.hd / "hd.json", baseline)
|
||||
candidate = copy.deepcopy(self.jobs[0]["candidate"])
|
||||
identity = candidate["carried_registered_identity"]
|
||||
identity["baseline"] = self.ref(self.hd / "hd.json")
|
||||
identity["frames"]["2059d"]["frame_binding"] = frame
|
||||
identity["frame_bindings_sha256"] = binding_digest({**self.carried, "2059d": frame})
|
||||
self.resign(candidate)
|
||||
with self.assertRaises(ValueError): self.approve()
|
||||
|
||||
def test_current_baseline_and_accepted_registration_receipts_are_exact(self):
|
||||
for mutation in ["wrong_baseline_path", "stale_baseline_hash", "missing_baseline", "wrong_provenance_path",
|
||||
"stale_provenance", "wrong_record_index", "boolean_record_index", "other_job_record", "wrong_count",
|
||||
"wrong_accepted_path", "changed_accepted_hash", "changed_accepted_digest", "changed_accepted_source",
|
||||
"changed_accepted_scale", "duplicate_accepted_scale", "duplicate_accepted_id"]:
|
||||
with self.subTest(mutation=mutation), self.original_fixture():
|
||||
candidate = copy.deepcopy(self.jobs[0]["candidate"])
|
||||
identity = candidate["carried_registered_identity"]
|
||||
if mutation == "wrong_baseline_path": identity["baseline"] = self.ref(self.hd / "context-baseline.json")
|
||||
elif mutation == "stale_baseline_hash": identity["baseline"]["sha256"] = "0" * 64
|
||||
elif mutation == "missing_baseline": identity.pop("baseline")
|
||||
elif mutation == "wrong_provenance_path": identity["provenance"] = self.ref(self.hd / "context-baseline.json")
|
||||
elif mutation == "stale_provenance": identity["provenance"]["sha256"] = "0" * 64
|
||||
elif mutation == "wrong_record_index": identity["registration_record_index"] = -1
|
||||
elif mutation == "boolean_record_index": identity["registration_record_index"] = False
|
||||
elif mutation in ["other_job_record", "wrong_count"]:
|
||||
path = self.hd / "provenance.json"; provenance = json.loads(path.read_text())
|
||||
entry = provenance["reviewed_sequence_additions"][0]
|
||||
if mutation == "other_job_record": entry["job_id"] = "terrain-10530-1053f"
|
||||
else: entry["frames"] = 13
|
||||
self.write(path, provenance); identity["provenance"] = self.ref(path)
|
||||
elif mutation == "wrong_accepted_path": identity["accepted_runtime_qa"] = self.ref(self.jobs[0]["directory"] / "runtime-qa.json")
|
||||
elif mutation == "changed_accepted_hash": identity["accepted_runtime_qa"]["sha256"] = "0" * 64
|
||||
else:
|
||||
accepted = json.loads(self.accepted_path.read_text())
|
||||
if mutation == "changed_accepted_digest": accepted["frame_bindings_sha256"] = "0" * 64
|
||||
elif mutation == "changed_accepted_source": accepted["frames"][0]["source_sha256"] = "0" * 64
|
||||
elif mutation == "changed_accepted_scale": accepted["frames"][0]["hd_masks"][0]["scale"] = True
|
||||
elif mutation == "duplicate_accepted_scale": accepted["frames"][0]["hd_masks"].append({"scale": 4})
|
||||
else: accepted["frames"].append(accepted["frames"][0])
|
||||
self.write(self.accepted_path, accepted); identity["accepted_runtime_qa"] = self.ref(self.accepted_path)
|
||||
self.resign(candidate)
|
||||
with self.assertRaises((ValueError, OSError)): self.approve()
|
||||
|
||||
def test_baseline_cannot_hide_carried_tile_in_boolean_or_partial_animation(self):
|
||||
for mutation in ["boolean_count", "boolean_frame", "partial_animation", "full_animation", "empty_sequence", "missing_sibling", "duplicate_sibling", "already_selected"]:
|
||||
with self.subTest(mutation=mutation), self.original_fixture():
|
||||
baseline = copy.deepcopy(self.baseline)
|
||||
if mutation == "boolean_count": baseline["sequences"][0]["frame_count"] = True
|
||||
elif mutation == "boolean_frame": baseline["sequences"][0]["frames"][0]["frame"] = False
|
||||
elif mutation == "partial_animation": baseline["sequences"][0]["frame_count"] = 2
|
||||
elif mutation == "full_animation":
|
||||
baseline["sequences"][0]["frame_count"] = 2
|
||||
baseline["sequences"][0]["frames"].append({**baseline["sequences"][1]["frames"][0], "frame": 1})
|
||||
baseline["sequences"].pop(1)
|
||||
elif mutation == "empty_sequence": baseline["sequences"].append({"base_asset": 0x2059d, "frame_count": 1, "frames": []})
|
||||
elif mutation == "missing_sibling": baseline["sequences"].pop()
|
||||
elif mutation == "duplicate_sibling": baseline["sequences"].append(baseline["sequences"][0])
|
||||
else: baseline["sequences"].append(self.jobs[0]["candidate"]["sequences"][0])
|
||||
self.write(self.hd / "hd.json", baseline)
|
||||
candidate = copy.deepcopy(self.jobs[0]["candidate"])
|
||||
candidate["carried_registered_identity"]["baseline"] = self.ref(self.hd / "hd.json")
|
||||
self.resign(candidate)
|
||||
with self.assertRaises(ValueError): self.approve()
|
||||
|
||||
def test_correctly_rehashed_future_projection_cannot_change_any_carried_or_new_binding(self):
|
||||
for mutation in ["omitted_carry", "omitted_new", "foreign_asset", "changed_carry", "changed_new", "boolean_frame", "boolean_count", "partial_animation", "empty_sequence", "duplicate_id", "wrong_baseline", "registered_context_overlap"]:
|
||||
with self.subTest(mutation=mutation), self.original_fixture():
|
||||
path = self.hd / "context-future_projection.json"
|
||||
future = json.loads(path.read_text())
|
||||
if mutation == "omitted_carry": future["sequences"].pop(0)
|
||||
elif mutation == "omitted_new": future["sequences"].pop()
|
||||
elif mutation == "foreign_asset": future["sequences"][0]["base_asset"] = 0x10530
|
||||
elif mutation == "changed_carry": future["sequences"][0]["frames"][0]["crop"] = [0, 0, 4, 4]
|
||||
elif mutation == "changed_new": future["sequences"][-1]["frames"][0]["origin"] = [1, 0]
|
||||
elif mutation == "boolean_frame": future["sequences"][0]["frames"][0]["frame"] = False
|
||||
elif mutation == "boolean_count": future["sequences"][0]["frame_count"] = True
|
||||
elif mutation == "partial_animation": future["sequences"][0]["frame_count"] = 2
|
||||
elif mutation == "empty_sequence": future["sequences"].append({"base_asset": 0x205a3, "frame_count": 1, "frames": []})
|
||||
elif mutation == "duplicate_id": future["sequences"].append(future["sequences"][0])
|
||||
elif mutation == "wrong_baseline":
|
||||
self.reseal_context(baseline=self.ref(self.hd / "context-baseline.json"))
|
||||
else:
|
||||
proposal = json.loads(self.context_candidate_path.read_text())
|
||||
proposal["sequences"].append(self.baseline["sequences"][0])
|
||||
self.reseal_context(proposal=proposal)
|
||||
if mutation not in ["wrong_baseline", "registered_context_overlap"]:
|
||||
self.write(path, future); self.reseal_context(future=self.ref(path))
|
||||
with self.assertRaises((ValueError, KeyError)): self.approve()
|
||||
|
||||
def test_carried_original_source_symlink_or_bytes_cannot_be_replaced(self):
|
||||
target = self.jobs[0]["job"]["targets"][0]
|
||||
source = self.data / target["source_path"]
|
||||
payload = source.read_bytes()
|
||||
source.write_bytes(payload[:-1] + b"\xff")
|
||||
with self.assertRaisesRegex(ValueError, "original inventory source differs"): self.approve()
|
||||
source.write_bytes(payload)
|
||||
outside = self.data.parent / "outside-carried-source"
|
||||
outside.write_bytes(payload)
|
||||
source.unlink(); source.symlink_to(outside)
|
||||
with self.assertRaisesRegex(ValueError, "INDEX path escapes"): self.approve()
|
||||
|
||||
def test_pending_or_missing_new_material_and_context_are_not_authorized_by_carried_content(self):
|
||||
for mutation in ["pending", "missing_material", "missing_context", "wrong_material", "same_reviewer"]:
|
||||
with self.subTest(mutation=mutation), self.original_fixture():
|
||||
review = copy.deepcopy(self.jobs[0]["review"])
|
||||
if mutation == "pending": review["status"] = "PENDING"
|
||||
elif mutation == "missing_material": review["standalone_selection"]["material_approvals"].pop("205a3")
|
||||
elif mutation == "missing_context": review["standalone_selection"].pop("contextual_approval")
|
||||
elif mutation == "wrong_material": review["standalone_selection"]["material_approvals"]["205a3"]["source_sha256"] = "0" * 64
|
||||
else:
|
||||
peer_path = self.jobs[0]["directory"] / "peer.json"; peer = json.loads(peer_path.read_text()); peer["reviewer"] = peer["author"]
|
||||
self.write(peer_path, peer)
|
||||
for approval in review["standalone_selection"]["material_approvals"].values(): approval["independent_review"] = self.ref(peer_path)
|
||||
self.save("review", review)
|
||||
with self.assertRaises(ValueError): self.approve()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user