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

Code before:
        anchor_tx.vout.append(CTxOut(0, PAY_TO_ANCHOR))
        anchor_tx.rehash()  # Rehash after modifying anchor_tx
        anchor_spend = sender.create_self_transfer(version=3)["tx"]
        anchor_spend.vin.append(CTxIn(COutPoint(anchor_tx.sha256, 1), b""))
        anchor_spend.wit.vtxinwit.append(CTxInWitness())
        anchor_spend.rehash()  # Rehash after modifying anchor_spend
        submit_res = self.nodes[0].submitpackage([anchor_tx.serialize().hex(), anchor_spend.serialize().hex()])
        assert_equal(submit_res["package_msg"], "success")
        anchor_txid = anchor_tx.hash
        anchor_spend_txid = anchor_spend.hash

        # Mine each tx in separate blocks
        self.generateblock(self.nodes[0], sender.get_address(), [anchor_tx.serialize().hex()])
        anchor_tx_height = self.nodes[0].getblockcount()
        self.generateblock(self.nodes[0], sender.get_address(), [anchor_spend.serialize().hex()])

        # Mock time forward and generate some blocks to avoid rescanning of latest blocks
        self.nodes[0].setmocktime(int(time.time()) + MAX_FUTURE_BLOCK_TIME + 1)
        self.generate(self.nodes[0], 10)

        self.nodes[0].createwallet(wallet_name="anchor", disable_private_keys=True)
        wallet = self.nodes[0].get_wallet_rpc("anchor")

        if self.options.descriptors:
            import_res = wallet.importdescriptors([{"desc": descsum_create(f"addr({ANCHOR_ADDRESS})"), "timestamp": "now"}])
            assert_equal(import_res[0]["success"], True)

            # The wallet should have no UTXOs, and not know of the anchor tx or its spend
            assert_equal(wallet.listunspent(), [])
            assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_txid)
            assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_spend_txid)

            # Rescanning the block containing the anchor so that listunspent will list the output
            wallet.rescanblockchain(0, anchor_tx_height)
            utxos = wallet.listunspent()
            assert_equal(len(utxos), 1)
            assert_equal(utxos[0]["txid"], anchor_txid)
            assert_equal(utxos[0]["address"], ANCHOR_ADDRESS)
            assert_equal(utxos[0]["amount"], 0)
            wallet.gettransaction(anchor_txid)
            assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_spend_txid)

            # Rescan the rest of the blockchain to see the anchor was spent
            wallet.rescanblockchain()
            assert_equal(wallet.listunspent(), [])
            wallet.gettransaction(anchor_spend_txid)

    def test_cannot_sign_anchors(self):
        self.log.info("Test that the wallet cannot spend anchor outputs")
        for disable_privkeys in [False, True]:
            self.nodes[0].createwallet(wallet_name=f"anchor_spend_{disable_privkeys}", disable_private_keys=disable_privkeys)

Review comment (suggestion):
In b605cc8c73a10162f47e0aeb361144e4913a2a54 "test: Add a test for anchor outputs in the wallet"

Using `wallet.importaddress(ANCHOR_ADDRESS, rescan=False)` instead of `importdescriptors` here allows this test case to work for both descriptors and legacy wallets.

Code after:
        anchor_tx.vout.append(CTxOut(0, PAY_TO_ANCHOR))
        anchor_tx.rehash()  # Rehash after modifying anchor_tx
        anchor_spend = sender.create_self_transfer(version=3)["tx"]
        anchor_spend.vin.append(CTxIn(COutPoint(anchor_tx.sha256, 1), b""))
        anchor_spend.wit.vtxinwit.append(CTxInWitness())
        anchor_spend.rehash()  # Rehash after modifying anchor_spend
        submit_res = self.nodes[0].submitpackage([anchor_tx.serialize().hex(), anchor_spend.serialize().hex()])
        assert_equal(submit_res["package_msg"], "success")
        anchor_txid = anchor_tx.hash
        anchor_spend_txid = anchor_spend.hash

        # Mine each tx in separate blocks
        self.generateblock(self.nodes[0], sender.get_address(), [anchor_tx.serialize().hex()])
        anchor_tx_height = self.nodes[0].getblockcount()
        self.generateblock(self.nodes[0], sender.get_address(), [anchor_spend.serialize().hex()])

        # Mock time forward and generate some blocks to avoid rescanning of latest blocks
        self.nodes[0].setmocktime(int(time.time()) + MAX_FUTURE_BLOCK_TIME + 1)
        self.generate(self.nodes[0], 10)

        self.nodes[0].createwallet(wallet_name="anchor", disable_private_keys=True)
        wallet = self.nodes[0].get_wallet_rpc("anchor")

        wallet.importaddress(ANCHOR_ADDRESS, rescan=False)

        # The wallet should have no UTXOs, and not know of the anchor tx or its spend
        assert_equal(wallet.listunspent(), [])
        assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_txid)
        assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_spend_txid)

        # Rescanning the block containing the anchor so that listunspent will list the output
        wallet.rescanblockchain(0, anchor_tx_height)
        utxos = wallet.listunspent()
        assert_equal(len(utxos), 1)
        assert_equal(utxos[0]["txid"], anchor_txid)
        assert_equal(utxos[0]["address"], ANCHOR_ADDRESS)
        assert_equal(utxos[0]["amount"], 0)
        wallet.gettransaction(anchor_txid)
        assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_spend_txid)

        # Rescan the rest of the blockchain to see the anchor was spent
        wallet.rescanblockchain()
        assert_equal(wallet.listunspent(), [])
        wallet.gettransaction(anchor_spend_txid)

    def test_cannot_sign_anchors(self):
        self.log.info("Test that the wallet cannot spend anchor outputs")
        for disable_privkeys in [False, True]:
            self.nodes[0].createwallet(wallet_name=f"anchor_spend_{disable_privkeys}", disable_private_keys=disable_privkeys)
            wallet = self.nodes[0].get_wallet_rpc(f"anchor_spend_{disable_privkeys}")
            if self.options.descriptors: