#!/usr/bin/env python3 # Copyright (c) 2019 The Bitcoin Unlimited developers """ Tests to check if basic electrum server integration works """ import asyncio from decimal import Decimal from test_framework.util import wait_for, assert_equal, assert_raises_async from test_framework.electrumutil import ( get_txid_from_idem, ElectrumTestFramework, ) from test_framework.electrumconnection import ElectrumConnection from test_framework.connectrum.exc import ElectrumErrorResponse from test_framework.constants import COIN from test_framework.environment import on_nex, on_bch async def address_to_scripthash(cli, address): return await cli.call("blockchain.address.get_scripthash", address) class ElectrumBlockchainAddress(ElectrumTestFramework): """ Basic blockchain.address.* testing, mostly to check that the function handle an address correctly. The blockchain.scripthash.* equivalents are more thoroughly tested. """ async def run_test(self): n = self.nodes[0] n.generate(200) await self.test_get_frist_use(n) cli = ElectrumConnection() await cli.connect() try: await self.test_invalid_args(cli) await self.test_get_balance(n, cli) await self.test_get_history(n, cli) await self.test_list_unspent(n, cli) finally: cli.disconnect() async def test_invalid_args(self, cli): hash_param_methods = ( "blockchain.address.get_balance", "blockchain.address.get_history", "blockchain.address.listunspent", ) for method in hash_param_methods: await assert_raises_async( ElectrumErrorResponse, cli.call, method, "invalidaddress" ) async def test_get_balance(self, n, cli): addr = n.getnewaddress() balance = 11.42 n.sendtoaddress(addr, balance) async def check_address(address, unconfirmed=0, confirmed=0): res = await cli.call("blockchain.address.get_balance", address) return ( res["unconfirmed"] == unconfirmed * COIN and res["confirmed"] == confirmed * COIN ) await wait_for(10, lambda: check_address(addr, unconfirmed=balance)) n.generate(1) await wait_for(10, lambda: check_address(addr, confirmed=balance)) async def sendtoaddr(self, n, cli, addr): utxo = n.listunspent().pop() if on_bch(): inputs = [{"txid": utxo["txid"], "vout": utxo["vout"]}] elif on_nex(): inputs = [{"outpoint": utxo["outpoint"], "amount": utxo["amount"]}] else: raise NotImplementedError() fee = Decimal(300) / COIN outputs = { addr: utxo["amount"] - fee, } tx = n.createrawtransaction(inputs, outputs) signed = n.signrawtransaction(tx) txid = await cli.call("blockchain.transaction.broadcast", signed["hex"]) return txid async def test_get_frist_use(self, n): cli = ElectrumConnection() await cli.connect() # New address that has never received coins. Should return an error. addr = n.getnewaddress() scripthash = await address_to_scripthash(cli, addr) await assert_raises_async( ElectrumErrorResponse, cli.call, "blockchain.address.get_first_use", addr ) await assert_raises_async( ElectrumErrorResponse, cli.call, "blockchain.scripthash.get_first_use", scripthash, ) # Send coin to the new address txid = await get_txid_from_idem(n, await self.sendtoaddr(n, cli, addr)) # Wait for electrum server to see the utxo. async def wait_for_utxo(): utxo = await cli.call("blockchain.address.listunspent", addr) if len(utxo) == 1: return utxo return None await wait_for(10, wait_for_utxo) # Observe that get_first_use returns the tx when it's in the mempool res = await cli.call("blockchain.address.get_first_use", addr) res2 = await cli.call( "blockchain.scripthash.get_first_use", await address_to_scripthash(cli, addr), ) assert_equal(res, res2) assert_equal( "0000000000000000000000000000000000000000000000000000000000000000", res["block_hash"], ) assert_equal(0, res["height"]) assert_equal(txid, res["tx_hash"]) # Confirm tx, observe that block height and gets set. n.generate(1) await self.sync_height(cli) res = await cli.call("blockchain.address.get_first_use", addr) res2 = await cli.call( "blockchain.scripthash.get_first_use", await address_to_scripthash(cli, addr), ) assert_equal(res, res2) assert_equal(n.getbestblockhash(), res["block_hash"]) assert_equal(n.getblockcount(), res["height"]) assert_equal(txid, res["tx_hash"]) # Send another tx, observe that the first one is till returned. await self.sendtoaddr(n, cli, addr) res = await cli.call("blockchain.address.get_first_use", addr) assert_equal(txid, res["tx_hash"]) # Also when the second tx is confirmed, the first is returned. n.generate(1) await self.sync_height(cli) res = await cli.call("blockchain.address.get_first_use", addr) assert_equal(txid, res["tx_hash"]) cli.disconnect() async def test_list_unspent(self, n, cli): addr = n.getnewaddress() utxo = await cli.call("blockchain.address.listunspent", addr) assert_equal(0, len(utxo)) txidem = n.sendtoaddress(addr, 21) txid = await get_txid_from_idem(n, txidem) async def fetch_utxo(): utxo = await cli.call("blockchain.address.listunspent", addr) if len(utxo) > 0: return utxo return None utxo = await wait_for(10, fetch_utxo) assert_equal(1, len(utxo)) assert_equal(0, utxo[0]["height"]) assert_equal(txid, utxo[0]["tx_hash"]) assert_equal(21 * COIN, utxo[0]["value"]) assert utxo[0]["tx_pos"] in [0, 1] n.generate(1) async def wait_for_confheight(): utxo = await cli.call("blockchain.address.listunspent", addr) return len(utxo) == 1 and utxo[0]["height"] == n.getblockcount() await wait_for(10, wait_for_confheight) async def test_get_history(self, n, cli): addr = n.getnewaddress() txidem = n.sendtoaddress(addr, 11) txid = await get_txid_from_idem(n, txidem) async def fetch_history(): h = await cli.call("blockchain.address.get_history", addr) if len(h) > 0: return h return None history = await wait_for(10, fetch_history) assert_equal(1, len(history)) unconfirmed_height = 0 assert_equal(unconfirmed_height, history[0]["height"]) assert_equal(txid, history[0]["tx_hash"]) n.generate(1) async def wait_for_confheight(): h = await cli.call("blockchain.address.get_history", addr) return len(h) == 1 and h[0]["height"] == n.getblockcount() await wait_for(10, wait_for_confheight) if __name__ == "__main__": asyncio.run(ElectrumBlockchainAddress().main())