From d512c585430c35d32cee3c3b9d820bacd0be5f01 Mon Sep 17 00:00:00 2001 From: zebra-lucky Date: Tue, 19 Aug 2025 01:43:37 +0300 Subject: [PATCH] fix test/unified/test_bumpfee.py --- test/unified/common.py | 62 ++- test/unified/test_bumpfee.py | 807 ++++++++++++++++++----------------- 2 files changed, 454 insertions(+), 415 deletions(-) diff --git a/test/unified/common.py b/test/unified/common.py index 723a01a..f96cfbe 100644 --- a/test/unified/common.py +++ b/test/unified/common.py @@ -10,6 +10,9 @@ from decimal import Decimal data_dir = os.path.dirname(os.path.dirname(os.path.realpath(__file__))) sys.path.insert(0, os.path.join(data_dir)) +from unittest import IsolatedAsyncioTestCase +from twisted.trial.unittest import TestCase as TrialTestCase + from jmbase import get_log from jmclient import open_test_wallet_maybe, BIP32Wallet, SegwitWallet, \ estimate_tx_fee, jm_single, WalletService, BaseWallet, WALLET_IMPLEMENTATIONS @@ -19,13 +22,24 @@ from jmbase import chunks log = get_log() -def make_sign_and_push(ins_full, - wallet_service, - amount, - output_addr=None, - change_addr=None, - hashcode=btc.SIGHASH_ALL, - estimate_fee = False): + +class TrialAsyncioTestCase(TrialTestCase, IsolatedAsyncioTestCase): + + def __init__(self, methodName='runTest'): + IsolatedAsyncioTestCase.__init__(self, methodName) + TrialTestCase.__init__(self, methodName) + + def __call__(self, *args, **kwds): + return IsolatedAsyncioTestCase.run(self, *args, **kwds) + + +async def make_sign_and_push(ins_full, + wallet_service, + amount, + output_addr=None, + change_addr=None, + hashcode=btc.SIGHASH_ALL, + estimate_fee = False): """Utility function for easily building transactions from wallets. """ @@ -33,8 +47,12 @@ def make_sign_and_push(ins_full, total = sum(x['value'] for x in ins_full.values()) ins = ins_full.keys() #random output address and change addr - output_addr = wallet_service.get_new_addr(1, BaseWallet.ADDRESS_TYPE_INTERNAL) if not output_addr else output_addr - change_addr = wallet_service.get_new_addr(0, BaseWallet.ADDRESS_TYPE_INTERNAL) if not change_addr else change_addr + output_addr = await wallet_service.get_new_addr( + 1, + BaseWallet.ADDRESS_TYPE_INTERNAL) if not output_addr else output_addr + change_addr = wallet_service.get_new_addr( + 0, + BaseWallet.ADDRESS_TYPE_INTERNAL) if not change_addr else change_addr fee_est = estimate_tx_fee(len(ins), 2) if estimate_fee else 10000 outs = [{'value': amount, 'address': output_addr}, {'value': total - amount - fee_est, @@ -57,17 +75,17 @@ def make_sign_and_push(ins_full, else: return False -def make_wallets(n, - wallet_structures=None, - mean_amt=1, - sdev_amt=0, - start_index=0, - fixed_seeds=None, - test_wallet=False, - passwords=None, - walletclass=SegwitWallet, - mixdepths=5, - fb_indices=[]): +async def make_wallets(n, + wallet_structures=None, + mean_amt=1, + sdev_amt=0, + start_index=0, + fixed_seeds=None, + test_wallet=False, + passwords=None, + walletclass=SegwitWallet, + mixdepths=5, + fb_indices=[]): '''n: number of wallets to be created wallet_structure: array of n arrays , each subarray specifying the number of addresses to be populated with coins @@ -105,8 +123,8 @@ def make_wallets(n, print("for index: {}, we got wallet type: {}".format(i, wc)) else: wc = walletclass - w = open_test_wallet_maybe(seeds[i], seeds[i], mixdepths - 1, - test_wallet_cls=wc) + w = await open_test_wallet_maybe(seeds[i], seeds[i], mixdepths - 1, + test_wallet_cls=wc) wallet_service = WalletService(w) wallets[i + start_index] = {'seed': seeds[i].decode('ascii'), diff --git a/test/unified/test_bumpfee.py b/test/unified/test_bumpfee.py index e2bfa61..1bf095b 100644 --- a/test/unified/test_bumpfee.py +++ b/test/unified/test_bumpfee.py @@ -7,6 +7,8 @@ from scripts.bumpfee import ( check_valid_candidate, compute_bump_fee, create_bumped_tx, sign_transaction, sign_psbt) +from common import TrialAsyncioTestCase + def fund_wallet_addr(wallet, addr, value_btc=1): # special case, grab_coins returns hex from rpc: txin_id = hextobin(jm_single().bc_interface.grab_coins(addr, value_btc)) @@ -16,396 +18,415 @@ def fund_wallet_addr(wallet, addr, value_btc=1): assert len(utxo_in) == 1 return list(utxo_in.keys())[0] -def test_tx_vsize(setup_wallet): - # tests that we correctly compute the transaction size - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": 10**8 - amount_sats - 142}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - - assert btc.tx_vsize(tx) in (142, 143) # transaction size may vary due to signature - -def test_check_valid_candidate_confirmed_tx(setup_wallet): - # test that the replaceable transaction is unconfirmed - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": 10**8 - amount_sats - 142}]) - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - jm_single().bc_interface.tick_forward_chain(1) - - with pytest.raises(RuntimeWarning, match="Transaction already confirmed. Nothing to do."): - check_valid_candidate(tx, wallet) - -def test_check_valid_candidate_unowned_input(setup_wallet): - # tests that all inputs in the replaceable transaction belong to the wallet - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - - mnemonic = 'abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about' - entropy = SegwitWallet.entropy_from_mnemonic(mnemonic) - storage = VolatileStorage() - SegwitWallet.initialize( - storage, get_network(), entropy=entropy, max_mixdepth=0) - wallet_ext = SegwitWallet(storage) - addr_ext = wallet_ext.get_external_addr(0) - utxo_ext = fund_wallet_addr(wallet_ext, addr_ext) - - tx = btc.mktx([utxo, utxo_ext], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": (2 * 10**8) - amount_sats - 210}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success, msg = wallet_ext.sign_tx(tx, {1: (wallet_ext.addr_to_script(addr_ext), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - - with pytest.raises(ValueError, match="Transaction inputs should belong to the wallet."): - check_valid_candidate(tx, wallet) - -def test_check_valid_candidate_explicit_output_index(setup_wallet): - # tests that there's at least one output that we own and can deduct fees - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": 10**8 - amount_sats - 143}, - {"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x01").to_p2sh_scriptPubKey())), - "value": amount_sats}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - - assert check_valid_candidate(tx, wallet, 0) == None - -def test_check_valid_candidate_one_output(setup_wallet): - # tests that there's at least one output that we own and can deduct fees - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": 10**8 - 111}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - - assert check_valid_candidate(tx, wallet) == None - -def test_check_valid_candidate_no_owned_outputs(setup_wallet): - # tests that there's at least one output that we own and can deduct fees - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": 10**8 - amount_sats - 143}, - {"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x01").to_p2sh_scriptPubKey())), - "value": amount_sats}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - - with pytest.raises(ValueError, match="Transaction has no obvious output we can deduct fees from. " - "Specify the output to pay from using the -o option."): - check_valid_candidate(tx, wallet) - -def test_check_valid_candidate(setup_wallet): - # tests that all checks are passed for a valid replaceable transaction - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": 10**8 - amount_sats - 142}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - - assert check_valid_candidate(tx, wallet) == None - -def test_compute_bump_fee(setup_wallet): - # tests that the compute_bump_fee method correctly calculates - # the fee by which to bump the transaction - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": 10**8 - amount_sats - 142}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - - assert compute_bump_fee(tx, 2000) in (142, 144) # will vary depending on signature size - -def test_create_bumped_tx(setup_wallet): - # tests that the bumped transaction has a change output with amount - # less the bump fee - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": 10**8 - amount_sats - 142}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 2000, wallet) - - assert orig_tx.vin[0] == bumped_tx.vin[0] - assert orig_tx.vout[0] == bumped_tx.vout[0] - assert (orig_tx.vout[1].nValue - bumped_tx.vout[1].nValue) in (142, 144) - -def test_create_bumped_tx_dust_change(setup_wallet): - # tests that the change output gets dropped when it's at or below dust - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**8 - jm_single().BITCOIN_DUST_THRESHOLD - 142 - change_sats = 10**8 - amount_sats - 142 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": change_sats}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 2000, wallet) - - assert orig_tx.vin[0] == bumped_tx.vin[0] - assert orig_tx.vout[0] == bumped_tx.vout[0] - assert len(bumped_tx.vout) == 1 - -def test_create_bumped_tx_multi_dust_change(setup_wallet): - # tests that several change outputs get dropped when they are at or below dust - # to fulfill fee requirements - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**8 - (546*18) - 669 - change_sats = 546 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}] + - [{"address": wallet.get_internal_addr(0), - "value": change_sats} for ix in range(18)]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 3000, wallet) - - assert orig_tx.vin[0] == bumped_tx.vin[0] - assert orig_tx.vout[0] == bumped_tx.vout[0] - assert len(bumped_tx.vout) == 16 - -def test_create_bumped_tx_single_output(setup_wallet): - # tests that fees are deducted from the only output available - # in the transaction - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**8 - 111 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 2000, wallet) - - assert orig_tx.vin[0] == bumped_tx.vin[0] - assert (orig_tx.vout[0].nValue - bumped_tx.vout[0].nValue) in (111, 113) - -def test_create_bumped_tx_output_index(setup_wallet): - # tests that the bumped transaction deducts its fee from the specified - # output even if it is an external wallet address - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**7 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": 10**8 - amount_sats - 142}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 2000, wallet, 0) - - assert orig_tx.vin[0] == bumped_tx.vin[0] - assert orig_tx.vout[1] == bumped_tx.vout[1] - assert (orig_tx.vout[0].nValue - bumped_tx.vout[0].nValue) in (142, 144) - -def test_create_bumped_tx_no_change(setup_wallet): - # tests that the bumped transaction is the same as the original if fees - # cannot be deducted - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr, 0.00002843) - amount_sats = 2730 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 2843)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 3000, wallet) - - assert orig_tx.vin[0] == bumped_tx.vin[0] - assert orig_tx.vout[0] == bumped_tx.vout[0] - -def test_sign_and_broadcast(setup_wallet): - # tests that we can correctly sign and broadcast a replaced transaction - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**8 - jm_single().BITCOIN_DUST_THRESHOLD - 142 - change_sats = 10**8 - amount_sats - 142 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": change_sats}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 2000, wallet) - sign_transaction(bumped_tx, orig_tx, wallet_service) - - assert jm_single().bc_interface.pushtx(bumped_tx.serialize()) == True - -def test_sign_psbt_broadcast(setup_wallet): - # tests that we can correctly sign and broadcast a replaced psbt transaction - wallet = setup_wallet[0] - wallet_service = setup_wallet[1] - wallet_service.resync_wallet() - addr = wallet.get_external_addr(0) - utxo = fund_wallet_addr(wallet, addr) - amount_sats = 10**8 - jm_single().BITCOIN_DUST_THRESHOLD - 142 - change_sats = 10**8 - amount_sats - 142 - tx = btc.mktx([utxo], - [{"address": str(btc.CCoinAddress.from_scriptPubKey( - btc.CScript(b"\x00").to_p2sh_scriptPubKey())), - "value": amount_sats}, - {"address": wallet.get_internal_addr(0), - "value": change_sats}]) - tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable - success, msg = wallet.sign_tx(tx, {0: (wallet.addr_to_script(addr), 10**8)}) - success = jm_single().bc_interface.pushtx(tx.serialize()) - orig_tx = tx.clone() - - bumped_tx = create_bumped_tx(tx, 2000, wallet) - psbt = sign_psbt(bumped_tx, orig_tx, wallet_service) - - assert jm_single().bc_interface.pushtx(psbt.extract_transaction().serialize()) == True - - -@pytest.fixture(scope='module') -def setup_wallet(request): - load_test_config() - btc.select_chain_params("bitcoin/regtest") - #see note in cryptoengine.py: - cryptoengine.BTC_P2WPKH.VBYTE = 100 - jm_single().bc_interface.tick_forward_chain_interval = 2 - jm_single().config.set('BLOCKCHAIN', 'network', 'testnet') - mnemonic = 'zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo abstract' - entropy = SegwitWallet.entropy_from_mnemonic(mnemonic) - storage = VolatileStorage() - SegwitWallet.initialize( - storage, get_network(), entropy=entropy, max_mixdepth=1) - wallet = SegwitWallet(storage) - wallet_service = WalletService(wallet) - return [wallet, wallet_service] + +class BumpFeeTests(TrialAsyncioTestCase): + + async def asyncSetUp(self): + load_test_config() + btc.select_chain_params("bitcoin/regtest") + #see note in cryptoengine.py: + cryptoengine.BTC_P2WPKH.VBYTE = 100 + jm_single().bc_interface.tick_forward_chain_interval = 2 + jm_single().config.set('BLOCKCHAIN', 'network', 'testnet') + mnemonic = 'zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo abstract' + entropy = SegwitWallet.entropy_from_mnemonic(mnemonic) + storage = VolatileStorage() + SegwitWallet.initialize( + storage, get_network(), entropy=entropy, max_mixdepth=1) + self.wallet = wallet = SegwitWallet(storage) + await wallet.async_init(storage) + self.wallet_service = WalletService(wallet) + + async def test_tx_vsize(self): + # tests that we correctly compute the transaction size + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": 10**8 - amount_sats - 142}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + + assert btc.tx_vsize(tx) in (142, 143) # transaction size may vary due to signature + + async def test_check_valid_candidate_confirmed_tx(self): + # test that the replaceable transaction is unconfirmed + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": 10**8 - amount_sats - 142}]) + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + jm_single().bc_interface.tick_forward_chain(1) + + with pytest.raises(RuntimeWarning, match="Transaction already confirmed. Nothing to do."): + check_valid_candidate(tx, wallet) + + async def test_check_valid_candidate_unowned_input(self): + # tests that all inputs in the replaceable transaction belong to the wallet + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + + mnemonic = 'abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about' + entropy = SegwitWallet.entropy_from_mnemonic(mnemonic) + storage = VolatileStorage() + SegwitWallet.initialize( + storage, get_network(), entropy=entropy, max_mixdepth=0) + wallet_ext = SegwitWallet(storage) + await wallet_ext.async_init(storage) + addr_ext = await wallet_ext.get_external_addr(0) + utxo_ext = fund_wallet_addr(wallet_ext, addr_ext) + + tx = btc.mktx([utxo, utxo_ext], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": (2 * 10**8) - amount_sats - 210}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success, msg = await wallet_ext.sign_tx( + tx, {1: (wallet_ext.addr_to_script(addr_ext), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + + with pytest.raises(ValueError, match="Transaction inputs should belong to the wallet."): + check_valid_candidate(tx, wallet) + + async def test_check_valid_candidate_explicit_output_index(self): + # tests that there's at least one output that we own and can deduct fees + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": 10**8 - amount_sats - 143}, + {"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x01").to_p2sh_scriptPubKey())), + "value": amount_sats}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + + assert check_valid_candidate(tx, wallet, 0) == None + + async def test_check_valid_candidate_one_output(self): + # tests that there's at least one output that we own and can deduct fees + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": 10**8 - 111}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + + assert check_valid_candidate(tx, wallet) == None + + async def test_check_valid_candidate_no_owned_outputs(self): + # tests that there's at least one output that we own and can deduct fees + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": 10**8 - amount_sats - 143}, + {"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x01").to_p2sh_scriptPubKey())), + "value": amount_sats}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + + with pytest.raises(ValueError, match="Transaction has no obvious output we can deduct fees from. " + "Specify the output to pay from using the -o option."): + check_valid_candidate(tx, wallet) + + async def test_check_valid_candidate(self): + # tests that all checks are passed for a valid replaceable transaction + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": 10**8 - amount_sats - 142}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + + assert check_valid_candidate(tx, wallet) == None + + async def test_compute_bump_fee(self): + # tests that the compute_bump_fee method correctly calculates + # the fee by which to bump the transaction + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": 10**8 - amount_sats - 142}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + + assert compute_bump_fee(tx, 2000) in (142, 144) # will vary depending on signature size + + async def test_create_bumped_tx(self): + # tests that the bumped transaction has a change output with amount + # less the bump fee + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": 10**8 - amount_sats - 142}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 2000, wallet) + + assert orig_tx.vin[0] == bumped_tx.vin[0] + assert orig_tx.vout[0] == bumped_tx.vout[0] + assert (orig_tx.vout[1].nValue - bumped_tx.vout[1].nValue) in (142, 144) + + async def test_create_bumped_tx_dust_change(self): + # tests that the change output gets dropped when it's at or below dust + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**8 - jm_single().BITCOIN_DUST_THRESHOLD - 142 + change_sats = 10**8 - amount_sats - 142 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": change_sats}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 2000, wallet) + + assert orig_tx.vin[0] == bumped_tx.vin[0] + assert orig_tx.vout[0] == bumped_tx.vout[0] + assert len(bumped_tx.vout) == 1 + + async def test_create_bumped_tx_multi_dust_change(self): + # tests that several change outputs get dropped when they are at or below dust + # to fulfill fee requirements + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**8 - (546*18) - 669 + change_sats = 546 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}] + + [{"address": await wallet.get_internal_addr(0), + "value": change_sats} for ix in range(18)]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 3000, wallet) + + assert orig_tx.vin[0] == bumped_tx.vin[0] + assert orig_tx.vout[0] == bumped_tx.vout[0] + assert len(bumped_tx.vout) == 16 + + async def test_create_bumped_tx_single_output(self): + # tests that fees are deducted from the only output available + # in the transaction + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**8 - 111 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 2000, wallet) + + assert orig_tx.vin[0] == bumped_tx.vin[0] + assert (orig_tx.vout[0].nValue - bumped_tx.vout[0].nValue) in (111, 113) + + async def test_create_bumped_tx_output_index(self): + # tests that the bumped transaction deducts its fee from the specified + # output even if it is an external wallet address + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**7 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": 10**8 - amount_sats - 142}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 2000, wallet, 0) + + assert orig_tx.vin[0] == bumped_tx.vin[0] + assert orig_tx.vout[1] == bumped_tx.vout[1] + assert (orig_tx.vout[0].nValue - bumped_tx.vout[0].nValue) in (142, 144) + + async def test_create_bumped_tx_no_change(self): + # tests that the bumped transaction is the same as the original if fees + # cannot be deducted + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr, 0.00002843) + amount_sats = 2730 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 2843)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 3000, wallet) + + assert orig_tx.vin[0] == bumped_tx.vin[0] + assert orig_tx.vout[0] == bumped_tx.vout[0] + + async def test_sign_and_broadcast(self): + # tests that we can correctly sign and broadcast a replaced transaction + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**8 - jm_single().BITCOIN_DUST_THRESHOLD - 142 + change_sats = 10**8 - amount_sats - 142 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": change_sats}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 2000, wallet) + await sign_transaction(bumped_tx, orig_tx, wallet_service) + + assert jm_single().bc_interface.pushtx(bumped_tx.serialize()) == True + + async def test_sign_psbt_broadcast(self): + # tests that we can correctly sign and broadcast a replaced psbt transaction + wallet = self.wallet + wallet_service = self.wallet_service + await wallet_service.resync_wallet() + addr = await wallet.get_external_addr(0) + utxo = fund_wallet_addr(wallet, addr) + amount_sats = 10**8 - jm_single().BITCOIN_DUST_THRESHOLD - 142 + change_sats = 10**8 - amount_sats - 142 + tx = btc.mktx([utxo], + [{"address": str(btc.CCoinAddress.from_scriptPubKey( + btc.CScript(b"\x00").to_p2sh_scriptPubKey())), + "value": amount_sats}, + {"address": await wallet.get_internal_addr(0), + "value": change_sats}]) + tx.vin[0].nSequence = 0xffffffff - 2 # mark as replaceable + success, msg = await wallet.sign_tx( + tx, {0: (wallet.addr_to_script(addr), 10**8)}) + success = jm_single().bc_interface.pushtx(tx.serialize()) + orig_tx = tx.clone() + + bumped_tx = create_bumped_tx(tx, 2000, wallet) + psbt = await sign_psbt(bumped_tx, orig_tx, wallet_service) + + assert jm_single().bc_interface.pushtx(psbt.extract_transaction().serialize()) == True