raptorreplica
6a3c465da7
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.
331 lines
21 KiB
Python
331 lines
21 KiB
Python
"""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()
|