Add 23 reviewed infantry and industrial terrain HD pictures
This commit is contained in:
raptorreplica committed
1 parent
ff4ebf35eb
commit
83a7c8cb54
260 files changed
+573128
-69
No files matched your search
@@ -0,0 +1,374 @@
|
||||
"""Intrinsic admission for complete, currently dormant Original enemy animations.
|
||||
|
||||
This is a Source/receipt validator, not a renderer or a secret-mode implementation.
|
||||
Ordinary retained-art and contextual Terrain approval paths do not use it.
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path, PurePosixPath
|
||||
import re
|
||||
import stat
|
||||
import struct
|
||||
from types import SimpleNamespace
|
||||
|
||||
from hd_art_audit import checked_palette, validate_hd_frame
|
||||
from hd_art_source import decode
|
||||
from hd_frame_policy import _indices
|
||||
from hd_candidate_review import binding_digest, proposal_frames
|
||||
|
||||
DORMANT_ENEMY_SCOPE = "catalog_dormant_original_enemy_animation"
|
||||
DORMANT_ENEMY_TRACE = "original-dormant-enemy-animation-source-v1"
|
||||
DORMANT_ENEMY_MATERIAL = "original-dormant-enemy-animation-material-v1"
|
||||
MAX_PHASES = 96
|
||||
_MAX_FILE_BYTES = 64 * 1024 * 1024
|
||||
# These exact current Source implementations preserve raw MAP difficulty and
|
||||
# enable only normal codes3/4/5. A gameplay change requires a new reviewed gate.
|
||||
SUPPORTED_GATE_SOURCES = {
|
||||
"src/assets.rs": "44515745330b47a6a99761c449225c67cbee95c7b1f9d653d919ed09032b1beb",
|
||||
"src/content.rs": "80021611102f22b3152164e88a3ee7ecc8c16e9b3a4563551a047c089f2ba7d2",
|
||||
"src/simulation.rs": "2537120f280a1cfcfcc73a42a4103a4435ba8aeb98e69c87fde3b885b97a9925",
|
||||
}
|
||||
|
||||
|
||||
def _fail(message):
|
||||
raise ValueError("dormant enemy animation: " + message)
|
||||
|
||||
|
||||
def _read(root, record, route=None, source=False):
|
||||
if not isinstance(record, dict) or not isinstance(record.get("path"), str):
|
||||
_fail("missing portable hash-bound file")
|
||||
text = record["path"]
|
||||
relative = PurePosixPath(text)
|
||||
if (relative.is_absolute() or not relative.parts or
|
||||
any(p in {".", ".."} for p in relative.parts) or
|
||||
relative.as_posix() != text or "\\" in text or ":" in text or
|
||||
route is not None and text != route or
|
||||
re.fullmatch("[0-9a-f]{64}", str(record.get("sha256", ""))) is None):
|
||||
_fail("unsafe or mismatched portable file pin")
|
||||
path = root.resolve()
|
||||
for part in relative.parts:
|
||||
path = path / part
|
||||
if path.is_symlink():
|
||||
_fail("symlink in proof or Source route")
|
||||
before = path.lstat()
|
||||
if not stat.S_ISREG(before.st_mode) or before.st_size > _MAX_FILE_BYTES:
|
||||
_fail("file must be bounded and regular")
|
||||
flags = (os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) |
|
||||
getattr(os, "O_NONBLOCK", 0))
|
||||
identity = lambda s: (s.st_dev, s.st_ino, s.st_mode, s.st_size, s.st_mtime_ns)
|
||||
with os.fdopen(os.open(path, flags), "rb") as stream:
|
||||
opened = os.fstat(stream.fileno())
|
||||
if (not stat.S_ISREG(opened.st_mode) or opened.st_size > _MAX_FILE_BYTES or
|
||||
identity(before) != identity(opened)):
|
||||
_fail("opened file must remain bounded, regular and unchanged")
|
||||
payload = stream.read(_MAX_FILE_BYTES + 1)
|
||||
if len(payload) > _MAX_FILE_BYTES or stream.read(1):
|
||||
_fail("file exceeds bounded read or did not reach EOF")
|
||||
finished = os.fstat(stream.fileno())
|
||||
if (identity(before) != identity(opened) or identity(before) != identity(finished)
|
||||
or identity(before) != identity(path.lstat())
|
||||
or hashlib.sha256(payload).hexdigest() != record["sha256"]):
|
||||
_fail("file changed or hash differs")
|
||||
if source and (type(record.get("bytes")) is not int or record["bytes"] != len(payload)):
|
||||
_fail("Source byte count differs")
|
||||
return payload
|
||||
|
||||
|
||||
def _proof(root, record):
|
||||
payload = _read(root, record)
|
||||
if len(payload) > 8 * 1024 * 1024:
|
||||
_fail("proof JSON exceeds8MiB")
|
||||
value = json.loads(payload)
|
||||
if not isinstance(value, dict):
|
||||
_fail("proof must contain an object")
|
||||
return value
|
||||
|
||||
|
||||
def approved_dormant_enemy_animation(root, project, candidate, job, review, rows, qa):
|
||||
"""Recompute Original family/caller facts, then require explicit intrinsic authority."""
|
||||
declaration = review.get("retained_original_detail_review", {})
|
||||
sequences = candidate.get("sequences", [])
|
||||
if (candidate.get("approval_scope") != DORMANT_ENEMY_SCOPE or
|
||||
declaration.get("hybrid_review_scope") != DORMANT_ENEMY_SCOPE or
|
||||
job.get("category") != "Enemies and ground objects" or
|
||||
candidate.get("category") != job.get("category") or candidate.get("kind") != "complete_family" or
|
||||
job.get("kind") != "complete_family" or candidate.get("complete_family") is not True or
|
||||
candidate.get("complete_standalone_selection") or len(sequences) != 1):
|
||||
_fail("requires one complete Original enemy animation, never singleton slices")
|
||||
sequence = sequences[0]
|
||||
base, count = sequence.get("base_asset"), sequence.get("frame_count")
|
||||
if (type(base) is not int or type(count) is not int or not 2 <= count <= MAX_PHASES or not 0 <= base <= 0xfffff-count+1 or
|
||||
set(rows) != {f"{base+i:05x}" for i in range(count)} or
|
||||
len(sequence.get("frames", [])) != count or
|
||||
any(type(f.get("frame")) is not int for f in sequence["frames"]) or
|
||||
[f.get("frame") for f in sequence["frames"]] != list(range(count))):
|
||||
_fail("phase inventory must be complete, contiguous and ordered")
|
||||
if (declaration.get("hybrid_allscale_and_genuine_context_independent_review") is not False or
|
||||
declaration.get("catalog_dormant_no_genuine_runtime_context_claimed") is not True or
|
||||
declaration.get("catalog_only_independent_whole_material_and_allscale_review") is not True):
|
||||
_fail("intrinsic-only declaration cannot claim genuine gameplay context")
|
||||
assets = sorted(rows)
|
||||
if any(frame.get("palette_constraint") is not None for frame in rows.values()):
|
||||
_fail("palette constraints retain their ordinary genuine-context approval requirement")
|
||||
targets = {t["asset"]: t for t in job.get("targets", [])}
|
||||
if len(targets) != len(job.get("targets", [])) or set(targets) != set(rows):
|
||||
_fail("Original job inventory differs")
|
||||
trace_ref = declaration.get("dormant_source_trace")
|
||||
trace = _proof(root, trace_ref)
|
||||
if (trace.get("contract_version") != DORMANT_ENEMY_TRACE or
|
||||
trace.get("no_genuine_original_runtime_context_claimed") is not True):
|
||||
_fail("explicit dormant Source trace required; no fabricated context")
|
||||
index = _read(project, trace.get("catalogue"), "data/INDEX", True)
|
||||
entries, names = {}, {}
|
||||
for line in index.decode().splitlines():
|
||||
if not line.strip():
|
||||
continue
|
||||
parts = line.split()
|
||||
if (len(parts) != 3 or re.fullmatch("[0-9a-fA-F]{1,8}", parts[0]) is None or
|
||||
not parts[1].isdigit() or int(parts[0], 16) in entries):
|
||||
_fail("invalid or duplicate INDEX record")
|
||||
identity = int(parts[0], 16)
|
||||
relative = PurePosixPath(parts[2])
|
||||
if relative.is_absolute() or ".." in relative.parts or relative.as_posix() != parts[2] or "\\" in parts[2]:
|
||||
_fail("unsafe INDEX route")
|
||||
entries[identity] = (int(parts[1]), "data/" + parts[2])
|
||||
names.setdefault(relative.name, []).append(identity)
|
||||
|
||||
def named(name):
|
||||
ids = names.get(name, [])
|
||||
if len(ids) != 1:
|
||||
_fail("missing or ambiguous named Original Source")
|
||||
return ids[0], entries[ids[0]]
|
||||
|
||||
family = trace.get("family", {})
|
||||
name = family.get("name")
|
||||
if (not isinstance(name, str) or re.fullmatch("[A-Za-z0-9_]{1,15}", name) is None or
|
||||
named(name)[0] != base or family.get("assets") != assets or
|
||||
type(family.get("frame_count")) is not int or family["frame_count"] != count):
|
||||
_fail("named Original family differs")
|
||||
named_family = {identity for stem, ids in names.items()
|
||||
if re.fullmatch(re.escape(name) + r"(?:\.[0-9]+)?", stem)
|
||||
for identity in ids}
|
||||
if named_family != {base+i for i in range(count)}:
|
||||
_fail("Original named family has omitted or additional phases")
|
||||
phases = trace.get("phases", [])
|
||||
if not isinstance(phases, list) or [p.get("asset") for p in phases] != assets:
|
||||
_fail("complete phase Source records required")
|
||||
palette_id, (palette_size, palette_route) = named("PALETTE_DAT")
|
||||
palette = _read(project, trace.get("palette"), palette_route, True)
|
||||
if len(palette) != palette_size:
|
||||
_fail("catalogue palette size differs")
|
||||
colors = checked_palette(palette)
|
||||
source_reserved, source_visible = {}, {}
|
||||
for i, record in enumerate(phases):
|
||||
asset = assets[i]
|
||||
identity, (size, route) = named(name if i == 0 else f"{name}.{i}")
|
||||
target, frame = targets[asset], rows[asset]
|
||||
payload = _read(project, record.get("Source"), route, True)
|
||||
if len(payload) < 20 or len(payload) != size or identity != base+i:
|
||||
_fail("phase Source size or ID differs")
|
||||
header = list(struct.unpack_from("<5i", payload))
|
||||
claimed = record.get("header_le_i32")
|
||||
if (not isinstance(claimed, list) or any(type(v) is not int for v in claimed) or
|
||||
claimed != header or header[0] not in (0, 1) or
|
||||
target.get("source_path") != route.removeprefix("data/") or
|
||||
target.get("source_sha256") != record["Source"]["sha256"] or
|
||||
json.dumps(target.get("logical_canvas")) != json.dumps(header[3:]) or
|
||||
json.dumps(target.get("origin")) != "[0, 0]" or json.dumps(frame.get("origin")) != "[0, 0]" or
|
||||
target.get("source_palette") != "PALETTE_DAT" or
|
||||
target.get("source_palette_sha256") != trace["palette"]["sha256"]):
|
||||
_fail("complete phase header, canvas, palette or Source binding differs")
|
||||
validate_hd_frame(root, frame, payload, decode(SimpleNamespace(read_bytes=lambda: payload), colors))
|
||||
_, _, indices = _indices(payload)
|
||||
source_reserved[asset] = [indices.count(i) for i in range(240, 256)]
|
||||
source_visible[asset] = sum(i < 256 for i in indices)
|
||||
|
||||
libraries = trace.get("sprite_libraries", [])
|
||||
if not isinstance(libraries, list) or len(libraries) != 4:
|
||||
_fail("all four Original SPRITE libraries required")
|
||||
callers, record_total = [], 0
|
||||
for game, record in enumerate(libraries):
|
||||
_, (size, route) = named(f"SPRITE{game+1}_ITM")
|
||||
payload = _read(project, record, route, True)
|
||||
if not payload or len(payload) != size or len(payload) % 528:
|
||||
_fail("truncated Original SPRITE library")
|
||||
record_total += len(payload) // 528
|
||||
for ordinal in range(len(payload) // 528):
|
||||
raw = payload[ordinal*528:(ordinal+1)*528]
|
||||
stem = raw[:16].split(b"\0", 1)[0].decode()
|
||||
if stem not in names:
|
||||
continue
|
||||
start, _ = named(stem)
|
||||
frames = max(1, struct.unpack_from("<i", raw, 44)[0])
|
||||
if start < base+count and base < start+frames:
|
||||
if start != base or frames != count or stem != name or struct.unpack_from("<i", raw, 64)[0] != 0:
|
||||
_fail("SPRITE library contains a partial, aliased or boss family")
|
||||
callers.append(dict(game=game, ordinal=ordinal, name=stem, base_asset=f"{base:05x}",
|
||||
frame_count=frames, enemy_id=f"base:enemy_{game}_{ordinal}",
|
||||
record_sha256=hashlib.sha256(raw).hexdigest()))
|
||||
if not callers or trace.get("sprite_callers") != callers:
|
||||
_fail("exact complete-family SPRITE callers differ")
|
||||
|
||||
source_records = trace.get("current_rust_sources", [])
|
||||
current_paths = {p.relative_to(project).as_posix() for p in (project / "src").rglob("*.rs")}
|
||||
if (not isinstance(source_records, list) or len(source_records) != len(current_paths) or
|
||||
{r.get("path") for r in source_records} != current_paths or len(current_paths) > 256):
|
||||
_fail("complete current Rust Source inventory required")
|
||||
current_sources = {}
|
||||
direct = re.compile("|".join([re.escape(name)] + [re.escape(c["enemy_id"]) for c in callers] +
|
||||
[rf"\b(?:0x)?{asset}\b|\b{int(asset,16)}\b" for asset in assets]), re.I)
|
||||
for record in source_records:
|
||||
payload = _read(project, record, record["path"], True)
|
||||
text = payload.decode()
|
||||
if re.search(r"\b(?:secret|bday)\w*\b", text, re.I) or direct.search(text):
|
||||
_fail("current Rust has a secret activation or direct family caller requiring review")
|
||||
current_sources[record["path"]] = record["sha256"]
|
||||
if any(current_sources.get(path) != sha for path, sha in SUPPORTED_GATE_SOURCES.items()):
|
||||
_fail("current raw-decode, registry or normal-difficulty gate Source changed")
|
||||
measured_sources = qa.get("current_game_source_sha256", qa.get("actual_sourceG2_source18_sha256"))
|
||||
if (measured_sources != current_sources or qa.get("source_newer_than_library") is not False or
|
||||
qa.get("any_game_source_newer_than_library") is not False):
|
||||
_fail("actual current QA does not bind all current Rust Source bytes")
|
||||
library = trace.get("current_library", {})
|
||||
route = qa.get("library", "")
|
||||
if (not route.startswith("target/") or library.get("path") != route or
|
||||
library.get("sha256") != qa.get("library_sha256")):
|
||||
_fail("current compiled library differs")
|
||||
_read(project, library, route, True)
|
||||
modified = (project / route).stat().st_mtime_ns
|
||||
if (type(qa.get("library_modified_ns")) is not int or modified != qa["library_modified_ns"] or
|
||||
any((project / path).stat().st_mtime_ns > modified for path in current_paths)):
|
||||
_fail("current compiled library is older than a game Source")
|
||||
|
||||
expected_maps = [named(f"MAP{wave}G{sector}_MAP")[1][1]
|
||||
for sector in range(1, 4) for wave in range(1, 10)]
|
||||
maps = trace.get("maps", [])
|
||||
if not isinstance(maps, list) or len(maps) != 27 or [r.get("path") for r in maps] != expected_maps:
|
||||
_fail("all27 distinct ordered Original MAP records required")
|
||||
declarations, spawn_total = [], 0
|
||||
target_callers = {(c["game"], c["ordinal"]) for c in callers}
|
||||
for record in maps:
|
||||
route = record["path"]
|
||||
_, (size, _) = named(PurePosixPath(route).name)
|
||||
payload = _read(project, record, route, True)
|
||||
if len(payload) < 5412:
|
||||
_fail("truncated Original MAP")
|
||||
length, offset, number = struct.unpack_from("<3i", payload)
|
||||
if length != size or length != len(payload) or offset != 5412 or not 0 <= number < 65536 or 5412+24*number > length:
|
||||
_fail("invalid Original MAP header")
|
||||
for tile in range(1350):
|
||||
flat, game = struct.unpack_from("<hh", payload, 12+4*tile)
|
||||
if flat < 0 or not 0 <= game < 4:
|
||||
_fail("invalid Original MAP layout")
|
||||
spawn_total += number
|
||||
for ordinal in range(number):
|
||||
location = 5412+24*ordinal
|
||||
fields = list(struct.unpack_from("<6i", payload, location))
|
||||
link, lib, x, row, game, difficulty = fields
|
||||
if lib < 0 or not 0 <= game < 4:
|
||||
_fail("invalid Original MAP spawn reference")
|
||||
if (game, lib) not in target_callers:
|
||||
continue
|
||||
enabled = [difficulty == 3 or difficulty == 4 and pilot >= 2 or difficulty == 5 and pilot >= 3
|
||||
for pilot in range(4)]
|
||||
if any(enabled) or difficulty not in (0, 1, 2):
|
||||
_fail("family has an ordinary or unsupported Original MAP caller")
|
||||
declarations.append(dict(map=route, ordinal=ordinal, byte_offset=location, fields=fields,
|
||||
raw_bytes_hex=payload[location:location+24].hex(), enabled_by_pilot_difficulty=enabled))
|
||||
counts = dict(maps=27, sprite_libraries=4, sprite_records=record_total,
|
||||
family_library_callers=len(callers), spawn_records=spawn_total,
|
||||
family_map_declarations=len(declarations), ordinary_difficulty_callers=0)
|
||||
if (trace.get("map_declarations") != declarations or trace.get("counts") != counts or
|
||||
any(type(trace.get("counts", {}).get(k)) is not int for k in counts)):
|
||||
_fail("false or incomplete recomputed Original caller census")
|
||||
|
||||
digest = binding_digest(rows)
|
||||
cells = {a: f.get("original_detail_cells", []) for a, f in rows.items()}
|
||||
sources = {a: f["source_sha256"] for a, f in rows.items()}
|
||||
artwork = {f["file"]: qa["artwork_sha256"][f["file"]] for f in rows.values()}
|
||||
if candidate.get("artwork_derivation"):
|
||||
lineage = _proof(root, candidate["artwork_derivation"])
|
||||
raw_records = [p["raw_png"] for p in lineage["parents"]]
|
||||
else:
|
||||
raw_records = [dict(path=p, sha256=sha) for p, sha in artwork.items()]
|
||||
raw_hashes = sorted({record["sha256"] for record in raw_records})
|
||||
for record in raw_records:
|
||||
if not _read(root, record).startswith(b"\x89PNG\r\n\x1a\n"):
|
||||
_fail("genuine parent binding is not a PNG")
|
||||
qa_path = review.get("runtime_qa", "")
|
||||
qa_record = declaration.get("dormant_actual_allscale_QA", {})
|
||||
if qa_record.get("path") != qa_path:
|
||||
_fail("actual allscale QA reference differs")
|
||||
qa_bytes = _read(root, qa_record)
|
||||
qa_sha = hashlib.sha256(qa_bytes).hexdigest()
|
||||
if (json.loads(qa_bytes) != qa or type(qa.get("validated_frames")) is not int or
|
||||
qa["validated_frames"] != count or qa.get("diagnostics") != [] or
|
||||
qa.get("static_card_palette_mode") is not False):
|
||||
_fail("current complete-family runtime QA differs")
|
||||
for asset in assets:
|
||||
reports = [r for r in qa["frames"] if r["asset"] == asset]
|
||||
if len(reports) != 1 or [s.get("scale") for s in reports[0]["hd_masks"]] != [1, 2, 3, 4]:
|
||||
_fail("current QA omits or duplicates a frame/native scale")
|
||||
report = reports[0]
|
||||
if (report.get("source_palette") != "PALETTE_DAT" or
|
||||
type(report.get("source_mask", {}).get("visible_pixels")) is not int or
|
||||
report["source_mask"]["visible_pixels"] != source_visible[asset] or
|
||||
json.dumps(report.get("source_mask", {}).get("palette_indices_240_to_255_counts")) !=
|
||||
json.dumps(source_reserved[asset])):
|
||||
_fail("runtime Source palette/mask differs from actual Original bytes")
|
||||
for scale in report["hd_masks"]:
|
||||
k = scale.get("scale")
|
||||
mask = scale.get("mask", {})
|
||||
palette_counts = mask.get("palette_indices_240_to_255_counts")
|
||||
visible = mask.get("visible_pixels")
|
||||
if (type(k) is not int or type(visible) is not int or not 0 < visible <=
|
||||
rows[asset]["logical_width"] * rows[asset]["logical_height"] * k * k or
|
||||
not isinstance(palette_counts, list) or len(palette_counts) != 16 or
|
||||
any(type(n) is not int or n < 0 for n in palette_counts) or sum(palette_counts) > visible or
|
||||
any(n and not original for n, original in zip(palette_counts, source_reserved[asset]))):
|
||||
_fail("native masks or functional palette labels differ")
|
||||
peer_ref = declaration.get("dormant_independent_material")
|
||||
peer = _proof(root, peer_ref)
|
||||
artist, preparer, reviewer = peer.get("artist"), peer.get("preparer"), peer.get("reviewer")
|
||||
common = dict(scope=DORMANT_ENEMY_SCOPE, assets=assets, frame_bindings_sha256=digest,
|
||||
source_sha256_by_asset=sources, artwork_sha256=artwork, original_generated_raw_sha256=raw_hashes,
|
||||
retained_cells_by_asset=cells, native_scales=[1, 2, 3, 4],
|
||||
all_frames_whole_viewed=True, no_genuine_original_runtime_context_claimed=True,
|
||||
runtime_qa_sha256=qa_sha)
|
||||
if (peer.get("contract_version") != DORMANT_ENEMY_MATERIAL or peer.get("status") != "qualified" or
|
||||
any(json.dumps(peer.get(k), sort_keys=True) != json.dumps(v, sort_keys=True) for k, v in common.items()) or
|
||||
not all(isinstance(v, str) and v for v in (artist, preparer, reviewer)) or
|
||||
reviewer in (artist, preparer) or peer.get("different_from_artist_and_preparer") is not True or
|
||||
type(peer.get("qualified_frames")) is not int or peer["qualified_frames"] != count or
|
||||
type(peer.get("held_frames")) is not int or peer["held_frames"] != 0 or
|
||||
proposal_frames(_proof(root, peer.get("candidate"))) != rows):
|
||||
_fail("exact different whole-family intrinsic approval required")
|
||||
expected = [dict(asset=a, scale=k, status="qualified", retained_SourceNx_exact=True,
|
||||
retained_samples=len(cells[a])*k*k) for a in assets for k in range(1, 5)]
|
||||
if json.dumps(peer.get("native_decisions"), sort_keys=True) != json.dumps(expected, sort_keys=True):
|
||||
_fail("independent all-frame/native-scale/Original-retention decisions differ")
|
||||
if review.get("independent_review") not in (peer_ref, peer.get("actual_whole_material_review")):
|
||||
_fail("coordinator independent citation differs from whole-family authority")
|
||||
actual = _proof(root, peer.get("actual_whole_material_review"))
|
||||
actual_decisions = actual.get("decisions", [])
|
||||
actual_views = [r for r in actual.get("physical_views", []) if r.get("kind") == "native1to4"]
|
||||
if (not isinstance(actual_decisions, list) or len(actual_decisions) != count or
|
||||
len(actual_views) != count or actual.get("reviewer") != reviewer or
|
||||
not str(actual.get("status", "")).startswith("QUALIFIED_") or
|
||||
any(re.fullmatch(r"NOT|UNQUALIFIED|HOLD[0-9]*|FAIL(?:ED)?|PENDING", word)
|
||||
for word in re.split(r"[^A-Z0-9]+", str(actual.get("status", "")).upper())) or
|
||||
{r.get("asset") for r in actual.get("decisions", [])
|
||||
if r.get("intrinsic") in ("QUALIFIED", "QUALIFIED_WITH_LIMITS")} != set(assets) or
|
||||
{r.get("asset") for r in actual.get("physical_views", [])
|
||||
if r.get("kind") == "native1to4" and r.get("whole_frame_personally_viewed") is True} != set(assets)):
|
||||
_fail("independent actual whole-native material review differs")
|
||||
approval = _proof(root, declaration.get("dormant_root_approval"))
|
||||
if (approval.get("status") != "approved" or approval.get("authority_role") != "project_root" or
|
||||
not isinstance(approval.get("reviewer"), str) or not approval["reviewer"] or
|
||||
approval["reviewer"] in (artist, preparer, reviewer) or
|
||||
any(json.dumps(approval.get(k), sort_keys=True) != json.dumps(v, sort_keys=True) for k, v in common.items()) or
|
||||
approval.get("source_trace_sha256") != trace_ref["sha256"] or
|
||||
approval.get("independent_material_sha256") != peer_ref["sha256"]):
|
||||
_fail("explicit Root whole-family intrinsic/no-context approval required")
|
||||
return counts
|
||||
@@ -0,0 +1,207 @@
|
||||
"""Reproduce PERSON01's exact Source-alias crop of generated PLATOON1 detail.
|
||||
|
||||
All6 are required. Parent rendering uses unchanged authenticated code in memory;
|
||||
child output is its fixed pixel-exact crop. No Source painting, resampling, gains,
|
||||
new donors or new image generation. Reproduction alone is not art approval.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
import argparse
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image, __version__ as PILLOW_VERSION
|
||||
|
||||
METHOD = 'derived-PERSON01-complete6-exact-Source-crop-of-PLATOON1-generated-top-bodies-v1'
|
||||
PARENT_SHA = '5368774adb98af834df4fddb7ebe24310e6b35be110adf56b9961a6d36edd866'
|
||||
DECODER_SHA = '66184a4758027dadbadb5e20fe90b2add2b9fc218b7810304b1a1ea2de74f58e'
|
||||
ASSETS = [f'{0x200f8+i:05x}' for i in range(6)]
|
||||
|
||||
|
||||
def digest(payload):
|
||||
return hashlib.sha256(payload).hexdigest()
|
||||
|
||||
|
||||
def exact(value, fields, label):
|
||||
if type(value) is not dict or set(value) != set(fields):
|
||||
raise ValueError(f'{label}: exact fields required')
|
||||
|
||||
|
||||
def bootstrap_regular_bytes(path):
|
||||
"""Bounded nonblocking descriptor read before the verified parent can help."""
|
||||
path = Path(path)
|
||||
if path.is_symlink() or any(p.is_symlink() for p in path.parents):
|
||||
raise ValueError('symlink dependency')
|
||||
before = path.lstat()
|
||||
if not stat.S_ISREG(before.st_mode) or before.st_size > 1_048_576:
|
||||
raise ValueError('regular bounded parent dependency required')
|
||||
flags = os.O_RDONLY | getattr(os, 'O_NOFOLLOW', 0) | getattr(os, 'O_NONBLOCK', 0)
|
||||
with os.fdopen(os.open(path, flags), 'rb') as stream:
|
||||
opened = os.fstat(stream.fileno())
|
||||
if not stat.S_ISREG(opened.st_mode) or opened.st_size > 1_048_576:
|
||||
raise ValueError('regular bounded parent dependency required')
|
||||
payload = stream.read(1_048_577)
|
||||
after = os.fstat(stream.fileno())
|
||||
def identity(value):
|
||||
return value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_mode
|
||||
if (len(payload) > 1_048_576 or identity(before) != identity(opened)
|
||||
or identity(opened) != identity(after) or identity(path.lstat()) != identity(before)):
|
||||
raise ValueError('parent dependency changed during bounded read')
|
||||
return payload
|
||||
|
||||
|
||||
def dependencies(project):
|
||||
"""Use the checked parent reader, then execute only verified buffers."""
|
||||
parent_path = Path(project)/'tools/hd_infantry_cleanup.py'
|
||||
data = bootstrap_regular_bytes(parent_path)
|
||||
if digest(data) != PARENT_SHA:
|
||||
raise ValueError('parent helper hash mismatch')
|
||||
scope = {'__name__': '_verified_infantry_buffer', '__file__': str(parent_path)}
|
||||
exec(compile(data, '<verified-infantry-buffer>', 'exec'), scope)
|
||||
decoder_bytes = scope['checked']({'path': 'tools/hd_frame_policy.py', 'sha256': DECODER_SHA}, project)
|
||||
decoder = {'__name__': '_verified_decoder_buffer', '__file__': '<verified-decoder-buffer>'}
|
||||
exec(compile(decoder_bytes, '<verified-decoder-buffer>', 'exec'), decoder)
|
||||
return scope, decoder['_indices']
|
||||
|
||||
|
||||
def png_bytes(rgba):
|
||||
target = io.BytesIO()
|
||||
Image.fromarray(rgba).save(target, format='PNG', compress_level=9)
|
||||
return target.getvalue()
|
||||
|
||||
|
||||
def run_recipe(recipe_path, output, project_root=None):
|
||||
recipe_path, output = Path(recipe_path).absolute(), Path(output).absolute()
|
||||
if output.exists() or output.is_symlink() or any(p.is_symlink() for p in output.parents):
|
||||
raise ValueError('fresh nonsymlink output required')
|
||||
project = Path(project_root) if project_root else Path(__file__).resolve().parents[1]
|
||||
parent, decode = dependencies(project)
|
||||
checked, regular, keys = parent['checked'], parent['regular_bytes'], parent['keys']
|
||||
rb = regular(recipe_path)
|
||||
own_sha = digest(regular(Path(__file__)))
|
||||
recipe = json.loads(rb)
|
||||
exact(recipe, {'format_version', 'method', 'parent_recipe', 'parent_execution_receipt', 'frames'}, 'recipe')
|
||||
if type(recipe['format_version']) is not int or recipe['format_version'] != 1 or recipe['method'] != METHOD:
|
||||
raise ValueError('fixed PERSON01 recipe version/method required')
|
||||
frames = recipe['frames']
|
||||
if type(frames) is not list or len(frames) != 6 or [f.get('asset') for f in frames if type(f) is dict] != ASSETS:
|
||||
raise ValueError('complete ordered PERSON01 family required')
|
||||
parent_payload = checked(recipe['parent_recipe'], recipe_path.parent)
|
||||
parent_recipe = json.loads(parent_payload)
|
||||
exact(parent_recipe, {'format_version', 'method', 'raw_png', 'palette', 'frames'}, 'parent recipe')
|
||||
if type(parent_recipe['format_version']) is not int or parent_recipe['format_version'] != 1 or parent_recipe['method'] != parent['METHOD']:
|
||||
raise ValueError('parent method/version mismatch')
|
||||
pf = parent_recipe['frames']
|
||||
if type(pf) is not list or len(pf) != 6 or [f.get('asset') for f in pf if type(f) is dict] != parent['ASSETS']:
|
||||
raise ValueError('complete ordered parent family required')
|
||||
parent_base = (recipe_path.parent/recipe['parent_recipe']['path']).parent
|
||||
raw_payload = checked(parent_recipe['raw_png'], parent_base)
|
||||
with Image.open(io.BytesIO(raw_payload)) as opened:
|
||||
if opened.format != 'PNG' or opened.size != (1536, 1024) or opened.mode != 'RGBA':
|
||||
raise ValueError('exact parent atlas layout required')
|
||||
raw = np.asarray(opened).copy()
|
||||
palette_payload = checked(parent_recipe['palette'], parent_base)
|
||||
if len(palette_payload) != 768 or max(palette_payload) > 63:
|
||||
raise ValueError('genuine VGA palette required')
|
||||
p6 = np.frombuffer(palette_payload, np.uint8).reshape(256, 3).astype(np.uint16)
|
||||
palette = ((p6 << 2) | (p6 >> 4)).astype(np.uint8)
|
||||
receipt = json.loads(checked(recipe['parent_execution_receipt'], recipe_path.parent))
|
||||
if type(receipt) is not dict or type(receipt.get('execution')) is not dict:
|
||||
raise ValueError('parent execution receipt required')
|
||||
execution = receipt['execution']
|
||||
exact(execution, {'status', 'method', 'recipe_sha256', 'tool_sha256', 'source_sha256',
|
||||
'parent_raw_sha256', 'output_sha256'}, 'parent execution')
|
||||
if (execution['status'] != 'success' or execution['method'] != parent['METHOD']
|
||||
or execution['tool_sha256'] != PARENT_SHA
|
||||
or execution['recipe_sha256'] != digest(parent_payload)
|
||||
or execution['parent_raw_sha256'] != [digest(raw_payload)]
|
||||
or type(execution['output_sha256']) is not dict
|
||||
or set(execution['output_sha256']) != set(parent['ASSETS'])
|
||||
or any(not parent['hash_string'](v) for v in execution['output_sha256'].values())):
|
||||
raise ValueError('parent execution provenance mismatch')
|
||||
child_sources, parent_sources = [], []
|
||||
for i, (frame, parent_frame) in enumerate(zip(frames, pf)):
|
||||
exact(frame, {'asset', 'parent_asset', 'source', 'functional_original_cells'}, 'child frame')
|
||||
keys(parent_frame, {'asset', 'source', 'components', 'functional_original_cells'}, 'parent frame')
|
||||
if (frame['parent_asset'] != parent['ASSETS'][i]
|
||||
or type(parent_frame['components']) is not list or len(parent_frame['components']) != 6
|
||||
or frame['functional_original_cells'] != [[3, 3]]
|
||||
or any(type(v) is not int for v in frame['functional_original_cells'][0])):
|
||||
raise ValueError('fixed corresponding parent phase and one child face required')
|
||||
width, height, pixels = decode(checked(parent_frame['source'], parent_base))
|
||||
if (width, height) != (24, 24):
|
||||
raise ValueError('parent24x24 required')
|
||||
ps = np.array(pixels, np.uint16).reshape(24, 24)
|
||||
width, height, pixels = decode(checked(frame['source'], recipe_path.parent))
|
||||
if (width, height) != (8, 8):
|
||||
raise ValueError('child8x8 required')
|
||||
cs = np.array(pixels, np.uint16).reshape(8, 8)
|
||||
if (not np.array_equal(cs, ps[0:8, 5:13]) or cs[3, 3] >= 256
|
||||
or 4 > int((cs < 256).sum())):
|
||||
raise ValueError('entire64-index Source crop/face/quarter bound mismatch')
|
||||
if not np.array_equal(parent['scale2'](cs), parent['scale2'](ps)[0:16, 10:26]):
|
||||
raise ValueError('SourceScale2 crop mismatch')
|
||||
if not np.array_equal(parent['scale2'](parent['scale2'](cs)),
|
||||
parent['scale2'](parent['scale2'](ps))[0:32, 20:52]):
|
||||
raise ValueError('SourceScale4 crop mismatch')
|
||||
child_sources.append(cs)
|
||||
parent_sources.append(ps)
|
||||
if execution['source_sha256'] != {f['asset']: f['source']['sha256'] for f in pf}:
|
||||
raise ValueError('parent source provenance mismatch')
|
||||
payloads, reports = [], []
|
||||
for i, (frame, parent_frame, ps, cs) in enumerate(zip(frames, pf, parent_sources, child_sources)):
|
||||
rgba, material_report = parent['render_frame'](parent_frame, ps, palette,
|
||||
raw[i//3*512:(i//3+1)*512, i%3*512:(i%3+1)*512])
|
||||
reproduced_parent = png_bytes(rgba)
|
||||
if digest(reproduced_parent) != execution['output_sha256'][parent_frame['asset']]:
|
||||
raise ValueError('parent PNG not exactly reproduced')
|
||||
crop = rgba[0:32, 20:52].copy()
|
||||
child_scale4 = parent['scale2'](parent['scale2'](cs))
|
||||
if not np.array_equal(crop[:, :, 3], np.where(child_scale4 < 256, 255, 0)):
|
||||
raise ValueError('child alpha differs from own SourceScale4')
|
||||
child_png = png_bytes(crop)
|
||||
payloads.append((frame['asset'], child_png))
|
||||
reports.append(dict(asset=frame['asset'], parent_asset=parent_frame['asset'],
|
||||
exact_parent_RGBA_crop=[20, 0, 32, 32], parent_PNG_sha256=digest(reproduced_parent),
|
||||
own_source_sha256=frame['source']['sha256'], own_SourceScale4_alpha_exact=True,
|
||||
functional_Original_cells=[[3, 3]], Original_master_samples=16,
|
||||
RGBA_sha256=digest(crop.tobytes()), parent_material_report=material_report))
|
||||
for cls in (0, 1, 2):
|
||||
rows = [c for report in reports for c in report['parent_material_report']['components'] if c['material_class'] == cls]
|
||||
if sum(c['repaired'] for c in rows)/sum(c['samples'] for c in rows) > parent['CAPS'][cls][1]+1e-12:
|
||||
raise ValueError('complete parent aggregate cap exceeded')
|
||||
if digest(regular(Path(__file__))) != own_sha:
|
||||
raise ValueError('dispatcher changed during execution')
|
||||
final = dict(execution=dict(status='success', method=METHOD, recipe_sha256=digest(rb),
|
||||
tool_sha256=own_sha, source_sha256={f['asset']: f['source']['sha256'] for f in frames},
|
||||
parent_raw_sha256=[digest(raw_payload)], output_sha256={a: digest(p) for a, p in payloads}),
|
||||
parent_helper_sha256=PARENT_SHA, decoder_sha256=DECODER_SHA,
|
||||
parent_recipe_sha256=digest(parent_payload), parent_execution_receipt_sha256=recipe['parent_execution_receipt']['sha256'],
|
||||
reports=reports, Python=sys.version.split()[0], NumPy=np.__version__, Pillow=PILLOW_VERSION,
|
||||
limits=['Source crop aliases are not encoded-payload aliases.',
|
||||
'Original alpha and one Source face per frame get zero generated detail credit.',
|
||||
'Substantial inherited near-black functional support is not new generated texture.',
|
||||
'Parent and child material/context approval remain separate.'])
|
||||
output.mkdir(parents=True)
|
||||
for asset, payload in payloads:
|
||||
(output/f'{asset}.png').write_bytes(payload)
|
||||
(output/'execution-receipt.json').write_text(json.dumps(final, indent=2, sort_keys=True)+'\n')
|
||||
return final
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('recipe', type=Path)
|
||||
parser.add_argument('output', type=Path)
|
||||
parser.add_argument('--project-root', type=Path)
|
||||
args = parser.parse_args()
|
||||
print(json.dumps(run_recipe(args.recipe, args.output, args.project_root)['execution'], sort_keys=True))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -21,6 +21,7 @@ from hd_art_audit import checked_palette, validate_hd_frame
|
||||
from hd_art_source import decode
|
||||
from hd_art_derivation import validate_derivation
|
||||
from hd_frame_policy import validate_frame_policy, retained_review_version_ok, original_palette_indices
|
||||
from hd_dormant_enemy_review import DORMANT_ENEMY_SCOPE, approved_dormant_enemy_animation
|
||||
|
||||
|
||||
CARRIED_TERRAIN_V2 = "standalone-terrain-carried-identity-v2"
|
||||
@@ -885,17 +886,24 @@ def approved_job(root, identity, project):
|
||||
raise ValueError(f"missing complete visual/independent review: {identity}")
|
||||
retained_details = {asset: frame["original_detail_cells"] for asset, frame in rows.items()
|
||||
if frame.get("original_detail_cells")}
|
||||
declaration = review.get("retained_original_detail_review", {})
|
||||
dormant_enemy = candidate.get("approval_scope") == DORMANT_ENEMY_SCOPE
|
||||
if (declaration.get("hybrid_review_scope") == DORMANT_ENEMY_SCOPE) != dormant_enemy:
|
||||
raise ValueError("dormant enemy scope must be explicit in candidate and review")
|
||||
if retained_details:
|
||||
declaration = review.get("retained_original_detail_review", {})
|
||||
catalog_only = declaration.get("hybrid_review_scope") == "catalog_only_no_known_original_runtime_caller"
|
||||
unused_terrain = declaration.get("hybrid_review_scope") == "catalog_only_unused_original_terrain"
|
||||
if (catalog_only or unused_terrain) and selection is not None:
|
||||
if (catalog_only or unused_terrain or dormant_enemy) and selection is not None:
|
||||
raise ValueError("catalog-only scope cannot replace contextual Terrain approval")
|
||||
qualified_scope = declaration.get("hybrid_allscale_and_genuine_context_independent_review") is True
|
||||
if catalog_only or unused_terrain:
|
||||
qualified_scope = (declaration.get("hybrid_allscale_and_genuine_context_independent_review") is False
|
||||
and declaration.get("catalog_only_no_original_runtime_caller") is True
|
||||
and declaration.get("catalog_only_independent_whole_material_and_allscale_review") is True)
|
||||
if dormant_enemy:
|
||||
qualified_scope = (declaration.get("hybrid_allscale_and_genuine_context_independent_review") is False
|
||||
and declaration.get("catalog_dormant_no_genuine_runtime_context_claimed") is True
|
||||
and declaration.get("catalog_only_independent_whole_material_and_allscale_review") is True)
|
||||
if (not same_json(declaration.get("cells_by_asset"), retained_details)
|
||||
or declaration.get("retained_original_not_generated_HD") is not True
|
||||
or not retained_review_version_ok(declaration.get("policy_version"), retained_details)
|
||||
@@ -955,6 +963,8 @@ def approved_job(root, identity, project):
|
||||
for asset, frame in rows.items():
|
||||
validate_hd_frame(root, frame, payloads[asset], pictures[asset])
|
||||
approved_standalone_evidence(root, selection, review, rows, qa)
|
||||
elif dormant_enemy:
|
||||
approved_dormant_enemy_animation(root, project, candidate, job, review, rows, qa)
|
||||
elif retained_details and catalog_only:
|
||||
approved_catalog_only_retained_evidence(root, project, declaration, rows, qa)
|
||||
elif retained_details and unused_terrain:
|
||||
@@ -1025,6 +1035,10 @@ def main():
|
||||
})
|
||||
if candidate.get("artwork_derivation"):
|
||||
additions[-1]["artwork_derivation"] = candidate["artwork_derivation"]
|
||||
if candidate.get("approval_scope") == DORMANT_ENEMY_SCOPE:
|
||||
additions[-1]["approval_scope"] = DORMANT_ENEMY_SCOPE
|
||||
additions[-1]["no_genuine_original_runtime_context_claimed"] = True
|
||||
additions[-1]["dormant_source_trace"] = review["retained_original_detail_review"]["dormant_source_trace"]
|
||||
if candidate.get("complete_standalone_selection"):
|
||||
additions[-1]["approval_scope"] = "complete_standalone_terrain_selection"
|
||||
additions[-1]["original_fallback_assets"] = candidate["original_fallback_assets"]
|
||||
|
||||
@@ -0,0 +1,467 @@
|
||||
"""Synthetic whole-animation dormant admission; no real art/engine/registrar main."""
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from PIL import Image
|
||||
import hd_dormant_enemy_review as dormant
|
||||
import hd_register_reviewed as registrar
|
||||
from hd_candidate_review import binding_digest
|
||||
|
||||
|
||||
class DormantEnemyAnimationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.temp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.temp.cleanup)
|
||||
self.project = Path(self.temp.name) / "Renamed 游戏 bundle"
|
||||
self.data = self.project / "data"
|
||||
self.root = self.data / "hd"
|
||||
self.directory = self.root / "jobs/dormant"
|
||||
for p in [self.directory / "raw", self.root / "proof", self.data / "sprites",
|
||||
self.data / "misc", self.data / "maps", self.project / "src", self.project / "target"]:
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
for name in ["assets", "content", "simulation", "hd"]:
|
||||
(self.project / f"src/{name}.rs").write_text(f"synthetic {name} Source gate fixture")
|
||||
gates = {p: self.sha(self.project / p) for p in dormant.SUPPORTED_GATE_SOURCES}
|
||||
gate_patch = patch.dict(dormant.SUPPORTED_GATE_SOURCES, gates, clear=True)
|
||||
gate_patch.start()
|
||||
self.addCleanup(gate_patch.stop)
|
||||
lib = self.project / "target/fixture.rlib"
|
||||
lib.write_bytes(b"synthetic immutable library, not an actual game execution")
|
||||
os.utime(lib, ns=(2_000_000_000_000_000_000,) * 2)
|
||||
palette = self.data / "misc/PALETTE_DAT"
|
||||
palette.write_bytes(bytes(v // 4 for v in range(256) for _ in range(3)))
|
||||
self.index = [(1, "misc/PALETTE_DAT")]
|
||||
self.assets = [f"{0x100+i:05x}" for i in range(6)]
|
||||
self.rows, targets, reports, artwork = {}, [], [], {}
|
||||
for i, asset in enumerate(self.assets):
|
||||
route = "sprites/DORMANT_PIC" + (f".{i}" if i else "")
|
||||
source = self.data / route
|
||||
source.write_bytes(struct.pack("<5i", 1, 0, 0, 4, 4) + bytes([i+1] * 16))
|
||||
self.index.append((0x100+i, route))
|
||||
target = dict(asset=asset, source_path=route, source_sha256=self.sha(source),
|
||||
logical_canvas=[4, 4], origin=[0, 0], source_palette="PALETTE_DAT",
|
||||
source_palette_sha256=self.sha(palette))
|
||||
targets.append(target)
|
||||
raw = self.directory / f"raw/{asset}.png"
|
||||
Image.new("RGBA", (8, 8), (4*(i+1),)*3 + (255,)).save(raw)
|
||||
frame = dict(frame=i, file=f"jobs/dormant/raw/{asset}.png", logical_width=4,
|
||||
logical_height=4, origin=[0, 0], source_sha256=self.sha(source),
|
||||
original_detail_cells=[[0, 0]])
|
||||
self.rows[asset] = frame
|
||||
artwork[frame["file"]] = self.sha(raw)
|
||||
mask = dict(palette_indices_240_to_255_counts=[0]*16, visible_pixels=16)
|
||||
reports.append(dict(asset=asset, source_sha256=frame["source_sha256"], logical_canvas=[4, 4],
|
||||
source_mask=mask, source_palette="PALETTE_DAT", hd_masks=[dict(scale=k, mask=mask) for k in range(1, 5)]))
|
||||
self.sprite = bytearray(528)
|
||||
self.sprite[:11] = b"DORMANT_PIC"
|
||||
struct.pack_into("<i", self.sprite, 44, 6)
|
||||
for game in range(4):
|
||||
route = f"misc/SPRITE{game+1}_ITM"
|
||||
(self.data / route).write_bytes(bytes(self.sprite) if game == 0 else bytes(528))
|
||||
self.index.append((0x200+game, route))
|
||||
for sector in range(1, 4):
|
||||
for wave in range(1, 10):
|
||||
route = f"maps/MAP{wave}G{sector}_MAP"
|
||||
spawn = struct.pack("<6i", -1, 0, 0, 31, 0, 1) if (sector, wave) == (1, 1) else b""
|
||||
payload = struct.pack("<3i", 5412+len(spawn), 5412, len(spawn)//24) + bytes(5400) + spawn
|
||||
(self.data / route).write_bytes(payload)
|
||||
self.index.append((0x300+16*sector+wave, route))
|
||||
(self.data / "INDEX").write_text("".join(f"{i:x} {(self.data/r).stat().st_size} {r}\n" for i, r in self.index))
|
||||
self.candidate = dict(format_version=1, job_id="dormant", complete_family=True,
|
||||
category="Enemies and ground objects", kind="complete_family",
|
||||
approval_scope=dormant.DORMANT_ENEMY_SCOPE,
|
||||
sequences=[dict(name="DORMANT", base_asset=0x100, frame_count=6,
|
||||
frames=list(self.rows.values()))])
|
||||
self.job = dict(job_id="dormant", kind="complete_family", category="Enemies and ground objects",
|
||||
owner="artist", targets=targets)
|
||||
self.qa = dict(frame_bindings_sha256=binding_digest(self.rows), validated_frames=6, diagnostics=[], static_card_palette_mode=False,
|
||||
library="target/fixture.rlib", library_sha256=self.sha(lib),
|
||||
library_modified_ns=lib.stat().st_mtime_ns, source_newer_than_library=False,
|
||||
any_game_source_newer_than_library=False, frames=reports, artwork_sha256=artwork,
|
||||
current_game_source_sha256={p.relative_to(self.project).as_posix(): self.sha(p)
|
||||
for p in (self.project / "src").rglob("*.rs")})
|
||||
self.trace = dict(contract_version=dormant.DORMANT_ENEMY_TRACE,
|
||||
no_genuine_original_runtime_context_claimed=True,
|
||||
family=dict(name="DORMANT_PIC", assets=self.assets, frame_count=6),
|
||||
catalogue=self.pin("data/INDEX"), palette=self.pin("data/misc/PALETTE_DAT"),
|
||||
phases=[dict(asset=a, Source=self.pin("data/"+t["source_path"]),
|
||||
header_le_i32=[1, 0, 0, 4, 4]) for a, t in zip(self.assets, targets)],
|
||||
sprite_libraries=[self.pin(f"data/misc/SPRITE{g}_ITM") for g in range(1, 5)],
|
||||
sprite_callers=[dict(game=0, ordinal=0, name="DORMANT_PIC", base_asset="00100",
|
||||
frame_count=6, enemy_id="base:enemy_0_0",
|
||||
record_sha256=hashlib.sha256(self.sprite).hexdigest())],
|
||||
current_rust_sources=[self.pin(p.relative_to(self.project).as_posix())
|
||||
for p in sorted((self.project / "src").rglob("*.rs"))],
|
||||
current_library=self.pin("target/fixture.rlib"),
|
||||
maps=[self.pin(f"data/maps/MAP{w}G{s}_MAP") for s in range(1, 4) for w in range(1, 10)],
|
||||
map_declarations=[dict(map="data/maps/MAP1G1_MAP", ordinal=0, byte_offset=5412,
|
||||
fields=[-1, 0, 0, 31, 0, 1],
|
||||
raw_bytes_hex=struct.pack("<6i", -1, 0, 0, 31, 0, 1).hex(),
|
||||
enabled_by_pilot_difficulty=[False]*4)],
|
||||
counts=dict(maps=27, sprite_libraries=4, sprite_records=4, family_library_callers=1,
|
||||
spawn_records=1, family_map_declarations=1, ordinary_difficulty_callers=0))
|
||||
self.common = dict(scope=dormant.DORMANT_ENEMY_SCOPE, assets=self.assets,
|
||||
frame_bindings_sha256=binding_digest(self.rows),
|
||||
source_sha256_by_asset={a: f["source_sha256"] for a, f in self.rows.items()},
|
||||
artwork_sha256=artwork, original_generated_raw_sha256=sorted(set(artwork.values())),
|
||||
retained_cells_by_asset={a: f["original_detail_cells"] for a, f in self.rows.items()},
|
||||
native_scales=[1, 2, 3, 4], all_frames_whole_viewed=True,
|
||||
no_genuine_original_runtime_context_claimed=True)
|
||||
self.peer = dict(**self.common, contract_version=dormant.DORMANT_ENEMY_MATERIAL, status="qualified",
|
||||
artist="artist", preparer="preparer", reviewer="different", qualified_frames=6,
|
||||
held_frames=0, different_from_artist_and_preparer=True,
|
||||
native_decisions=[dict(asset=a, scale=k, status="qualified", retained_SourceNx_exact=True,
|
||||
retained_samples=k*k) for a in self.assets for k in range(1, 5)])
|
||||
self.actual = dict(status="QUALIFIED_COMPLETE_ANIMATION_INTRINSIC", reviewer="different",
|
||||
decisions=[dict(asset=a, intrinsic="QUALIFIED_WITH_LIMITS") for a in self.assets],
|
||||
physical_views=[dict(asset=a, kind="native1to4", whole_frame_personally_viewed=True)
|
||||
for a in self.assets])
|
||||
self.approval = dict(**self.common, status="approved", authority_role="project_root", reviewer="root")
|
||||
|
||||
@staticmethod
|
||||
def sha(path):
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
def pin(self, route):
|
||||
path = self.project / route
|
||||
return dict(path=route, sha256=self.sha(path), bytes=path.stat().st_size)
|
||||
|
||||
def proof(self, route, value):
|
||||
path = self.root / route
|
||||
path.write_text(json.dumps(value, indent=2) + "\n")
|
||||
return dict(path=route, sha256=self.sha(path))
|
||||
|
||||
def declarations(self):
|
||||
self.proof("jobs/dormant/job.json", self.job)
|
||||
candidate = self.proof("jobs/dormant/candidate.json", self.candidate)
|
||||
qa_ref = self.proof("proof/qa.json", self.qa)
|
||||
trace_ref = self.proof("proof/trace.json", self.trace)
|
||||
peer = copy.deepcopy(self.peer)
|
||||
peer.update(candidate=candidate, runtime_qa_sha256=qa_ref["sha256"],
|
||||
actual_whole_material_review=self.proof("proof/actual.json", self.actual))
|
||||
peer_ref = self.proof("proof/peer.json", peer)
|
||||
approval = dict(self.approval, runtime_qa_sha256=qa_ref["sha256"], source_trace_sha256=trace_ref["sha256"],
|
||||
independent_material_sha256=peer_ref["sha256"])
|
||||
declaration = dict(hybrid_review_scope=dormant.DORMANT_ENEMY_SCOPE,
|
||||
hybrid_allscale_and_genuine_context_independent_review=False,
|
||||
catalog_dormant_no_genuine_runtime_context_claimed=True,
|
||||
catalog_only_independent_whole_material_and_allscale_review=True,
|
||||
cells_by_asset=self.common["retained_cells_by_asset"], retained_original_not_generated_HD=True,
|
||||
policy_version="retained-original-details-v2-painted0255",
|
||||
painted_Source_palette_domain=[0, 255], functional_palette_effects_are_retained_Original=True,
|
||||
frame_bindings_sha256=binding_digest(self.rows), dormant_source_trace=trace_ref,
|
||||
dormant_independent_material=peer_ref, dormant_actual_allscale_QA=qa_ref,
|
||||
dormant_root_approval=self.proof("proof/root.json", approval))
|
||||
review = dict(status="approved", source_and_all_scale_visual_review=True, independent_review=peer_ref,
|
||||
frame_bindings_sha256=binding_digest(self.rows), runtime_qa="proof/qa.json",
|
||||
retained_original_detail_review=declaration)
|
||||
self.proof("jobs/dormant/coordinator-review.json", review)
|
||||
return review
|
||||
|
||||
def check(self, review=None):
|
||||
return dormant.approved_dormant_enemy_animation(self.root, self.project, self.candidate, self.job,
|
||||
self.declarations() if review is None else review,
|
||||
self.rows, self.qa)
|
||||
|
||||
def test_whole_synthetic_six_phase_secret_only_family_accepts_relocated_unicode(self):
|
||||
result = self.check()
|
||||
self.assertEqual(result["family_map_declarations"], 1)
|
||||
self.assertEqual(result["ordinary_difficulty_callers"], 0)
|
||||
|
||||
def test_actual_registrar_dispatch_accepts_explicit_whole_family(self):
|
||||
self.declarations()
|
||||
_, _, rows = registrar.approved_job(self.root, "dormant", self.project)
|
||||
self.assertEqual(set(rows), set(self.assets))
|
||||
|
||||
def test_normal_caller_rejected_even_with_updated_pin_and_false_zero_claim(self):
|
||||
path = self.data / "maps/MAP1G1_MAP"
|
||||
payload = bytearray(path.read_bytes())
|
||||
struct.pack_into("<i", payload, 5412+20, 3)
|
||||
path.write_bytes(payload)
|
||||
self.trace["maps"][0] = self.pin("data/maps/MAP1G1_MAP")
|
||||
with self.assertRaisesRegex(ValueError, "ordinary or unsupported"): self.check()
|
||||
|
||||
def test_partial_sequence_singleton_and_phase_limit_rejected(self):
|
||||
for count in [1, 5, 97, True]:
|
||||
with self.subTest(count=count):
|
||||
self.candidate["sequences"][0]["frame_count"] = count
|
||||
with self.assertRaises(ValueError): self.check()
|
||||
|
||||
def test_six_singletons_cannot_fake_whole_animation(self):
|
||||
self.candidate["sequences"] = [dict(base_asset=0x100+i, frame_count=1, frames=[dict(f, frame=0)])
|
||||
for i, f in enumerate(self.rows.values())]
|
||||
with self.assertRaisesRegex(ValueError, "one complete"): self.check()
|
||||
|
||||
def test_phase_header_false_or_bool_rejected(self):
|
||||
for value in [True, 2]:
|
||||
self.trace["phases"][0]["header_le_i32"][0] = value
|
||||
with self.assertRaisesRegex(ValueError, "phase header"): self.check()
|
||||
|
||||
def test_duplicate_map_or_missing_fourth_library_rejected(self):
|
||||
self.trace["maps"][-1] = self.trace["maps"][0]
|
||||
with self.assertRaisesRegex(ValueError, "all27"): self.check()
|
||||
self.trace["maps"][-1] = self.pin("data/maps/MAP9G3_MAP")
|
||||
self.trace["sprite_libraries"].pop()
|
||||
with self.assertRaisesRegex(ValueError, "all four"): self.check()
|
||||
|
||||
def test_wrong_sprite_frame_count_and_alias_rejected(self):
|
||||
path = self.data / "misc/SPRITE1_ITM"
|
||||
payload = bytearray(path.read_bytes())
|
||||
struct.pack_into("<i", payload, 44, 5)
|
||||
path.write_bytes(payload)
|
||||
self.trace["sprite_libraries"][0] = self.pin("data/misc/SPRITE1_ITM")
|
||||
with self.assertRaisesRegex(ValueError, "partial, aliased or boss"): self.check()
|
||||
|
||||
def test_updated_gate_source_receipt_cannot_bypass_supported_source_epoch(self):
|
||||
path = self.project / "src/simulation.rs"
|
||||
path.write_text("changed normal gate predicate")
|
||||
for row in self.trace["current_rust_sources"]:
|
||||
if row["path"] == "src/simulation.rs": row.update(self.pin("src/simulation.rs"))
|
||||
self.qa["current_game_source_sha256"]["src/simulation.rs"] = self.sha(path)
|
||||
with self.assertRaisesRegex(ValueError, "gate Source changed"): self.check()
|
||||
|
||||
def test_missing_source_or_stale_qa_rejected(self):
|
||||
self.trace["current_rust_sources"].pop()
|
||||
with self.assertRaisesRegex(ValueError, "Rust Source inventory"): self.check()
|
||||
|
||||
def test_qa_false_freshness_and_wrong_source_hash_rejected(self):
|
||||
self.qa["any_game_source_newer_than_library"] = None
|
||||
with self.assertRaisesRegex(ValueError, "current QA"): self.check()
|
||||
|
||||
def test_old_material_cannot_bind_different_runtime_raw(self):
|
||||
self.peer["artwork_sha256"] = dict(self.peer["artwork_sha256"])
|
||||
self.peer["artwork_sha256"][next(iter(self.peer["artwork_sha256"]))] = "0"*64
|
||||
with self.assertRaisesRegex(ValueError, "different whole-family"): self.check()
|
||||
|
||||
def test_peer_not_independent_or_held_cannot_qualify(self):
|
||||
self.peer["reviewer"] = "preparer"
|
||||
with self.assertRaisesRegex(ValueError, "different whole-family"): self.check()
|
||||
|
||||
def test_missing_scale_face_counts_and_boolean_view_rejected(self):
|
||||
self.peer["native_decisions"].pop()
|
||||
with self.assertRaisesRegex(ValueError, "native-scale"): self.check()
|
||||
|
||||
def test_root_cannot_claim_fabricated_context_or_false_whole_view(self):
|
||||
self.approval["no_genuine_original_runtime_context_claimed"] = False
|
||||
with self.assertRaisesRegex(ValueError, "Root whole-family"): self.check()
|
||||
|
||||
def test_actual_review_missing_whole_phase_rejected(self):
|
||||
self.actual["physical_views"].pop()
|
||||
with self.assertRaisesRegex(ValueError, "actual whole-native"): self.check()
|
||||
|
||||
def test_source_and_proof_paths_cannot_escape_or_use_symlink(self):
|
||||
self.trace["phases"][0]["Source"]["path"] = "../outside"
|
||||
with self.assertRaisesRegex(ValueError, "portable file pin"): self.check()
|
||||
|
||||
def test_dormant_scope_cannot_silently_replace_ordinary_retained_contract(self):
|
||||
self.declarations()
|
||||
path = self.directory / "coordinator-review.json"
|
||||
review = json.loads(path.read_text())
|
||||
review["retained_original_detail_review"]["hybrid_review_scope"] = "ordinary"
|
||||
path.write_text(json.dumps(review))
|
||||
with self.assertRaisesRegex(ValueError, "explicit in candidate and review"):
|
||||
registrar.approved_job(self.root, "dormant", self.project)
|
||||
|
||||
|
||||
def test_false_census_and_boolean_count_are_not_status_authority(self):
|
||||
for value in [0, True]:
|
||||
self.trace["counts"]["family_map_declarations"] = value
|
||||
with self.assertRaisesRegex(ValueError, "caller census"): self.check()
|
||||
|
||||
def test_classic_secret_code_zero_and_two_remain_dormant(self):
|
||||
for code in [0, 2]:
|
||||
path = self.data / "maps/MAP1G1_MAP"
|
||||
payload = bytearray(path.read_bytes())
|
||||
struct.pack_into("<i", payload, 5412+20, code)
|
||||
path.write_bytes(payload)
|
||||
self.trace["maps"][0] = self.pin("data/maps/MAP1G1_MAP")
|
||||
self.trace["map_declarations"][0]["fields"][-1] = code
|
||||
self.trace["map_declarations"][0]["raw_bytes_hex"] = bytes(payload[5412:5436]).hex()
|
||||
self.check()
|
||||
|
||||
def test_static_card_palette_bypass_is_not_enemy_qa(self):
|
||||
self.qa["static_card_palette_mode"] = True
|
||||
with self.assertRaisesRegex(ValueError, "runtime QA differs"): self.check()
|
||||
|
||||
def test_missing_root_and_untrue_whole_view_rejected(self):
|
||||
review = self.declarations()
|
||||
review["retained_original_detail_review"].pop("dormant_root_approval")
|
||||
with self.assertRaisesRegex(ValueError, "portable hash-bound"): self.check(review)
|
||||
|
||||
def test_fake_context_declaration_rejected_even_without_retention_dispatch(self):
|
||||
review = self.declarations()
|
||||
review["retained_original_detail_review"]["hybrid_allscale_and_genuine_context_independent_review"] = True
|
||||
with self.assertRaisesRegex(ValueError, "cannot claim genuine"): self.check(review)
|
||||
|
||||
def test_boolean_native_sample_count_is_not_one_source_sample(self):
|
||||
self.peer["native_decisions"][0]["retained_samples"] = True
|
||||
with self.assertRaisesRegex(ValueError, "native-scale"): self.check()
|
||||
|
||||
def test_numeric_one_cannot_fake_whole_family_view(self):
|
||||
self.peer["all_frames_whole_viewed"] = 1
|
||||
with self.assertRaisesRegex(ValueError, "different whole-family"): self.check()
|
||||
|
||||
def test_ordinary_source_cannot_gain_functional_palette_labels(self):
|
||||
self.qa["frames"][0]["hd_masks"][0]["mask"] = dict(self.qa["frames"][0]["hd_masks"][0]["mask"])
|
||||
self.qa["frames"][0]["hd_masks"][0]["mask"]["palette_indices_240_to_255_counts"] = [1] + [0]*15
|
||||
with self.assertRaisesRegex(ValueError, "functional palette"): self.check()
|
||||
|
||||
def test_changed_raw_artwork_and_stale_source_mask_rejected(self):
|
||||
raw = self.root / self.rows[self.assets[0]]["file"]
|
||||
Image.new("RGBA", (8, 8), (50, 50, 50, 255)).save(raw)
|
||||
with self.assertRaisesRegex(ValueError, "hash differs"): self.check()
|
||||
|
||||
def test_direct_native_family_or_activation_code_requires_new_review(self):
|
||||
path = self.project / "src/hd.rs"
|
||||
path.write_text("DORMANT_PIC")
|
||||
for row in self.trace["current_rust_sources"]:
|
||||
if row["path"] == "src/hd.rs": row.update(self.pin("src/hd.rs"))
|
||||
with self.assertRaisesRegex(ValueError, "direct family caller"): self.check()
|
||||
|
||||
def test_source_symlink_and_library_hash_cannot_bypass_epoch(self):
|
||||
source = self.data / "sprites/DORMANT_PIC"
|
||||
payload = source.read_bytes()
|
||||
outside = self.project.parent / "outside"
|
||||
outside.write_bytes(payload)
|
||||
source.unlink()
|
||||
source.symlink_to(outside)
|
||||
with self.assertRaisesRegex(ValueError, "symlink"): self.check()
|
||||
|
||||
def test_extra_named_source_phase_cannot_be_omitted(self):
|
||||
path = self.data / "sprites/DORMANT_PIC.6"
|
||||
payload = (self.data / "sprites/DORMANT_PIC").read_bytes()
|
||||
path.write_bytes(payload)
|
||||
index = self.data / "INDEX"
|
||||
index.write_text(index.read_text() + f"106 {len(payload)} sprites/DORMANT_PIC.6\n")
|
||||
self.trace["catalogue"] = self.pin("data/INDEX")
|
||||
with self.assertRaisesRegex(ValueError, "additional phases"): self.check()
|
||||
|
||||
|
||||
|
||||
def test_pending_phase_material_word_is_not_whole_qualification(self):
|
||||
self.actual["decisions"][0]["intrinsic"] = "QUALIFIED_PENDING"
|
||||
with self.assertRaisesRegex(ValueError, "actual whole-native"): self.check()
|
||||
|
||||
def test_global_not_qualified_cannot_match_qualified_substring(self):
|
||||
self.actual["status"] = "QUALIFIED_NOT_APPROVED"
|
||||
with self.assertRaisesRegex(ValueError, "actual whole-native"): self.check()
|
||||
|
||||
def test_not_approved_hyphen_separator_cannot_qualify(self):
|
||||
self.actual["status"] = "QUALIFIED_NOT-APPROVED"
|
||||
with self.assertRaisesRegex(ValueError, "actual whole-native"): self.check()
|
||||
|
||||
def test_hold_number_hyphen_separator_cannot_qualify(self):
|
||||
self.actual["status"] = "QUALIFIED_WITH_HOLD-1"
|
||||
with self.assertRaisesRegex(ValueError, "actual whole-native"): self.check()
|
||||
|
||||
@unittest.skipUnless(hasattr(os, "mkfifo"), "requires POSIX FIFO")
|
||||
def test_swapped_fifo_is_rejected_before_read_without_blocking(self):
|
||||
path = self.directory / "proof-fifo-race.bin"
|
||||
path.write_bytes(b"original")
|
||||
code = r"""
|
||||
import hashlib,os,sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
import hd_dormant_enemy_review as d
|
||||
root=Path(sys.argv[1]); path=root/"proof-fifo-race.bin"
|
||||
record=dict(path=path.name,sha256=hashlib.sha256(path.read_bytes()).hexdigest())
|
||||
actual_open=os.open
|
||||
|
||||
def swap_to_fifo(name,flags):
|
||||
path.unlink();os.mkfifo(path)
|
||||
return actual_open(name,flags)
|
||||
|
||||
with patch.object(d.os,"open",swap_to_fifo):
|
||||
try:d._read(root,record)
|
||||
except ValueError as error:
|
||||
assert "bounded, regular and unchanged" in str(error),str(error)
|
||||
print("actual swapped FIFO rejected before descriptor read")
|
||||
else:raise AssertionError("FIFO accepted")
|
||||
"""
|
||||
env = dict(os.environ, PYTHONPATH=os.pathsep.join(
|
||||
[str(Path(dormant.__file__).parent)] + sys.path))
|
||||
try:
|
||||
actual = subprocess.run([sys.executable, "-B", "-c", code, str(self.directory)],
|
||||
capture_output=True, text=True, env=env, timeout=3,
|
||||
cwd=Path(dormant.__file__).parent)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.fail("swapped FIFO blocked while opening the descriptor")
|
||||
self.assertEqual(actual.returncode, 0, actual.stderr)
|
||||
self.assertIn("actual swapped FIFO rejected", actual.stdout)
|
||||
|
||||
def test_file_growth_is_rejected_with_a_bounded_read(self):
|
||||
path = self.directory / "proof-growth.bin"
|
||||
path.write_bytes(b"old")
|
||||
record = dict(path=path.name, sha256=self.sha(path))
|
||||
actual_fdopen = os.fdopen
|
||||
read_sizes = []
|
||||
|
||||
class GrowDuringRead:
|
||||
def __init__(self, stream): self.stream = stream
|
||||
def __enter__(self): return self
|
||||
def __exit__(self, *unused): self.stream.close()
|
||||
def fileno(self): return self.stream.fileno()
|
||||
def read(self, size=-1):
|
||||
read_sizes.append(size)
|
||||
if len(read_sizes) == 1:
|
||||
with path.open("ab") as writer: writer.write(b"x" * 65)
|
||||
return self.stream.read(size)
|
||||
|
||||
with patch.object(dormant, "_MAX_FILE_BYTES", 64, create=True), \
|
||||
patch.object(dormant.os, "fdopen", lambda *args: GrowDuringRead(actual_fdopen(*args))):
|
||||
with self.assertRaisesRegex(ValueError, "exceeds bounded read|file changed"):
|
||||
dormant._read(self.directory, record)
|
||||
self.assertEqual(read_sizes, [65], "growth must never use an unbounded read")
|
||||
self.assertEqual(path.stat().st_size, 68)
|
||||
|
||||
def test_palette_constraint_cannot_bypass_existing_genuine_context_contract(self):
|
||||
self.rows[self.assets[0]]["palette_constraint"] = "original-painted-ordinary-v1"
|
||||
with self.assertRaisesRegex(ValueError, "ordinary genuine-context"): self.check()
|
||||
|
||||
def test_current_library_replacement_is_detected_without_trusting_qa_flags(self):
|
||||
path = self.project / "target/fixture.rlib"
|
||||
path.write_bytes(b"replacement library")
|
||||
with self.assertRaisesRegex(ValueError, "hash differs"): self.check()
|
||||
|
||||
def test_wrong_category_and_boolean_origin_rejected(self):
|
||||
self.candidate["category"] = "Terrain"
|
||||
with self.assertRaisesRegex(ValueError, "complete Original enemy"): self.check()
|
||||
self.candidate["category"] = "Enemies and ground objects"
|
||||
self.job["targets"][0]["origin"] = [False, False]
|
||||
with self.assertRaisesRegex(ValueError, "phase header"): self.check()
|
||||
|
||||
|
||||
|
||||
def test_actual_sourceG2_qa_field_is_accepted_without_relabeling_old_sourceF(self):
|
||||
self.qa["actual_sourceG2_source18_sha256"] = self.qa.pop("current_game_source_sha256")
|
||||
self.check()
|
||||
|
||||
def test_source_mask_visible_count_must_come_from_actual_source_bytes(self):
|
||||
self.qa["frames"][0]["source_mask"] = dict(self.qa["frames"][0]["source_mask"], visible_pixels=15)
|
||||
with self.assertRaisesRegex(ValueError, "Source palette/mask"): self.check()
|
||||
|
||||
def test_standalone_selection_cannot_use_animation_exception(self):
|
||||
self.candidate["complete_standalone_selection"] = True
|
||||
with self.assertRaisesRegex(ValueError, "one complete"): self.check()
|
||||
|
||||
def test_normal_retained_path_still_requires_its_own_genuine_context_declaration(self):
|
||||
self.candidate.pop("approval_scope")
|
||||
review = self.declarations()
|
||||
review["retained_original_detail_review"]["hybrid_review_scope"] = "ordinary"
|
||||
self.proof("jobs/dormant/coordinator-review.json", review)
|
||||
with self.assertRaisesRegex(ValueError, "retained Original detail provenance"):
|
||||
registrar.approved_job(self.root, "dormant", self.project)
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,213 @@
|
||||
"""Real synthetic RLE8/24 aliases and actual unmocked parent rendering/receipt."""
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
spec = importlib.util.spec_from_file_location('person01_under_test', HERE/'hd_person01_cleanup.py')
|
||||
tool = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(tool)
|
||||
# Public tools layout uses its existing parent test fixture. The private CODE
|
||||
# packet carries an exact small copy under dependency-project for review only.
|
||||
PARENT_TOOLS = HERE if (HERE/'test_hd_infantry_cleanup.py').is_file() else HERE/'dependency-project/tools'
|
||||
ps = importlib.util.spec_from_file_location('actual_synthetic_parent_fixture', PARENT_TOOLS/'test_hd_infantry_cleanup.py')
|
||||
parent_fixture_module = importlib.util.module_from_spec(ps)
|
||||
ps.loader.exec_module(parent_fixture_module)
|
||||
|
||||
|
||||
class PersonRecipeTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.fixture = parent_fixture_module.CompleteRecipeTests(methodName='runTest')
|
||||
self.fixture.setUp()
|
||||
self.root, self.base, self.project = self.fixture.root, self.fixture.base, self.fixture.project
|
||||
shutil.copyfile(PARENT_TOOLS/'hd_infantry_cleanup.py', self.project/'tools/hd_infantry_cleanup.py')
|
||||
self.parent_output = self.root/'parent-output'
|
||||
parent_receipt = parent_fixture_module.tool.run_recipe(self.fixture.path, self.parent_output, self.project)
|
||||
receipt = self.base/'parent-execution.json'
|
||||
receipt.write_text(json.dumps(parent_receipt))
|
||||
frames = []
|
||||
for i, asset in enumerate(tool.ASSETS):
|
||||
child = self.fixture.sources[i][0:8, 5:13].copy()
|
||||
path = self.base/f'{asset}.bin'
|
||||
path.write_bytes(parent_fixture_module.rle(child))
|
||||
frames.append(dict(asset=asset, parent_asset=parent_fixture_module.tool.ASSETS[i],
|
||||
source=self.fixture.bind(path), functional_original_cells=[[3, 3]]))
|
||||
self.recipe = dict(format_version=1, method=tool.METHOD,
|
||||
parent_recipe=self.fixture.bind(self.fixture.path),
|
||||
parent_execution_receipt=self.fixture.bind(receipt), frames=frames)
|
||||
self.path = self.base/'person-recipe.json'
|
||||
self.output = self.root/'child-output'
|
||||
|
||||
def tearDown(self):
|
||||
self.fixture.tearDown()
|
||||
|
||||
def write(self):
|
||||
self.path.write_text(json.dumps(self.recipe))
|
||||
|
||||
def run_real(self):
|
||||
self.write()
|
||||
return tool.run_recipe(self.path, self.output, self.project)
|
||||
|
||||
def reject(self, message=None):
|
||||
self.write()
|
||||
with self.assertRaisesRegex((ValueError, FileNotFoundError), message or '.*'):
|
||||
tool.run_recipe(self.path, self.output, self.project)
|
||||
self.assertFalse(self.output.exists())
|
||||
|
||||
def mutate_receipt(self, operation):
|
||||
path = self.base/'parent-execution.json'
|
||||
value = json.loads(path.read_text())
|
||||
operation(value['execution'])
|
||||
path.write_text(json.dumps(value))
|
||||
self.recipe['parent_execution_receipt'] = self.fixture.bind(path)
|
||||
|
||||
def test_real_six8x8_RLE_and_exact_parent_RGBA_crop_and_receipt(self):
|
||||
receipt = self.run_real()
|
||||
self.assertEqual(set(receipt['execution']), {'status', 'method', 'recipe_sha256', 'tool_sha256',
|
||||
'source_sha256', 'parent_raw_sha256', 'output_sha256'})
|
||||
self.assertEqual(receipt, json.loads((self.output/'execution-receipt.json').read_text()))
|
||||
self.assertEqual(len(list(self.output.glob('*.png'))), 6)
|
||||
for i, frame in enumerate(self.recipe['frames']):
|
||||
actual = np.asarray(Image.open(self.output/(frame['asset']+'.png')))
|
||||
parent = np.asarray(Image.open(self.parent_output/(frame['parent_asset']+'.png')))
|
||||
self.assertEqual(actual.shape, (32, 32, 4))
|
||||
self.assertTrue(np.array_equal(actual, parent[0:32, 20:52]))
|
||||
source = self.fixture.sources[i][0:8, 5:13]
|
||||
s4 = parent_fixture_module.tool.scale2(parent_fixture_module.tool.scale2(source))
|
||||
self.assertTrue(np.array_equal(actual[:, :, 3], np.where(s4 < 256, 255, 0)))
|
||||
self.assertTrue(np.array_equal(actual[12:16, 12:16, :3], self.fixture.palette[s4[12:16, 12:16]]))
|
||||
|
||||
def test_portable_replay_byte_identical(self):
|
||||
first = self.run_real()
|
||||
self.output = self.root/'replay'
|
||||
second = self.run_real()
|
||||
self.assertEqual(first['execution'], second['execution'])
|
||||
|
||||
def test_one_child_Source_index_difference_rejected_before_output(self):
|
||||
source = self.fixture.sources[5][0:8, 5:13].copy()
|
||||
source[0, 0] = 16
|
||||
path = self.base/(tool.ASSETS[5]+'.bin')
|
||||
path.write_bytes(parent_fixture_module.rle(source))
|
||||
self.recipe['frames'][5]['source'] = self.fixture.bind(path)
|
||||
self.reject('entire64-index')
|
||||
|
||||
def test_missing_or_reordered_child_rejected(self):
|
||||
self.recipe['frames'].reverse()
|
||||
self.reject('complete ordered')
|
||||
|
||||
def test_actual_missing_child_rejected(self):
|
||||
self.recipe['frames'].pop()
|
||||
self.reject('complete ordered')
|
||||
|
||||
def test_child_wrong8x8_canvas_rejected(self):
|
||||
source = self.fixture.sources[5][0:8, 5:12].copy()
|
||||
path = self.base/(tool.ASSETS[5]+'.bin')
|
||||
path.write_bytes(parent_fixture_module.rle(source))
|
||||
self.recipe['frames'][5]['source'] = self.fixture.bind(path)
|
||||
self.reject('child8x8')
|
||||
|
||||
def test_parent_wrong_atlas_layout_rejected(self):
|
||||
raw = self.base/'raw.png'
|
||||
im = Image.open(raw)
|
||||
im.crop((0, 0, 1536, 1023)).save(raw)
|
||||
self.fixture.recipe['raw_png'] = self.fixture.bind(raw)
|
||||
self.fixture.write()
|
||||
self.recipe['parent_recipe'] = self.fixture.bind(self.fixture.path)
|
||||
self.reject('exact parent atlas layout')
|
||||
|
||||
def test_parent_phase_swap_rejected(self):
|
||||
self.recipe['frames'][5]['parent_asset'] = '200ec'
|
||||
self.reject('fixed corresponding')
|
||||
|
||||
def test_unknown_crop_or_transform_rejected(self):
|
||||
self.recipe['crop'] = [19, 0, 32, 32]
|
||||
self.reject('exact fields')
|
||||
|
||||
def test_extra_Original_cells_rejected(self):
|
||||
self.recipe['frames'][5]['functional_original_cells'] = [[3, 3], [2, 2]]
|
||||
self.reject('fixed corresponding')
|
||||
|
||||
def test_child_source_hash_mismatch_rejected(self):
|
||||
self.recipe['frames'][5]['source']['sha256'] = '0'*64
|
||||
self.reject('hash mismatch')
|
||||
|
||||
def test_parent_tool_hash_mismatch_rejected_without_execution(self):
|
||||
(self.project/'tools/hd_infantry_cleanup.py').write_text('raise RuntimeError("must not execute")')
|
||||
self.reject('parent helper hash mismatch')
|
||||
|
||||
def test_parent_receipt_recipe_hash_mismatch(self):
|
||||
self.mutate_receipt(lambda x: x.update(recipe_sha256='0'*64))
|
||||
self.reject('parent execution provenance')
|
||||
|
||||
def test_parent_receipt_raw_hash_mismatch(self):
|
||||
self.mutate_receipt(lambda x: x.update(parent_raw_sha256=['0'*64]))
|
||||
self.reject('parent execution provenance')
|
||||
|
||||
def test_parent_receipt_Source_map_mismatch(self):
|
||||
self.mutate_receipt(lambda x: x.update(source_sha256={}))
|
||||
self.reject('parent source provenance')
|
||||
|
||||
def test_parent_PNG_provenance_mismatch_after_render_emits_no_child(self):
|
||||
self.mutate_receipt(lambda x: x['output_sha256'].update({'200f1': '0'*64}))
|
||||
self.reject('parent PNG not exactly reproduced')
|
||||
|
||||
def test_traversal_absolute_and_symlink_inputs_rejected(self):
|
||||
original = self.recipe['parent_recipe']['path']
|
||||
for bad in ['../recipe.json', '/recipe.json', 'x/../recipe.json', 'C:/recipe.json']:
|
||||
self.recipe['parent_recipe']['path'] = bad
|
||||
self.reject('path')
|
||||
self.recipe['parent_recipe']['path'] = original
|
||||
link = self.base/'link.json'
|
||||
link.symlink_to(self.fixture.path)
|
||||
self.recipe['parent_recipe']['path'] = 'link.json'
|
||||
self.reject('symlink')
|
||||
|
||||
def test_existing_child_output_not_overwritten(self):
|
||||
self.output.mkdir()
|
||||
(self.output/'sentinel').write_text('keep')
|
||||
self.write()
|
||||
with self.assertRaisesRegex(ValueError, 'fresh nonsymlink'):
|
||||
tool.run_recipe(self.path, self.output, self.project)
|
||||
self.assertEqual((self.output/'sentinel').read_text(), 'keep')
|
||||
|
||||
@unittest.skipUnless(hasattr(os, 'mkfifo'), 'POSIX FIFO negative')
|
||||
def test_parent_bootstrap_FIFO_rejected_promptly_without_hash_read(self):
|
||||
path = self.project/'tools/hd_infantry_cleanup.py'
|
||||
path.unlink()
|
||||
os.mkfifo(path)
|
||||
self.write()
|
||||
completed = subprocess.run([sys.executable, str(HERE/'hd_person01_cleanup.py'),
|
||||
str(self.path), str(self.output), '--project-root', str(self.project)],
|
||||
capture_output=True, text=True, timeout=3)
|
||||
self.assertNotEqual(completed.returncode, 0)
|
||||
self.assertIn('regular bounded parent dependency', completed.stderr)
|
||||
self.assertFalse(self.output.exists())
|
||||
|
||||
def test_parent_bootstrap_directory_rejected_without_execution(self):
|
||||
path = self.project/'tools/hd_infantry_cleanup.py'
|
||||
path.unlink()
|
||||
path.mkdir()
|
||||
self.reject('regular bounded parent dependency')
|
||||
|
||||
def test_parent_dependency_consumed_from_verified_buffer(self):
|
||||
original = tool.bootstrap_regular_bytes
|
||||
def mutate_after_read(path):
|
||||
data = original(path)
|
||||
path.write_text('raise RuntimeError("path reopened")')
|
||||
return data
|
||||
with patch.object(tool, 'bootstrap_regular_bytes', mutate_after_read):
|
||||
receipt = self.run_real()
|
||||
self.assertEqual(receipt['execution']['status'], 'success')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user