universe/scripts/send-test-txs.py

#!/usr/bin/env python3
"""Send test transactions on Lux EVM chains.

Usage:
    PRIVATE_KEY=deadbeef... python send-test-txs.py
    PRIVATE_KEY=deadbeef... python send-test-txs.py --testnet --chains C Zoo --count 5
"""

import argparse
import os
import sys
import time

from web3 import Web3

CHAINS = {
    "mainnet": {
        "C":     {"chain_id": 96369,  "port": 19630, "path": "/v1/chain/c/rpc"},
        "Zoo":   {"chain_id": 200200, "port": 19630, "path": "/v1/chain/Zoo/rpc"},
        "Hanzo": {"chain_id": 36963,  "port": 19630, "path": "/v1/chain/Hanzo/rpc"},
        "SPC":   {"chain_id": 36911,  "port": 19630, "path": "/v1/chain/SPC/rpc"},
        "Pars":  {"chain_id": 494949, "port": 19630, "path": "/v1/chain/Pars/rpc"},
    },
    "testnet": {
        "C":     {"chain_id": 96368,  "port": 19640, "path": "/v1/chain/c/rpc"},
        "Zoo":   {"chain_id": 200201, "port": 19640, "path": "/v1/chain/Zoo/rpc"},
        "Hanzo": {"chain_id": 36962,  "port": 19640, "path": "/v1/chain/Hanzo/rpc"},
        "SPC":   {"chain_id": 36910,  "port": 19640, "path": "/v1/chain/SPC/rpc"},
        "Pars":  {"chain_id": 494950, "port": 19640, "path": "/v1/chain/Pars/rpc"},
    },
}

VALUE_WEI = Web3.to_wei(0.001, "ether")


def make_rpc_url(cfg):
    return f"http://localhost:{cfg['port']}{cfg['path']}"


def supports_eip1559(w3):
    try:
        latest = w3.eth.get_block("latest")
        return "baseFeePerGas" in latest
    except Exception:
        return False


def send_txs(name, cfg, account, count):
    """Send count self-transfer transactions on a single chain. Returns list of (tx_hash, success)."""
    url = make_rpc_url(cfg)
    w3 = Web3(Web3.HTTPProvider(url, request_kwargs={"timeout": 30}))

    if not w3.is_connected():
        print(f"  [{name}] SKIP - cannot connect to {url}")
        return []

    chain_id = cfg["chain_id"]
    addr = account.address
    balance = w3.eth.get_balance(addr)
    print(f"  [{name}] chain_id={chain_id} balance={Web3.from_wei(balance, 'ether')} LUX")

    if balance < VALUE_WEI * count:
        print(f"  [{name}] SKIP - insufficient balance for {count} txs")
        return []

    eip1559 = supports_eip1559(w3)
    nonce = w3.eth.get_transaction_count(addr, "pending")
    results = []

    for i in range(count):
        tx = {
            "to": addr,
            "value": VALUE_WEI,
            "nonce": nonce + i,
            "chainId": chain_id,
            "gas": 21000,
        }

        if eip1559:
            base_fee = w3.eth.get_block("latest")["baseFeePerGas"]
            tx["maxFeePerGas"] = base_fee * 2 + Web3.to_wei(1, "gwei")
            tx["maxPriorityFeePerGas"] = Web3.to_wei(1, "gwei")
            tx["type"] = 2
        else:
            tx["gasPrice"] = w3.eth.gas_price or Web3.to_wei(25, "gwei")

        signed = account.sign_transaction(tx)
        tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
        tx_hex = tx_hash.hex()
        print(f"  [{name}] tx {i+1}/{count} sent: 0x{tx_hex}")

        try:
            receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
            status = "OK" if receipt["status"] == 1 else "FAIL"
        except Exception as e:
            status = f"TIMEOUT ({e})"

        results.append((tx_hex, status))
        print(f"  [{name}] tx {i+1}/{count} status: {status}")

    return results


def main():
    parser = argparse.ArgumentParser(description="Send test transactions on Lux EVM chains")
    parser.add_argument("--testnet", action="store_true", help="Use testnet chain IDs and port")
    parser.add_argument("--chains", nargs="+", choices=["C", "Zoo", "Hanzo", "SPC", "Pars"],
                        help="Chains to send on (default: all)")
    parser.add_argument("--count", type=int, default=10, help="Number of txs per chain (default: 10)")
    args = parser.parse_args()

    pk = os.environ.get("PRIVATE_KEY")
    if not pk:
        print("ERROR: PRIVATE_KEY environment variable not set", file=sys.stderr)
        sys.exit(1)

    pk = pk.strip().removeprefix("0x")

    network = "testnet" if args.testnet else "mainnet"
    chains = CHAINS[network]
    selected = args.chains or list(chains.keys())

    account = Web3().eth.account.from_key(pk)
    print(f"Network: {network}")
    print(f"Address: {account.address}")
    print(f"Chains:  {', '.join(selected)}")
    print(f"Count:   {args.count} txs per chain")
    print()

    summary = {}
    t0 = time.monotonic()

    for name in selected:
        if name not in chains:
            print(f"  [{name}] SKIP - not configured for {network}")
            continue
        print(f"--- {name} ---")
        results = send_txs(name, chains[name], account, args.count)
        summary[name] = results
        print()

    elapsed = time.monotonic() - t0

    # Print summary
    print("=" * 60)
    print("SUMMARY")
    print("=" * 60)
    total_sent = 0
    total_ok = 0
    for name in selected:
        results = summary.get(name, [])
        ok = sum(1 for _, s in results if s == "OK")
        total_sent += len(results)
        total_ok += ok
        if results:
            print(f"  {name:6s}: {ok}/{len(results)} confirmed")
        else:
            print(f"  {name:6s}: skipped")
    print(f"  Total:  {total_ok}/{total_sent} confirmed in {elapsed:.1f}s")


if __name__ == "__main__":
    main()