-
Notifications
You must be signed in to change notification settings - Fork 5
/
feature_assets_mempool.py
executable file
·153 lines (115 loc) · 5.55 KB
/
feature_assets_mempool.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
#!/usr/bin/env python3
# Copyright (c) 2017 The Bitcoin Core developers
# Copyright (c) 2017-2020 The Raven Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Testing asset mempool use cases"""
from test_framework.test_framework import RavenTestFramework
from test_framework.util import assert_equal, disconnect_all_nodes, connect_all_nodes_bi
class AssetMempoolTest(RavenTestFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.num_nodes = 2
def activate_assets(self):
self.log.info("Generating RVN and activating assets...")
n0, n1 = self.nodes[0], self.nodes[1]
n0.generate(1)
self.sync_all()
n0.generate(216)
self.sync_all()
n1.generate(216)
self.sync_all()
assert_equal("active", n0.getblockchaininfo()['bip9_softforks']['assets']['status'])
def issue_mempool_test(self):
self.log.info("Testing (issue_mempool_test) mempool state after asset issuance on two chains(only one is mined, the other is in the mempool), and then connecting the nodes together")
n0, n1 = self.nodes[0], self.nodes[1]
disconnect_all_nodes(self.nodes)
asset_name = "MEMPOOL"
# Issue asset on chain 1 and mine it into the blocks
n0.issue(asset_name)
n0.generate(15)
# Issue asset on chain 2 but keep it in the mempool. No mining
n1.issue(asset_name)
connect_all_nodes_bi(self.nodes, True)
# Assert that the reorg was successful
assert_equal(n0.getblockcount(), n1.getblockcount())
assert_equal(n0.getbestblockhash(), n1.getbestblockhash())
# Verify that the mempool are empty ( No extra transactions from the old chain made it into the mempool)
assert_equal(0, n1.getmempoolinfo()['size'])
assert_equal(0, n0.getmempoolinfo()['size'])
def issue_mempool_test_extended(self):
self.log.info("Testing (issue_mempool_test_extended) mempool state after, asset issuance, reissue, transfer, issue sub asset, getting reorged, with asset override on other chain")
n0, n1 = self.nodes[0], self.nodes[1]
disconnect_all_nodes(self.nodes)
# Create new asset for testing
asset_name = "MEMPOOL_2"
n0.issue(asset_name)
n0.generate(15)
# Reissue that asset
address1 = n0.getnewaddress()
n0.reissue(asset_name=asset_name, qty=2000, to_address=address1, change_address='', reissuable=True, new_units=-1)
n0.generate(15)
# Get a transfer address
n1_address = n1.getnewaddress()
# Transfer the asset
n0.transfer(asset_name, 2, n1_address)
# Issue sub asset
n0.issue(asset_name + '/SUB')
n0.generate(15)
# Issue the same asset on the other node
n1.issue(asset_name)
# Create enough blocks so that n0 will reorg to n1 chain
n1.generate(55)
# Connect the nodes, a reorg should occur
connect_all_nodes_bi(self.nodes, True)
# Asset the reorg occurred
assert_equal(n0.getblockcount(), n1.getblockcount())
assert_equal(n0.getbestblockhash(), n1.getbestblockhash())
# Verify that the mempool are empty ( No extra transactions from the old chain made it into the mempool)
assert_equal(0, n1.getmempoolinfo()['size'])
assert_equal(0, n0.getmempoolinfo()['size'])
def issue_mempool_test_extended_sub(self):
self.log.info("Testing (issue_mempool_test) mempool state after asset issuance on two chains(only one is mined, the other is in the mempool), and then connecting the nodes together")
n0, n1 = self.nodes[0], self.nodes[1]
disconnect_all_nodes(self.nodes)
asset_name = "MEMPOOL_3"
# Issue asset on chain 1 and mine it into the blocks
n0.issue(asset_name)
n0.generate(15)
# Issue asset on chain 2 but keep it in the mempool. No mining
n1.issue(asset_name)
# Mine a block on n1 chain
n1.generate(1)
# Issue sub assets and unique assets but only have it in the mempool
n1.issue(asset_name + '/SUB')
n1.issue(asset_name + '/SUB/1')
n1.issue(asset_name + '/SUB/2')
n1.issue(asset_name + '/SUB/3')
n1.issue(asset_name + '/SUB/4')
n1.issue(asset_name + '/SUB/5')
n1.issue(asset_name + '/SUB/6')
n1.issue(asset_name + '/SUB#1')
n1.issue(asset_name + '/SUB#2')
n1.issue(asset_name + '/SUB#3')
n1.issue(asset_name + '/SUB#4')
n1.issue(asset_name + '/SUB#5')
n1.issue(asset_name + '/SUB#6')
n1.issue(asset_name + '/SUB#7')
n1.issue(asset_name + '/SUB/SUB')
n1.issue(asset_name + '/SUB/SUB/SUB')
n1.issue(asset_name + '/SUB/SUB/SUB/SUB')
assert_equal(17, n1.getmempoolinfo()['size'])
connect_all_nodes_bi(self.nodes, True)
# Assert that the reorg was successful
assert_equal(n0.getblockcount(), n1.getblockcount())
assert_equal(n0.getbestblockhash(), n1.getbestblockhash())
# Verify that the mempools are empty ( No extra transactions from the old chain made it into the mempool)
assert_equal(0, n1.getmempoolinfo()['size'])
assert_equal(0, n0.getmempoolinfo()['size'])
def run_test(self):
self.activate_assets()
self.issue_mempool_test()
self.issue_mempool_test_extended()
self.issue_mempool_test_extended_sub()
if __name__ == '__main__':
AssetMempoolTest().main()