#!/usr/bin/env python3"""Checks on the built outputs. Each tamper test first proves the check can fail: a guard that never fires proves nothing."""import csv, gzip, json, math, pathlib, re, sys, tempfileROOT = pathlib.Path(__file__).resolve().parent.parentsys.path.insert(0, str(ROOT / "pipeline"))import grade  # noqa: E402M = json.loads((ROOT / "docs" / "data" / "metrics.json").read_text())C = json.loads((ROOT / "docs" / "data" / "charts.json").read_text())def v(k):    return M[k]["value"]def test_every_metric_has_a_source():    for key, rec in M.items():        assert rec["source"], key        assert rec["label"], key        if isinstance(rec["value"], float):            assert math.isfinite(rec["value"]), keydef test_supply_identity():    """Supply at the end must equal supply at the start plus mint minus burn, over the whole header series."""    with gzip.open(ROOT / "raw" / "bigquery" / "01_supply_daily.csv.gz", "rt") as fh:        rows = [r for r in csv.DictReader(fh)]    a, b = rows[100], rows[-2]    between = rows[101:-1]    mint = sum(float(r["minted_near"]) for r in between)    burn = sum(float(r["burned_near"]) for r in between)    gap = float(b["supply_near"]) - (float(a["supply_near"]) + mint - burn)    # one block-to-block difference per month is dropped by the LAG partition, about 75 months of it    assert abs(gap) / float(b["supply_near"]) < 2e-4, gapdef test_issuance_matches_the_protocol_cap():    assert 1.4 < v("gross_issuance_pct") < 2.7  # 2.5% cap today, 1.6% target if the ramp passes    assert 4.5 < v("mint_per_day_before_cut") * 365 / 1.28e9 * 100 < 5.3def test_flows_add_up():    p = C["per_100"]    assert abs(p["gross"] - p["captured"] - p["others"]) < 1e-3    fl = C["flows_near_per_day"]    assert abs((fl["burn"] + fl["revenue_equiv"]) / fl["issuance"] * 100 - v("offset_30d_pct")) < 0.2def test_revenue_never_exceeds_fees():    """A month where revenue passes fees would mean a partial or mismatched series slipped in."""    for row in C["intents_monthly"]:        if row["fees"] > 100000:            assert row["rev"] <= row["fees"], row["m"]def test_flow_range_is_the_true_range():    weeks = sorted(r["ratio"] for r in C["flow_reconciliation"][-26:])    assert v("flow_ratio_low") == weeks[0] and v("flow_ratio_high") == weeks[-1]def test_kaiching_mass_deletions_are_out_of_the_signer_count():    assert v("signers_last") < v("signers_last_raw")def test_shares_sum():    for key in ("flow_assets",):        assert abs(sum(r["share"] for r in C[key]) - 100) < 1    for key in ("n_share", "usd_share"):        assert abs(sum(r[key] for r in C["withdrawal_sizes"]) - 100) < 1    if C.get("ex_fee_owners"):        assert abs(sum(r["share"] for r in C["ex_fee_owners"]) - 100) < 1def test_treasury_trap():    """The RPC field protocol_reward_rate reads 0 on mainnet. nearcore hardcodes 1/10 and the treasury receives it."""    assert v("treasury_share_of_issuance_pct") == 10.0    assert 4.5 < v("staking_yield_gross_pct") < 5.5def test_commission_excludes_self_stake():    assert 2 < v("commission_wavg_pct") < 10def test_ledger_chain_and_tamper_detection():    with tempfile.TemporaryDirectory() as tmp:        path = pathlib.Path(tmp) / "l.jsonl"        prev = "0" * 64        lines = []        for i in range(3):            rec = {"prev": prev, "graded_on": f"2026-11-0{i + 1}", "statuses": {"T1": "live"}}            rec["hash"] = grade.line_hash(prev, rec)            prev = rec["hash"]            lines.append(json.dumps(rec))        path.write_text("\n".join(lines) + "\n")        assert grade.verify_chain(path) == 3        path.write_text(path.read_text().replace('"live"', '"tripped"', 1))        try:            grade.verify_chain(path)        except SystemExit:            return        raise AssertionError("tampered ledger was not detected")def test_real_ledger():    assert grade.verify_chain(ROOT / "ledger" / "grades.jsonl") >= 1def test_page_is_fully_rendered():    html = (ROOT / "docs" / "index.html").read_text()    assert "{{" not in html    assert "NEAR Toll Audit" in htmldef test_page_loads_no_third_party_script():    for name in ("index.html", "thesis.html"):        path = ROOT / "docs" / name        if path.exists():            assert not re.search(r'<script[^>]+src="https?:', path.read_text()), namedef test_no_secrets_committed():    bad = re.compile(r"BEGIN [A-Z ]*PRIVATE KEY|\"private_key\"\s*:\s*\"-|eyJ[A-Za-z0-9_-]{20,}\.eyJ|ghp_[A-Za-z0-9]{20,}|AIza[0-9A-Za-z_-]{30,}")    for path in ROOT.rglob("*"):        if not path.is_file() or ".git" in path.parts or "raw_private" in path.parts or path.name == "test_pipeline.py" or "__pycache__" in path.parts:            continue        data = gzip.open(path, "rb").read() if path.suffix == ".gz" else path.read_bytes()        assert not bad.search(data.decode("utf8", "replace")), pathdef test_redirects_are_fenced():    """A redirect to a local service must be refused, and one to an untrusted host must never carry a token."""    import common, urllib.request    handler = common.SafeRedirect()    req = urllib.request.Request("https://api.llama.fi/x", headers={"Authorization": "Bearer fake"})    for target in ("http://127.0.0.1:9222/json/list", "https://192.168.5.2/x", "https://evil.example/x"):        try:            handler.redirect_request(req, None, 302, "Found", {}, target)        except common.StopSource:            continue        raise AssertionError(f"redirect to {target} was allowed")    out = handler.redirect_request(req, None, 302, "Found", {}, "https://coins.llama.fi/y")    assert out is not None and not out.has_header("Authorization")def test_memo_links_cannot_break_out():    import build    html_out = build.inline('[x](https://a"onmouseover="alert)')    assert "onmouseover=" not in html_out.replace("&quot;onmouseover=&quot;", "")def test_memo_numbered_list_is_a_list():    import build    out = build.markdown("Needs:\n\n1. One.\n2. Two.\n\n- a\n- b")    assert out.count("<li>") == 4 and "<ol>" in out and "</ol>" in out and "</ul>" in out, out    # the built memo itself: the "What has to be true" items are list items, not one paragraph    page = (ROOT / "docs" / "thesis.html").read_text()    assert not re.search(r"<p>1\. .*? 2\. ", page), "numbered items flattened into one paragraph"def test_page_reads_without_javascript():    # A reader or crawler with scripts off must still see every kill row and the Test three sentence.    page = (ROOT / "docs" / "index.html").read_text()    spec = json.loads((ROOT / "thesis" / "kill_criteria.json").read_text())    for c in spec["criteria"]:        assert f'<span class="kill-id">{c["id"]}</span>' in page, c["id"]    assert page.count('class="kill-row"') == len(spec["criteria"])    m = re.search(r'id="elast-text">([^<]+)<', page)    assert m and len(m.group(1)) > 40, "Test three is empty without JavaScript"    assert re.search(r'id="killlede">[^<]{20,}<', page), "kill list lede is empty"    # positive control: the check fails on a page with the static list stripped out    stripped = re.sub(r'<div class="kill-row">.*?</div></div>', "", page, flags=re.S)    assert stripped.count('class="kill-row"') != len(spec["criteria"])if __name__ == "__main__":    for name, fn in list(globals().items()):        if name.startswith("test_"):            fn()            print("ok", name)