#!/usr/bin/env python3
"""Verify wallet-scoped RPC calls and PSBT pipeline."""
import json
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "wallet-engine"))
from wallet import WalletEngine
eng = WalletEngine()
print("=" * 60)
print("Testing wallet-scoped RPC pipeline")
print("=" * 60)
# --- Test 1: Basic RPC (no wallet needed) ---
print("\n--- Test 1: getblockcount (base URL) ---")
try:
height = eng.get_block_count()
print(f" Block height: {height} [OK]")
except Exception as e:
print(f" FAIL: {e}")
# --- Test 2: import_wallet_watchonly ---
print("\n--- Test 2: import_wallet_watchonly (wallet-scoped) ---")
try:
# Generate a test mnemonic
test_mnemonic = eng.generate_mnemonic(128)
bip84 = eng.derive_bip84(test_mnemonic)
xpub = bip84["xpub"]
print(f" Test XPUB: {xpub[:30]}...")
# Create wallet from dict
wallet = eng.create_wallet("test_wallet", test_mnemonic)
# Import to node
result = eng.import_wallet_watchonly(xpub, "test_watch")
print(f" Import result: {json.dumps(result.get('imported', []), indent=2)}")
print(f" [OK] Wallet imported")
except Exception as e:
print(f" FAIL: {e}")
# --- Test 3: get_balance (wallet-scoped) ---
print("\n--- Test 3: get_balance (wallet-scoped via listunspent) ---")
try:
addrs = eng.get_addresses(wallet, count=3)
addr = addrs[0]["address"]
print(f" Querying balance for: {addr}")
balance = eng.get_balance(addr)
print(f" Balance: {balance}")
print(f" [OK] Balance query worked")
except Exception as e:
print(f" FAIL: {e}")
# --- Test 4: get_utxos (wallet-scoped) ---
print("\n--- Test 4: get_utxos (wallet-scoped) ---")
try:
addrs = eng.get_addresses(wallet, count=3)
addr_list = [a["address"] for a in addrs]
utxos = eng.get_utxos(addr_list)
print(f" UTXOs found: {len(utxos)}")
print(f" [OK] UTXO query worked")
except Exception as e:
print(f" FAIL: {e}")
# --- Test 5: create_psbt ---
print("\n--- Test 5: create_psbt (base URL) ---")
try:
# Use a dummy output - expect "insufficient balance" on test wallet
result = eng.create_psbt(
wallet,
outputs=[{"address": "bc1qar0srrr7xfkvy5l643lydnw9re59gtzzwf5mdq", "amount": 0.001}],
fee_rate=2.0,
)
print(f" PSBT created: {result.get('psbt', '')[:50]}...")
print(f" [OK] PSBT created")
except ValueError as e:
# Expected for empty wallet
print(f" Expected (no funds): {e}")
print(f" [OK] Balance check works")
except Exception as e:
print(f" FAIL: {e}")
# --- Test 6: sign_psbt structure ---
print("\n--- Test 6: sign_psbt method exists ---")
print(f" sign_psbt: {hasattr(eng, 'sign_psbt')} [OK]")
print(f" finalize_psbt: {hasattr(eng, 'finalize_psbt')} [OK]")
print(f" sign_and_finalize_psbt: {hasattr(eng, 'sign_and_finalize_psbt')} [OK]")
print(f" broadcast_tx: {hasattr(eng, 'broadcast_tx')} [OK]")
# --- Test 7: Verify RPC URL routing ---
print("\n--- Test 7: RPC URL routing ---")
test_methods = [
("getblockcount", "base"),
("importdescriptors", "wallet/watchonly"),
("listunspent", "wallet/watchonly"),
("listtransactions", "wallet/watchonly"),
("createpsbt", "base"),
("signrawtransactionwithkey", "base"),
("finalizepsbt", "base"),
("sendrawtransaction", "base"),
]
for method, expected_scope in test_methods:
if method in eng._WALLET_RPC_METHODS:
actual = "wallet/watchonly"
else:
actual = "base"
status = "OK" if actual == expected_scope else "MISMATCH"
print(f" {method}: {actual} (expected {expected_scope}) [{status}]")
print("\n" + "=" * 60)
print("All structural tests complete")
print("=" * 60)