-
Notifications
You must be signed in to change notification settings - Fork 388
/
Copy pathwallet_groups.py
executable file
·184 lines (162 loc) · 8.64 KB
/
wallet_groups.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
#!/usr/bin/env python3
# Copyright (c) 2018-2021 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test wallet group functionality."""
from test_framework.blocktools import COINBASE_MATURITY
from test_framework.test_framework import BitcoinTestFramework
from test_framework.messages import (
tx_from_hex,
)
from test_framework.util import (
assert_approx,
assert_equal,
)
class WalletGroupTest(BitcoinTestFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.num_nodes = 5
self.extra_args = [
[],
[],
["-avoidpartialspends"],
["-maxapsfee=0.00002739"], # ELEMENTS: these and a few other numbers increased for larger transactions
["-maxapsfee=0.00002740"],
]
self.rpc_timeout = 480
def skip_test_if_missing_module(self):
self.skip_if_no_wallet()
def run_test(self):
self.log.info("Setting up")
# Mine some coins
self.generate(self.nodes[0], COINBASE_MATURITY + 1)
# Get some addresses from the two nodes
addr1 = [self.nodes[1].getnewaddress() for _ in range(3)]
addr2 = [self.nodes[2].getnewaddress() for _ in range(3)]
addrs = addr1 + addr2
# Send 1 + 0.5 coin to each address
[self.nodes[0].sendtoaddress(addr, 1.0) for addr in addrs]
[self.nodes[0].sendtoaddress(addr, 0.5) for addr in addrs]
self.generate(self.nodes[0], 1)
# For each node, send 0.2 coins back to 0;
# - node[1] should pick one 0.5 UTXO and leave the rest
# ELEMENTS: since SRD was added, node[1] sometimes picks a 1.0 UTXO
# - node[2] should pick one (1.0 + 0.5) UTXO group corresponding to a
# given address, and leave the rest
self.log.info("Test sending transactions picks one UTXO group and leaves the rest")
txid1 = self.nodes[1].sendtoaddress(self.nodes[0].getnewaddress(), 0.2)
tx1 = self.nodes[1].getrawtransaction(txid1, True)
# txid1 should have 1 input and 2 outputs
assert_equal(1, len(tx1["vin"]))
assert_equal(2+1, len(tx1["vout"]))
# one output should be 0.2, the other should be ~0.3
v = [vout["value"] for vout in tx1["vout"] if vout["scriptPubKey"]["type"] != "fee"]
v.sort()
assert_approx(v[0], vexp=0.2, vspan=0.0001)
# ELEMENTS
try:
assert_approx(v[1], vexp=0.3, vspan=0.0001)
except AssertionError:
assert_approx(v[1], vexp=0.8, vspan=0.0001)
txid2 = self.nodes[2].sendtoaddress(self.nodes[0].getnewaddress(), 0.2)
tx2 = self.nodes[2].getrawtransaction(txid2, True)
# txid2 should have 2 inputs and 2 outputs
assert_equal(2, len(tx2["vin"]))
assert_equal(2+1, len(tx2["vout"]))
# one output should be 0.2, the other should be ~1.3
v = [vout["value"] for vout in tx2["vout"] if vout["scriptPubKey"]["type"] != "fee"]
v.sort()
assert_approx(v[0], vexp=0.2, vspan=0.0001)
assert_approx(v[1], vexp=1.3, vspan=0.0001)
self.log.info("Test avoiding partial spends if warranted, even if avoidpartialspends is disabled")
self.sync_all()
self.generate(self.nodes[0], 1)
# Nodes 1-2 now have confirmed UTXOs (letters denote destinations):
# Node #1: Node #2:
# - A 1.0 - D0 1.0
# - B0 1.0 - D1 0.5
# - B1 0.5 - E0 1.0
# - C0 1.0 - E1 0.5
# - C1 0.5 - F ~1.3
# - D ~0.3
assert_approx(self.nodes[1].getbalance()['bitcoin'], vexp=4.3, vspan=0.0001)
assert_approx(self.nodes[2].getbalance()['bitcoin'], vexp=4.3, vspan=0.0001)
# Sending 1.4 btc should pick one 1.0 + one more. For node #1,
# this could be (A / B0 / C0) + (B1 / C1 / D). We ensure that it is
# B0 + B1 or C0 + C1, because this avoids partial spends while not being
# detrimental to transaction cost
txid3 = self.nodes[1].sendtoaddress(self.nodes[0].getnewaddress(), 1.4)
tx3 = self.nodes[1].getrawtransaction(txid3, True)
# tx3 should have 2 inputs and 2 outputs
assert_equal(2, len(tx3["vin"]))
assert_equal(3, len(tx3["vout"])) ## ELEMENTS: 2 outputs plus fee
# the accumulated value should be 1.5, so the outputs should be
# ~0.1 and 1.4 and should come from the same destination
values = [vout["value"] for vout in tx3["vout"]]
values.sort()
assert_approx(values[0], vexp=0.00006, vspan=0.00001)
assert_approx(values[1], vexp=0.1, vspan=0.0001)
assert_approx(values[2], vexp=1.4, vspan=0.0001)
input_txids = [vin["txid"] for vin in tx3["vin"]]
input_addrs = [self.nodes[1].gettransaction(txid)['details'][0]['address'] for txid in input_txids]
assert_equal(input_addrs[0], input_addrs[1])
# Node 2 enforces avoidpartialspends so needs no checking here
tx4_ungrouped_fee = 5140
tx4_grouped_fee = 6520
tx5_6_ungrouped_fee = 7880
tx5_6_grouped_fee = 10620
self.log.info("Test wallet option maxapsfee")
addr_aps = self.nodes[3].getnewaddress()
self.nodes[0].sendtoaddress(addr_aps, 1.0)
self.nodes[0].sendtoaddress(addr_aps, 1.0)
self.generate(self.nodes[0], 1)
with self.nodes[3].assert_debug_log([f'Fee non-grouped = {tx4_ungrouped_fee}, grouped = {tx4_grouped_fee}, using grouped']):
txid4 = self.nodes[3].sendtoaddress(self.nodes[0].getnewaddress(), 0.1)
tx4 = self.nodes[3].getrawtransaction(txid4, True)
# tx4 should have 2 inputs and 2 outputs although one output would
# have been enough and the transaction caused higher fees
assert_equal(2, len(tx4["vin"]))
assert_equal(3, len(tx4["vout"])) # ELEMENTS: plus fee
addr_aps2 = self.nodes[3].getnewaddress()
[self.nodes[0].sendtoaddress(addr_aps2, 1.0) for _ in range(5)]
self.generate(self.nodes[0], 1)
with self.nodes[3].assert_debug_log([f'Fee non-grouped = {tx5_6_ungrouped_fee}, grouped = {tx5_6_grouped_fee}, using non-grouped']):
txid5 = self.nodes[3].sendtoaddress(self.nodes[0].getnewaddress(), 2.95)
tx5 = self.nodes[3].getrawtransaction(txid5, True)
# tx5 should have 3 inputs (1.0, 1.0, 1.0) and 2 outputs
assert_equal(3, len(tx5["vin"]))
assert_equal(3, len(tx5["vout"])) # ELEMENTS: plus fee
# Test wallet option maxapsfee with node 4, which sets maxapsfee
# 1 sat higher, crossing the threshold from non-grouped to grouped.
self.log.info("Test wallet option maxapsfee threshold from non-grouped to grouped")
addr_aps3 = self.nodes[4].getnewaddress()
[self.nodes[0].sendtoaddress(addr_aps3, 1.0) for _ in range(5)]
self.generate(self.nodes[0], 1)
with self.nodes[4].assert_debug_log([f'Fee non-grouped = {tx5_6_ungrouped_fee}, grouped = {tx5_6_grouped_fee}, using grouped']):
txid6 = self.nodes[4].sendtoaddress(self.nodes[0].getnewaddress(), 2.95)
tx6 = self.nodes[4].getrawtransaction(txid6, True)
# tx6 should have 5 inputs and 2 outputs
assert_equal(5, len(tx6["vin"]))
assert_equal(3, len(tx6["vout"])) # ELEMENTS: plus fee
# Empty out node2's wallet
self.nodes[2].sendtoaddress(address=self.nodes[0].getnewaddress(), amount=self.nodes[2].getbalance()['bitcoin'], subtractfeefromamount=True)
self.sync_all()
self.generate(self.nodes[0], 1)
self.log.info("Fill a wallet with 10,000 outputs corresponding to the same scriptPubKey")
for _ in range(5):
raw_tx = self.nodes[0].createrawtransaction([{"txid":"0"*64, "vout":0}], [{addr2[0]: 0.10}])
tx = tx_from_hex(raw_tx)
tx.vin = []
# ELEMENTS: lower number because outputs are bigger (otherwise tx gets too large)
tx.vout = [tx.vout[0]] * 1000
funded_tx = self.nodes[0].fundrawtransaction(tx.serialize().hex())
signed_tx = self.nodes[0].signrawtransactionwithwallet(funded_tx['hex'])
self.nodes[0].sendrawtransaction(signed_tx['hex'])
self.generate(self.nodes[0], 1)
# Check that we can create a transaction that only requires ~100 of our
# utxos, without pulling in all outputs and creating a transaction that
# is way too big.
self.log.info("Test creating txn that only requires ~100 of our UTXOs without pulling in all outputs")
assert self.nodes[2].sendtoaddress(address=addr2[0], amount=5)
if __name__ == '__main__':
WalletGroupTest().main()