| 88 | |
| 89 | |
| 90 | class MiniWallet: |
| 91 | def __init__(self, test_node, *, mode=MiniWalletMode.ADDRESS_OP_TRUE, tag_name=None): |
| 92 | self._test_node = test_node |
| 93 | self._utxos = [] |
| 94 | self._mode = mode |
| 95 | |
| 96 | assert isinstance(mode, MiniWalletMode) |
| 97 | if mode == MiniWalletMode.RAW_OP_TRUE: |
| 98 | assert tag_name is None |
| 99 | self._scriptPubKey = bytes(CScript([OP_TRUE])) |
| 100 | elif mode == MiniWalletMode.RAW_P2PK: |
| 101 | # use simple deterministic private key (k=1) |
| 102 | assert tag_name is None |
| 103 | self._priv_key = ECKey() |
| 104 | self._priv_key.set((1).to_bytes(32, 'big'), True) |
| 105 | pub_key = self._priv_key.get_pubkey() |
| 106 | self._scriptPubKey = key_to_p2pk_script(pub_key.get_bytes()) |
| 107 | elif mode == MiniWalletMode.ADDRESS_OP_TRUE: |
| 108 | internal_key = None if tag_name is None else compute_xonly_pubkey(hash256(tag_name.encode()))[0] |
| 109 | self._address, self._taproot_info = create_deterministic_address_bcrt1_p2tr_op_true(internal_key) |
| 110 | self._scriptPubKey = address_to_scriptpubkey(self._address) |
| 111 | |
| 112 | # When the pre-mined test framework chain is used, it contains coinbase |
| 113 | # outputs to the MiniWallet's default address in blocks 76-100 |
| 114 | # (see method BitcoinTestFramework._initialize_chain()) |
| 115 | # The MiniWallet needs to rescan_utxos() in order to account |
| 116 | # for those mature UTXOs, so that all txs spend confirmed coins |
| 117 | self.rescan_utxos() |
| 118 | |
| 119 | def _create_utxo(self, *, txid, vout, value, height, coinbase, confirmations): |
| 120 | return {"txid": txid, "vout": vout, "value": value, "height": height, "coinbase": coinbase, "confirmations": confirmations} |
| 121 | |
| 122 | def _bulk_tx(self, tx, target_vsize): |
| 123 | """Pad a transaction with extra outputs until it reaches a target vsize. |
| 124 | returns the tx |
| 125 | """ |
| 126 | tx.vout.append(CTxOut(nValue=0, scriptPubKey=CScript([OP_RETURN]))) |
| 127 | bulk_vout(tx, target_vsize) |
| 128 | |
| 129 | |
| 130 | def get_balance(self): |
| 131 | return sum(u['value'] for u in self._utxos) |
| 132 | |
| 133 | def rescan_utxos(self, *, include_mempool=True): |
| 134 | """Drop all utxos and rescan the utxo set""" |
| 135 | self._utxos = [] |
| 136 | res = self._test_node.scantxoutset(action="start", scanobjects=[self.get_descriptor()]) |
| 137 | assert_equal(True, res['success']) |
| 138 | for utxo in res['unspents']: |
| 139 | self._utxos.append( |
| 140 | self._create_utxo(txid=utxo["txid"], |
| 141 | vout=utxo["vout"], |
| 142 | value=utxo["amount"], |
| 143 | height=utxo["height"], |
| 144 | coinbase=utxo["coinbase"], |
| 145 | confirmations=res["height"] - utxo["height"] + 1)) |
| 146 | if include_mempool: |
| 147 | mempool = self._test_node.getrawmempool(verbose=True) |
no outgoing calls