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
|
#!/usr/bin/env python3
# Copyright (c) 2014-2024 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 the getorphantxs RPC."""
from test_framework.mempool_util import tx_in_orphanage
from test_framework.messages import msg_tx
from test_framework.p2p import P2PInterface
from test_framework.util import assert_equal
from test_framework.test_framework import BitcoinTestFramework
from test_framework.wallet import MiniWallet
class GetOrphanTxsTest(BitcoinTestFramework):
def set_test_params(self):
self.num_nodes = 1
def run_test(self):
self.wallet = MiniWallet(self.nodes[0])
self.test_orphan_activity()
self.test_orphan_details()
def test_orphan_activity(self):
self.log.info("Check that orphaned transactions are returned with getorphantxs")
node = self.nodes[0]
self.log.info("Create two 1P1C packages, but only broadcast the children")
tx_parent_1 = self.wallet.create_self_transfer()
tx_child_1 = self.wallet.create_self_transfer(utxo_to_spend=tx_parent_1["new_utxo"])
tx_parent_2 = self.wallet.create_self_transfer()
tx_child_2 = self.wallet.create_self_transfer(utxo_to_spend=tx_parent_2["new_utxo"])
peer = node.add_p2p_connection(P2PInterface())
peer.send_and_ping(msg_tx(tx_child_1["tx"]))
peer.send_and_ping(msg_tx(tx_child_2["tx"]))
self.log.info("Check that neither parent is in the mempool")
assert_equal(node.getmempoolinfo()["size"], 0)
self.log.info("Check that both children are in the orphanage")
orphanage = node.getorphantxs(verbosity=0)
self.log.info("Check the size of the orphanage")
assert_equal(len(orphanage), 2)
self.log.info("Check that negative verbosity is treated as 0")
assert_equal(orphanage, node.getorphantxs(verbosity=-1))
assert tx_in_orphanage(node, tx_child_1["tx"])
assert tx_in_orphanage(node, tx_child_2["tx"])
self.log.info("Broadcast parent 1")
peer.send_and_ping(msg_tx(tx_parent_1["tx"]))
self.log.info("Check that parent 1 and child 1 are in the mempool")
raw_mempool = node.getrawmempool()
assert_equal(len(raw_mempool), 2)
assert tx_parent_1["txid"] in raw_mempool
assert tx_child_1["txid"] in raw_mempool
self.log.info("Check that orphanage only contains child 2")
orphanage = node.getorphantxs()
assert_equal(len(orphanage), 1)
assert tx_in_orphanage(node, tx_child_2["tx"])
peer.send_and_ping(msg_tx(tx_parent_2["tx"]))
self.log.info("Check that all parents and children are now in the mempool")
raw_mempool = node.getrawmempool()
assert_equal(len(raw_mempool), 4)
assert tx_parent_1["txid"] in raw_mempool
assert tx_child_1["txid"] in raw_mempool
assert tx_parent_2["txid"] in raw_mempool
assert tx_child_2["txid"] in raw_mempool
self.log.info("Check that the orphanage is empty")
assert_equal(len(node.getorphantxs()), 0)
self.log.info("Confirm the transactions (clears mempool)")
self.generate(node, 1)
assert_equal(node.getmempoolinfo()["size"], 0)
def test_orphan_details(self):
self.log.info("Check the transaction details returned from getorphantxs")
node = self.nodes[0]
self.log.info("Create two orphans, from different peers")
tx_parent_1 = self.wallet.create_self_transfer()
tx_child_1 = self.wallet.create_self_transfer(utxo_to_spend=tx_parent_1["new_utxo"])
tx_parent_2 = self.wallet.create_self_transfer()
tx_child_2 = self.wallet.create_self_transfer(utxo_to_spend=tx_parent_2["new_utxo"])
peer_1 = node.add_p2p_connection(P2PInterface())
peer_2 = node.add_p2p_connection(P2PInterface())
peer_1.send_and_ping(msg_tx(tx_child_1["tx"]))
peer_2.send_and_ping(msg_tx(tx_child_2["tx"]))
orphanage = node.getorphantxs(verbosity=2)
assert tx_in_orphanage(node, tx_child_1["tx"])
assert tx_in_orphanage(node, tx_child_2["tx"])
self.log.info("Check that orphan 1 and 2 were from different peers")
assert orphanage[0]["from"][0] != orphanage[1]["from"][0]
self.log.info("Unorphan child 2")
peer_2.send_and_ping(msg_tx(tx_parent_2["tx"]))
assert not tx_in_orphanage(node, tx_child_2["tx"])
self.log.info("Checking orphan details")
orphanage = node.getorphantxs(verbosity=1)
assert_equal(len(node.getorphantxs()), 1)
orphan_1 = orphanage[0]
self.orphan_details_match(orphan_1, tx_child_1, verbosity=1)
self.log.info("Checking orphan details (verbosity 2)")
orphanage = node.getorphantxs(verbosity=2)
orphan_1 = orphanage[0]
self.orphan_details_match(orphan_1, tx_child_1, verbosity=2)
def orphan_details_match(self, orphan, tx, verbosity):
self.log.info("Check txid/wtxid of orphan")
assert_equal(orphan["txid"], tx["txid"])
assert_equal(orphan["wtxid"], tx["wtxid"])
self.log.info("Check the sizes of orphan")
assert_equal(orphan["bytes"], len(tx["tx"].serialize()))
assert_equal(orphan["vsize"], tx["tx"].get_vsize())
assert_equal(orphan["weight"], tx["tx"].get_weight())
if verbosity == 2:
self.log.info("Check the transaction hex of orphan")
assert_equal(orphan["hex"], tx["hex"])
if __name__ == '__main__':
GetOrphanTxsTest(__file__).main()
|
