232 lines
13 KiB
Python
232 lines
13 KiB
Python
"""Family review references preserve queue decisions and external evidence."""
|
|
import argparse
|
|
import contextlib
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import hd_job_queue as queue
|
|
from hd_candidate_review import binding_digest, proposal_frames
|
|
|
|
|
|
class FamilyReviewLedgerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.data = Path(self.temporary.name) / "data"
|
|
self.hd = self.data / "hd"
|
|
self.jobs = self.hd / "jobs"
|
|
self.job_id = "enemies-11000-11002"
|
|
self.directory = self.jobs / self.job_id
|
|
self.directory.mkdir(parents=True)
|
|
pictures = []
|
|
entries = []
|
|
for asset in [0x11000, 0x11001, 0x11002, 0x21000]:
|
|
identity = f"{asset:05x}"
|
|
terrain = asset == 0x21000
|
|
name = "unused_ground" if terrain else f"flanker.{asset - 0x11000}"
|
|
payload = name.encode()
|
|
(self.data / name).write_bytes(payload)
|
|
entries.append(f"{identity} {len(payload)} {name}")
|
|
pictures.append({
|
|
"asset": identity, "path": name, "category": "Terrain" if terrain else "Enemies and ground objects",
|
|
"source_sha256": self.sha(payload), "logical_canvas": [32, 32],
|
|
"header_options": 0, "source_palette": "PALETTE_DAT",
|
|
"source_palette_sha256": "a" * 64, "decoded_rgba_sha256": "b" * 64,
|
|
"hd": asset == 0x11002, "registry_or_layout_referenced": not terrain,
|
|
})
|
|
for asset, name in [(0x31000, "intro_AGX.0"), (0x31001, "small_FNT")]:
|
|
(self.data / name).write_bytes(name.encode())
|
|
entries.append(f"{asset:05x} {len(name)} {name}")
|
|
(self.data / "INDEX").write_text("\n".join(entries) + "\n")
|
|
self.expected = {row["asset"]: row for row in pictures[:3]}
|
|
audit = {
|
|
"assets": pictures,
|
|
"sequences": {
|
|
"Enemies and ground objects": [{"name": "Flanker", "base_asset": 0x11000,
|
|
"frames": [0x11000, 0x11001, 0x11002]}],
|
|
"Terrain": [{"name": "Terrain set 1", "frames": [0x21000]}],
|
|
}, "terrain": {"runtime_hd_composition_supported": True},
|
|
}
|
|
self.write(self.hd / "coverage-audit.json", audit)
|
|
self.write(self.hd / "runtime-content-audit.json", {})
|
|
self.candidate = {"job_id": self.job_id, "complete_family": True, "sequences": [{
|
|
"base_asset": 0x11000, "frame_count": 3, "frames": [],
|
|
}]}
|
|
for offset, source in enumerate(pictures[:3]):
|
|
relative = f"jobs/{self.job_id}/generated/frame-{offset}.png"
|
|
artwork = self.hd / relative
|
|
artwork.parent.mkdir(parents=True, exist_ok=True)
|
|
# The queue deliberately checks presence/metadata, not PNG artistry.
|
|
artwork.write_bytes(b"synthetic artwork presence")
|
|
self.candidate["sequences"][0]["frames"].append({
|
|
"frame": offset, "file": relative, "logical_width": 32, "logical_height": 32,
|
|
"origin": [0, 0], "source_sha256": source["source_sha256"], "crop": [0, 0, 128, 128],
|
|
})
|
|
self.candidate_path = self.directory / "candidate.json"
|
|
self.write(self.candidate_path, self.candidate)
|
|
self.review = {
|
|
"status": "approved", "candidate_sha256": self.sha(self.candidate_path.read_bytes()),
|
|
"frame_bindings_sha256": binding_digest(proposal_frames(self.candidate, self.expected)),
|
|
"runtime_qa": f"jobs/{self.job_id}/runtime-review/runtime-qa.json",
|
|
"source_and_all_scale_visual_review": {"description": "retained full visual receipt " * 100},
|
|
"independent_review": {"report_sha256": "c" * 64},
|
|
}
|
|
self.review_path = self.directory / "coordinator-review.json"
|
|
self.write(self.review_path, self.review)
|
|
self.write(self.directory / "job.json", {"owner": "/root/fixture", "targets": []})
|
|
self.ledger_path = Path(self.temporary.name) / "ledger.json"
|
|
self.write(self.ledger_path, {"format_version": 1, "assets": {
|
|
"11002": {"source_sha256": pictures[2]["source_sha256"], "status": "renderreachable",
|
|
"renderer": {"availability": "verified", "evidence": ["existing exact scene"]}},
|
|
}})
|
|
self.args = argparse.Namespace(data=self.data, audit=self.hd / "coverage-audit.json",
|
|
runtime=self.hd / "runtime-content-audit.json", jobs=self.jobs,
|
|
ledger=self.ledger_path, terrain_batch=16)
|
|
|
|
@staticmethod
|
|
def sha(payload):
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
@staticmethod
|
|
def write(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, indent=2) + "\n")
|
|
|
|
def build(self, legacy=False):
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
if legacy:
|
|
with patch.object(queue, "normalize_ledger", side_effect=lambda ledger, *_: ledger):
|
|
queue.build(self.args)
|
|
else:
|
|
queue.build(self.args)
|
|
return json.loads(self.ledger_path.read_text())
|
|
|
|
def test_round_trip_keeps_decisions_sources_and_complete_external_receipt(self):
|
|
before_receipt = self.review_path.read_bytes()
|
|
legacy = self.build(legacy=True)
|
|
normalized = self.build()
|
|
self.assertEqual(normalized["format_version"], 1)
|
|
self.assertEqual(normalized["ledger_schema"], queue.LEDGER_SCHEMA)
|
|
self.assertEqual(set(normalized["family_reviews"]), {self.job_id})
|
|
family = normalized["family_reviews"][self.job_id]
|
|
self.assertEqual(family["receipt"], {"path": f"jobs/{self.job_id}/coordinator-review.json",
|
|
"sha256": self.sha(before_receipt)})
|
|
self.assertEqual(family["frame_bindings_sha256"], self.review["frame_bindings_sha256"])
|
|
self.assertNotIn("source_and_all_scale_visual_review", family)
|
|
for identity, old in legacy["assets"].items():
|
|
new = normalized["assets"][identity]
|
|
self.assertEqual(queue.coordinator_review_for_asset(legacy, identity, self.hd),
|
|
queue.coordinator_review_for_asset(normalized, identity, self.hd))
|
|
self.assertEqual({k: v for k, v in old.items() if k != "coordinator_review"},
|
|
{k: v for k, v in new.items() if k != "coordinator_review_ref"})
|
|
self.assertNotIn("coordinator_review", new)
|
|
self.assertEqual(legacy["picture_status_counts"], normalized["picture_status_counts"])
|
|
self.assertEqual(normalized, queue.normalize_ledger(normalized, self.hd))
|
|
self.assertEqual(normalized, self.build())
|
|
self.assertEqual(self.review_path.read_bytes(), before_receipt)
|
|
local_job = json.loads((self.directory / "job.json").read_text())
|
|
self.assertEqual(local_job["coordinator_review"], self.review)
|
|
self.assertLess(len(json.dumps(normalized)), len(json.dumps(legacy)))
|
|
|
|
def test_unreviewed_unused_and_native_preservation_entries_remain(self):
|
|
self.review["status"] = "hold"
|
|
self.write(self.review_path, self.review)
|
|
ledger = self.build()
|
|
self.assertEqual(ledger["assets"]["11000"]["status"], "candidate")
|
|
self.assertEqual(ledger["family_reviews"][self.job_id]["status"], "hold")
|
|
self.assertEqual(ledger["assets"]["11002"]["status"], "renderreachable")
|
|
unused = ledger["assets"]["21000"]
|
|
self.assertEqual(unused["status"], "missing")
|
|
self.assertFalse(unused["registry_or_layout_referenced"])
|
|
self.assertNotIn("coordinator_review_ref", unused)
|
|
self.assertEqual(ledger["assets"]["31000"]["kind"], "cinematic_delta")
|
|
self.assertEqual(ledger["assets"]["31001"]["kind"], "bitmap_font")
|
|
self.assertFalse(ledger["assets"]["31000"]["remaster_required"])
|
|
self.review_path.unlink()
|
|
without_review = self.build()
|
|
self.assertEqual(without_review["family_reviews"], {})
|
|
self.assertNotIn("coordinator_review_ref", without_review["assets"]["11000"])
|
|
self.assertEqual(without_review["assets"]["11000"]["status"], "candidate")
|
|
|
|
def test_missing_changed_and_stale_candidates_cannot_reuse_approval(self):
|
|
for mutation in ["missing", "changed_binding", "stale_review_hash"]:
|
|
with self.subTest(mutation=mutation):
|
|
candidate = copy.deepcopy(self.candidate)
|
|
review = copy.deepcopy(self.review)
|
|
if mutation == "changed_binding":
|
|
candidate["sequences"][0]["frames"][0]["crop"] = [0, 0, 256, 256]
|
|
if mutation == "stale_review_hash":
|
|
review.pop("frame_bindings_sha256")
|
|
review["candidate_sha256"] = "0" * 64
|
|
self.write(self.candidate_path, candidate)
|
|
self.write(self.review_path, review)
|
|
if mutation == "missing":
|
|
self.candidate_path.unlink()
|
|
ledger = self.build()
|
|
self.assertEqual(ledger["family_reviews"], {})
|
|
self.assertNotIn("coordinator_review_ref", ledger["assets"]["11000"])
|
|
self.assertNotEqual(ledger["assets"]["11000"]["status"], "reviewed")
|
|
self.assertEqual(ledger["assets"]["11002"]["status"], "renderreachable")
|
|
|
|
def test_complete_family_binds_every_job_target_and_rejects_wrong_sources(self):
|
|
for mutation in ["omitted_target", "foreign_target", "changed_source", "wrong_job"]:
|
|
with self.subTest(mutation=mutation):
|
|
candidate = copy.deepcopy(self.candidate)
|
|
if mutation == "omitted_target":
|
|
candidate["sequences"][0]["frames"].pop()
|
|
elif mutation == "foreign_target":
|
|
candidate["sequences"][0]["base_asset"] += 1
|
|
elif mutation == "changed_source":
|
|
candidate["sequences"][0]["frames"][0]["source_sha256"] = "0" * 64
|
|
else:
|
|
candidate["job_id"] = "another-job"
|
|
self.write(self.candidate_path, candidate)
|
|
ledger = self.build()
|
|
self.assertEqual(ledger["family_reviews"], {})
|
|
self.assertNotEqual(ledger["assets"]["11000"]["status"], "reviewed")
|
|
job = json.loads((self.directory / "job.json").read_text())
|
|
self.assertIn("candidate_error", job)
|
|
|
|
def test_metadata_only_candidate_change_retains_exact_frame_map_approval(self):
|
|
self.candidate["review_notes"] = "new artist notes; renderer bindings unchanged"
|
|
self.write(self.candidate_path, self.candidate)
|
|
ledger = self.build()
|
|
self.assertEqual(ledger["assets"]["11000"]["status"], "reviewed")
|
|
self.assertNotEqual(ledger["assets"]["11000"]["candidate"]["sha256"], self.review["candidate_sha256"])
|
|
self.assertEqual(ledger["family_reviews"][self.job_id]["candidate_sha256"], self.review["candidate_sha256"])
|
|
self.assertEqual(ledger["family_reviews"][self.job_id]["frame_bindings_sha256"], self.review["frame_bindings_sha256"])
|
|
|
|
def test_missing_changed_receipt_and_conflicting_identities_fail_safely(self):
|
|
legacy = self.build(legacy=True)
|
|
normalized = queue.normalize_ledger(legacy, self.hd)
|
|
changed = copy.deepcopy(legacy)
|
|
changed["assets"]["11001"]["candidate"]["sha256"] = "0" * 64
|
|
with self.assertRaisesRegex(ValueError, "different candidate identities"):
|
|
queue.normalize_ledger(changed, self.hd)
|
|
changed = copy.deepcopy(legacy)
|
|
changed["assets"]["11001"]["coordinator_review"]["status"] = "hold"
|
|
with self.assertRaisesRegex(ValueError, "external review differ"):
|
|
queue.normalize_ledger(changed, self.hd)
|
|
changed = copy.deepcopy(normalized)
|
|
changed["assets"]["11000"]["coordinator_review_ref"] = "another-job"
|
|
with self.assertRaisesRegex(ValueError, "differs from its job"):
|
|
queue.normalize_ledger(changed, self.hd)
|
|
self.write(self.review_path, {**self.review, "status": "hold"})
|
|
with self.assertRaisesRegex(ValueError, "receipt changed"):
|
|
queue.coordinator_review_for_asset(normalized, "11000", self.hd)
|
|
self.review_path.unlink()
|
|
with self.assertRaises(FileNotFoundError):
|
|
queue.normalize_ledger(legacy, self.hd)
|
|
with self.assertRaises(FileNotFoundError):
|
|
queue.coordinator_review_for_asset(normalized, "11000", self.hd)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|