-
Notifications
You must be signed in to change notification settings - Fork 2
/
single_node_undo.py
executable file
·108 lines (84 loc) · 4.04 KB
/
single_node_undo.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
#!/usr/bin/env python3
# Copyright (c) 2017 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
# Avoid wildcard * imports if possible
import time
from test_framework.blocktools import (create_coinbase, create_transaction, create_block_with_txns,
make_conform_to_ctor)
from test_framework.mininode import (
NetworkThread,
NodeConn,
NodeConnCB,
msg_block,
)
from test_framework.script import CScript, OP_TRUE
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
p2p_port,
)
class SingleNodeUndoTest(BitcoinTestFramework):
def set_test_params(self):
self.setup_clean_chain = True
self.num_nodes = 1
def setup_network(self):
self.extra_args = [['--whitelist=127.0.0.1']] * self.num_nodes
self.setup_nodes()
def run_test(self):
node0 = NodeConnCB()
node0.add_connection(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))
NetworkThread().start()
node0.wait_for_verack()
self.log.info("#1. generate 1 block by node0==================================================================")
self.nodes[0].generate(nblocks=1)
self.tip = int(self.nodes[0].getbestblockhash(), 16)
self.block_time = self.nodes[0].getblock(self.nodes[0].getbestblockhash())['time'] + 1
self.height = 1
self.coinbase_txs = []
self.log.info("#2. create 100 blocks and send to node0========================================================")
for i in range(100):
coinbase_tx = create_coinbase(self.height)
self.coinbase_txs.append(coinbase_tx)
self.create_block_and_send([coinbase_tx], node0)
self.nodes[0].waitforblockheight(101)
self.fork_point_hash = self.tip
self.fork_height = self.height
self.log.info("#3. create one fork chain with one block=======================================================")
for i in range(1):
block_fee, txns = self.create_txns_from(self.coinbase_txs[i], 2)
coinbase = create_coinbase(self.height, None, block_fee)
self.create_block_and_send([coinbase] + txns, node0)
self.nodes[0].waitforblockheight(102)
self.log.info("#4. create another fork chain with two blocks==================================================")
self.tip = self.fork_point_hash
self.height = self.fork_height
for i in range(2):
block_fee, txns = self.create_txns_from(self.coinbase_txs[i], 2)
coinbase = create_coinbase(self.height, None, block_fee)
self.create_block_and_send([coinbase] + txns, node0)
self.log.info("#5. expect node0 switch to new chain===========================================================")
self.nodes[0].waitforblockheight(103)
def create_txns_from(self, input_txn, num=1):
input_value = input_txn.vout[0].nValue
tx_fee = 10000
assert (num * tx_fee <= input_value)
txn_input_values = [ input_value - i*tx_fee for i in range(num)]
txn_output_values = [ txn_input_values[i] - tx_fee for i in range(num)]
block_fee = sum(txn_input_values) - sum(txn_output_values)
txns = [create_transaction(input_txn, 0, b'', txn_output_values[0], CScript([OP_TRUE]))]
for i in range(1, num):
txn = create_transaction(txns[i - 1], 0, b'', txn_output_values[i], CScript([OP_TRUE]))
txns.append(txn)
return block_fee, txns
def create_block_and_send(self, txs, node):
block = create_block_with_txns(self.tip, txs, self.block_time)
make_conform_to_ctor(block)
block.hashMerkleRoot = block.calc_merkle_root()
block.calc_sha256()
block.solve()
node.send_message(msg_block(block))
self.tip = block.sha256
self.block_time += 1
self.height += 1
if __name__ == '__main__':
SingleNodeUndoTest().main()