Remaster 23 classic terrain and structure pictures

Add reviewed native terrain artwork, preserve original detail where required, and validate versioned carried Terrain identities. Bundle the 5407-frame pack for Linux and Windows and record current validation results.

tests/test_hd_carried_identity.py

@@ -0,0 +1,330 @@
+"""Portable synthetic identity tests; no engine, image generation or registration."""
+import copy
+import hashlib
+import json
+import os
+from pathlib import Path
+import struct
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
+import hd_register_reviewed as reg
+
+
+def sha(data):
+    return hashlib.sha256(data).hexdigest()
+
+
+class CarriedFixture:
+    def __init__(self, directory, legacy=False):
+        self.base = Path(directory)
+        self.root = self.base / "data/hd"
+        self.project = self.base / "project"
+        self.root.mkdir(parents=True)
+        (self.project / "src").mkdir(parents=True)
+        (self.project / "target").mkdir()
+        for name in ["assets.rs", "hd.rs"]:
+            p = self.project / "src" / name
+            p.write_text("// synthetic fixture, never compiled\n")
+            os.utime(p, ns=(1000000000, 1000000000))
+        self.library = self.project / "target/test.rlib"
+        self.library.write_bytes(b"synthetic QA identity, not an engine library")
+        self.legacy = legacy
+        self.ids = [f"{x:05x}" for x in range(0x2059d, 0x205ad)] if legacy else [f"{x:05x}" for x in range(0x17400, 0x17406)]
+        self.selected = ["205a3", "205a4"] if legacy else self.ids[:2]
+        self.carried = [a for a in self.ids if a not in self.selected] if legacy else self.ids[2:4]
+        self.fallback = [] if legacy else self.ids[4:]
+        self.job_id = f"terrain-{self.ids[0]}-{self.ids[-1]}"
+        self.payload = struct.pack("<5i", 1, 0, 0, 32, 32) + bytes([31]) * 1024
+        palette = bytes([1]) * 768
+        self.targets, self.frames = {}, {}
+        lines = ["10000 0 misc/STARTG1TILES", "20000 0 misc/ENDG1TILES",
+                 "20000 0 misc/STARTG2TILES", "30000 0 misc/ENDG2TILES"]
+        # One non-overlapping actual Terrain marker pair suffices per fixture.
+        lines = lines[2:] if legacy else lines[:2]
+        (self.base / "data/misc").mkdir()
+        (self.base / "data/misc/PALETTE_DAT").write_bytes(palette)
+        (self.base / "data/sprites").mkdir()
+        for asset in self.ids:
+            rel = f"sprites/tile-{asset}"
+            (self.base / "data" / rel).write_bytes(self.payload)
+            lines.append(f"{asset} {len(self.payload)} {rel}")
+            self.targets[asset] = dict(asset=asset, source_path=rel, source_sha256=sha(self.payload),
+                    logical_canvas=[32, 32], origin=[0, 0], source_palette="PALETTE_DAT",
+                    source_palette_sha256=sha(palette))
+            raw = f"raw-{asset}.bin"
+            (self.root / raw).write_bytes(f"synthetic untouched raw {asset}".encode())
+            self.frames[asset] = dict(frame=0, file=raw, logical_width=32, logical_height=32,
+                    origin=[0, 0], source_sha256=sha(self.payload))
+        (self.base / "data/INDEX").write_text("\n".join(lines) + "\n")
+        if not legacy:
+            self.frames[self.carried[0]].update(crop=None, preserve_reserved_coverage=False,
+                                               original_detail_cells=[[0, 0]])
+            self.frames[self.carried[1]].update(native_nearest_cues=[])
+        self.job = dict(job_id=self.job_id, category="Terrain", kind="contextual_terrain_shard",
+                        targets=list(self.targets.values()))
+        self.active = {a: copy.deepcopy(self.frames[a]) for a in self.carried}
+        self.manifest = self.pack(self.active)
+        groups = [self.carried] if legacy else [[self.carried[0]], [self.carried[1]]]
+        self.additions = []
+        self.accepted = []
+        for index, group in enumerate(groups):
+            bindings = {a: self.active[a] for a in group}
+            qa = self.qa(bindings)
+            path = f"accepted-{index}.json"
+            self.write(path, qa)
+            self.accepted.append(qa)
+            self.additions.append(dict(job_id=self.job_id if legacy else f"different-prior-job-{index}",
+                    assets=group, frames=len(group), runtime_qa=path,
+                    frame_bindings_sha256=reg.binding_digest(bindings)))
+        self.provenance = dict(reviewed_sequence_additions=self.additions)
+        self.write("hd.json", self.manifest)
+        self.write("provenance.json", self.provenance)
+        identity = dict(baseline=self.receipt("hd.json"), provenance=self.receipt("provenance.json"),
+                        frames={}, frame_bindings_sha256=reg.binding_digest(self.active))
+        for index, group in enumerate(groups):
+            for asset in group:
+                identity["frames"][asset] = dict(frame_binding=copy.deepcopy(self.active[asset]),
+                        source_sha256=sha(self.payload), raw_sha256=sha((self.root / self.active[asset]["file"]).read_bytes()))
+                if not legacy:
+                    identity["frames"][asset]["registration_record_index"] = index
+        if legacy:
+            identity.update(registration_record_index=0, accepted_runtime_qa=self.receipt("accepted-0.json"))
+        else:
+            identity.update(contract_version=reg.CARRIED_TERRAIN_V2, registration_records=[dict(
+                    registration_record_index=i, registration_record_sha256=reg.binding_digest(r),
+                    assets=r["assets"], frame_bindings_sha256=r["frame_bindings_sha256"],
+                    accepted_runtime_qa=self.receipt(r["runtime_qa"])) for i, r in enumerate(self.additions)])
+        self.candidate = dict(job_id=self.job_id, category="Terrain", kind="contextual_terrain_shard",
+                complete_family=False, complete_standalone_selection=True, full_inventory=self.ids,
+                selected_assets=self.selected, original_fallback_assets=self.fallback,
+                carried_registered_assets=self.carried, carried_registered_identity=identity,
+                sequences=self.pack({a:self.frames[a] for a in self.selected})["sequences"])
+        self.rows = reg.proposal_frames(self.candidate)
+        self.review = {"standalone_selection": dict(reg.standalone_selection(self.candidate, self.job, self.rows),
+                    carried_registered_identity=copy.deepcopy(identity))}
+
+    @staticmethod
+    def pack(rows):
+        return dict(format_version=1, sequences=[dict(name=f"tile-{a}", base_asset=int(a,16),
+                    frame_count=1, frames=[copy.deepcopy(f)]) for a,f in rows.items()])
+
+    def write(self, name, value):
+        p=self.root/name;p.parent.mkdir(parents=True,exist_ok=True);p.write_text(json.dumps(value)+"\n")
+
+    def receipt(self, name):
+        return dict(path=name, sha256=sha((self.root/name).read_bytes()))
+
+    def qa(self, rows):
+        return dict(frame_bindings_sha256=reg.binding_digest(rows), diagnostics=[], frames=[dict(
+                    asset=a, source_sha256=f["source_sha256"], logical_canvas=[32,32],
+                    source_mask=dict(palette_indices_240_to_255_counts=[0]*16),
+                    hd_masks=[dict(scale=s,mask=dict(palette_indices_240_to_255_counts=[0]*16)) for s in [1,2,3,4]])
+                    for a,f in rows.items()], artwork_sha256={f["file"]:sha((self.root/f["file"]).read_bytes()) for f in rows.values()})
+
+    def sync_identity(self):
+        self.review["standalone_selection"]["carried_registered_identity"] = copy.deepcopy(self.candidate["carried_registered_identity"])
+
+    def validate(self):
+        return reg.validate_carried_registered_identity(self.root,self.candidate,self.job,self.rows,self.review)
+
+    def approved_setup(self):
+        directory=f"jobs/{self.job_id}"
+        self.write(f"{directory}/candidate.json",self.candidate)
+        qa=self.qa(self.rows)
+        qa.update(candidate_sha256=self.receipt(f"{directory}/candidate.json")["sha256"],
+                  source_newer_than_library=False,any_game_source_newer_than_library=False,
+                  library="target/test.rlib",library_sha256=sha(self.library.read_bytes()),
+                  library_modified_ns=self.library.stat().st_mtime_ns)
+        self.write(f"{directory}/runtime-qa.json",qa)
+        approvals={}
+        for asset,frame in self.rows.items():
+            peer=dict(reviewer="synthetic-different-peer",author="synthetic-artist",records=[dict(
+                asset=asset,standalone_verdict="QUALIFIED_APPROVE",selected_frame_binding=frame,
+                source_payload_sha256=frame["source_sha256"],full_untouched_raw_sha256=qa["artwork_sha256"][frame["file"]],actual_scales=[1,2,3,4])])
+            path=f"peer-{asset}.json";self.write(path,peer)
+            approvals[asset]=dict(status="approved",frame_bindings_sha256=reg.binding_digest({asset:frame}),
+                source_sha256=frame["source_sha256"],raw_sha256=qa["artwork_sha256"][frame["file"]],independent_review=self.receipt(path),record_index=0)
+        self.write("context-candidate.json",self.pack(self.rows))
+        future=copy.deepcopy(self.manifest);future["sequences"].extend(self.pack(self.rows)["sequences"])
+        self.write("future.json",future);self.write("changed-proof.json",dict(synthetic=True))
+        authority=dict(candidate=self.receipt("context-candidate.json"),selected_assets=list(self.rows),
+            frame_bindings_sha256=reg.binding_digest(self.rows),technical_verdict="QUALIFIED_CONTEXT_INTEGRATION",
+            baseline=self.receipt("hd.json"),future_projection=self.receipt("future.json"))
+        closure=dict(status="CLOSED_QUALIFIED_CONTEXT",candidate_sha256=authority["candidate"]["sha256"],
+            frame_bindings_sha256=authority["frame_bindings_sha256"],baseline_sha256=authority["baseline"]["sha256"],
+            future_projection_sha256=authority["future_projection"]["sha256"],changed_case_proof=self.receipt("changed-proof.json"))
+        self.write("closure.json",closure);authority["changed_case_closure"]=self.receipt("closure.json")
+        self.write("context-peer.json",dict(reviewer="synthetic-context-peer",author="synthetic-producer",standalone_selection_context=authority))
+        self.review.update(status="approved",frame_bindings_sha256=reg.binding_digest(self.rows),
+            candidate_sha256=qa["candidate_sha256"],source_and_all_scale_visual_review="synthetic review fixture, not real approval",
+            independent_review="synthetic material/context fixture",runtime_qa=f"{directory}/runtime-qa.json")
+        self.review["standalone_selection"].update(material_approvals=approvals,contextual_approval=dict(
+            selected_assets=list(self.rows),selected_frame_bindings_sha256=reg.binding_digest(self.rows),receipt=self.receipt("context-peer.json")))
+        self.write(f"{directory}/coordinator-review.json",self.review);self.write(f"{directory}/job.json",self.job)
+
+
+class CarriedIdentityTests(unittest.TestCase):
+    def setUp(self):
+        self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
+        self.f=CarriedFixture(self.temp.name)
+        self.process_trap=patch('subprocess.run',side_effect=AssertionError('external process forbidden'))
+        self.process_trap.start();self.addCleanup(self.process_trap.stop)
+
+    def test_new_multi_record_partition_and_optional_policies(self):
+        before={p:p.read_bytes()for p in self.f.root.rglob('*')if p.is_file()}
+        self.f.validate()
+        self.assertEqual(before,{p:p.read_bytes()for p in self.f.root.rglob('*')if p.is_file()})
+
+    def test_legacy_master74_roundtrip(self):
+        with tempfile.TemporaryDirectory() as d:
+            f=CarriedFixture(d,legacy=True);before=copy.deepcopy(f.candidate)
+            f.validate();self.assertEqual(before,f.candidate)
+            self.assertNotIn('contract_version',f.candidate['carried_registered_identity'])
+
+    def test_legacy_full_approved_job_readonly_roundtrip(self):
+        with tempfile.TemporaryDirectory() as d:
+            f=CarriedFixture(d,legacy=True);f.approved_setup()
+            before={p:p.read_bytes()for p in f.root.rglob('*')if p.is_file()}
+            with patch.object(reg,'atomic_json',side_effect=AssertionError('writer forbidden')),patch.object(reg,'main',side_effect=AssertionError('main forbidden')),patch.object(reg,'validate_hd_frame',return_value=None) as decoder:
+                _,_,rows=reg.approved_job(f.root,f.job_id,f.project)
+            self.assertEqual(set(rows),set(f.selected));self.assertEqual(decoder.call_count,2)
+            self.assertEqual(before,{p:p.read_bytes()for p in f.root.rglob('*')if p.is_file()})
+
+    def test_partial_carried_subset_authenticates_whole_prior_record(self):
+        extra='17ff0';frame=copy.deepcopy(self.f.active[self.f.carried[0]])
+        frame['file']='unrelated-prior-raw.bin';(self.f.root/frame['file']).write_bytes(b'unchanged unrelated prior raw')
+        self.f.active[extra]=frame;self.f.manifest=self.f.pack(self.f.active)
+        record=self.f.additions[0];record['assets'].append(extra);record['frames']=2
+        bindings={a:self.f.active[a]for a in record['assets']};record['frame_bindings_sha256']=reg.binding_digest(bindings)
+        self.f.write('accepted-0.json',self.f.qa(bindings));self.f.write('hd.json',self.f.manifest);self.f.write('provenance.json',self.f.provenance)
+        identity=self.f.candidate['carried_registered_identity'];identity['baseline']=self.f.receipt('hd.json');identity['provenance']=self.f.receipt('provenance.json')
+        descriptor=identity['registration_records'][0];descriptor.update(registration_record_sha256=reg.binding_digest(record),assets=record['assets'],frame_bindings_sha256=record['frame_bindings_sha256'],accepted_runtime_qa=self.f.receipt('accepted-0.json'))
+        self.f.sync_identity();self.f.validate()
+        descriptor['assets']=[self.f.carried[0]];self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'asset set'):self.f.validate()
+
+    def test_unversioned_generic_scope_is_still_rejected(self):
+        self.f.candidate['carried_registered_identity'].pop('contract_version')
+        with self.assertRaisesRegex(ValueError,'master74'):reg.standalone_selection(self.f.candidate,self.f.job,self.f.rows)
+
+    def test_unknown_version_rejected(self):
+        self.f.candidate['carried_registered_identity']['contract_version']='future'
+        with self.assertRaisesRegex(ValueError,'unknown'):reg.standalone_selection(self.f.candidate,self.f.job,self.f.rows)
+
+    def test_optional_absent_is_not_null(self):
+        self.f.candidate['carried_registered_identity']['frames'][self.f.carried[0]]['frame_binding'].pop('crop')
+        self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'binding/policy'):self.f.validate()
+
+    def test_optional_false_is_not_absent(self):
+        self.f.candidate['carried_registered_identity']['frames'][self.f.carried[0]]['frame_binding'].pop('preserve_reserved_coverage')
+        self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'binding/policy'):self.f.validate()
+
+    def test_changed_raw_bytes(self):
+        (self.f.root/self.f.active[self.f.carried[0]]['file']).write_bytes(b'changed')
+        with self.assertRaisesRegex(ValueError,'raw artwork'):self.f.validate()
+
+    def test_forged_registration_record_hash(self):
+        self.f.candidate['carried_registered_identity']['registration_records'][0]['registration_record_sha256']='0'*64
+        self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'record changed'):self.f.validate()
+
+    def test_wrong_registration_assignment(self):
+        self.f.candidate['carried_registered_identity']['frames'][self.f.carried[0]]['registration_record_index']=1
+        self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'foreign registration'):self.f.validate()
+
+    def test_duplicate_record_index_bool_and_wrong_asset_set(self):
+        for field,value in [('registration_record_index',True),('assets',[self.f.carried[1]])]:
+            with self.subTest(field=field):
+                old=copy.deepcopy(self.f.candidate['carried_registered_identity'])
+                self.f.candidate['carried_registered_identity']['registration_records'][0][field]=value;self.f.sync_identity()
+                with self.assertRaises(ValueError):self.f.validate()
+                self.f.candidate['carried_registered_identity']=old;self.f.sync_identity()
+
+    def test_bad_partition_overlap_or_omission(self):
+        for fallback in [[self.f.carried[0]],self.f.fallback[:-1]]:
+            with self.subTest(fallback=fallback):
+                self.f.candidate['original_fallback_assets']=fallback
+                with self.assertRaisesRegex(ValueError,'partition'):self.f.validate()
+
+    def test_selected_already_registered(self):
+        self.f.manifest['sequences']+=self.f.pack({self.f.selected[0]:self.f.frames[self.f.selected[0]]})['sequences']
+        self.f.write('hd.json',self.f.manifest)
+        self.f.candidate['carried_registered_identity']['baseline']=self.f.receipt('hd.json');self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'partition'):self.f.validate()
+
+    def test_fallback_already_registered(self):
+        self.f.manifest['sequences']+=self.f.pack({self.f.fallback[0]:self.f.frames[self.f.fallback[0]]})['sequences']
+        self.f.write('hd.json',self.f.manifest)
+        self.f.candidate['carried_registered_identity']['baseline']=self.f.receipt('hd.json');self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'partition'):self.f.validate()
+
+    def test_carried_animation_or_partial_sequence(self):
+        self.f.manifest['sequences'][0]['frame_count']=2
+        self.f.write('hd.json',self.f.manifest)
+        self.f.candidate['carried_registered_identity']['baseline']=self.f.receipt('hd.json');self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'animation'):self.f.validate()
+
+    def test_source_payload_and_palette_mutations(self):
+        for path in [self.f.base/'data'/self.f.targets[self.f.carried[0]]['source_path'],self.f.base/'data/misc/PALETTE_DAT']:
+            with self.subTest(path=str(path)):
+                old=path.read_bytes();path.write_bytes(old+b'x')
+                with self.assertRaisesRegex(ValueError,'source differs|palette changed'):self.f.validate()
+                path.write_bytes(old)
+
+    def test_false_terrain_source_inventory(self):
+        index=self.f.base/'data/INDEX';index.write_text(index.read_text().replace('20000 0 misc/ENDG1TILES','173ff 0 misc/ENDG1TILES'))
+        with self.assertRaisesRegex(ValueError,'not catalog Terrain'):self.f.validate()
+
+    def test_source_origin_and_native_canvas_are_literal(self):
+        for key,value in [('origin',[False,0]),('logical_canvas',[31,32])]:
+            with self.subTest(key=key):
+                target=next(t for t in self.f.job['targets']if t['asset']==self.f.carried[0]);old=target[key];target[key]=value
+                with self.assertRaises(ValueError):self.f.validate()
+                target[key]=old
+
+    def test_selected_animation_is_rejected(self):
+        self.f.candidate['sequences'][0]['frame_count']=2
+        with self.assertRaisesRegex(ValueError,'animation'):self.f.validate()
+
+    def test_independent_review_cannot_omit_or_change_identity(self):
+        self.f.review['standalone_selection']['carried_registered_identity']['frame_bindings_sha256']='0'*64
+        with self.assertRaisesRegex(ValueError,'independent review'):self.f.validate()
+
+    def test_receipt_hash_and_current_pair_are_mandatory(self):
+        self.f.candidate['carried_registered_identity']['provenance']['path']='other.json';self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'canonical pair'):self.f.validate()
+
+    def test_accepted_allscale_qa_and_raw_mapping_are_not_invented(self):
+        qa=self.f.accepted[0];qa['frames'][0]['hd_masks'][3]['scale']=3
+        self.f.write('accepted-0.json',qa)
+        self.f.candidate['carried_registered_identity']['registration_records'][0]['accepted_runtime_qa']=self.f.receipt('accepted-0.json');self.f.sync_identity()
+        with self.assertRaisesRegex(ValueError,'Source/scales'):self.f.validate()
+
+    def test_complete_approved_job_reaches_all_existing_selected_gates(self):
+        self.f.approved_setup();seen=[]
+        with patch.object(reg,'atomic_json',side_effect=AssertionError('writer forbidden')),patch.object(reg,'main',side_effect=AssertionError('main forbidden')),patch.object(reg,'validate_hd_frame',side_effect=lambda root,frame,payload,picture:seen.append((payload,picture.size))):
+            result=reg.approved_job(self.f.root,self.f.job_id,self.f.project)
+        self.assertEqual(set(result[2]),set(self.f.selected));self.assertEqual(seen,[(self.f.payload,(32,32))]*2)
+        self.assertEqual(set(reg.proposal_frames(json.loads((self.f.root/'hd.json').read_text()))),set(self.f.carried))
+
+    def test_selected_freshness_and_material_context_remain_required(self):
+        self.f.approved_setup();directory=f'jobs/{self.f.job_id}';qa=json.loads((self.f.root/f'{directory}/runtime-qa.json').read_text())
+        qa['any_game_source_newer_than_library']=True;self.f.write(f'{directory}/runtime-qa.json',qa)
+        with self.assertRaisesRegex(ValueError,'freshness'):reg.approved_job(self.f.root,self.f.job_id,self.f.project)
+        qa['any_game_source_newer_than_library']=False;self.f.write(f'{directory}/runtime-qa.json',qa)
+        review=copy.deepcopy(self.f.review);review['standalone_selection']['material_approvals']={};self.f.write(f'{directory}/coordinator-review.json',review)
+        with patch.object(reg,'validate_hd_frame',return_value=None),self.assertRaisesRegex(ValueError,'material map'):
+            reg.approved_job(self.f.root,self.f.job_id,self.f.project)
+        review=copy.deepcopy(self.f.review);review['standalone_selection'].pop('contextual_approval');self.f.write(f'{directory}/coordinator-review.json',review)
+        with patch.object(reg,'validate_hd_frame',return_value=None),self.assertRaises(ValueError):reg.approved_job(self.f.root,self.f.job_id,self.f.project)
+
+
+if __name__=='__main__':
+    unittest.main()