f4bbfb229d
measure.py run now checks a result first and writes nothing on a finding, where it used to overwrite result.json and only then exit 1. check takes the rule from rule.json, the archive and split from corpus.json, and each dictionary's length and digest from dictionaries.tar.xz, which is new, so a rule, split or dictionary length rewritten inside a result is refused. The page says the verdict is this workload's, with the object counts at which each family would reach 10 %, and that the median object saves at 64 and 112 KiB in every family but records. Co-Authored-By: Claude Opus 5.5 <[email protected]>
503 lines
22 KiB
Python
503 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""The dictionary measurement of research/text-dictionary.md, and the check of its record.
|
|
|
|
python3 measure.py run [--out FILE] # needs compression.zstd: Python 3.14 or later
|
|
python3 measure.py check FILE [TEXT.md]
|
|
python3 measure.py selftest FILE
|
|
|
|
`run` reads corpus.tar.xz (each object against corpus.json's digest) and rule.json, trains
|
|
each family's dictionaries on its train split, codes each held-out object plain and with
|
|
each dictionary at each level of the rule, and decodes every dictionary form back. It then
|
|
checks the result, and only a result with no finding is written, with dictionaries.tar.xz
|
|
beside it: on a finding it exits 1 and the files stay as they were, so a total that leaves
|
|
the dictionary's bytes out never reaches result.json.
|
|
|
|
`check` needs no Zstandard. It holds the result to what the result cannot state for itself:
|
|
rule.json, corpus.json's archive and split, and each dictionary's length and digest in
|
|
dictionaries.tar.xz. From the per-object lengths alone it then recomputes every total, each
|
|
by a formula of its own, the verdicts by the rule, and each figure TEXT.md quotes (each is
|
|
followed by a `txd` marker naming its place in the result). The per-object lengths
|
|
themselves are `run`'s: only `run` codes an object. `selftest` changes copies of the result
|
|
and requires `check` to refuse each, and `run`'s last step to leave its file alone. Exit 0
|
|
when everything holds, 1 with one line a finding.
|
|
"""
|
|
import contextlib
|
|
import copy
|
|
import functools
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import lzma
|
|
import math
|
|
import os
|
|
import pathlib
|
|
import platform
|
|
import re
|
|
import statistics
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
FAMILIES = ("notes", "spec", "code", "records")
|
|
DICTIONARIES = "dictionaries.tar.xz"
|
|
ZSTD_DICTIONARY_MAGIC = bytes.fromhex("37a430ec")
|
|
|
|
|
|
def sha(b):
|
|
return hashlib.sha256(b).hexdigest()
|
|
|
|
|
|
def pin_bytes():
|
|
"""The manifest bytes of a minimal `prior` (SPEC 3.1: `id`, `sha256` and `determinism` are
|
|
required), with the comma that joins it to `conditioning`'s `kind`."""
|
|
return len(json.dumps({"id": "d", "sha256": "0" * 64, "determinism": "z"}, separators=(",", ":"))) + len(',"prior":')
|
|
|
|
|
|
def charge(plain, coded, pin, dictionary):
|
|
"""What one sink moves for its objects, by prior-carriage 17.2.
|
|
|
|
An object with a saving takes its dictionary form; the others stay plain. The dictionary's
|
|
bytes are charged once, in full.
|
|
"""
|
|
savings = [max(p - (c + pin), 0) for p, c in zip(plain, coded)]
|
|
saving = sum(savings)
|
|
plain_total = sum(plain)
|
|
return {
|
|
"plain_total": plain_total,
|
|
"saving": saving,
|
|
"dictionary_form_objects": sum(1 for s in savings if s > 0),
|
|
"total": plain_total - saving + dictionary,
|
|
"net": saving - dictionary,
|
|
"median_saving": statistics.median(savings),
|
|
}
|
|
|
|
|
|
def cell(level, size, digest, dictionary, plain, coded, pin, rule):
|
|
c = {"level": level, "size_asked": size, "dictionary_bytes": dictionary, "dictionary_sha256": digest, "coded": coded}
|
|
c.update(charge(plain, coded, pin, dictionary))
|
|
c["net_fraction"] = c["net"] / c["plain_total"]
|
|
c["net_fraction_no_pin"] = charge(plain, coded, 0, dictionary)["net"] / c["plain_total"]
|
|
mean = c["saving"] / len(plain)
|
|
c["break_even_objects"] = dictionary / mean if mean > 0 else None
|
|
c["pass"] = c["net_fraction"] >= rule["net_fraction_min"] and c["median_saving"] > 0
|
|
# A longer workload, which the rule does not read: the dictionary is charged once, so with
|
|
# objects of the held-out mean the net tends to `saving_fraction` as they grow in number.
|
|
c["saving_fraction"] = c["saving"] / c["plain_total"]
|
|
room = mean - rule["net_fraction_min"] * c["plain_total"] / len(plain)
|
|
c["objects_to_reach_min"] = math.ceil(dictionary / room) if room > 0 else None
|
|
return c
|
|
|
|
|
|
def verdict(family_doc, rule):
|
|
levels = {}
|
|
for level in rule["levels"]:
|
|
levels[str(level)] = [c["dictionary_bytes"] for c in family_doc["cells"] if c["level"] == level and c["pass"]]
|
|
return {"passing_sizes": levels, "go": all(levels.values())}
|
|
|
|
|
|
def unpack(raw):
|
|
bodies = {}
|
|
with tarfile.open(fileobj=io.BytesIO(lzma.decompress(raw)), mode="r:") as tar:
|
|
for m in tar.getmembers():
|
|
bodies[m.name] = tar.extractfile(m).read()
|
|
return bodies
|
|
|
|
|
|
def pack(bodies):
|
|
"""One archive, the same bytes for the same members, as corpus.tar.xz is packed."""
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w", format=tarfile.USTAR_FORMAT) as tar:
|
|
for name in sorted(bodies):
|
|
ti = tarfile.TarInfo(name)
|
|
ti.size = len(bodies[name])
|
|
tar.addfile(ti, io.BytesIO(bodies[name]))
|
|
return lzma.compress(buf.getvalue(), preset=9 | lzma.PRESET_EXTREME)
|
|
|
|
|
|
def corpus():
|
|
doc = json.loads((HERE / "corpus.json").read_text())
|
|
raw = (HERE / "corpus.tar.xz").read_bytes()
|
|
if sha(raw) != doc["archive_sha256"]:
|
|
raise SystemExit("corpus.tar.xz is not the archive corpus.json names")
|
|
bodies = unpack(raw)
|
|
for e in doc["objects"]:
|
|
body = bodies[e["name"]]
|
|
if sha(body) != e["sha256"] or len(body) != e["bytes"]:
|
|
raise SystemExit(f"{e['name']}: not the object corpus.json names")
|
|
return doc, bodies
|
|
|
|
|
|
@functools.cache
|
|
def committed():
|
|
"""What a result is held to and cannot state for itself: the rule, the corpus and its split."""
|
|
return json.loads((HERE / "rule.json").read_text()), corpus()[0]
|
|
|
|
|
|
@functools.cache
|
|
def committed_dictionaries():
|
|
return unpack((HERE / DICTIONARIES).read_bytes())
|
|
|
|
|
|
def dictionary_name(group, size, level):
|
|
return f"{group}/{size}-{level}"
|
|
|
|
|
|
def split_after_the_run(entries):
|
|
"""§6's split: `records` by object, so a held-out record's near-duplicates train."""
|
|
return [dict(e, split="held-out" if int(e["sha256"][:8], 16) % 4 == 0 else "train") for e in entries]
|
|
|
|
|
|
def entries_of(doc, group):
|
|
"""A family's objects, or an after-the-run group's, each on its side of the split."""
|
|
if group == "records_by_object":
|
|
return split_after_the_run([e for e in doc["objects"] if e["family"] == "records"])
|
|
return [e for e in doc["objects"] if e["family"] == group]
|
|
|
|
|
|
def publish(result, dictionaries, out):
|
|
"""`run`'s last step. The check comes first: nothing is written unless it finds nothing."""
|
|
findings = check(result, dictionaries=dictionaries)
|
|
for f in findings:
|
|
print(f, file=sys.stderr)
|
|
if findings:
|
|
return 1
|
|
files = ((out.with_name(DICTIONARIES), pack(dictionaries)),
|
|
(out, (json.dumps(result, indent=0, sort_keys=True) + "\n").encode()))
|
|
for path, raw in files:
|
|
spare = path.with_name(path.name + ".new")
|
|
spare.write_bytes(raw)
|
|
os.replace(spare, path)
|
|
print(f"wrote {out}; GO for {result['go'] or 'no family'}")
|
|
return 0
|
|
|
|
|
|
def run(out):
|
|
from compression import zstd
|
|
|
|
rule = json.loads((HERE / "rule.json").read_text())
|
|
doc, bodies = corpus()
|
|
pin = pin_bytes()
|
|
dictionaries = {}
|
|
P = zstd.CompressionParameter
|
|
|
|
def options(level):
|
|
return {P.compression_level: level, P.checksum_flag: 0, P.dict_id_flag: 0}
|
|
|
|
def measure(group):
|
|
entries = entries_of(doc, group)
|
|
train = [bodies[e["name"]] for e in entries if e["split"] == "train"]
|
|
held = [e for e in entries if e["split"] == "held-out"]
|
|
data = [bodies[e["name"]] for e in held]
|
|
plain, solid = {}, {}
|
|
for level in rule["levels"]:
|
|
comp = zstd.ZstdCompressor(options=options(level))
|
|
plain[str(level)] = [len(comp.compress(b, mode=comp.FLUSH_FRAME)) for b in data]
|
|
solid[str(level)] = len(zstd.ZstdCompressor(options=options(level)).compress(b"".join(data), mode=comp.FLUSH_FRAME))
|
|
cells = []
|
|
for size in rule["dictionary_bytes"]:
|
|
base = zstd.train_dict(train, size)
|
|
for level in rule["levels"]:
|
|
d = zstd.finalize_dict(base, train, size, level)
|
|
comp = zstd.ZstdCompressor(options=options(level), zstd_dict=d)
|
|
frames = [comp.compress(b, mode=comp.FLUSH_FRAME) for b in data]
|
|
for b, f in zip(data, frames):
|
|
if zstd.decompress(f, zstd_dict=d) != b:
|
|
raise SystemExit("a dictionary form does not decode to its object")
|
|
dictionaries[dictionary_name(group, size, level)] = d.dict_content
|
|
cells.append(cell(level, size, sha(d.dict_content), len(d.dict_content), plain[str(level)],
|
|
[len(f) for f in frames], pin, rule))
|
|
f = {"objects_train": len(train), "bytes_train": sum(map(len, train)),
|
|
"objects_held_out": len(data), "bytes_held_out": sum(map(len, data)),
|
|
"held_out": [e["name"] for e in held], "plain": plain, "solid": solid, "cells": cells}
|
|
f["verdict"] = verdict(f, rule)
|
|
return f
|
|
|
|
families = {fam: measure(fam) for fam in FAMILIES}
|
|
after = {"records_by_object": measure("records_by_object")}
|
|
result = {"format": "wai.research.text-dictionary/1", "corpus_sha256": doc["archive_sha256"], "rule": rule,
|
|
"pin_bytes": pin, "tool": {"python": platform.python_version(), "libzstd": zstd.zstd_version},
|
|
"families": families, "go": [f for f in FAMILIES if families[f]["verdict"]["go"]],
|
|
"after_the_run": after}
|
|
return publish(result, dictionaries, out)
|
|
|
|
|
|
def check(r, text=None, dictionaries=None):
|
|
"""Every total and verdict, recomputed from the per-object lengths, under the committed
|
|
rule, over the committed split, with the committed dictionaries' lengths."""
|
|
bad = []
|
|
rule, doc = committed()
|
|
if dictionaries is None:
|
|
dictionaries = committed_dictionaries()
|
|
pin = r["pin_bytes"]
|
|
if r["rule"] != rule:
|
|
bad.append("rule: the result's rule is not rule.json")
|
|
if r["corpus_sha256"] != doc["archive_sha256"]:
|
|
bad.append(f"corpus_sha256 is {r['corpus_sha256']}, corpus.json names {doc['archive_sha256']}")
|
|
if pin != len(',"prior":{"id":"d","sha256":"","determinism":"z"}') + 64:
|
|
bad.append(f"pin_bytes {pin}: not a minimal prior's length")
|
|
if set(r["after_the_run"]) != {"records_by_object"}:
|
|
bad.append(f"after_the_run holds {sorted(r['after_the_run'])}, not the one reading §6 names")
|
|
want_go = []
|
|
named = set()
|
|
groups = [(fam, r["families"][fam]) for fam in FAMILIES] + list(r["after_the_run"].items())
|
|
for fam, f in groups:
|
|
n = f["objects_held_out"]
|
|
entries = entries_of(doc, fam)
|
|
train = [e for e in entries if e["split"] == "train"]
|
|
held = [e for e in entries if e["split"] == "held-out"]
|
|
for key, value in (("objects_train", len(train)), ("bytes_train", sum(e["bytes"] for e in train)),
|
|
("objects_held_out", len(held)), ("bytes_held_out", sum(e["bytes"] for e in held))):
|
|
if f[key] != value:
|
|
bad.append(f"{fam}: {key} is {f[key]}, corpus.json gives {value}")
|
|
if f["held_out"] != [e["name"] for e in held]:
|
|
bad.append(f"{fam}: its held-out objects are not corpus.json's split")
|
|
seen = set()
|
|
for c in f["cells"]:
|
|
where = f"{fam} level {c['level']} dictionary {c['dictionary_bytes']}"
|
|
seen.add((c["level"], c["size_asked"]))
|
|
plain = f["plain"][str(c["level"])]
|
|
coded = c["coded"]
|
|
if len(plain) != n or len(coded) != n:
|
|
bad.append(f"{where}: {len(coded)} readings for {n} objects")
|
|
continue
|
|
if c["dictionary_bytes"] <= 0 or c["dictionary_bytes"] > c["size_asked"]:
|
|
bad.append(f"{where}: a dictionary of {c['dictionary_bytes']} bytes for {c['size_asked']} asked")
|
|
name = dictionary_name(fam, c["size_asked"], c["level"])
|
|
named.add(name)
|
|
body = dictionaries.get(name)
|
|
if body is None:
|
|
bad.append(f"{where}: no dictionary {name} in {DICTIONARIES}")
|
|
elif len(body) != c["dictionary_bytes"] or sha(body) != c["dictionary_sha256"]:
|
|
bad.append(f"{where}: {DICTIONARIES} holds {len(body)} bytes for it, with another digest or length")
|
|
elif body[:4] != ZSTD_DICTIONARY_MAGIC:
|
|
bad.append(f"{where}: {name} is not a Zstandard dictionary")
|
|
# What moves, object by object: the smaller form, then the dictionary, whole.
|
|
moved = 0
|
|
per_object = []
|
|
taken = 0
|
|
for p, k in zip(plain, coded):
|
|
cost = k + pin
|
|
if cost < p:
|
|
moved += cost
|
|
taken += 1
|
|
per_object.append(p - cost)
|
|
else:
|
|
moved += p
|
|
per_object.append(0)
|
|
total = moved + c["dictionary_bytes"]
|
|
base = sum(plain)
|
|
per_object.sort()
|
|
mid = len(per_object) // 2
|
|
median = per_object[mid] if len(per_object) % 2 else (per_object[mid - 1] + per_object[mid]) / 2
|
|
net = base - total
|
|
for key, value in (("plain_total", base), ("total", total), ("net", net), ("saving", base - moved),
|
|
("dictionary_form_objects", taken), ("median_saving", median)):
|
|
if c[key] != value:
|
|
bad.append(f"{where}: {key} is {c[key]}, the readings give {value}")
|
|
frac = net / base
|
|
if abs(c["net_fraction"] - frac) > 1e-12:
|
|
bad.append(f"{where}: net_fraction is {c['net_fraction']}, the readings give {frac}")
|
|
no_pin = (sum(min(p, k) for p, k in zip(plain, coded)) + c["dictionary_bytes"])
|
|
if abs(c["net_fraction_no_pin"] - (base - no_pin) / base) > 1e-12:
|
|
bad.append(f"{where}: net_fraction_no_pin is {c['net_fraction_no_pin']}, the readings give {(base - no_pin) / base}")
|
|
be = c["dictionary_bytes"] * n / (base - moved) if base > moved else None
|
|
if (be is None) != (c["break_even_objects"] is None) or (be is not None and abs(be - c["break_even_objects"]) > 1e-9):
|
|
bad.append(f"{where}: break_even_objects is {c['break_even_objects']}, the readings give {be}")
|
|
if abs(c["saving_fraction"] - (base - moved) / base) > 1e-12:
|
|
bad.append(f"{where}: saving_fraction is {c['saving_fraction']}, the readings give {(base - moved) / base}")
|
|
# The fewest objects of the held-out mean whose net is the rule's bar of their plain
|
|
# bytes: k objects save k/n of (base - moved) and cost one dictionary.
|
|
spare = (base - moved) - rule["net_fraction_min"] * base
|
|
reach = math.ceil(c["dictionary_bytes"] * n / spare) if spare > 0 else None
|
|
if c["objects_to_reach_min"] != reach:
|
|
bad.append(f"{where}: objects_to_reach_min is {c['objects_to_reach_min']}, the readings give {reach}")
|
|
passes = frac >= rule["net_fraction_min"] and median > 0
|
|
if c["pass"] != passes:
|
|
bad.append(f"{where}: pass is {c['pass']}, the rule gives {passes}")
|
|
want = {(lv, s) for lv in rule["levels"] for s in rule["dictionary_bytes"]}
|
|
if seen != want:
|
|
bad.append(f"{fam}: cells {sorted(seen)}, the rule asks {sorted(want)}")
|
|
by_level = {str(lv): [c["dictionary_bytes"] for c in f["cells"] if c["level"] == lv and c["pass"]] for lv in rule["levels"]}
|
|
go = all(by_level.values())
|
|
if f["verdict"] != {"passing_sizes": by_level, "go": go}:
|
|
bad.append(f"{fam}: verdict is {f['verdict']}, the rule gives {by_level} and {go}")
|
|
if go and fam in FAMILIES:
|
|
want_go.append(fam)
|
|
if r["go"] != want_go:
|
|
bad.append(f"go is {r['go']}, the rule gives {want_go}")
|
|
if set(dictionaries) != named:
|
|
bad.append(f"{DICTIONARIES} and the result's cells do not name the same dictionaries")
|
|
if text is not None:
|
|
bad += check_text(r, text)
|
|
return bad
|
|
|
|
|
|
MARK = re.compile(r"([-+]?[0-9][0-9,]*(?:\.[0-9]+)?)\s*(%|bytes|KiB|objects)?<!--txd:([^>]+)-->")
|
|
|
|
|
|
def value_at(r, path):
|
|
"""A marker's place: dots between members, `cell:<family>:<level>:<dictionary size asked>` for a cell."""
|
|
parts = path.split(".")
|
|
if parts[0].startswith("cell:"):
|
|
_, fam, level, size = parts[0].split(":")
|
|
cells = (r["families"].get(fam) or r["after_the_run"][fam])["cells"]
|
|
node = next(c for c in cells if c["level"] == int(level) and c["size_asked"] == int(size))
|
|
parts = parts[1:]
|
|
else:
|
|
node = r
|
|
for p in parts:
|
|
node = node[int(p)] if isinstance(node, list) else node[p]
|
|
return node
|
|
|
|
|
|
def check_text(r, text):
|
|
bad = []
|
|
n = 0
|
|
for m in MARK.finditer(text):
|
|
n += 1
|
|
printed, unit, path = m.group(1), m.group(2), m.group(3)
|
|
try:
|
|
want = value_at(r, path)
|
|
except (KeyError, IndexError, StopIteration, ValueError) as e:
|
|
bad.append(f"text: marker {path}: no such place ({e!r})")
|
|
continue
|
|
got = float(printed.replace(",", ""))
|
|
if unit == "%":
|
|
want = want * 100
|
|
places = len(printed.split(".")[1]) if "." in printed else 0
|
|
if round(float(want), places) != got:
|
|
bad.append(f"text: {printed} at {path}: the result gives {want}")
|
|
if "<!--txd:" in text and n != text.count("<!--txd:"):
|
|
bad.append(f"text: {text.count('<!--txd:')} markers, {n} follow a figure")
|
|
return bad
|
|
|
|
|
|
def selftest(r):
|
|
"""Each change must be refused, for the reason it is planted for."""
|
|
fails = []
|
|
rule, doc = committed()
|
|
|
|
def expect(name, change, says=""):
|
|
m = copy.deepcopy(r)
|
|
change(m)
|
|
found = check(m)
|
|
if not any(says in f for f in found):
|
|
fails.append(f"selftest: {name} was not refused" + (f" ({says})" if found else ""))
|
|
|
|
def groups(m):
|
|
return list(m["families"].items()) + list(m["after_the_run"].items())
|
|
|
|
def recount(m, use, length=None):
|
|
"""Every cell and verdict written again, consistently, under the rule `use`."""
|
|
for _, f in groups(m):
|
|
f["cells"] = [cell(c["level"], c["size_asked"], c["dictionary_sha256"], length or c["dictionary_bytes"],
|
|
f["plain"][str(c["level"])], c["coded"], m["pin_bytes"], use) for c in f["cells"]]
|
|
f["verdict"] = verdict(f, use)
|
|
m["go"] = [fam for fam in FAMILIES if m["families"][fam]["verdict"]["go"]]
|
|
|
|
def leave_out_dictionary(m):
|
|
for fam in FAMILIES:
|
|
for c in m["families"][fam]["cells"]:
|
|
c["total"] -= c["dictionary_bytes"]
|
|
c["net"] += c["dictionary_bytes"]
|
|
c["net_fraction"] = c["net"] / c["plain_total"]
|
|
|
|
def shrink(m):
|
|
c = m["families"]["records"]["cells"][0]
|
|
c["coded"][0] -= 1
|
|
|
|
def no_pin(m):
|
|
m["pin_bytes"] = 0
|
|
|
|
def claim_go(m):
|
|
m["go"] = list(FAMILIES)
|
|
|
|
def flip(m):
|
|
c = m["families"]["spec"]["cells"][0]
|
|
c["pass"] = not c["pass"]
|
|
|
|
def one_byte_dictionaries(m):
|
|
recount(m, rule, length=1)
|
|
|
|
def easier_rule(m):
|
|
m["rule"] = dict(rule, net_fraction_min=0.05)
|
|
recount(m, m["rule"])
|
|
|
|
def one_level_at_two_percent(m):
|
|
m["rule"] = dict(rule, levels=[19], net_fraction_min=0.02)
|
|
for _, f in groups(m):
|
|
f["cells"] = [c for c in f["cells"] if c["level"] == 19]
|
|
f["plain"] = {"19": f["plain"]["19"]}
|
|
f["solid"] = {"19": f["solid"]["19"]}
|
|
recount(m, m["rule"])
|
|
|
|
def another_corpus(m):
|
|
m["corpus_sha256"] = "0" * 64
|
|
|
|
def another_split(m):
|
|
f = m["families"]["notes"]
|
|
f["held_out"] = [e["name"] for e in doc["objects"] if e["family"] == "notes" and e["split"] == "train"][:f["objects_held_out"]]
|
|
|
|
def another_train_size(m):
|
|
m["families"]["code"]["bytes_train"] += 1
|
|
|
|
def another_digest(m):
|
|
m["families"]["spec"]["cells"][0]["dictionary_sha256"] = "0" * 64
|
|
|
|
expect("a total without the dictionary's bytes", leave_out_dictionary, "the readings give")
|
|
expect("a changed reading", shrink, "the readings give")
|
|
expect("a pin of no bytes", no_pin, "pin_bytes")
|
|
expect("a verdict the rule does not give", claim_go, "the rule gives")
|
|
expect("a cell's verdict flipped", flip, "the rule gives")
|
|
expect("dictionaries recorded as one byte, every total recounted", one_byte_dictionaries, DICTIONARIES)
|
|
expect("a rule lowered to 5 %, every verdict recounted", easier_rule, "rule.json")
|
|
expect("a rule of one level at 2 %, every verdict recounted", one_level_at_two_percent, "rule.json")
|
|
expect("a corpus digest that is not corpus.json's", another_corpus, "corpus_sha256")
|
|
expect("held-out objects that are not the split's", another_split, "corpus.json's split")
|
|
expect("a train size that is not the corpus's", another_train_size, "corpus.json gives")
|
|
expect("a dictionary digest that is not the archive's", another_digest, DICTIONARIES)
|
|
|
|
# `run`'s last step: a result that fails the check must leave what is there as it was.
|
|
bad = copy.deepcopy(r)
|
|
leave_out_dictionary(bad)
|
|
with tempfile.TemporaryDirectory() as d:
|
|
out = pathlib.Path(d) / "result.json"
|
|
out.write_text("the result before\n")
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
status = publish(bad, committed_dictionaries(), out)
|
|
if status != 1:
|
|
fails.append(f"selftest: run's last step gave {status} on a result that fails")
|
|
if out.read_text() != "the result before\n" or sorted(p.name for p in pathlib.Path(d).iterdir()) != ["result.json"]:
|
|
fails.append("selftest: a result that fails the check was written")
|
|
with contextlib.redirect_stdout(io.StringIO()):
|
|
status = publish(r, committed_dictionaries(), out)
|
|
if status != 0 or json.loads(out.read_text()) != r:
|
|
fails.append("selftest: run's last step did not write a result that holds")
|
|
return fails
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
if not args:
|
|
raise SystemExit(__doc__)
|
|
if args[0] == "run":
|
|
out = pathlib.Path(args[args.index("--out") + 1]) if "--out" in args else HERE / "result.json"
|
|
return run(out)
|
|
r = json.loads(pathlib.Path(args[1]).read_text())
|
|
if args[0] == "check":
|
|
text = pathlib.Path(args[2]).read_text() if len(args) > 2 else None
|
|
findings = check(r, text)
|
|
elif args[0] == "selftest":
|
|
findings = check(r) + selftest(r)
|
|
else:
|
|
raise SystemExit(__doc__)
|
|
for f in findings:
|
|
print(f)
|
|
if not findings:
|
|
print(f"{args[0]}: {pathlib.Path(args[1]).name} holds")
|
|
return 1 if findings else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|