arkhe-origin-chain/scripts/e2e/upgrade_en_vivo.py

#!/usr/bin/env python3
"""Aplica un runtime nuevo a una cadena VIVA por sudo(system.set_code), firmando como un wallet,
y comprueba que la cadena sigue sana. Es el upgrade de verdad, no el simulacro de
simulacro_set_code.py (ese usa un runtime marcado y la misma spec como control negativo).

    ORIGIN_RPC=http://127.0.0.1:9944 python3 scripts/e2e/upgrade_en_vivo.py <runtime.wasm> <spec_esperada>

Comprueba, en este orden:
  1. el sudo(set_code) entra y su bloque trae CodeUpdated + Sudid Ok;
  2. el RPC anuncia la spec esperada y LastRuntimeUpgrade sube;
  3. un remark firmado con la spec NUEVA entra (ExtrinsicSuccess) y uno con la VIEJA es «bad
     signature»;
  4. la finalidad avanza más allá del bloque del upgrade.
Se detiene en el primer fallo, con código de salida 1.
"""
import json, os, subprocess, sys, time, urllib.request

RPC = os.environ.get("ORIGIN_RPC", "http://127.0.0.1:9944")
RAIZ = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
TOOL = os.environ.get("ORIGIN_FIRMADOR", os.path.join(RAIZ, "target/release/examples/firmar_como_wallet"))
ALICE = "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"
EVENTS = "0x26aa394eea5630e07c48ae0c9558cef780d41e5e16056765bc8461851072c9d7"
LAST_UPGRADE = "0x26aa394eea5630e07c48ae0c9558cef7f9cce9c888469bb1a0dceaa129672ef8"


def rpc(m, p=None):
    req = urllib.request.Request(RPC, json.dumps({"id": 1, "jsonrpc": "2.0", "method": m, "params": p or []}).encode(),
                                 {"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)


def best():
    return int(rpc("chain_getHeader")["result"]["number"], 16)


def finalizado():
    return int(rpc("chain_getHeader", [rpc("chain_getFinalizedHead")["result"]])["result"]["number"], 16)


def firmar(llamada, nonce, genesis, v):
    return subprocess.run([TOOL, "firmar", llamada, str(nonce), genesis, str(v["specVersion"]), str(v["transactionVersion"])],
                          capture_output=True, text=True, check=True).stdout.strip()


def eventos(n):
    h = rpc("chain_getBlockHash", [n])["result"]
    ev = rpc("state_getStorage", [EVENTS, h])["result"]
    return subprocess.run([TOOL, "eventos", ev], capture_output=True, text=True).stdout if ev else ""


def buscar(n0, texto, espera=60):
    t0 = time.time()
    while time.time() - t0 < espera:
        for n in range(n0 + 1, best() + 1):
            e = eventos(n)
            if texto in e:
                return n, e
        time.sleep(2)
    return None, ""


def falla(msg):
    print("❌", msg)
    sys.exit(1)


def main(wasm, spec_esperada):
    genesis = rpc("chain_getBlockHash", [0])["result"]
    v = rpc("state_getRuntimeVersion")["result"]
    print(f"antes: spec {v['specVersion']} · best #{best()} · finalizado #{finalizado()} · LastRuntimeUpgrade {rpc('state_getStorage', [LAST_UPGRADE])['result']}")
    if v["specVersion"] >= spec_esperada:
        falla(f"la cadena ya va en spec {v['specVersion']}")

    n0 = best()
    nonce = rpc("system_accountNextIndex", [ALICE])["result"]
    r = rpc("author_submitExtrinsic", [firmar(f"set_code:{wasm}", nonce, genesis, v)])
    if "error" in r:
        falla(f"set_code rechazado: {r['error']}")
    n, e = buscar(n0, "CodeUpdated")
    if not n or "Sudid { sudo_result: Ok" not in e:
        falla("no apareció CodeUpdated + Sudid Ok\n" + e)
    print(f"1 ✅ bloque #{n}: CodeUpdated + Sudid Ok")

    while best() < n + 1:
        time.sleep(1)
    v2 = rpc("state_getRuntimeVersion")["result"]
    lru = rpc("state_getStorage", [LAST_UPGRADE])["result"]
    if v2["specVersion"] != spec_esperada:
        falla(f"el RPC anuncia spec {v2['specVersion']}, no {spec_esperada}")
    print(f"2 ✅ spec {v2['specVersion']} · LastRuntimeUpgrade {lru}")

    nonce = rpc("system_accountNextIndex", [ALICE])["result"]
    viejo = rpc("author_submitExtrinsic", [firmar("remark", nonce, genesis, v)])
    if "bad signature" not in json.dumps(viejo):
        falla(f"una firma con la spec vieja NO fue rechazada: {viejo}")
    m0 = best()
    nuevo = rpc("author_submitExtrinsic", [firmar("remark", nonce, genesis, v2)])
    if "error" in nuevo:
        falla(f"la firma con la spec nueva fue rechazada: {nuevo['error']}")
    m, e = buscar(m0, "ExtrinsicSuccess { dispatch_info: DispatchEventInfo { weight: Weight { ref_time", 60)
    exitos = [l for l in e.splitlines() if "ExtrinsicSuccess" in l and "Normal" in l]
    if not m or not exitos:
        falla("el remark con la spec nueva no se incluyó con éxito")
    print(f"3 ✅ spec vieja → bad signature · spec nueva → incluido en #{m}: {exitos[0][:110]}")

    t0 = time.time()
    while finalizado() <= n and time.time() - t0 < 120:
        time.sleep(2)
    if finalizado() <= n:
        falla("la finalidad no pasó del bloque del upgrade en 2 min")
    print(f"4 ✅ finalizado #{finalizado()} > bloque del upgrade #{n} · best #{best()}")


if __name__ == "__main__":
    if len(sys.argv) != 3:
        sys.exit(__doc__)
    main(os.path.abspath(sys.argv[1]), int(sys.argv[2]))