SniperGold_ML/tests/golden/generate_cases.py

"""Deterministic golden-corpus generator.

Computes expected outputs for G01-G17 using the INDEPENDENT verifier
(engine.verify.vparse + vinvariants) and hand-anchored primitives, then writes
the committed fixtures (cases/*.txt) and expected outputs (expected/*.json).

Run once to materialize fixtures:  python tests/golden/generate_cases.py
"""
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.abspath(__file__)))))

from tests.golden.common import (  # noqa: E402
    case_bytes, case_chunk_nominal, golden_cfg,
)
from engine.chunkmap import build_chunk_map  # noqa: E402
from engine.verify.vparse import v_parse_chunk  # noqa: E402
from engine.verify.vinvariants import independent_ticks_content_id  # noqa: E402
from engine.parse import ALL_MALFORMED_CLASSES  # noqa: E402
from engine.certify import certify_source  # noqa: E402

HERE = os.path.dirname(os.path.abspath(__file__))
CASES_DIR = os.path.join(HERE, "cases")
EXPECTED_DIR = os.path.join(HERE, "expected")

CERTIFY_NOW_MS = 4_102_444_800_000  # fixed 2100 reference for tests


def _compute_expected(case_id, path, data):
    cfg = golden_cfg(chunk_bytes=case_chunk_nominal(case_id))
    cm = build_chunk_map(path, cfg["chunk_bytes_nominal"],
                         cfg["workload_bytes_nominal"])
    total_records = []
    counters = {cls: 0 for cls in ALL_MALFORMED_CLASSES}
    ids = {}
    rows_parsed_total = 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:]
        records, _ms, ctr, flags = 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)
        total_records.extend(records)
        rows_parsed_total += flags["rows_parsed"]
        for cls, n in ctr.items():
            counters[cls] += n
        ids[str(ch["index"])] = independent_ticks_content_id(
            [(r[0], r[1], r[2], r[4]) for r in records])
    return {
        "case": case_id,
        "chunks_expected": len(cm["chunks"]),
        "rows_parsed": rows_parsed_total,
        "rows_canonical": len(total_records),
        "malformed_expected": counters,
        "content_ids": ids,
    }


def generate(force=False):
    os.makedirs(CASES_DIR, exist_ok=True)
    os.makedirs(EXPECTED_DIR, exist_ok=True)
    ids = ["G01", "G02", "G03", "G04", "G05", "G06", "G07", "G08", "G09",
           "G10", "G11", "G12", "G13", "G17"]
    for case in ids:
        data = case_bytes(case)
        txt = os.path.join(CASES_DIR, case + ".txt")
        if force or not os.path.exists(txt):
            with open(txt, "wb") as fh:
                fh.write(data)
        exp_path = os.path.join(EXPECTED_DIR, case + ".json")
        expected = _compute_expected(case, txt, data)
        if force or not os.path.exists(exp_path):
            with open(exp_path, "w", encoding="utf-8") as fh:
                json.dump(expected, fh, indent=2, sort_keys=True)
        print("generated", case)


if __name__ == "__main__":
    generate(force=("--force" in sys.argv))