SniperGold_ML/tests/cli_tests.py

"""Headless CLI qualification (spec 14): all commands + exit codes, run on
synthetic fixtures only, completely without any interactive session."""

import json
import os
import subprocess
import sys
import tempfile

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PY = r"C:\Users\defau\AppData\Local\Python\bin\python.exe"
CLI = os.path.join(REPO, "sniper-data.py")

G01 = os.path.join(REPO, "tests", "golden", "cases", "G01.txt")
LEGACY_EVIDENCE = os.path.join(REPO, "ml", "p3", "p3_s251_external_ingest")


def run_cli(args, cwd=REPO):
    return subprocess.run([PY, CLI] + args, cwd=cwd, capture_output=True,
                          text=True, timeout=300)


def write_json(path, obj):
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(obj, fh, indent=2, sort_keys=True)


def run():
    os.makedirs(os.path.join(REPO, "qualification"), exist_ok=True)
    with tempfile.TemporaryDirectory(
            dir=os.path.join(REPO, "qualification")) as runroot:
        cfg = {
            "source_path": os.path.abspath(G01),
            "source_grammar": "dotted_v1",
            "source_tz_offset_minutes": 0,
            "has_volume": False,
            "timeframes": ["M1", "M5", "M15", "M30", "H1"],
            "chunk_bytes_nominal": 25_165_824,
            "workload_bytes_nominal": 536_870_912,
            "output_root": runroot,
            "workers_requested": 1,
            "retry_limit": 2,
        }
        cfg_path = os.path.join(runroot, "engine_config.json")
        write_json(cfg_path, cfg)
        research = {
            "dataset_id": "xau_cli_v1",
            "source_dataset_version": "DS_V1.0.0",
            "timeframe": "M1",
            "prediction_horizon_min": 1,
            "target": {"type": "direction", "horizon": "1m", "from": "close"},
            "features": [{"name": "ret1", "primitive": "return", "window": 1}],
            "split": {"policy": "time_ordered", "train": 0.7, "val": 0.15,
                       "test": 0.15, "oos_barrier": False,
                       "barrier_gap_min": 1},
        }
        rcfg_path = os.path.join(runroot, "research.json")
        write_json(rcfg_path, research)

        results = []

        def check(name, expected_code, args):
            r = run_cli(args)
            results.append((name, r.returncode, expected_code,
                            r.stderr.strip()[-300:]))

        # P3-DE-004 CLI regression: status/resume on a run-less directory must
        # answer the spec exit code 4 (BLOCKED) without a traceback (the
        # P3-DE-003 audit observed exit 1 + TypeError on this path).
        emptyroot = os.path.join(runroot, "empty_runless")
        os.makedirs(emptyroot)
        check("cli_status_runless_dir", 4, ["status", "--out-root", emptyroot])
        check("cli_resume_runless_dir", 4, ["resume", "--out-root", emptyroot])
        check("cli_version", 0, ["--version"])
        check("cli_certify", 0, ["certify", "--source", G01,
                                 "--out-root", runroot])
        check("cli_init", 0, ["init", "--config", cfg_path])
        check("cli_ingest", 3, ["ingest", "--out-root", runroot])
        check("cli_status_before_finalize", 3,
              ["status", "--out-root", runroot])
        check("cli_verify_golden", 0, ["verify", "--mode", "golden"])
        check("cli_verify_dataset", 0,
              ["verify", "--mode", "dataset", "--out-root", runroot])
        check("cli_run_complete_exists", 0,
              ["verify", "--mode", "run-complete", "--out-root", runroot])
        check("cli_manifest", 0, ["manifest", "--out-root", runroot])
        check("cli_build_dataset", 0,
              ["build-dataset", "--config", rcfg_path, "--out-root", runroot])
        check("cli_diff_legacy", 0,
              ["diff-legacy", "--legacy-evidence", LEGACY_EVIDENCE,
               "--legacy-checkpoint", os.path.join("output",
                                                   "s251_real_checkpoint.json"),
               "--out-root", runroot])
        check("cli_status_completed", 0, ["status", "--out-root", runroot])
        check("cli_resume_blocked_on_completed", 4,
              ["resume", "--out-root", runroot])
        check("cli_pause", 0, ["pause", "--out-root", runroot])

        run_complete = os.path.join(runroot, "RUN_COMPLETE.json")
        rc_exists = os.path.exists(run_complete)
        verdict = all(code == expected for _n, code, expected, _e in results) \
            and rc_exists
        return {"ok": verdict, "run_complete_created": rc_exists,
                "results": [{"command": n, "code": c, "expected": e,
                             "stderr_tail": t}
                            for n, c, e, t in results]}


if __name__ == "__main__":
    res = run()
    for r in res["results"]:
        ok = r["code"] == r["expected"]
        print("%-32s code=%d expected=%d %s %s"
              % (r["command"], r["code"], r["expected"], "PASS" if ok else "FAIL",
                 r["stderr_tail"]))
    print("run_complete_created", res["run_complete_created"])
    print("OK" if res["ok"] else "FAIL")
    sys.exit(0 if res["ok"] else 1)