tool_bitcoin_chainstate.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2022-present The Bitcoin Core developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """Test bitcoin-chainstate tool functionality
6
7 Test basic block processing via bitcoin-chainstate tool, including detecting
8 duplicates and malformed input.
9
10 Test that bitcoin-chainstate can load a datadir initialized with an assumeutxo
11 snapshot and extend the snapshot chain with new blocks.
12 """
13
14 import subprocess
15
16 from test_framework.test_framework import BitcoinTestFramework
17 from test_framework.util import assert_equal
18 from test_framework.wallet import MiniWallet
19
20 START_HEIGHT = 199
21 # Hardcoded in regtest chainparams
22 SNAPSHOT_BASE_BLOCK_HEIGHT = 299
23 SNAPSHOT_BASE_BLOCK_HASH = "0c552ced4721c249a389eb9b08cb8da261cd46f0e7b5f9d064d48f3113406853"
24
25
26 class BitcoinChainstateTest(BitcoinTestFramework):
27 def skip_test_if_missing_module(self):
28 self.skip_if_no_bitcoin_chainstate()
29
30 def set_test_params(self):
31 """Use the pregenerated, deterministic chain up to height 199."""
32 self.num_nodes = 2
33
34 def setup_network(self):
35 """Start with the nodes disconnected so that one can generate a snapshot
36 including blocks the other hasn't yet seen."""
37 self.add_nodes(2)
38 self.start_nodes()
39
40 def generate_snapshot_chain(self):
41 self.log.info(f"Generate deterministic chain up to block {SNAPSHOT_BASE_BLOCK_HEIGHT} for node0 while node1 disconnected")
42 n0 = self.nodes[0]
43 assert_equal(n0.getblockcount(), START_HEIGHT)
44 n0.setmocktime(n0.getblockheader(n0.getbestblockhash())['time'])
45 mini_wallet = MiniWallet(n0)
46 for i in range(SNAPSHOT_BASE_BLOCK_HEIGHT - n0.getblockchaininfo()["blocks"]):
47 if i % 3 == 0:
48 mini_wallet.send_self_transfer(from_node=n0)
49 self.generate(n0, nblocks=1, sync_fun=self.no_op)
50 assert_equal(n0.getblockcount(), SNAPSHOT_BASE_BLOCK_HEIGHT)
51 assert_equal(n0.getbestblockhash(), SNAPSHOT_BASE_BLOCK_HASH)
52 return n0.dumptxoutset('utxos.dat', "latest")
53
54 def add_block(self, datadir, input, *, expected_stderr=None, expected_stdout=None):
55 proc = subprocess.Popen(
56 self.get_binaries().chainstate_argv() + ["-regtest", datadir],
57 stdin=subprocess.PIPE,
58 stdout=subprocess.PIPE,
59 stderr=subprocess.PIPE,
60 text=True,
61 )
62 stdout, stderr = proc.communicate(input=input + "\n", timeout=5 * self.options.timeout_factor)
63 self.log.debug("STDOUT: {0}".format(stdout.strip("\n")))
64 self.log.info("STDERR: {0}".format(stderr.strip("\n")))
65
66 if expected_stderr is not None and expected_stderr not in stderr:
67 raise AssertionError(f"Expected stderr output '{expected_stderr}' does not partially match stderr:\n{stderr}")
68 if expected_stdout is not None and expected_stdout not in stdout:
69 raise AssertionError(f"Expected stdout output '{expected_stdout}' does not partially match stdout:\n{stdout}")
70
71 def basic_test(self):
72 n0 = self.nodes[0]
73 n1 = self.nodes[1]
74 datadir = n1.chain_path
75 n1.stop_node()
76 block = n0.getblock(n0.getblockhash(START_HEIGHT+1), 0)
77 self.log.info(f"Test bitcoin-chainstate {self.get_binaries().chainstate_argv()} with datadir: {datadir}")
78 self.add_block(datadir, block, expected_stderr="Block has not yet been rejected")
79 self.add_block(datadir, block, expected_stderr="duplicate")
80 self.add_block(datadir, "00", expected_stderr="Block decode failed")
81 self.add_block(datadir, "", expected_stderr="Empty line found")
82
83 def assumeutxo_test(self, dump_output_path):
84 n0 = self.nodes[0]
85 n1 = self.nodes[1]
86 self.start_node(1)
87 self.log.info("Submit headers for new blocks to node1, then load the snapshot so it activates")
88 for height in range(START_HEIGHT+2, SNAPSHOT_BASE_BLOCK_HEIGHT+1):
89 block = n0.getblock(n0.getblockhash(height), 0)
90 n1.submitheader(block)
91 assert_equal(n1.getblockcount(), START_HEIGHT+1)
92 loaded = n1.loadtxoutset(dump_output_path)
93 assert_equal(loaded['base_height'], SNAPSHOT_BASE_BLOCK_HEIGHT)
94 datadir = n1.chain_path
95 n1.stop_node()
96 self.log.info(f"Test bitcoin-chainstate {self.get_binaries().chainstate_argv()} with an assumeutxo datadir: {datadir}")
97 new_tip_hash = self.generate(n0, nblocks=1, sync_fun=self.no_op)[0]
98 self.add_block(datadir, n0.getblock(new_tip_hash, 0), expected_stdout="Block tip changed")
99
100 def run_test(self):
101 dump_output = self.generate_snapshot_chain()
102 self.basic_test()
103 self.assumeutxo_test(dump_output['path'])
104
105 if __name__ == "__main__":
106 BitcoinChainstateTest(__file__).main()
107