#!/usr/bin/env python3 # Copyright (c) 2018-2024, The Monero Project # # All rights reserved. # # Redistribution and use in source and binary forms, with or without modification, are # permitted provided that the following conditions are met: # # 1. Redistributions of source code must retain the above copyright notice, this list of # conditions and the following disclaimer. # # 2. Redistributions in binary form must reproduce the above copyright notice, this list # of conditions and the following disclaimer in the documentation and/or other # materials provided with the distribution. # # 3. Neither the name of the copyright holder nor the names of its contributors may be # used to endorse or promote products derived from this software without specific # prior written permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY # EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF # MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL # THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, # STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF # THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import time """Test daemon blockchain RPC calls Test the following RPCs: - get_info - generateblocks - misc block retrieval - pop_blocks - [TODO: many tests still need to be written] """ from framework.daemon import Daemon from framework.wallet import Wallet BASE_FEE_ESTIMATE = 1200000 class BlockchainTest(): def run_test(self): self.reset() self._test_generateblocks(5) self._test_reorg_handling() self.test_get_blocks_fast() def reset(self): print('Resetting blockchain') daemon = Daemon() res = daemon.get_height() daemon.pop_blocks(res.height - 1) daemon.flush_txpool() def _check_blocktemplate_reserved_offset(self, daemon, address, reserve_size): res = daemon.getblocktemplate(address, reserve_size = reserve_size) assert res.reserved_offset > 0 block_blob = bytes.fromhex(res.blocktemplate_blob) reserved_bytes = block_blob[res.reserved_offset:res.reserved_offset + reserve_size] assert reserved_bytes == bytes(reserve_size) def _test_generateblocks(self, blocks): assert blocks >= 2 print("Test generating", blocks, 'blocks') daemon = Daemon() address = '42ey1afDFnn4886T7196doS9GPMzexD9gXpsZJDwVjeRVdFCSoHnv7KPbBeGpzJBzHRCAs9UxqeoyFQMYbqSWYTfJJQAWDm' # check info/height before generating blocks res_info = daemon.get_info() height = res_info.height prev_block = res_info.top_block_hash res_height = daemon.get_height() assert res_height.height == height assert int(res_info.wide_cumulative_difficulty, 16) == (res_info.cumulative_difficulty_top64 << 64) + res_info.cumulative_difficulty cumulative_difficulty = int(res_info.wide_cumulative_difficulty, 16) # we should not see a block at height ok = False try: daemon.getblock(height = height) except: ok = True assert ok res = daemon.get_fee_estimate() assert res.fee == BASE_FEE_ESTIMATE, res.fee assert res.quantization_mask == 10000 res = daemon.get_fee_estimate(10) assert res.fee <= BASE_FEE_ESTIMATE, res.fee # generate blocks res_generateblocks = daemon.generateblocks(address, blocks) # check info/height after generateblocks blocks assert res_generateblocks.height == height + blocks - 1 res_info = daemon.get_info() assert res_info.height == height + blocks assert res_info.top_block_hash != prev_block res_height = daemon.get_height() assert res_height.height == height + blocks # get the blocks, check they have the right height res_getblock = [] for n in range(blocks): res_getblock.append(daemon.getblock(height = height + n)) block_header = res_getblock[n].block_header assert abs(block_header.timestamp - time.time()) < 60 # within 60 seconds assert block_header.height == height + n assert block_header.orphan_status == False assert block_header.depth == blocks - n - 1 assert block_header.prev_hash == prev_block, prev_block assert int(block_header.wide_difficulty, 16) == (block_header.difficulty_top64 << 64) + block_header.difficulty assert int(block_header.wide_cumulative_difficulty, 16) == (block_header.cumulative_difficulty_top64 << 64) + block_header.cumulative_difficulty assert block_header.reward >= 600000000000 # tail emission cumulative_difficulty += int(block_header.wide_difficulty, 16) assert cumulative_difficulty == int(block_header.wide_cumulative_difficulty, 16) assert block_header.block_size > 0 assert block_header.block_weight >= block_header.block_size assert block_header.long_term_weight > 0 prev_block = block_header.hash # we should not see a block after that ok = False try: daemon.getblock(height = height + blocks) except: ok = True assert ok # getlastblockheader and by height/hash should return the same block res_getlastblockheader = daemon.getlastblockheader() assert res_getlastblockheader.block_header == block_header res_getblockheaderbyhash = daemon.getblockheaderbyhash(prev_block) assert res_getblockheaderbyhash.block_header == block_header res_getblockheaderbyheight = daemon.getblockheaderbyheight(height + blocks - 1) assert res_getblockheaderbyheight.block_header == block_header # getting a block template after that should have the right height, etc res_getblocktemplate = daemon.getblocktemplate(address) assert res_getblocktemplate.height == height + blocks assert res_getblocktemplate.reserved_offset > 0 assert res_getblocktemplate.prev_hash == res_info.top_block_hash assert res_getblocktemplate.expected_reward >= 600000000000 assert res_getblocktemplate.cumulative_weight > 0 assert len(res_getblocktemplate.blocktemplate_blob) > 0 assert len(res_getblocktemplate.blockhashing_blob) > 0 assert int(res_getblocktemplate.wide_difficulty, 16) == (res_getblocktemplate.difficulty_top64 << 64) + res_getblocktemplate.difficulty # 255 matches TX_EXTRA_NONCE_MAX_COUNT in the C++ code. for reserve_size in (1, 127, 128, 255): self._check_blocktemplate_reserved_offset(daemon, address, reserve_size) # diff etc should be the same assert res_getblocktemplate.prev_hash == res_info.top_block_hash res_getlastblockheader = daemon.getlastblockheader() # pop a block res_popblocks = daemon.pop_blocks(1) assert res_popblocks.height == height + blocks - 1 res_info = daemon.get_info() assert res_info.height == height + blocks - 1 # getlastblockheader and by height/hash should return the previous block block_header = res_getblock[blocks - 2].block_header block_header.depth = 0 # this will be different, ignore it res_getlastblockheader = daemon.getlastblockheader() assert res_getlastblockheader.block_header == block_header res_getblockheaderbyhash = daemon.getblockheaderbyhash(block_header.hash) assert res_getblockheaderbyhash.block_header == block_header res_getblockheaderbyheight = daemon.getblockheaderbyheight(height + blocks - 2) assert res_getblockheaderbyheight.block_header == block_header # we should not see the popped block anymore ok = False try: daemon.getblock(height = height + blocks - 1) except: ok = True assert ok # get transactions res = daemon.get_info() assert res.height == height + blocks - 1 nblocks = height + blocks - 1 res = daemon.getblockheadersrange(0, nblocks - 1) assert len(res.headers) == nblocks assert res.headers[-1] == block_header txids = [x.miner_tx_hash for x in res.headers] res = daemon.get_transactions(txs_hashes = txids) assert len(res.txs) == nblocks assert not 'missed_txs' in res or len(res.missed_txs) == 0 running_output_index = 0 for i in range(len(txids)): tx = res.txs[i] assert tx.tx_hash == txids[i] assert not tx.double_spend_seen assert not tx.in_pool assert tx.block_height == i if i > 0: for idx in tx.output_indices: assert idx == running_output_index running_output_index += 1 res_out = daemon.get_outs([{'amount': 0, 'index': idx} for idx in tx.output_indices], get_txid = True) assert len(res_out.outs) == len(tx.output_indices) for out in res_out.outs: assert len(out.key) == 64 assert len(out.mask) == 64 assert not out.unlocked assert out.height == i assert out.txid == txids[i] for i in range(height + nblocks - 1): res_sum = daemon.get_coinbase_tx_sum(i, 1) res_header = daemon.getblockheaderbyheight(i) assert res_sum.emission_amount == res_header.block_header.reward assert res_sum.emission_amount_top64 == 0 assert res_sum.emission_amount == int(res_sum.wide_emission_amount, 16) assert res_sum.fee_amount == int(res_sum.wide_fee_amount, 16) res = daemon.get_coinbase_tx_sum(0, 1) assert res.emission_amount == 17592186044415 assert res.emission_amount_top64 == 0 assert res.fee_amount == 0 assert res.fee_amount_top64 == 0 sum_blocks = height + nblocks - 1 res = daemon.get_coinbase_tx_sum(0, sum_blocks) extrapolated = 17592186044415 + 17592186044415 * 2 * (sum_blocks - 1) assert res.emission_amount < extrapolated and res.emission_amount > extrapolated - 1e12 assert res.fee_amount == 0 sum_blocks_emission = res.emission_amount res = daemon.get_coinbase_tx_sum(1, sum_blocks - 1) assert res.emission_amount == sum_blocks_emission - 17592186044415 assert res.fee_amount == 0 # height + count must not exceed the blockchain height res = daemon.get_coinbase_tx_sum(1, sum_blocks) assert res.status != 'OK' res = daemon.get_output_distribution([0, 1, 17592186044415], 0, 0) assert len(res.distributions) == 3 for a in range(3): assert res.distributions[a].amount == [0, 1, 17592186044415][a] assert res.distributions[a].start_height == 0 assert res.distributions[a].base == 0 assert len(res.distributions[a].distribution) == height + nblocks - 1 assert res.distributions[a].binary == False for i in range(height + nblocks - 1): assert res.distributions[a].distribution[i] == (1 if i > 0 and a == 0 else 1 if a == 2 and i == 0 else 0) res = daemon.get_output_histogram([], min_count = 0, max_count = 0) assert len(res.histogram) == 2 for i in range(2): assert res.histogram[i].amount in [0, 17592186044415] assert res.histogram[i].total_instances in [height + nblocks - 2, 1] assert res.histogram[i].unlocked_instances == 0 assert res.histogram[i].recent_instances == 0 res = daemon.get_fee_estimate() assert res.fee == BASE_FEE_ESTIMATE, res.fee assert res.quantization_mask == 10000 res = daemon.get_fee_estimate(10) assert res.fee <= BASE_FEE_ESTIMATE, res.fee def _test_reorg_handling(self): print('Testing reorg handling') daemon0 = Daemon(idx = 2) daemon1 = Daemon(idx = 3) miner_addr = '42ey1afDFnn4886T7196doS9GPMzexD9gXpsZJDwVjeRVdFCSoHnv7KPbBeGpzJBzHRCAs9UxqeoyFQMYbqSWYTfJJQAWDm' # Wait until daemons are on the same chain, waiting a max of 60s def wait_until_same_chain(): i = 0 retry_attempts = 60 while i < retry_attempts: res0_info = daemon0.get_info() res1_info = daemon1.get_info() if res0_info.height == res1_info.height: assert res0_info.top_block_hash == res1_info.top_block_hash return (res0_info, res1_info) i += 1 time.sleep(1) assert False wait_until_same_chain() # Test various n blocks, some around the default spendable age # Note: starting with 100 because we want the test to include unlocked coinbase outputs for n_block_reorg in [100, 11, 10, 9, 2, 1]: print('Checking', n_block_reorg, 'block reorg') # Disconnect the daemons from their peers daemon0.out_peers(0) daemon1.out_peers(0) # Generate blocks on each daemon, have the second daemon mine 1 extra block so it has the heavier chain # Note: these two calls could be concurrent to speed this test up daemon0.generateblocks(miner_addr, n_block_reorg) genblocks1_res = daemon1.generateblocks(miner_addr, n_block_reorg+1) # Make sure they're on different chains res0_info = daemon0.get_info() res1_info = daemon1.get_info() assert res0_info.top_block_hash != res1_info.top_block_hash assert res1_info.height == (res0_info.height + 1) assert res1_info.height == (genblocks1_res.height + 1) # Now re-connect the daemons and wait until daemon0 gets onto daemon1's heavier chain daemon0.out_peers(12) daemon1.out_peers(12) (res0_info, res1_info) = wait_until_same_chain() assert res0_info.height == (genblocks1_res.height + 1) def test_get_blocks_fast(self): print('Testing the /get_blocks.bin RPC endpoint') daemon = Daemon() n_blocks = daemon.get_height()['height'] assert(n_blocks >= 3) genesis_block_id = daemon.getblockheaderbyheight(0)['block_header']['hash'] target_block_index = n_blocks-1 target_block_id = daemon.getblockheaderbyheight(target_block_index)['block_header']['hash'] print('First, mining to wallet and creating 2 transactions in the top block') wallet = Wallet() seed = 'velvet lymph giddy number token physics poetry unquoted nibs useful sabotage limits benches lifestyle eden nitrogen anvil fewest avoid batch vials washing fences goat unquoted' main_address = '42ey1afDFnn4886T7196doS9GPMzexD9gXpsZJDwVjeRVdFCSoHnv7KPbBeGpzJBzHRCAs9UxqeoyFQMYbqSWYTfJJQAWDm' try: wallet.close_wallet() except: pass wallet.auto_refresh(enable = False) wallet.restore_deterministic_wallet(seed) assert wallet.get_transfers() == {} assert wallet.get_address().address == main_address wallet.refresh() res = wallet.get_transfers() assert len(res['in']) > 0 first_recv_height = res['in'][0]['height'] N_TO_MINE = 20 + max(0, first_recv_height + 60 - n_blocks) daemon.generateblocks(main_address, N_TO_MINE) wallet.refresh() dst = {'address': '8BQKgTSSqJjP14AKnZUBwnXWj46MuNmLvHfPTpmry52DbfNjjHVvHUk4mczU8nj8yZ57zBhksTJ8kM5xKeJXw55kCMVqyG7', 'amount': 1000000000000} res = wallet.get_transfers() assert len(res['in']) >= N_TO_MINE last_recv_height = max(x['height'] for x in res['in']) assert last_recv_height == n_blocks + N_TO_MINE - 1, last_recv_height assert wallet.get_balance().unlocked_balance > dst['amount'] * 2 res = wallet.transfer([dst]) res = wallet.transfer([dst]) assert len(daemon.get_transaction_pool_hashes()['tx_hashes']) == 2 daemon.generateblocks(main_address, 1) print('Calling /get_blocks.bin (blocks only) and testing response...') res = daemon.get_blocks_fast(0, [target_block_id, genesis_block_id]) assert len(res.blocks) == N_TO_MINE + 1 + 1 assert len(res.blocks) == len(res.output_indices) for i in range(len(res.blocks)): is_last = i == len(res.blocks) - 1 block = res.blocks[i] assert block.pruned block_indices = res.output_indices[i]['indices'] n_txs = len(block['txs']) if 'txs' in block else 0 if is_last: assert n_txs == 2 assert len(block_indices) == 3 else: assert n_txs == 0 assert len(block_indices) == 1 for tx_idx in range(n_txs): tx = block.txs[tx_idx] tx_indices = block_indices[1 + tx_idx]['indices'] assert tx.prunable_hash != (b'\0' * 32) # non-null assert len(tx_indices) == 2 print('Calling /get_blocks.bin (blocks and incremental pool, no new blocks) and testing response...') time.sleep(1) res = daemon.get_blocks_fast(0, [], requested_info = 2) assert res.pool_info_extent == 2 pool_info_since = res.daemon_time wallet.refresh() pending_txid = wallet.transfer([dst]).tx_hash current_height = daemon.get_height().height res = daemon.get_blocks_fast( current_height, [], requested_info = 1, pool_info_since = pool_info_since) assert not res.get('blocks') assert res.pool_info_extent == 1 assert [tx.tx_hash.hex() for tx in res.added_pool_txs] == [pending_txid] daemon.flush_txpool([pending_txid]) if __name__ == '__main__': BlockchainTest().run_test()