"""Property / invariant tests (spec 18.3 + session requirement 3). At minimum: deterministic serialization, deterministic aggregation, worker scheduling does not change logical output, no gaps, no overlaps, monotonic source ordering, checkpoint monotonicity, resume equivalence, content-ID stability, no silent malformed-row transformation, no empty-bar fabrication. """ import json import os import sys import tempfile sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from engine.chunkmap import build_chunk_map # noqa: E402 from engine.config import default_config # 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 independent_ticks_content_id, \ independent_bars_content_id # noqa: E402 from engine.aggregate import Aggregator # noqa: E402 from engine.verify.vaggregate import v_aggregate, v_aggregate_eof # noqa: E402 from engine.checkpoint import write_checkpoint, load_checkpoint # noqa: E402 from engine.storage import ( # noqa: E402 ticks_chunk_path, bar_part_path, read_ticks_chunk, read_bar_part, checkpoint_path, cert_path, ) from engine.certify import certify_source # noqa: E402 from engine.dispatcher import ( # noqa: E402 init_run, IngestRunner, resume_checks, finalize_completed, new_run_id, ) from engine.journal import append_commit, read_journal, seq_after # noqa: E402 from engine.util import sha256_bytes # noqa: E402 from tests.golden.common import G01_ROWS # noqa: E402 from tests.golden.run_golden import mini_setup, mini_run # noqa: E402 CERT_NOW = 4_102_444_800_000 def rows_source(n=60, base_ms=1_672_900_000_000): """Deterministic synthetic tick stream (valid prices around 100.0..100.3). Prices are rendered from integer micro-unit values (no float formatting).""" import datetime lines = [] ts = base_ms for i in range(n): p = 100000000 + (i % 30) * 10000 ask = p + 100 dt = datetime.datetime.fromtimestamp(ts / 1000, datetime.timezone.utc) stamp = dt.strftime("%Y.%m.%d %H:%M:%S.") + ("%03d" % (ts % 1000)) bid_txt = "%d.%06d" % (p // 1000000, p % 1000000) ask_txt = "%d.%06d" % (ask // 1000000, ask % 1000000) lines.append("%s,%s,%s" % (stamp, bid_txt, ask_txt)) ts += 60_000 // 4 return ("\n".join(lines) + "\n").encode("utf-8") def run(): cfg = default_config("C:\\unused\\src.csv", output_root="C:\\unused\\out", source_tz_offset_minutes=0) results = [] # 1. deterministic serialization / content-ID stability (producer==verifier) data = rows_source(40) ids_prod, ids_ver = [], [] for _ in range(3): recs, _, _, _ = parse_chunk(data, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) vrecs, _, _, _ = v_parse_chunk(data, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) ids_prod.append(tick_chunk_content_id([(r[0], r[1], r[2], r[4]) for r in recs])) ids_ver.append(independent_ticks_content_id([(r[0], r[1], r[2], r[4]) for r in vrecs])) nn = len(recs) results.append(("deterministic_serialization", all(x == ids_prod[0] for x in ids_prod) and all(x == ids_ver[0] for x in ids_ver), "")) results.append(("content_id_stability_producer_verifier", ids_prod[0] == ids_ver[0] and nn > 0, "")) # 2. deterministic aggregation: producer vs independent verifier recs, _, _, _ = parse_chunk(data, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) agg = Aggregator(cfg) for r in recs: agg.consume(r) prod_bars = {tf: agg.finish_eof(tf) for tf in cfg["timeframes"]} vbars, vcarry, vcounts = v_aggregate([r for r in recs], None, cfg) veof, _, veof_counts = v_aggregate_eof(cfg, vcarry, vcounts) ok = True for tf in cfg["timeframes"]: if vbars[tf] + veof[tf] != prod_bars[tf]: ok = False agg2 = Aggregator(cfg) for r in recs: agg2.consume(r) prod2 = {tf: agg2.finish_eof(tf) for tf in cfg["timeframes"]} ok2 = all(prod_bars[tf] == prod2[tf] for tf in cfg["timeframes"]) results.append(("deterministic_aggregation", ok, "")) results.append(("aggregation_producer_verifier_equal", ok2, "")) # 3. no gaps / no overlaps in chunkmap on a mixed-length file with tempfile.TemporaryDirectory() as td: src = os.path.join(td, "mix.txt") lines = [b"2023.01.01 00:00:%02d.00%d,100.0%d,100.0%d" % (i % 60, i % 10, i % 9, i % 9 + 1) for i in range(200)] data_mix = b"\n".join(lines) + b"\n" with open(src, "wb") as fh: fh.write(data_mix) cm = build_chunk_map(src, 500, 5_000_000) gap_ok = cm["chunks"][0]["byte_start"] == 0 and \ cm["chunks"][-1]["byte_end"] == cm["total_bytes"] for i in range(1, len(cm["chunks"])): gap_ok = gap_ok and \ cm["chunks"][i]["byte_start"] == cm["chunks"][i - 1]["byte_end"] results.append(("no_gaps_no_overlaps", gap_ok and len(cm["chunks"]) > 1, "")) # 4. monotonic source ordering mono = True prev = None for r in recs: if prev is not None and r[0] < prev: mono = False prev = r[0] results.append(("monotonic_source_order", mono, "")) # 5. checkpoint monotonicity + journal seq monotonicity with tempfile.TemporaryDirectory() as td: jp = os.path.join(td, "commits.jsonl") prev_hash = sha256_bytes(b"") seqs = [] for i in range(5): prev_hash = append_commit(jp, i, i, "c%d" % i, "s%d" % i, "t%d" % i, prev_hash) seqs.append(i) recs_j, tail = read_journal(jp) ok_seq = [r["seq"] for r in recs_j] == seqs # checkpoint writes with increasing next_chunk che1 = {"n": 0, "next_chunk": 1} che2 = {"n": 1, "next_chunk": 2} cp = os.path.join(td, "checkpoint.json") write_checkpoint(cp, che1) write_checkpoint(cp, che2) loaded = load_checkpoint(cp) results.append(("journal_seq_monotonic", ok_seq, "")) results.append(("checkpoint_next_chunk_monotonic", loaded["next_chunk"] == 2, "")) # 6. resume equivalence: full run vs interrupted/resumed run data6 = rows_source(120) with tempfile.TemporaryDirectory() as tdA, tempfile.TemporaryDirectory() as tdB: cfgA, certA, cmA, ridA = mini_setup(data6, tdA, chunk_bytes=700, workload_bytes=1400) stA = mini_run(cfgA, certA, cmA, tdA, ridA) idsA = _file_ids(cmA, tdA, cfgA) cfgB, certB, cmB, ridB = mini_setup(data6, tdB, chunk_bytes=700, workload_bytes=1400) stB1 = mini_run(cfgB, certB, cmB, tdB, ridB, pause_after=1) _c, reason = resume_checks(cfgB, tdB, cfgB["source_path"], cmB, certB) if reason is None: rr = IngestRunner(cfgB, tdB, cfgB["source_path"], cmB["source_id"], cmB, certB, run_id=ridB) stB2 = rr.run() else: stB2 = None idsB = _file_ids(cmA, tdA, cfgA) idsB2 = _file_ids(cmB, tdB, cfgB) results.append(("resume_equivalence", stB2 == "CHUNK_COMMITTED" and idsA == idsB2, "stB2=%r" % stB2)) # 7. no silent malformed-row transformation bad = G01_ROWS + ["2023.13.01 00:00:00.000,100.0,101.0", "x,1,2", "2023.01.01 00:00:00.000,200,100", "2023.01.01 00:00:00.000,100.000000,100.000100"] data_bad = (b"\n".join(x.encode() for x in bad) + b"\n") brecs, _m, ctr, _f = parse_chunk(data_bad, chunk_index=0, byte_start=0, global_line_start=0, cfg=cfg, expect_header=True, certify_time_ms=CERT_NOW) n_mal = sum(ctr.values()) every_valid = all(0 < r[1] <= r[2] for r in brecs) results.append(("no_silent_malformed_transformation", n_mal == 4 and every_valid and len(brecs) == 4, "mal=%d rows=%d" % (n_mal, len(brecs)))) # 8. no empty-bar fabrication (property scale) agg3 = Aggregator(cfg) agg3.consume((0, 100000000, 100000100, 100, 1, 1, 0)) agg3.consume((91 * 86400000, 100000000, 100000100, 100, 1, 2, 1)) rows_m1 = agg3.finish_eof("M1") results.append(("no_empty_bar_fabrication", len(rows_m1) == 2 and [r[1] for r in rows_m1] == [0, 91 * 1440], [str(r[1]) for r in rows_m1])) return results 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 if __name__ == "__main__": res = run() for name, ok, detail in res: print("%-40s %s %s" % (name, "PASS" if ok else "FAIL", detail)) sys.exit(0 if all(r[1] for r in res) else 1)