forked from chiki2bum2/SniperGold_ML
Chiki Bum
0d5ca3c197
Implements the frozen P3_DATA_ENGINE_V1_SPEC (SHA 84bf0f217ffba51197112a6bbacbcc297058e04b5ca47f0459028fe33e0321e5). Components: engine/ producer (certify, chunkmap, parse, canonical, worker, dispatcher, aggregate, storage, journal, checkpoint, lock, evidence, manifest, run_complete, dataset_builder, cli) + engine/verify independent verifier (vparse, vaggregate, vinvariants, vcompare); headless CLI sniper-data; golden corpus G01-G17; unit/property/adversarial/mutation/legacy-diff/CLI suites. Qualification verdict: QUALIFIED (all 9 mandatory gates pass; independent verifier accepted). Spec, governance record, and legacy checkpoint untouched. G-14 NOT AUTHORIZED honored; no real-data processing, no pilot, no workload 46, no chunk 760 access.
346 lines
No EOL
16 KiB
Python
346 lines
No EOL
16 KiB
Python
"""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) |