143 lines
8.5 KiB
Python
143 lines
8.5 KiB
Python
"""Verify actual rendered feedback bytes against a separate Process checks package."""
|
|
from pathlib import Path, PurePosixPath
|
|
from datetime import datetime
|
|
import argparse
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import stat
|
|
import zipfile
|
|
from PIL import Image
|
|
|
|
COMMIT = "b15345123a7feaf630ef24d45f591c76c6715778"
|
|
INPUT_SHA = "7f5b3de4d5bb70282e424fbe90211af2ac259965aef10b05438b46076467a767"
|
|
TEST = "production_process_multiselect_reverse_click_group_drop_has_one_history_step_and_stable_camera"
|
|
SOURCES = ("crates/oa-desktop/src/behavior/topology/layout_tests.rs", "crates/oa-desktop/src/behavior/canvas_workspace/tests.rs")
|
|
NAMES = [f"process-{layer}-{stage}" for layer in ("Oa", "Sa") for stage in ("before-press", "after-motion")]
|
|
MEMBERS = {"observations.md", "encode-png.py", "capture-1/png-readback.json"} | {f"capture-1/{name}.{ext}" for name in NAMES for ext in ("rgba", "png", "json")}
|
|
OUTER = {"prepare.py", "verify.py", "README.md", "manifest.json", "evidence.zip"}
|
|
|
|
|
|
def need(ok, message):
|
|
if not ok:
|
|
raise ValueError(message)
|
|
|
|
|
|
def pin(data):
|
|
return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}
|
|
|
|
|
|
def safe(name):
|
|
return isinstance(name, str) and bool(name) and not PurePosixPath(name).is_absolute() and all(c not in name for c in ("\\", ":")) and all(ord(c) >= 32 for c in name) and all(p not in ("", ".", "..") for p in name.split("/"))
|
|
|
|
|
|
def parsed(data):
|
|
def unique(items):
|
|
result = {}
|
|
for key, value in items:
|
|
need(key not in result, "duplicate JSON key")
|
|
result[key] = value
|
|
return result
|
|
def constant(value):
|
|
raise ValueError("nonfinite JSON: " + value)
|
|
return json.loads(data, object_pairs_hook=unique, parse_constant=constant)
|
|
|
|
|
|
def inventory(archive, expected):
|
|
names = archive.namelist()
|
|
need(len(names) == len(set(names)) and set(names) == set(expected), "exact ZIP inventory")
|
|
for row in archive.infolist():
|
|
need(safe(row.filename) and not row.is_dir() and not stat.S_ISLNK(row.external_attr >> 16), "unsafe ZIP entry")
|
|
need(row.file_size == expected[row.filename]["bytes"], "ZIP size")
|
|
|
|
|
|
def checks_data(root):
|
|
"""Read referenced gate bytes, without importing or executing package code."""
|
|
files = parsed((root / "files.json").read_bytes())
|
|
need(set(files) == OUTER, "checks outer inventory")
|
|
for name, expected in files.items():
|
|
need(pin((root / name).read_bytes()) == expected, "checks outer pin: " + name)
|
|
manifest = parsed((root / "manifest.json").read_bytes())
|
|
need(manifest["schema"] == 1 and manifest["kind"] == "multiselect-cargo-gates", "checks format")
|
|
with zipfile.ZipFile(root / "evidence.zip") as archive:
|
|
inventory(archive, manifest["members"])
|
|
def raw(logical):
|
|
member = manifest["logicalFiles"][logical]
|
|
data = archive.read(member)
|
|
need(pin(data) == manifest["members"][member], "referenced checks pin")
|
|
return data
|
|
prefix = "cargo/feedback-1/"
|
|
gate = {name: raw(prefix + name) for name in ("started.json", "process.json", "result.json", "stdout.log", "stderr.log", "inputs.json", "inputs.zip")}
|
|
spec = manifest["runs"]["feedback-1"]
|
|
gate["helper"] = raw(spec["helper"])
|
|
result, start, process = [parsed(gate[name + ".json"]) for name in ("result", "started", "process")]
|
|
need(all(result.get(k) == v for k, v in start.items()) and result["command"] == process["command"], "feedback start/process/result")
|
|
times = [datetime.fromisoformat(t) for t in (start["start"], process["observed"], result["finish"])]
|
|
need(all(t.utcoffset() is not None for t in times) and times == sorted(times), "closed feedback chronology")
|
|
need(result["exitCode"] == 0 and result["inputsUnchanged"] is True and TEST in result["command"], "passed exact feedback gate")
|
|
need(result["environment"]["ARCADIA_RENDER_EVIDENCE_DIR"].replace("\\", "/") == result["cwd"].replace("\\", "/") + "/.local/process-feedback-review/capture-1", "actual output directory provenance")
|
|
for key, name in (("inputs", "inputs.json"), ("archive", "inputs.zip"), ("stdout", "stdout.log"), ("stderr", "stderr.log"), ("helper", "helper")):
|
|
need(result[key] == pin(gate[name]), "feedback receipt pin: " + key)
|
|
need(result["inputs"]["sha256"] == INPUT_SHA, "fixed b153 input map identity")
|
|
inputs = parsed(gate["inputs.json"])
|
|
need(len(inputs) == result["inputCount"] == 742 and all(safe(name) for name in inputs), "captured source inventory")
|
|
with zipfile.ZipFile(io.BytesIO(gate["inputs.zip"])) as capsule:
|
|
inventory(capsule, inputs)
|
|
source = {name: capsule.read(name) for name in inputs}
|
|
need(all(pin(source[name]) == expected for name, expected in inputs.items()), "all captured source bytes")
|
|
need(("test behavior::topology::view::layout_tests::" + TEST + " ... ok").encode() in gate["stdout.log"], "actual test result")
|
|
return gate, source
|
|
|
|
|
|
def verify_data(manifest, read, checks):
|
|
need(manifest["schema"] == 1 and manifest["kind"] == "process-render-feedback" and manifest["sourceCommit"] == COMMIT, "feedback manifest")
|
|
need(set(manifest["members"]) == MEMBERS, "exact feedback files")
|
|
for name, expected in manifest["members"].items():
|
|
need(pin(read(name)) == expected, "feedback pin: " + name)
|
|
gate, source = checks
|
|
need(manifest["checksRun"] == "feedback-1" and manifest["checksPins"] == {name: pin(data) for name, data in gate.items()}, "exact checks references")
|
|
need(manifest["capturingSources"] == {name: pin(source[name]) for name in SOURCES}, "capturing test and renderer source identity")
|
|
report = parsed(read("capture-1/png-readback.json"))
|
|
need(set(report["captures"]) == set(NAMES), "four readback records")
|
|
for layer in ("Oa", "Sa"):
|
|
previous = None
|
|
for stage in ("before-press", "after-motion"):
|
|
name = f"process-{layer}-{stage}"
|
|
metadata = parsed(read("capture-1/" + name + ".json"))
|
|
raw = read("capture-1/" + name + ".rgba")
|
|
need(metadata["layer"] == layer and metadata["stage"] == stage and metadata["format"] == "RGBA8", "capture identity")
|
|
need(metadata["width"] == 1440 and metadata["height"] == 900 and len(raw) == 1440 * 900 * 4, "exact original dimensions")
|
|
need(metadata["selectionCount"] == 2 and isinstance(metadata["selection"], str) and "history" not in metadata, "selection metadata; no invented history field")
|
|
bundle = metadata["source"]
|
|
need(safe(bundle["entry"]) and bundle["entry"] in bundle["files"] and all(safe(n) and isinstance(s, str) for n, s in bundle["files"].items()), "complete captured source bundle")
|
|
with Image.open(io.BytesIO(read("capture-1/" + name + ".png"))) as png:
|
|
need(png.format == "PNG" and png.size == (1440, 900) and png.convert("RGBA").tobytes() == raw, "decoded PNG equals actual raw RGBA")
|
|
expected = {"bytes": len(raw), "rgbaSha256": pin(raw)["sha256"], "width": 1440, "height": 900}
|
|
need(report["captures"][name] == expected, "original PNG readback receipt")
|
|
if previous:
|
|
for field in ("source", "selection", "selectionCount", "canvas", "camera", "scale", "width", "height"):
|
|
need(metadata[field] == previous[0][field], layer + " changed " + field + " before release")
|
|
need(raw != previous[1], "actual before/after pixel difference")
|
|
previous = metadata, raw
|
|
return {"captures": 4, "pngPixelsExact": True, "stableMetadataPairs": 2, "checksRun": "feedback-1", "nativeAcceptance": False}
|
|
|
|
|
|
def verify_package(root, checks):
|
|
files = parsed((root / "files.json").read_bytes())
|
|
need(set(files) == OUTER, "feedback outer inventory")
|
|
for name, expected in files.items():
|
|
need(pin((root / name).read_bytes()) == expected, "feedback outer pin")
|
|
manifest = parsed((root / "manifest.json").read_bytes())
|
|
with zipfile.ZipFile(root / "evidence.zip") as archive:
|
|
inventory(archive, manifest["members"])
|
|
return verify_data(manifest, archive.read, checks)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("package", nargs="?", type=Path, default=Path(__file__).resolve().parent)
|
|
parser.add_argument("--checks", type=Path, required=True, help="Process checks package containing closed feedback-1")
|
|
args = parser.parse_args()
|
|
print("PASS feedback byte proof:", verify_package(args.package, checks_data(args.checks)))
|
|
print("History is asserted by the captured test, not recorded in image metadata. No native acceptance claim.")
|