#!/usr/bin/env python3 # Copyright (c) 2020 The Bitcoin Unlimited developers """ Tests the electrum call 'blockchain.transaction.get' """ import asyncio from test_framework.util import assert_equal from test_framework.electrumutil import ElectrumTestFramework from test_framework.serialize import to_hex from test_framework.utiltx import pad_tx from test_framework.script import ( CScript, OP_CHECKSIG, OP_DROP, OP_DUP, OP_EQUAL, OP_EQUALVERIFY, OP_FALSE, OP_HASH160, OP_TRUE, ) from test_framework.constants import COIN from test_framework.environment import on_bch, on_nex, node, NodeFeature, node_supports from test_framework.electrumconnection import ElectrumConnection if on_bch(): from test_framework.bch.blocktools import create_transaction elif on_nex(): from test_framework.nex.blocktools import create_transaction else: raise NotImplementedError() TX_GET = "blockchain.transaction.get" DUMMY_HASH = 0x1111111111111111111111111111111111111111 class ElectrumTransactionGet(ElectrumTestFramework): async def run_test(self): n = self.nodes[0] await self.bootstrap_p2p() cli = ElectrumConnection() await cli.connect() coinbases = await self.mine_blocks(cli, n, 104) fee = 300 # non-coinbase transactions prevtx = coinbases[0] nonstandard_tx = create_transaction( prevtx=prevtx, value=prevtx.vout[0].n_value - fee, n=0, sig=CScript([OP_TRUE]), out=CScript([OP_FALSE, OP_DROP]), ) prevtx = coinbases[1] p2sh_tx = create_transaction( prevtx=prevtx, value=prevtx.vout[0].n_value - fee, n=0, sig=CScript([OP_TRUE]), out=CScript([OP_HASH160, DUMMY_HASH, OP_EQUAL]), ) prevtx = coinbases[2] p2pkh_tx = create_transaction( prevtx=prevtx, value=prevtx.vout[0].n_value - fee, n=0, sig=CScript([OP_TRUE]), out=CScript([OP_DUP, OP_HASH160, DUMMY_HASH, OP_EQUALVERIFY, OP_CHECKSIG]), ) prevtx = coinbases[3] unconfirmed_tx = create_transaction( prevtx=prevtx, value=prevtx.vout[0].n_value - fee, n=0, sig=CScript([OP_TRUE]), out=CScript([OP_DUP, OP_HASH160, DUMMY_HASH, OP_EQUALVERIFY, OP_CHECKSIG]), ) for tx in [nonstandard_tx, p2sh_tx, p2pkh_tx, unconfirmed_tx]: pad_tx(tx) coinbases.extend( await self.mine_blocks(cli, n, 1, [nonstandard_tx, p2sh_tx, p2pkh_tx]) ) n.sendrawtransaction(to_hex(unconfirmed_tx)) await self.sync_mempool_count(cli, n) await asyncio.gather( self.test_verbose( n, cli, nonstandard_tx.get_rpc_hex_id(), p2sh_tx.get_rpc_hex_id(), p2pkh_tx.get_rpc_hex_id(), unconfirmed_tx.get_rpc_hex_id(), ), self.test_non_verbose(cli, coinbases, unconfirmed_tx), self.test_input_amounts( cli, coinbases[0].get_rpc_hex_id(), unconfirmed_tx.get_rpc_hex_id() ), self.test_by_txidem(cli, p2sh_tx), ) if node_supports(node(), NodeFeature.TOKENS): await self.test_token(n, cli) await self.test_by_txidem(cli, p2sh_tx) cli.disconnect() async def test_non_verbose(self, cli, coinbases, unconfirmed): for tx in coinbases + [unconfirmed]: assert_equal(to_hex(tx), await cli.call(TX_GET, tx.get_rpc_hex_id())) async def test_input_amounts(self, cli, coinbase_tx, unconfirmed_tx): """ Test that input amount are included for vin. """ # Nex coinbases don't have inputs. if not on_nex(): # Coinbase does not have a input amount tx = await cli.call(TX_GET, coinbase_tx, True) assert_equal(None, tx["vin"][0]["value_satoshi"]) assert_equal(None, tx["vin"][0]["value_coin"]) # This tx spends a coinbase transaction if on_nex(): coinbase_reward = 10000000 else: coinbase_reward = 50 tx = await cli.call(TX_GET, unconfirmed_tx, True) assert_equal(coinbase_reward, tx["vin"][0]["value_coin"]) assert_equal(coinbase_reward * COIN, tx["vin"][0]["value_satoshi"]) async def create_token_tx(self, n, cli): token_id = self.create_token(to_addr=n.getnewaddress(), mint_amount=42) await self.sync_mempool_count(cli, n) h = await cli.call("token.transaction.get_history", token_id) return h["history"] async def test_token(self, n, cli): # We use the node to generate token transactions, so # we need to give it a coin n.generate(101) txs = await self.create_token_tx(n, cli) txid = txs[0]["tx_hash"] def has_token_data(x): if on_bch(): return "tokenData" in x and not x["tokenData"] is None if on_nex(): return ( "group" in x["scriptPubKey"] and not x["scriptPubKey"]["group"] is None ) raise Exception("NYI") await self.sync_mempool_count(cli, n) bitcoind = list(filter(has_token_data, n.getrawtransaction(txid, True)["vout"])) rostrum = list( filter(has_token_data, (await cli.call(TX_GET, txid, True))["vout"]) ) assert len(rostrum) >= 1, f"{rostrum} {bitcoind}" assert len(bitcoind) >= 1, f"{rostrum} {bitcoind}" rostrum = rostrum[0] bitcoind = bitcoind[0] if on_bch(): assert_equal( bitcoind["tokenData"]["category"], rostrum["tokenData"]["category"] ) assert_equal( bitcoind["tokenData"]["amount"], rostrum["tokenData"]["amount"] ) if on_nex(): assert_equal( bitcoind["scriptPubKey"]["group"], rostrum["scriptPubKey"]["group"] ) assert_equal( bitcoind["scriptPubKey"]["groupQuantity"], rostrum["scriptPubKey"]["groupQuantity"], ) assert_equal( bitcoind["scriptPubKey"]["groupAuthority"], rostrum["scriptPubKey"]["groupAuthority"], ) group = rostrum["scriptPubKey"]["group"] quantity = rostrum["scriptPubKey"]["groupQuantity"] authority = rostrum["scriptPubKey"]["groupAuthority"] def has_input_token_data(x): return "group" in x and not x["group"] is None tx1 = await cli.call(TX_GET, txs[1]["tx_hash"], True) # possible minting tx rostrum = list(filter(has_input_token_data, tx1["vin"])) if len(rostrum) < 1: # history call might return unordered list and tx1 is actually the 'create' rostrum = list( filter( has_input_token_data, (await cli.call(TX_GET, txs[0]["tx_hash"], True))["vin"], ) ) # in this case we need to update the quantity to the create tx tx_out = list(filter(has_token_data, tx1["vout"])) assert len(tx_out) >= 1, f"{tx_out}" tx_out = tx_out[0] quantity = tx_out["scriptPubKey"]["groupQuantity"] authority = tx_out["scriptPubKey"]["groupAuthority"] assert len(rostrum) >= 1, f"{rostrum}" rostrum = rostrum[0] assert_equal(group, rostrum["group"]) assert_equal(quantity, rostrum["groupQuantity"]) assert_equal(authority, rostrum["groupAuthority"]) # pylint: disable=too-many-arguments,too-many-statements async def test_verbose( self, n, cli, nonstandard_tx, p2sh_tx, p2pkh_tx, unconfirmed_tx ): """ The spec is unclear. It states: "whatever the coin daemon returns when asked for a verbose form of the raw transaction" We've implemented this call in rostrum with "common denominators" between the bitcoind implementations and some extras. """ # All confirmed transactions are confirmed in the tip block = n.getbestblockhash() tipheight = n.getblockcount() if on_bch(): coinbase_tx = n.getblock(block, 1)["tx"][0] elif on_nex(): coinbase_tx = n.getblock(block, 1)["txid"][0] else: raise NotImplementedError() # pylint: disable=too-many-branches async def check_tx(txid, is_confirmed=True, check_output_type=False): electrum = await cli.call(TX_GET, txid, True) if is_confirmed: bitcoind = n.getrawtransaction(txid, True, block) else: bitcoind = n.getrawtransaction(txid, True) if not is_confirmed: # Transaction is unconfirmed. We handle this slightly different # than bitcoind. assert_equal(None, electrum["blockhash"]) assert_equal(None, electrum["confirmations"]) assert_equal(None, electrum["time"]) assert_equal(None, electrum["height"]) else: assert_equal(n.getbestblockhash(), electrum["blockhash"]) assert_equal(1, electrum["confirmations"]) assert_equal(bitcoind["time"], electrum["time"]) assert_equal(tipheight, electrum["height"]) assert_equal(bitcoind["txid"], electrum["txid"]) assert_equal(bitcoind["locktime"], electrum["locktime"]) assert_equal(bitcoind["size"], electrum["size"]) assert_equal(bitcoind["hex"], electrum["hex"]) assert_equal(bitcoind["version"], electrum["version"]) if "fee" in bitcoind: assert_equal(bitcoind["fee"], electrum["fee"]) else: assert "fee" in electrum # inputs assert_equal(len(bitcoind["vin"]), len(bitcoind["vin"])) for i in range(len(bitcoind["vin"])): if "coinbase" in bitcoind["vin"][i]: # bitcoind drops txid and other fields, butadds 'coinbase' for coinbase # inputs assert_equal( bitcoind["vin"][i]["coinbase"], electrum["vin"][i]["coinbase"] ) assert_equal( bitcoind["vin"][i]["sequence"], electrum["vin"][i]["sequence"] ) continue if on_nex(): assert_equal( bitcoind["vin"][i]["outpoint"], electrum["vin"][i]["outpoint"] ) assert "addresses" in electrum["vin"][i] assert "group" in electrum["vin"][i] if on_bch(): assert_equal(bitcoind["vin"][i]["txid"], electrum["vin"][i]["txid"]) assert_equal(bitcoind["vin"][i]["vout"], electrum["vin"][i]["vout"]) assert_equal( bitcoind["vin"][i]["sequence"], electrum["vin"][i]["sequence"] ) assert_equal( bitcoind["vin"][i]["scriptSig"]["hex"], electrum["vin"][i]["scriptSig"]["hex"], ) # There is more than one way to represent script as assembly. # For instance '51' can be represented as '1' or 'OP_PUSHNUM_1'. # Just check for existance. assert "asm" in electrum["vin"][i]["scriptSig"] # outputs assert_equal(len(bitcoind["vout"]), len(bitcoind["vout"])) for i in range(len(bitcoind["vout"])): assert_equal(bitcoind["vout"][i]["n"], electrum["vout"][i]["n"]) if on_nex(): assert_equal( bitcoind["vout"][i]["type"], electrum["vout"][i]["type"] ) if "scriptHash" in bitcoind["vout"][i]["scriptPubKey"]: assert_equal( bitcoind["vout"][i]["scriptPubKey"]["scriptHash"], electrum["vout"][i]["scriptPubKey"]["scriptHash"], ) assert_equal( bitcoind["vout"][i]["scriptPubKey"]["argsHash"], electrum["vout"][i]["scriptPubKey"]["argsHash"], ) else: assert_equal( None, electrum["vout"][i]["scriptPubKey"]["scriptHash"] ) assert_equal( None, electrum["vout"][i]["scriptPubKey"]["argsHash"] ) assert_equal( bitcoind["vout"][i]["value"] * COIN, electrum["vout"][i]["value_coin"] * COIN, ) assert_equal( bitcoind["vout"][i]["value"] * COIN, electrum["vout"][i]["value_satoshi"], ) assert_equal( bitcoind["vout"][i]["scriptPubKey"]["hex"], electrum["vout"][i]["scriptPubKey"]["hex"], ) assert "asm" in electrum["vout"][i]["scriptPubKey"] if on_nex(): if "addresses" in bitcoind["vout"][i]["scriptPubKey"]: assert_equal( bitcoind["vout"][i]["scriptPubKey"]["addresses"], electrum["vout"][i]["scriptPubKey"]["addresses"], ) assert "outpoint_hash" in electrum["vout"][i] if on_bch(): if "addresses" in bitcoind["vout"][i]["scriptPubKey"]: for addr in bitcoind["vout"][i]["scriptPubKey"]["addresses"]: assert ( addr in electrum["vout"][i]["scriptPubKey"]["addresses"] ), f"address {addr} missing in response" else: assert_equal( [], electrum["vout"][i]["scriptPubKey"]["addresses"] ) if check_output_type: assert_equal( bitcoind["vout"][i]["scriptPubKey"]["type"], electrum["vout"][i]["scriptPubKey"]["type"], ) await asyncio.gather( # rostrum cannot tell if it's nonstandard check_tx(nonstandard_tx, check_output_type=False), check_tx(p2sh_tx), check_tx(p2pkh_tx), check_tx(coinbase_tx), check_tx(unconfirmed_tx, is_confirmed=False), ) async def test_by_txidem(self, cli, confirmed_tx): if not on_nex(): return assert_equal( to_hex(confirmed_tx), await cli.call(TX_GET, confirmed_tx.get_rpc_hex_idem(), False), ) # Transactions fetched by idem should have height # (issue #214) verbose = await cli.call(TX_GET, confirmed_tx.get_rpc_hex_idem(), True) assert verbose["height"] >= 100 if __name__ == "__main__": asyncio.run(ElectrumTransactionGet().main())