wallet_anchor.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2025-present The Limenka developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or https://www.opensource.org/licenses/mit-license.php.
5
6 import time
7
8 from test_framework.blocktools import MAX_FUTURE_BLOCK_TIME
9 from test_framework.descriptors import descsum_create
10 from test_framework.messages import (
11 COutPoint,
12 CTxIn,
13 CTxInWitness,
14 CTxOut,
15 )
16 from test_framework.script_util import (
17 ANCHOR_ADDRESS,
18 PAY_TO_ANCHOR,
19 )
20 from test_framework.test_framework import LimenkaTestFramework
21 from test_framework.util import (
22 assert_equal,
23 assert_raises_rpc_error,
24 )
25 from test_framework.wallet import MiniWallet
26
27 class WalletAnchorTest(LimenkaTestFramework):
28 def add_options(self, parser):
29 self.add_wallet_options(parser)
30
31 def set_test_params(self):
32 self.num_nodes = 1
33
34 def skip_test_if_missing_module(self):
35 self.skip_if_no_wallet()
36
37 def test_0_value_anchor_listunspent(self):
38 self.log.info("Test that 0-value anchor outputs are detected as UTXOs")
39
40 # Create an anchor output, and spend it
41 sender = MiniWallet(self.nodes[0])
42 anchor_tx = sender.create_self_transfer(fee_rate=0, version=3)["tx"]
43 anchor_tx.vout.append(CTxOut(0, PAY_TO_ANCHOR))
44 anchor_tx.rehash() # Rehash after modifying anchor_tx
45 anchor_spend = sender.create_self_transfer(version=3)["tx"]
46 anchor_spend.vin.append(CTxIn(COutPoint(anchor_tx.sha256, 1), b""))
47 anchor_spend.wit.vtxinwit.append(CTxInWitness())
48 anchor_spend.rehash() # Rehash after modifying anchor_spend
49 submit_res = self.nodes[0].submitpackage([anchor_tx.serialize().hex(), anchor_spend.serialize().hex()])
50 assert_equal(submit_res["package_msg"], "success")
51 anchor_txid = anchor_tx.hash
52 anchor_spend_txid = anchor_spend.hash
53
54 # Mine each tx in separate blocks
55 self.generateblock(self.nodes[0], sender.get_address(), [anchor_tx.serialize().hex()])
56 anchor_tx_height = self.nodes[0].getblockcount()
57 self.generateblock(self.nodes[0], sender.get_address(), [anchor_spend.serialize().hex()])
58
59 # Mock time forward and generate some blocks to avoid rescanning of latest blocks
60 self.nodes[0].setmocktime(int(time.time()) + MAX_FUTURE_BLOCK_TIME + 1)
61 self.generate(self.nodes[0], 10)
62
63 self.nodes[0].createwallet(wallet_name="anchor", disable_private_keys=True)
64 wallet = self.nodes[0].get_wallet_rpc("anchor")
65
66 wallet.importaddress(ANCHOR_ADDRESS, rescan=False)
67
68 # The wallet should have no UTXOs, and not know of the anchor tx or its spend
69 assert_equal(wallet.listunspent(), [])
70 assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_txid)
71 assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_spend_txid)
72
73 # Rescanning the block containing the anchor so that listunspent will list the output
74 wallet.rescanblockchain(0, anchor_tx_height)
75 utxos = wallet.listunspent()
76 assert_equal(len(utxos), 1)
77 assert_equal(utxos[0]["txid"], anchor_txid)
78 assert_equal(utxos[0]["address"], ANCHOR_ADDRESS)
79 assert_equal(utxos[0]["amount"], 0)
80 wallet.gettransaction(anchor_txid)
81 assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", wallet.gettransaction, anchor_spend_txid)
82
83 # Rescan the rest of the blockchain to see the anchor was spent
84 wallet.rescanblockchain()
85 assert_equal(wallet.listunspent(), [])
86 wallet.gettransaction(anchor_spend_txid)
87
88 def test_cannot_sign_anchors(self):
89 self.log.info("Test that the wallet cannot spend anchor outputs")
90 for disable_privkeys in [False, True]:
91 self.nodes[0].createwallet(wallet_name=f"anchor_spend_{disable_privkeys}", disable_private_keys=disable_privkeys)
92 wallet = self.nodes[0].get_wallet_rpc(f"anchor_spend_{disable_privkeys}")
93 if self.options.descriptors:
94 import_res = wallet.importdescriptors([
95 {"desc": descsum_create(f"addr({ANCHOR_ADDRESS})"), "timestamp": "now"},
96 {"desc": descsum_create(f"raw({PAY_TO_ANCHOR.hex()})"), "timestamp": "now"}
97 ])
98 assert_equal(import_res[0]["success"], disable_privkeys)
99 assert_equal(import_res[1]["success"], disable_privkeys)
100 else:
101 wallet.importaddress(ANCHOR_ADDRESS)
102
103 anchor_txid = self.default_wallet.sendtoaddress(ANCHOR_ADDRESS, 1)
104 self.generate(self.nodes[0], 1)
105
106 wallet = self.nodes[0].get_wallet_rpc("anchor_spend_True")
107 utxos = wallet.listunspent()
108 assert_equal(len(utxos), 1)
109 assert_equal(utxos[0]["txid"], anchor_txid)
110 assert_equal(utxos[0]["address"], ANCHOR_ADDRESS)
111 assert_equal(utxos[0]["amount"], 1)
112
113 if self.options.descriptors:
114 assert_raises_rpc_error(-4, "Missing solving data for estimating transaction size", wallet.send, [{self.default_wallet.getnewaddress(): 0.9999}])
115 assert_raises_rpc_error(-4, "Unable to determine the size of the transaction, the wallet contains unsolvable descriptors", wallet.sendall, recipients=[self.default_wallet.getnewaddress()])
116 else:
117 assert_raises_rpc_error(-4, "Insufficient funds", wallet.send, [{self.default_wallet.getnewaddress(): 0.9999}])
118 assert_raises_rpc_error(-6, "Total value of UTXO pool too low to pay for transaction. Try using lower feerate or excluding uneconomic UTXOs with 'send_max' option.", wallet.sendall, recipients=[self.default_wallet.getnewaddress()])
119 assert_raises_rpc_error(-4, "Error: Private keys are disabled for this wallet", wallet.sendtoaddress, self.default_wallet.getnewaddress(), 0.9999)
120 assert_raises_rpc_error(-4, "Unable to determine the size of the transaction, the wallet contains unsolvable descriptors", wallet.sendall, recipients=[self.default_wallet.getnewaddress()], inputs=utxos)
121
122 def run_test(self):
123 self.default_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
124 self.test_0_value_anchor_listunspent()
125 self.test_cannot_sign_anchors()
126
127 if __name__ == '__main__':
128 WalletAnchorTest(__file__).main()
129