Repository: bitcoin/bitcoin
File: test/functional/test_framework/wallet.py (Python)

Code before:
            self._priv_key = ECKey()
            self._priv_key.set((1).to_bytes(32, 'big'), True)
            pub_key = self._priv_key.get_pubkey()
            self._scriptPubKey = key_to_p2pk_script(pub_key.get_bytes())
        elif mode == MiniWalletMode.ADDRESS_OP_TRUE:
            internal_key = None if tag_name is None else hash256(tag_name.encode())
            self._address, self._internal_key = create_deterministic_address_bcrt1_p2tr_op_true(internal_key)
            self._scriptPubKey = address_to_scriptpubkey(self._address)

        # When the pre-mined test framework chain is used, it contains coinbase
        # outputs to the MiniWallet's default address in blocks 76-100
        # (see method BitcoinTestFramework._initialize_chain())
        # The MiniWallet needs to rescan_utxos() in order to account
        # for those mature UTXOs, so that all txs spend confirmed coins
        self.rescan_utxos()

    def _create_utxo(self, *, txid, vout, value, height, coinbase, confirmations):
        return {"txid": txid, "vout": vout, "value": value, "height": height, "coinbase": coinbase, "confirmations": confirmations}

    def _bulk_tx(self, tx, target_weight):
        """Pad a transaction with extra outputs until it reaches a target weight (or higher).
        returns the tx
        """
        tx.vout.append(CTxOut(nValue=0, scriptPubKey=CScript([OP_RETURN, b'a'])))
        dummy_vbytes = (target_weight - tx.get_weight() + 3) // 4
        tx.vout[-1].scriptPubKey = CScript([OP_RETURN, b'a' * dummy_vbytes])
        # Lower bound should always be off by at most 3
        assert_greater_than_or_equal(tx.get_weight(), target_weight)
        # Higher bound should always be off by at most 3 + 12 weight (for encoding the length)
        assert_greater_than_or_equal(target_weight + 15, tx.get_weight())

    def get_balance(self):
        return sum(u['value'] for u in self._utxos)

    def rescan_utxos(self, *, include_mempool=True):
        """Drop all utxos and rescan the utxo set"""
        self._utxos = []
        res = self._test_node.scantxoutset(action="start", scanobjects=[self.get_descriptor()])
        assert_equal(True, res['success'])
        for utxo in res['unspents']:
            self._utxos.append(
                self._create_utxo(txid=utxo["txid"],
                                  vout=utxo["vout"],
                                  value=utxo["amount"],
                                  height=utxo["height"],
                                  coinbase=utxo["coinbase"],
                                  confirmations=res["height"] - utxo["height"] + 1))
        if include_mempool:
            mempool = self._test_node.getrawmempool(verbose=True)
            # Sort tx by ancestor count. See BlockAssembler::SortForBlock in src/node/miner.cpp
            sorted_mempool = sorted(mempool.items(), key=lambda item: (item[1]["ancestorcount"], int(item[0], 16)))

Review comment (suggestion):
In the previous implementation, it seemed a little unusual that `b'a'` was added first, and then more padding was added later. The newer approach seems cleaner where padding is added only once.

Code after:
            assert tag_name is None
            self._scriptPubKey = bytes(CScript([OP_TRUE]))
        elif mode == MiniWalletMode.RAW_P2PK:
            # use simple deterministic private key (k=1)
            assert tag_name is None
            self._priv_key = ECKey()
            self._priv_key.set((1).to_bytes(32, 'big'), True)
            pub_key = self._priv_key.get_pubkey()
            self._scriptPubKey = key_to_p2pk_script(pub_key.get_bytes())
        elif mode == MiniWalletMode.ADDRESS_OP_TRUE:
            internal_key = None if tag_name is None else hash256(tag_name.encode())
            self._address, self._internal_key = create_deterministic_address_bcrt1_p2tr_op_true(internal_key)
            self._scriptPubKey = address_to_scriptpubkey(self._address)

        # When the pre-mined test framework chain is used, it contains coinbase
        # outputs to the MiniWallet's default address in blocks 76-100
        # (see method BitcoinTestFramework._initialize_chain())
        # The MiniWallet needs to rescan_utxos() in order to account
        # for those mature UTXOs, so that all txs spend confirmed coins
        self.rescan_utxos()

    def _create_utxo(self, *, txid, vout, value, height, coinbase, confirmations):
        return {"txid": txid, "vout": vout, "value": value, "height": height, "coinbase": coinbase, "confirmations": confirmations}

    def _bulk_tx(self, tx, target_weight):
        """Pad a transaction with extra outputs until it reaches a target weight (or higher).
        returns the tx
        """
        tx.vout.append(CTxOut(nValue=0, scriptPubKey=CScript([OP_RETURN])))
        # determine number of needed padding bytes by converting weight difference to vbytes
        dummy_vbytes = (target_weight - tx.get_weight() + 3) // 4
        # compensate for the increase of the compact-size encoded script length
        # (note that the length encoding of the unpadded output script needs one byte)
        dummy_vbytes -= len(ser_compact_size(dummy_vbytes)) - 1
        tx.vout[-1].scriptPubKey = CScript([OP_RETURN] + [OP_1] * dummy_vbytes)
        # Actual weight should be at most 3 higher than target weight
        assert_greater_than_or_equal(tx.get_weight(), target_weight)
        assert_greater_than_or_equal(target_weight + 3, tx.get_weight())

    def get_balance(self):
        return sum(u['value'] for u in self._utxos)

    def rescan_utxos(self, *, include_mempool=True):
        """Drop all utxos and rescan the utxo set"""
        self._utxos = []
        res = self._test_node.scantxoutset(action="start", scanobjects=[self.get_descriptor()])
        assert_equal(True, res['success'])
        for utxo in res['unspents']:
            self._utxos.append(
                self._create_utxo(txid=utxo["txid"],
                                  vout=utxo["vout"],