PYTHON test_psbt_pipeline.py

#!/usr/bin/env python3
"""Test the full PSBT pipeline: generate wallet -> import watch-only -> create PSBT -> sign -> broadcast (dry run)."""
import json
import requests

BASE = "http://127.0.0.1:14195/api"

def post(path, payload=None):
    r = requests.post(f"{BASE}{path}", json=payload or {})
    r.raise_for_status()
    return r.json()

print("1. Generate seed...")
seed = post("/generate_seed", {"strength": 128})["seed"]
print(f"   Seed: {seed}")

print("2. Create wallet...")
wallet_resp = post("/create_wallet", {
    "seed": seed,
    "wallet_name": "TestWallet",
    "password": "test123",
})
wallet = wallet_resp["wallet"]
addresses = wallet_resp["addresses"]
print(f"   Wallet xpub: {wallet['xpub'][:20]}...")
print(f"   First address: {addresses[0]['address']}")

print("3. Import watch-only wallet...")
xpub = wallet["xpub"]
import_resp = post("/import_wallet", {
    "xpub": xpub,
    "wallet_name": "watchonly",
})
print(f"   Import result: {json.dumps(import_resp.get('imported', []), indent=2)[:200]}")

print("4. Get UTXOs...")
addrs = [a["address"] for a in addresses]
utxo_resp = post("/get_utxos", {"addresses": addrs})
utxos = utxo_resp["utxos"]
print(f"   UTXOs found: {len(utxos)}")

if not utxos:
    print("\n   No UTXOs — can't test PSBT creation (expected on testnet/fresh wallet)")
    print("   Pipeline structure verified. To test end-to-end, send BTC to one of the addresses.")
else:
    # Use a tiny output for testing
    addr = addresses[0]["address"]
    balance_btc = sum(u["amount"] for u in utxos)
    send_amount = balance_btc * 0.5  # Send half, keep half as change

    print("5. Create PSBT...")
    psbt_resp = post("/create_psbt", {
        "wallet": wallet,
        "outputs": [{"address": addr, "amount": send_amount}],
        "fee_rate": 1.0,
        "password": "test123",
    })
    print(f"   PSBT created: {psbt_resp['psbt'][:50]}...")
    print(f"   Fee: {psbt_resp['fee_btc']} BTC")

    print("6. Sign PSBT...")
    sign_resp = post("/sign_psbt", {
        "wallet": wallet,
        "psbt": psbt_resp["psbt"],
        "password": "test123",
    })
    print(f"   All signed: {sign_resp['all_signed']}")
    print(f"   Finalized: {sign_resp['finalized']}")
    if sign_resp.get("errors"):
        print(f"   Errors: {sign_resp['errors']}")
    if sign_resp.get("raw_tx_hex"):
        print(f"   Raw TX: {sign_resp['raw_tx_hex'][:50]}...")

        print("7. Decode TX (dry run, NOT broadcasting)...")
        decode_resp = post("/decode_psbt", {"psbt": sign_resp["raw_tx_hex"]})
        print(f"   TX decoded: {len(decode_resp)} keys returned")
        print(f"   Inputs: {len(decode_resp.get('vin', []))}")
        print(f"   Outputs: {len(decode_resp.get('vout', []))}")
        print(f"   TXID: {decode_resp.get('txid', 'N/A')}")

print("\nDone!")