"""Golden corpus runner G01-G17 (spec 19, Layer 13). Both the producer (engine.parse + engine.canonical) and the independent verifier (engine.verify.vparse + vinvariants) must reproduce the committed expected outputs byte-for-byte. Procedural cases: G13 (worker reorder), G14 (missing carry), G15 (altered state), G16 (repeated resume), G17 (already-transformed input) and the G11 dropped-row probe run through the engine end-to-end where state transitions are involved. Returns {"verdict", "cases": [{"case", "ok", "detail"}]}. """ import json import os import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname( os.path.abspath(__file__))))) from tests.golden.common import case_bytes, case_chunk_nominal, golden_cfg # noqa: E402 from engine.chunkmap import build_chunk_map # noqa: E402 from engine.parse import parse_chunk, ALL_MALFORMED_CLASSES # noqa: E402 from engine.verify.vparse import v_parse_chunk # noqa: E402 from engine.canonical import tick_chunk_content_id # noqa: E402 from engine.verify.vinvariants import ( # noqa: E402 independent_ticks_content_id, independent_bars_content_id, ) from engine.storage import ( # noqa: E402 cert_path, checkpoint_path, journal_path, ticks_chunk_path, bar_part_path, read_ticks_chunk, read_bar_part, ) from engine.certify import certify_source # noqa: E402 from engine.config import default_config # noqa: E402 from engine.dispatcher import ( # noqa: E402 init_run, IngestRunner, IngestError, resume_checks, new_run_id, ) from engine.checkpoint import ( # noqa: E402 load_checkpoint, write_checkpoint, STATUS_PAUSED, ) from engine.util import sha256_bytes # noqa: E402 CERTIFY_NOW_MS = 4_102_444_800_000 HERE = os.path.dirname(os.path.abspath(__file__)) def _case_path(case_id): return os.path.join(HERE, "cases", case_id + ".txt") def _expected_path(case_id): return os.path.join(HERE, "expected", case_id + ".json") def _chunk_parse(data, cm, cfg, parser): records, counters, ids = [], {c: 0 for c in ALL_MALFORMED_CLASSES}, {} rows_parsed = 0 for ch in cm["chunks"]: sub = data[ch["byte_start"]:ch["byte_end"]] if ch["index"] == 0 and sub.startswith(b"\xef\xbb\xbf"): sub = sub[3:] if parser == "producer": recs, _m, ctr, fl = parse_chunk( sub, chunk_index=ch["index"], byte_start=ch["byte_start"], global_line_start=0, cfg=cfg, expect_header=(ch["index"] == 0), certify_time_ms=CERTIFY_NOW_MS) ids[str(ch["index"])] = tick_chunk_content_id( [(r[0], r[1], r[2], r[4]) for r in recs]) else: recs, _m, ctr, fl = v_parse_chunk( sub, chunk_index=ch["index"], byte_start=ch["byte_start"], global_line_start=0, cfg=cfg, expect_header=(ch["index"] == 0), certify_time_ms=CERTIFY_NOW_MS) ids[str(ch["index"])] = independent_ticks_content_id( [(r[0], r[1], r[2], r[4]) for r in recs]) records.extend(recs) rows_parsed += fl["rows_parsed"] for cls, n in ctr.items(): counters[cls] += n return records, counters, ids, rows_parsed def _check_simple(case_id, data, expected): cfg = golden_cfg(chunk_bytes=case_chunk_nominal(case_id)) cm = build_chunk_map(_case_path(case_id), cfg["chunk_bytes_nominal"], cfg["workload_bytes_nominal"]) recs_p, ctr_p, ids_p, rp_p = _chunk_parse(data, cm, cfg, "producer") recs_v, ctr_v, ids_v, rp_v = _chunk_parse(data, cm, cfg, "verifier") diffs = [] if len(recs_p) != expected["rows_canonical"] or len(recs_v) != expected["rows_canonical"]: diffs.append("rows canonical p=%d v=%d expected=%d" % (len(recs_p), len(recs_v), expected["rows_canonical"])) if rp_p != expected["rows_parsed"] or rp_v != expected["rows_parsed"]: diffs.append("rows_parsed p=%d v=%d expected=%d" % (rp_p, rp_v, expected["rows_parsed"])) for cls in ALL_MALFORMED_CLASSES: exp = expected["malformed_expected"].get(cls, 0) if ctr_p[cls] != exp: diffs.append("producer %s=%d expected=%d" % (cls, ctr_p[cls], exp)) if ctr_v[cls] != exp: diffs.append("verifier %s=%d expected=%d" % (cls, ctr_v[cls], exp)) if ids_p != expected["content_ids"]: diffs.append("producer content ids != expected") if ids_v != expected["content_ids"]: diffs.append("verifier content ids != expected") if ids_p != ids_v: diffs.append("producer ids != verifier ids") return diffs # --------------------------------------------------------------------------- # mini end-to-end run helpers (synthetic fixtures only) # --------------------------------------------------------------------------- def mini_setup(data, out_root, chunk_bytes=None, workload_bytes=None, tz_offset=0): src = os.path.join(out_root, "source.txt") with open(src, "wb") as fh: fh.write(data) cfg = default_config( src, output_root=out_root, source_tz_offset_minutes=tz_offset, workers_requested=1, timeframes=["M1", "M5", "M15", "M30", "H1"], chunk_bytes_nominal=chunk_bytes or 25_165_824, workload_bytes_nominal=workload_bytes or 5_368_709_120) cert = certify_source(src, cert_path(out_root), certify_timeout_sec=120) cm = build_chunk_map(src, cfg["chunk_bytes_nominal"], cfg["workload_bytes_nominal"]) run_id = new_run_id() init_run(cfg, src, cm["source_id"], cert, run_id, out_root, cm) return cfg, cert, cm, run_id def _file_ids(cm, out_root, cfg): ids = {} for c in cm["chunks"]: rows = read_ticks_chunk(ticks_chunk_path(out_root, c["index"])) ids["ticks/%d" % c["index"]] = independent_ticks_content_id( [(r[0], r[1], r[2], r[4]) for r in rows]) for wl in cm["workloads"]: for tf in cfg["timeframes"]: p = bar_part_path(out_root, tf, wl["index"]) if os.path.exists(p): ids["bars/%s/%d" % (tf, wl["index"])] = \ independent_bars_content_id(read_bar_part(p)) return ids # --------------------------------------------------------------------------- # main runner # --------------------------------------------------------------------------- def run_golden(cases_dir=None, expected_dir=None): results = [] for case_id in ["G01", "G02", "G03", "G04", "G05", "G06", "G07", "G08", "G09", "G10", "G11", "G12"]: expected = json.load(open(_expected_path(case_id), encoding="utf-8")) diffs = _check_simple(case_id, case_bytes(case_id), expected) results.append({"case": case_id, "ok": not diffs, "detail": diffs}) e1 = json.load(open(_expected_path("G01"), encoding="utf-8")) e2 = json.load(open(_expected_path("G02"), encoding="utf-8")) same = e1["content_ids"] == e2["content_ids"] results.append({"case": "G01_equals_G02_ids", "ok": same, "detail": [] if same else ["CRLF/LF ids differ"]}) # G05/G06 chunkmap contiguity (exact-boundary and inside-line cuts) for cid, nominal in (("G05", 46), ("G06", 40)): cm = build_chunk_map(_case_path(cid), nominal, 5_368_709_120) ok = cm["chunks"][0]["byte_start"] == 0 and \ cm["chunks"][-1]["byte_end"] == cm["total_bytes"] for i in range(1, len(cm["chunks"])): ok = ok and cm["chunks"][i]["byte_start"] == \ cm["chunks"][i - 1]["byte_end"] results.append({"case": "%s_boundaries" % cid, "ok": ok, "detail": [] if ok else ["gap/overlap in chunk map"]}) # G11 dropped-row probe: the verifier-side count check must flag a drop data11 = case_bytes("G11") buf = data11.rstrip() lines = buf.split(b"\n") dropped = b"\n".join(lines[:-1]) + b"\n" cfg11 = golden_cfg(chunk_bytes=case_chunk_nominal("G11")) cm11 = build_chunk_map(_case_path("G11"), cfg11["chunk_bytes_nominal"], cfg11["workload_bytes_nominal"]) full_rows = _chunk_parse(data11, cm11, cfg11, "producer")[0] tmp = os.path.join(HERE, "cases", ".tmp_drop.txt") with open(tmp, "wb") as fh: fh.write(dropped) short_rows = _chunk_parse(dropped, build_chunk_map(tmp, cfg11["chunk_bytes_nominal"], cfg11["workload_bytes_nominal"]), cfg11, "producer")[0] os.remove(tmp) detected = len(full_rows) != len(short_rows) exp11 = json.load(open(_expected_path("G11"), encoding="utf-8")) results.append({"case": "G11_dropped_row", "ok": detected and len(full_rows) == exp11["rows_canonical"], "detail": [] if detected else ["drop not detected"]}) # G13 worker reorder: identical per-chunk ids and merged id data13 = case_bytes("G13") cm13 = build_chunk_map(_case_path("G13"), 46, 5_368_709_120) cfg13 = golden_cfg(chunk_bytes=46) def ids_in_order(order): out = {} for ci in order: ch = cm13["chunks"][ci] recs, *_ = parse_chunk( data13[ch["byte_start"]:ch["byte_end"]], chunk_index=ci, byte_start=ch["byte_start"], global_line_start=0, cfg=cfg13, expect_header=(ci == 0), certify_time_ms=CERTIFY_NOW_MS) out["chunk_%d" % ci] = tick_chunk_content_id( [(r[0], r[1], r[2], r[4]) for r in recs]) return out order_a = [c["index"] for c in cm13["chunks"]] order_b = list(reversed(order_a)) ia, ib = ids_in_order(order_a), ids_in_order(order_b) merged_a = sha256_bytes("".join(ia[k] for k in sorted(ia))) merged_b = sha256_bytes("".join(ib[k] for k in sorted(ib))) results.append({"case": "G13_worker_reorder", "ok": ia == ib and merged_a == merged_b, "detail": [] if ia == ib and merged_a == merged_b else ["reorder changed ids"]}) # G14 missing carry with tempfile.TemporaryDirectory() as td: cfg, cert, cm, run_id = mini_setup(case_bytes("G01"), td, chunk_bytes=47, workload_bytes=94) st = mini_run(cfg, cert, cm, td, run_id, pause_after=1) pay = load_checkpoint(checkpoint_path(td)) del pay["carries"] # hash NOT recomputed with open(checkpoint_path(td), "w", encoding="utf-8") as fh: json.dump(pay, fh, sort_keys=True) _c, reason = resume_checks(cfg, td, cfg["source_path"], cm, cert) results.append({"case": "G14_missing_carry", "ok": reason is not None, "detail": [reason or "resume not blocked"]}) with tempfile.TemporaryDirectory() as td2: cfg, cert, cm, run_id = mini_setup(case_bytes("G01"), td2, chunk_bytes=47, workload_bytes=94) st = mini_run(cfg, cert, cm, td2, run_id, pause_after=1) pay = load_checkpoint(checkpoint_path(td2)) pay["carries"][cfg["timeframes"][0]]["last_ts_ms"] += 8_000_000_000 write_checkpoint(checkpoint_path(td2), pay) # consistent hash, wrong carry _c, reason = resume_checks(cfg, td2, cfg["source_path"], cm, cert) if reason is None: runner = IngestRunner(cfg, td2, cfg["source_path"], cm["source_id"], cm, cert, run_id=run_id) final_status = runner.run() detection = (final_status == "FAILED") else: detection = True results.append({"case": "G14_incorrect_carry", "ok": detection, "detail": [] if detection else ["carry mismatch not failed"]}) # G15 altered state (journal chain / checkpoint content) with tempfile.TemporaryDirectory() as td: cfg, cert, cm, run_id = mini_setup(case_bytes("G01"), td, chunk_bytes=47, workload_bytes=94) mini_run(cfg, cert, cm, td, run_id, pause_after=1) with open(journal_path(td), "ab") as fh: fh.write(b"\xff\x00 bogus tail\n") _c, reason_a = resume_checks(cfg, td, cfg["source_path"], cm, cert) results.append({"case": "G15_journal_corrupt", "ok": reason_a is not None, "detail": [reason_a or "not blocked"]}) # restore journal by truncating the bogus tail with open(journal_path(td), "rb") as fh: content = fh.read() keep = content[:content.rfind(b"\xff\x00 bogus tail\n")] + b"\n" with open(journal_path(td), "wb") as fh: fh.write(keep) pay = load_checkpoint(checkpoint_path(td)) pay["status_reason"] = "tampered" with open(checkpoint_path(td), "w", encoding="utf-8") as fh: json.dump(pay, fh, sort_keys=True) # content tamper, no rehash _c, reason_b = resume_checks(cfg, td, cfg["source_path"], cm, cert) results.append({"case": "G15_checkpoint_corrupt", "ok": reason_b is not None, "detail": [reason_b or "not blocked"]}) # G16 repeated resume -> identical content ids with tempfile.TemporaryDirectory() as td: cfg, cert, cm, run_id = mini_setup(case_bytes("G01"), td, chunk_bytes=47, workload_bytes=94) mini_run(cfg, cert, cm, td, run_id, pause_after=1) _c, reason = resume_checks(cfg, td, cfg["source_path"], cm, cert) assert reason is None, "resume unexpectedly blocked: %s" % reason runner = IngestRunner(cfg, td, cfg["source_path"], cm["source_id"], cm, cert, run_id=run_id) st_a = runner.run() ids_a = _file_ids(cm, td, cfg) _c, reason2 = resume_checks(cfg, td, cfg["source_path"], cm, cert) if reason2 is None: runner2 = IngestRunner(cfg, td, cfg["source_path"], cm["source_id"], cm, cert, run_id=run_id) st_b = runner2.run() else: st_b = None ids_b = _file_ids(cm, td, cfg) ok16 = ids_a == ids_b and st_a == "CHUNK_COMMITTED" results.append({"case": "G16_repeated_resume", "ok": ok16, "detail": [] if ok16 else ["ids drifted: %s" % json.dumps( {k: (ids_a.get(k), ids_b.get(k)) for k in ids_a if ids_a.get(k) != ids_b.get(k)})]}) # G17 already-transformed (canonical-shaped) input -> blocked at init with tempfile.TemporaryDirectory() as td: src = os.path.join(td, "source.txt") with open(src, "wb") as fh: fh.write(case_bytes("G17")) cfg17 = default_config(src, output_root=td, source_tz_offset_minutes=0, workers_requested=1) cert17 = certify_source(src, cert_path(td), certify_timeout_sec=120) cm17 = build_chunk_map(src, cfg17["chunk_bytes_nominal"], cfg17["workload_bytes_nominal"]) fmt = cert17.get("source_format") refused = False try: run_id17 = new_run_id() init_run(cfg17, src, cm17["source_id"], cert17, run_id17, td, cm17) except IngestError: refused = True results.append({"case": "G17_already_transformed", "ok": fmt != "csv_tickstory_mt5" and refused, "detail": ["format=%r init_refused=%s" % (fmt, refused)]}) verdict = "VERIFIER_ACCEPTED" if all(r["ok"] for r in results) \ else "VERIFIER_REJECTED" return {"verdict": verdict, "cases": results} def mini_run(cfg, cert, cm, out_root, run_id, pause_after=None): runner = IngestRunner(cfg, out_root, cfg["source_path"], cm["source_id"], cm, cert, run_id=run_id, pause_after=pause_after) return runner.run() if __name__ == "__main__": res = run_golden() for c in res["cases"]: print("%-28s %s %s" % (c["case"], "PASS" if c["ok"] else "FAIL", c["detail"])) print("VERDICT:", res["verdict"]) sys.exit(0 if res["verdict"] == "VERIFIER_ACCEPTED" else 1)