Repository: bitcoin/bitcoin
File: test/functional/feature_assumeutxo.py (Python)

Code before:
        self.restart_node(0, ["-reindex"])

        # Ensure n0 is back at the tip
        assert_equal(n0.getblockchaininfo()["blocks"], FINAL_HEIGHT)

        self.test_snapshot_with_less_work(dump_output['path'])
        self.test_invalid_mempool_state(dump_output['path'])
        self.test_invalid_snapshot_scenarios(dump_output['path'])
        self.test_invalid_chainstate_scenarios()
        self.test_invalid_file_path()
        self.test_snapshot_block_invalidated(dump_output['path'])
        self.test_snapshot_not_on_most_work_chain(dump_output['path'])

        # Prune-node sanity check
        assert 'NETWORK' not in n1.getnetworkinfo()['localservicesnames']

        self.log.info(f"Loading snapshot into second node from {dump_output['path']}")
        # This node's tip is on an ancestor block of the snapshot, which should
        # be the normal case
        loaded = n1.loadtxoutset(dump_output['path'])
        assert_equal(loaded['coins_loaded'], SNAPSHOT_BASE_HEIGHT)
        assert_equal(loaded['base_height'], SNAPSHOT_BASE_HEIGHT)

        self.log.info("Confirm that local services remain unchanged")
        # Since n1 is a pruned node, the 'NETWORK' service flag must always be unset.
        prune_node_services = n1.getnetworkinfo()['localservicesnames']
        assert 'NETWORK' not in prune_node_services
        assert 'NETWORK_LIMITED' in prune_node_services

        self.log.info("Check that UTXO-querying RPCs operate on snapshot chainstate")
        snapshot_hash = loaded['tip_hash']
        snapshot_num_coins = loaded['coins_loaded']
        # coinstatsindex might be not caught up yet and is not relevant for this test, so don't use it
        utxo_info = n1.gettxoutsetinfo(use_index=False)
        assert_equal(utxo_info['txouts'], snapshot_num_coins)
        assert_equal(utxo_info['height'], SNAPSHOT_BASE_HEIGHT)
        assert_equal(utxo_info['bestblock'], snapshot_hash)

        # find coinbase output at snapshot height on node0 and scan for it on node1,
        # where the block is not available, but the snapshot was loaded successfully
        coinbase_tx = n0.getblock(snapshot_hash, verbosity=2)['tx'][0]
        assert_raises_rpc_error(-1, "Block not found on disk", n1.getblock, snapshot_hash)
        coinbase_output_descriptor = coinbase_tx['vout'][0]['scriptPubKey']['desc']
        scan_result = n1.scantxoutset('start', [coinbase_output_descriptor])
        assert_equal(scan_result['success'], True)
        assert_equal(scan_result['txouts'], snapshot_num_coins)
        assert_equal(scan_result['height'], SNAPSHOT_BASE_HEIGHT)
        assert_equal(scan_result['bestblock'], snapshot_hash)
        scan_utxos = [(coin['txid'], coin['vout']) for coin in scan_result['unspents']]
        assert (coinbase_tx['txid'], 0) in scan_utxos


Review comment (refactor):
Since this check is repeated six times it could be a reusable function

```python
def check_network_limited(node):
    node_services = node.getnetworkinfo()['localservicesnames']
    assert 'NETWORK' not in node_services
    assert 'NETWORK_LIMITED' in node_services
```

Code after:
        self.log.info("-- Testing assumeutxo + some indexes + pruning")

        assert_equal(n0.getblockcount(), SNAPSHOT_BASE_HEIGHT)
        assert_equal(n1.getblockcount(), START_HEIGHT)

        self.log.info(f"Creating a UTXO snapshot at height {SNAPSHOT_BASE_HEIGHT}")
        dump_output = n0.dumptxoutset('utxos.dat', "latest")

        self.log.info("Test loading snapshot when the node tip is on the same block as the snapshot")
        assert_equal(n0.getblockcount(), SNAPSHOT_BASE_HEIGHT)
        assert_equal(n0.getblockchaininfo()["blocks"], SNAPSHOT_BASE_HEIGHT)
        self.test_snapshot_with_less_work(dump_output['path'])

        self.log.info("Test loading snapshot when headers are not synced")
        self.test_headers_not_synced(dump_output['path'])

        # In order for the snapshot to activate, we have to ferry over the new
        # headers to n1 and n2 so that they see the header of the snapshot's
        # base block while disconnected from n0.
        for i in range(1, 300):
            block = n0.getblock(n0.getblockhash(i), 0)
            # make n1 and n2 aware of the new header, but don't give them the
            # block.
            n1.submitheader(block)
            n2.submitheader(block)
            n3.submitheader(block)

        # Ensure everyone is seeing the same headers.
        for n in self.nodes:
            assert_equal(n.getblockchaininfo()["headers"], SNAPSHOT_BASE_HEIGHT)

        assert_equal(n0.getblockchaininfo()["blocks"], SNAPSHOT_BASE_HEIGHT)

        def check_dump_output(output):
            assert_equal(
                output['txoutset_hash'],
                "a4bf3407ccb2cc0145c49ebba8fa91199f8a3903daf0883875941497d2493c27")
            assert_equal(output["nchaintx"], blocks[SNAPSHOT_BASE_HEIGHT].chain_tx)

        check_dump_output(dump_output)

        # Mine more blocks on top of the snapshot that n1 hasn't yet seen. This
        # will allow us to test n1's sync-to-tip on top of a snapshot.
        self.generate(n0, nblocks=100, sync_fun=self.no_op)

        assert_equal(n0.getblockcount(), FINAL_HEIGHT)
        assert_equal(n1.getblockcount(), START_HEIGHT)

        assert_equal(n0.getblockchaininfo()["blocks"], FINAL_HEIGHT)

        self.log.info(f"Check that dumptxoutset works for past block heights")