rpc_getblocklocations.py raw

   1  #!/usr/bin/env python3
   2  # Copyright (c) 2019-2020 The Limenka 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 the getblocklocations rpc call."""
   6  from test_framework.test_framework import LimenkaTestFramework
   7  from test_framework.util import (
   8      assert_equal,
   9      assert_raises_rpc_error,
  10      util_xor,
  11  )
  12  from test_framework.messages import ser_vector
  13  
  14  
  15  class GetblocklocationsTest(LimenkaTestFramework):
  16      def set_test_params(self):
  17          self.setup_clean_chain = True
  18          self.num_nodes = 1
  19  
  20      def run_test(self):
  21          """Test a trivial usage of the getblocklocations RPC command."""
  22          node = self.nodes[0]
  23          test_block_count = 7
  24          self.generate(node, test_block_count)
  25  
  26          NULL_HASH = '0000000000000000000000000000000000000000000000000000000000000000'
  27  
  28          block_hashes = [node.getblockhash(height) for height in range(test_block_count)]
  29          block_hashes.reverse()
  30  
  31          block_locations = {}
  32          def check_consistency(tip, a):
  33              for o in a:
  34                  if tip in block_locations:
  35                      assert_equal(block_locations[tip], o)
  36                  else:
  37                      block_locations[tip] = o
  38                  tip = o['prev']
  39  
  40          # Get blocks' locations using several batch sizes
  41          last_locations = None
  42          for batch_size in range(1, 10):
  43              locations = []
  44              tip = block_hashes[0]
  45              while tip != NULL_HASH:
  46                  locations.extend(node.getblocklocations(tip, batch_size))
  47                  check_consistency(block_hashes[0], locations)
  48                  tip = locations[-1]['prev']
  49              if last_locations: assert_equal(last_locations, locations)
  50              last_locations = locations
  51  
  52          xor_key = node.read_xor_key()
  53  
  54          # Read blocks' data from the file system
  55          blocks_dir = node.chain_path / 'blocks'
  56          with (blocks_dir / 'blk00000.dat').open('rb') as blkfile:
  57              for block_hash in block_hashes:
  58                  location = block_locations[block_hash]
  59                  block_bytes = bytes.fromhex(node.getblock(block_hash, 0))
  60                  assert_file_contains(blkfile, location['data'], block_bytes, xor_key)
  61  
  62  
  63          empty_undo = ser_vector([])  # empty blocks = no transactions to undo
  64          with (blocks_dir / 'rev00000.dat').open('rb') as revfile:
  65              for block_hash in block_hashes[:-1]:  # skip genesis block (has no undo)
  66                  location = block_locations[block_hash]
  67                  assert_file_contains(revfile, location['undo'], empty_undo, xor_key)
  68  
  69          # Fail getting unknown block
  70          unknown_block_hash = '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef'
  71          assert_raises_rpc_error(-5, 'Block not found', node.getblocklocations, unknown_block_hash, 3)
  72  
  73          # Fail in pruned mode
  74          self.restart_node(0, ['-prune=1'])
  75          tip = block_hashes[0]
  76          assert_raises_rpc_error(-1, 'Block locations are not available in prune mode', node.getblocklocations, tip, 3)
  77  
  78  
  79  def assert_file_contains(fileobj, offset, data, xor_key):
  80      fileobj.seek(offset)
  81      read_data = fileobj.read(len(data))
  82      read_data = util_xor(read_data, xor_key, offset=offset)
  83      assert_equal(read_data, data)
  84  
  85  if __name__ == '__main__':
  86      GetblocklocationsTest(__file__).main()
  87