mirror of
https://github.com/openglow-org/forgefirm.git
synced 2026-09-28 01:01:12 -07:00
ScottW514
c0f53a865f
A stdlib-only daemon on the dev image (HTTP :8090) that runs the acceptance catalog against the machine from a self-contained page, keeps the append-only result log under /data/forgetest, and exports the release artifact the gate reads. Tests declare kind (auto / operator / live), hardware (api / takeover), coverage globs, prerequisites, and core membership; a test's domain fingerprint is the hash of the manifest files its globs select plus the platform and its own implementation, so a PASS stays valid exactly while nothing it covers changed. Campaign rules: a FAIL ends the campaign, the core (image health, kernel latch and drills, one live emission witness) is never inherited, invalidate-all forces a full campaign, no SKIP. Live tests need the operator acknowledgment and the physical arm press through the controller; takeover tests stop forgectrl for the duration with a crash-recoverable marker; the tool never touches the laser latch. Catalog v1: 24 tests ported from the proven bench drills with their recorded pass criteria (image, kernel K1-K3 and fire A/B/U, forgectrl API and logs, motion incl. dead-man, cooling, live laser, camera, update, cloud). The bench tab lists every scripts/bench tool and runs the board-side ones as subprocesses (takeover tools wrapped). 44 host unit tests, including the gate verification fixtures. Installed only by forgefirm-image-dev, with the bench scripts under /usr/share/forgetest/bench.
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""The append-only result log (JSONL) and the export helpers.
|
|
|
|
One JSON object per line under the data directory (default
|
|
/data/forgetest/results.jsonl, override FORGETEST_DATA). Records:
|
|
|
|
campaign {"id","manifest_sha","catalog_hash","image"} a campaign opened
|
|
result {"campaign","test","result","fingerprint","manifest_sha",
|
|
"image","duration_s","evidence","answers","log","message"}
|
|
invalidate {"reason"} manual invalidate-all (full campaign required)
|
|
reset {"reason"} explicit campaign reset
|
|
export {"artifact_sha256","authorized","campaign"}
|
|
|
|
Every record carries "t" (type) and "ts" (UTC, ISO 8601, seconds).
|
|
The file is only ever appended; a corrupt line is skipped, counted, and
|
|
reported, never repaired in place.
|
|
"""
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
|
|
DEFAULT_DATA_DIR = "/data/forgetest"
|
|
|
|
|
|
def now_ts():
|
|
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
|
|
|
|
def data_dir():
|
|
return os.environ.get("FORGETEST_DATA") or DEFAULT_DATA_DIR
|
|
|
|
|
|
class Log:
|
|
def __init__(self, path=None):
|
|
self.path = path or os.path.join(data_dir(), "results.jsonl")
|
|
self._lock = threading.Lock()
|
|
self.corrupt = 0
|
|
|
|
def append(self, rec):
|
|
rec = dict(rec)
|
|
rec.setdefault("ts", now_ts())
|
|
line = json.dumps(rec, sort_keys=True, separators=(",", ":"))
|
|
with self._lock:
|
|
os.makedirs(os.path.dirname(self.path), exist_ok=True)
|
|
with open(self.path, "a", encoding="utf-8") as f:
|
|
f.write(line + "\n")
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
return rec
|
|
|
|
def read(self):
|
|
recs = []
|
|
self.corrupt = 0
|
|
if not os.path.exists(self.path):
|
|
return recs
|
|
with self._lock:
|
|
with open(self.path, "r", encoding="utf-8") as f:
|
|
lines = f.readlines()
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
rec = json.loads(line)
|
|
except ValueError:
|
|
self.corrupt += 1
|
|
continue
|
|
if isinstance(rec, dict) and "t" in rec:
|
|
recs.append(rec)
|
|
else:
|
|
self.corrupt += 1
|
|
return recs
|
|
|
|
def raw(self):
|
|
if not os.path.exists(self.path):
|
|
return ""
|
|
with self._lock:
|
|
with open(self.path, "r", encoding="utf-8") as f:
|
|
return f.read()
|