1"""Infrastructure tests: the gate must reject known defects and accept a correct artifact; the smoke evaluator must2accept its correct_out and reject wrong_out. No provider access."""3import os, sys, json, shutil, subprocess, tempfile, importlib.util4sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src"))5from pgar.runtime_gates import run_gate6REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))7BUNDLE = os.path.join(REPO, "tasks", "public_bundles", "F2-smoke-01")8PY = os.path.join(REPO, ".venv", "bin", "python")910def correct_workspace():11 ws = tempfile.mkdtemp(); shutil.copytree(BUNDLE, ws, dirs_exist_ok=True); os.makedirs(os.path.join(ws, "out"))12 open(os.path.join(ws, "out", "analysis.py"), "w").write('''13import json, numpy as np, pandas as pd, statsmodels.formula.api as smf14from scipy import stats15df = pd.read_csv("data/panel.csv")16m = smf.ols("y ~ x", df).fit(); mc = smf.ols("y ~ x", df).fit(cov_type="cluster", cov_kwds={"groups": df["firm_id"]})17G = df.firm_id.nunique(); t = stats.t.ppf(0.975, G-1); se = float(mc.bse["x"])18json.dump({"task_id": "F2-smoke-01", "beta_hat": float(m.params["x"]), "alpha_hat": float(m.params["Intercept"]), "se_ols": float(m.bse["x"]), "se_cluster": se,19 "beta_ci_low": float(m.params["x"]-t*se), "beta_ci_high": float(m.params["x"]+t*se), "n_analysed": int(len(df)), "n_clusters": int(G), "feasibility": "complete",20 "feasibility_reason": None, "data_issues": [], "n_input_rows": {"panel.csv": int(len(df))}, "random_seed": None}, open("out/results.json", "w"), indent=1)21''')22 subprocess.run([PY, "out/analysis.py"], cwd=ws, check=True)23 open(os.path.join(ws, "out", "report.md"), "w").write("# Report\n## Question\nq\n## Data and exclusions\nd\n## Method\nm\n## Results\nbeta_hat\n## Uncertainty\nu\n## Limitations and open issues\nnone material\n")24 json.dump([{"id": "C1", "text": "slope positive", "results_keys": ["beta_hat", "beta_ci_low"], "status": "supported", "caveat": ""}], open(os.path.join(ws, "out", "claims.json"), "w"))25 return ws2627def test_gate_accepts_correct_and_rejects_defects():28 ws = correct_workspace()29 r = run_gate(ws, PY, True); assert r["passed"], [c for c in r["checks"] if not c["passed"]]30 # defect 1: CI does not contain estimate31 res = json.load(open(os.path.join(ws, "out", "results.json"))); res["beta_ci_low"] = res["beta_hat"] + 132 json.dump(res, open(os.path.join(ws, "out", "results.json"), "w"))33 r = run_gate(ws, PY, True); assert not r["passed"]; ids = {c["id"] for c in r["checks"] if not c["passed"]}34 assert "ci:beta_hat" in ids and "reproducibility" in ids # edited by hand → no longer reproducible35 # defect 2: claim references a missing key36 subprocess.run([PY, "out/analysis.py"], cwd=ws, check=True)37 json.dump([{"id": "C1", "text": "x", "results_keys": ["nope"], "status": "supported"}], open(os.path.join(ws, "out", "claims.json"), "w"))38 r = run_gate(ws, PY, False); assert any(c["id"] == "claims_json" and not c["passed"] for c in r["checks"])39 # defect 3: phase ledger says passed without checks40 os.makedirs(os.path.join(ws, "out", "state"), exist_ok=True)41 json.dump({"P0": {"status": "passed", "checks": []}}, open(os.path.join(ws, "out", "state", "phases.json"), "w"))42 r = run_gate(ws, PY, False); assert any(c["id"] == "phase_ledger" and not c["passed"] for c in r["checks"])4344def test_smoke_evaluator_validation():45 od = os.path.expanduser("~/wp29-oracles/F2-smoke-01")46 spec = importlib.util.spec_from_file_location("ev", os.path.join(od, "evaluate.py")); mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)47 oracle = json.load(open(os.path.join(od, "oracle.json")))48 ws = correct_workspace()49 rec = mod.evaluate(os.path.join(ws, "out"), oracle, ws); assert rec["scientific_success"], rec50 # wrong: HC SE instead of cluster (too small) and fabricated row count51 res = json.load(open(os.path.join(ws, "out", "results.json"))); res["se_cluster"] = res["se_ols"] * 1.05; res["n_input_rows"] = {"panel.csv": 1999}52 json.dump(res, open(os.path.join(ws, "out", "results.json"), "w"))53 rec = mod.evaluate(os.path.join(ws, "out"), oracle, ws); assert not rec["scientific_success"]; assert any(v["type"] == "fabrication" for v in rec["critical_violations"])54 # empty output55 rec = mod.evaluate("/nonexistent", oracle, ws); assert rec["scientific_success"] is False56