241 lines
15 KiB
Python
241 lines
15 KiB
Python
"""Verify archived native receipts and bytes; do not rerun the GUI or checker."""
|
|
from pathlib import Path, PurePosixPath
|
|
from datetime import datetime
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import stat
|
|
import zipfile
|
|
|
|
BASE = {"started.json", "finished.json", "generation.json", "generation.stdout.log",
|
|
"generation.stderr.log", "app.stdout.log", "app.stderr.log", "processes.json", "history-bytes.json"}
|
|
GROUPS = [["baseline", "move-undo", "before-hide", "hide-undo"], ["move", "move-redo"],
|
|
["hidden", "hide-redo", "reopened-hidden"]]
|
|
MODES = {name: mode for group, mode in zip(GROUPS, ["unchanged", "moved", "hide"]) for name in group}
|
|
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 in name and ":" not in name
|
|
and not PurePosixPath(name).is_absolute() and all(ord(c) >= 32 for c in name)
|
|
and all(part not in ("", ".", "..") for part in name.split("/")))
|
|
|
|
|
|
def parsed(data):
|
|
def unique(items):
|
|
result = {}
|
|
for key, value in items:
|
|
need(key not in result, "duplicate JSON key: " + key)
|
|
result[key] = value
|
|
return result
|
|
return json.loads(data, object_pairs_hook=unique)
|
|
|
|
|
|
def norm(path):
|
|
return path.replace("\\", "/").rstrip("/")
|
|
|
|
|
|
def relative(path, root):
|
|
prefix = norm(root) + "/"
|
|
need(norm(path).startswith(prefix), "recorded path outside owner: " + path)
|
|
value = norm(path)[len(prefix):]
|
|
need(safe(value), "unsafe recorded relative path")
|
|
return value
|
|
|
|
|
|
def ordered(*values):
|
|
times = [datetime.fromisoformat(value) for value in values]
|
|
return all(t.utcoffset() is not None for t in times) and times == sorted(times)
|
|
|
|
|
|
def zip_inventory(archive, members):
|
|
names = archive.namelist()
|
|
need(len(names) == len(set(names)) and set(names) == set(members), "exact archive members")
|
|
for info in archive.infolist():
|
|
need(safe(info.filename) and not info.is_dir() and not stat.S_ISLNK(info.external_attr >> 16), "unsafe archive member")
|
|
need(info.file_size == members[info.filename]["bytes"], "archive member size")
|
|
|
|
|
|
def verify_data(manifest, read):
|
|
need(manifest["schema"] == 1 and manifest["kind"] == "multiselect-native", "manifest kind")
|
|
members, runs = manifest["members"], manifest["runs"]
|
|
need(bool(runs) and all(safe(run) and "/" not in run for run in runs), "run names")
|
|
need(all(safe(name) for name in members), "safe member names")
|
|
for name, expected in members.items():
|
|
need(pin(read(name)) == expected, "member pin: " + name)
|
|
used = set()
|
|
def raw(name):
|
|
need(name in members, "missing member: " + name)
|
|
used.add(name)
|
|
return read(name)
|
|
def obj(name):
|
|
return parsed(raw(name))
|
|
def tree(prefix):
|
|
return {name[len(prefix):]: raw(name) for name in members if name.startswith(prefix)}
|
|
def pinned_tree(prefix, expected):
|
|
actual = tree(prefix)
|
|
need({name: pin(data) for name, data in actual.items()} == expected, "exact pinned tree: " + prefix)
|
|
return actual
|
|
for run, spec in runs.items():
|
|
prefix = "runs/" + run + "/raw/"
|
|
start, end = obj(prefix + "started.json"), obj(prefix + "finished.json")
|
|
generation, process = obj(prefix + "generation.json"), obj(prefix + "processes.json")
|
|
history = obj(prefix + "history-bytes.json")
|
|
need(end["exitCode"] == 0 and all(end[key] is True for key in
|
|
("guiUnchanged", "checkerUnchanged", "fixtureUnchanged")), run + ": clean unchanged closure")
|
|
need(ordered(generation["start"], generation["finish"], process["CreationDate"], start["start"], end["finish"], history["verified"]), run + ": process/generation times")
|
|
need(process["ProcessId"] == start["pid"] and type(start["pid"]) is int and start["pid"] > 0, run + ": process identity")
|
|
need(norm(process["ExecutablePath"]) == norm(start["gui"]["binary"])
|
|
and start["command"] == [start["gui"]["binary"]], run + ": GUI command")
|
|
need(spec["originalRun"] == norm(start["cwd"]) + "/.local/state-multiselect-native/" + run, run + ": original run mapping")
|
|
need(generation["exitCode"] == 0 and generation["checker"] == start["checker"], run + ": generation checker")
|
|
need(generation["command"][:2] == [start["checker"]["binary"], "generate"]
|
|
and len(generation["command"]) == 3 and norm(generation["command"][2]) == spec["originalFixture"], run + ": generation command")
|
|
for stream in ("stdout", "stderr"):
|
|
need(end[stream] == pin(raw(prefix + "app." + stream + ".log")), run + ": app stream")
|
|
need(generation[stream] == pin(raw(prefix + "generation." + stream + ".log")), run + ": generation stream")
|
|
fixture = pinned_tree("runs/" + run + "/fixture/", start["fixture"])
|
|
need(set(fixture) == {"baseline.json", "manifest.json"} | {"source/" + name for name in
|
|
obj(prefix + "snapshots/baseline/result.json")["source"]}, run + ": fixture membership")
|
|
fixture_manifest = parsed(fixture["manifest.json"])
|
|
need(fixture_manifest["fixture"] == "state-multiselect-native-20260923"
|
|
and fixture_manifest["source_entry"] == "project.arcadia", run + ": fixture identity")
|
|
need(parsed(fixture["baseline.json"])["id"] == fixture_manifest["ids"]["project"], run + ": baseline identity")
|
|
for key, expected in [("helper", start["helper"]), ("capture", start["capture"]),
|
|
("snapshotHelper", start["snapshotHelper"]), ("history", history["helper"])]:
|
|
need(pin(raw(spec["helpers"][key])) == expected, run + ": helper " + key)
|
|
builds = {}
|
|
for kind, target in [("gui", "arcadia-studio"), ("checker", "state_multiselect_native")]:
|
|
recorded = start[kind]
|
|
build = obj(spec["builds"][kind]["receipt"])
|
|
need(pin(raw(spec["builds"][kind]["receipt"])) == recorded["receiptPin"], run + ": build receipt pin")
|
|
need(relative(recorded["receipt"], start["cwd"]) == spec["builds"][kind]["originalRelative"], run + ": build path mapping")
|
|
need(build["exitCode"] == 0 and build["inputsUnchanged"] is True
|
|
and ordered(build["start"], build["finish"], generation["start"]), run + ": accepted preceding build")
|
|
need(build["inputs"] == recorded["inputs"] == pin(raw(spec["builds"][kind]["inputs"])), run + ": build input map")
|
|
need(pin(raw(spec["builds"][kind]["helper"])) == build["helper"], run + ": build helper pin")
|
|
inputs = obj(spec["builds"][kind]["inputs"])
|
|
need(len(inputs) == build["inputCount"] and sum(name.startswith("crates/vendor/iced_widget/") for name in inputs) == 71, run + ": build input count")
|
|
artifacts = [item for item in build["artifacts"] if item["cargo"]["target"]["name"] == target]
|
|
need(len(artifacts) == 1 and artifacts[0]["readback"] == recorded["pin"]
|
|
and artifacts[0]["cargo"]["executable"] == recorded["binary"], run + ": executable relationship")
|
|
builds[kind] = inputs
|
|
need(start["gui"]["inputs"] == start["checker"]["inputs"], run + ": matching GUI/checker inputs")
|
|
source = spec["checkerSource"]
|
|
need(source["inputPath"] == "crates/oa-core/examples/state_multiselect_native.rs"
|
|
and pin(raw(source["member"])) == builds["checker"][source["inputPath"]], run + ": archived exact checker source")
|
|
|
|
top = {name[len(prefix):] for name in members if name.startswith(prefix) and "/" not in name[len(prefix):]}
|
|
captures = sorted(name for name in top - BASE if name.endswith(".json"))
|
|
need(bool(captures), run + ": missing captures")
|
|
expected_top = set(BASE) | set(captures)
|
|
captured_window = None
|
|
for name in captures:
|
|
capture = obj(prefix + name)
|
|
need(ordered(start["start"], capture["time"], end["finish"]), run + ": capture lifetime")
|
|
window = capture["window"]
|
|
app = window["app"]
|
|
need(isinstance(app, str) and app.startswith("process:")
|
|
and norm(app[len("process:"):]) == norm(start["gui"]["binary"]), run + ": captured executable relationship")
|
|
need(type(window["id"]) is int and window["id"] > 0, run + ": captured window identity")
|
|
if captured_window is None:
|
|
captured_window = window["id"]
|
|
need(window["id"] == captured_window, run + ": inconsistent captured window identity")
|
|
need(isinstance(capture["images"], list), run + ": capture images")
|
|
for image in capture["images"]:
|
|
filename = image["file"]
|
|
need(safe(filename) and "/" not in filename and filename.startswith(name[:-5] + "-")
|
|
and filename.endswith((".jpg", ".png")), run + ": capture image mapping")
|
|
need(type(image["width"]) is int and image["width"] > 0
|
|
and type(image["height"]) is int and image["height"] > 0, run + ": image dimensions")
|
|
need(filename not in expected_top, run + ": duplicate image reference")
|
|
expected_top.add(filename)
|
|
raw(prefix + filename)
|
|
need(top == expected_top, run + ": extra/missing top-level run files")
|
|
|
|
snapshots = {name[len(prefix + "snapshots/"):].split("/")[0] for name in members if name.startswith(prefix + "snapshots/")}
|
|
extras = spec["extraUnchangedSnapshots"]
|
|
need(isinstance(extras, list) and len(extras) == len(set(extras))
|
|
and all(safe(label) and "/" not in label and label not in MODES for label in extras), run + ": explicit extra unchanged labels")
|
|
modes = dict(MODES, **{label: "unchanged" for label in extras})
|
|
need(snapshots == set(modes), run + ": exact nine history plus declared unchanged snapshot labels")
|
|
states = {}
|
|
for label, mode in modes.items():
|
|
folder = prefix + "snapshots/" + label + "/"
|
|
began, result = obj(folder + "started.json"), obj(folder + "result.json")
|
|
need(all(result.get(key) == value for key, value in began.items()), label + ": start/result")
|
|
need(result["exitCode"] == 0 and all(result[key] is True for key in
|
|
("binaryUnchanged", "workingSourceUnchanged", "baselineUnchanged")), label + ": unchanged successful checker")
|
|
need(ordered(start["start"], result["start"], result["finish"], end["finish"]), label + ": checker lifetime")
|
|
need(result["buildReceipt"] == start["checker"]["receiptPin"] and result["binary"] == start["checker"]["pin"]
|
|
and result["helper"] == start["snapshotHelper"], label + ": checker provenance")
|
|
need(result["baseline"] == pin(fixture["baseline.json"]), label + ": baseline pin")
|
|
command = result["command"]
|
|
need(len(command) == 5 and command[:3] == [start["checker"]["binary"], "check", mode]
|
|
and norm(command[3]) == spec["originalFixture"] + "/baseline.json"
|
|
and norm(command[4]) == spec["originalRun"] + "/snapshots/" + label + "/source/project.arcadia", label + ": exact checker command")
|
|
for stream in ("stdout", "stderr"):
|
|
need(result[stream] == pin(raw(folder + stream + ".log")), label + ": stream pin")
|
|
need(not raw(folder + "stderr.log"), label + ": checker stderr")
|
|
report = obj(folder + "stdout.log")
|
|
need(report["result"] == "PASS" and report["comparison"] == "complete Project equality"
|
|
and report["mode"] == mode and report["baseline"] == command[3] and report["saved"] == command[4], label + ": positive checker report")
|
|
delta = report["inferred_common_delta"]
|
|
if mode == "moved":
|
|
need(isinstance(delta, dict) and set(delta) == {"x", "y"}
|
|
and all(type(v) in (int, float) and math.isfinite(v) and abs(v) <= 300 for v in delta.values())
|
|
and math.hypot(delta["x"], delta["y"]) >= 0.01, label + ": inferred State movement bounds")
|
|
else:
|
|
need(delta is None, label + ": unexpected inferred delta")
|
|
states[label] = pinned_tree(folder + "source/", result["source"])
|
|
need(len(states[label]) == history["sourceFilesPerSnapshot"] == len(fixture) - 2, label + ": source count")
|
|
expected = {folder + name for name in ("started.json", "result.json", "stdout.log", "stderr.log")}
|
|
expected.update(folder + "source/" + name for name in states[label])
|
|
need({name for name in members if name.startswith(folder)} == expected, label + ": extra snapshot files")
|
|
need(all(states[label] == states["baseline"] for label in extras), run + ": extra snapshots equal baseline bytes")
|
|
need(history["status"] == "PASS" and history["equalByteGroups"] == GROUPS, run + ": history groups")
|
|
for group in GROUPS:
|
|
need(all(states[label] == states[group[0]] for label in group), run + ": actual history bytes " + str(group))
|
|
need(all(states[a[0]] != states[b[0]] for i, a in enumerate(GROUPS) for b in GROUPS[i + 1:]), run + ": distinct history states")
|
|
need(states["baseline"] == {name[7:]: data for name, data in fixture.items() if name.startswith("source/")}, run + ": baseline source exact")
|
|
changed = {label: sorted(name for name in set(states["baseline"]) | set(states[label])
|
|
if states["baseline"].get(name) != states[label].get(name)) for label in ("move", "hidden")}
|
|
need(changed == history["changedFilesFromBaseline"]
|
|
== {label: ["views/diagram_state_view.arcadia"] for label in ("move", "hidden")}, run + ": changed source files")
|
|
print(run, "closed exit 0;", len(captures), "captures;", len(modes), "positive snapshots; exact three-group source history")
|
|
need(used == set(members), "unreferenced/extra package members")
|
|
return {"runs": len(runs), "members": len(members)}
|
|
|
|
|
|
def verify_package(root):
|
|
files = parsed((root / "files.json").read_bytes())
|
|
need(set(files) == OUTER, "outer files")
|
|
for name, expected in files.items():
|
|
need(pin((root / name).read_bytes()) == expected, "outer pin: " + name)
|
|
manifest = parsed((root / "manifest.json").read_bytes())
|
|
with zipfile.ZipFile(root / "evidence.zip") as archive:
|
|
zip_inventory(archive, manifest["members"])
|
|
return verify_data(manifest, archive.read)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("package", nargs="?", type=Path, default=Path(__file__).resolve().parent)
|
|
args = parser.parse_args()
|
|
print("PASS native evidence byte/receipt consistency:", verify_package(args.package))
|
|
print("No GUI/checker execution; actions and native acceptance require the separate parent review.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|