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

Code before:
            assert_raises_rpc_error(-32603, "Unable to load UTXO snapshot", node.loadtxoutset, dump_output_path)
        self.restart_node(0, extra_args=self.extra_args[0])

    def test_snapshot_in_a_divergent_chain(self, dump_output_path):
        n0 = self.nodes[0]
        n3 = self.nodes[3]
        n4 = self.nodes[4]
        assert_equal(n0.getblockcount(), FINAL_HEIGHT)
        assert_equal(n3.getblockcount(), START_HEIGHT)
        assert_equal(n4.getblockcount(), START_HEIGHT)

        self.log.info(f"Check importing a snapshot where current chain-tip is not an ancestor of the snapshot block but has less work")
        # Generate a divergent chain in n3 up to 298
        self.generate(n3, nblocks=99, sync_fun=self.no_op)
        assert_equal(n3.getblockcount(), SNAPSHOT_BASE_HEIGHT - 1)

        # Try importing the snapshot and assert its success
        self.log.info('Importing the snapshot into n3')
        loaded = n3.loadtxoutset(dump_output_path)
        assert_equal(loaded['coins_loaded'], SNAPSHOT_BASE_HEIGHT)
        assert_equal(loaded['base_height'], SNAPSHOT_BASE_HEIGHT)

        # Now lets sync the nodes and wait for the background validation to finish
        self.connect_nodes(0, 3)
        self.sync_blocks(nodes=(n0, n3))
        print('Ensuring background validation finishes')
        self.wait_until(lambda: len(n3.getchainstates()['chainstates']) == 1)

        self.log.info(f"Check importing a snapshot where current chain-tip is not an ancestor or a descendant of the snapshot block and has more work")
        # Generate a divergent chain in n4 that has more work than the snapshot
        # This covers the scenario where the snapshot block is not on the most-work chain
        self.generate(n4, nblocks=101, sync_fun=self.no_op)
        assert_equal(n4.getblockcount(), SNAPSHOT_BASE_HEIGHT + 1)
        # Import the snapshot and assert its failure
        with n4.assert_debug_log(expected_msgs=["[snapshot] activation failed - work does not exceed active chainstate"]):
            assert_raises_rpc_error(-32603, "Unable to load UTXO snapshot", n4.loadtxoutset, dump_output_path)

    def run_test(self):
        """
        Bring up two (disconnected) nodes, mine some new blocks on the first,
        and generate a UTXO snapshot.

        Load the snapshot into the second, ensure it syncs to tip and completes
        background validation when connected to the first.
        """
        n0 = self.nodes[0]
        n1 = self.nodes[1]
        n2 = self.nodes[2]
        n3 = self.nodes[3]
        n4 = self.nodes[4]


Review comment (suggestion):
```suggestion
        self.log.info('Ensuring background validation finishes')
```

Code after:
        assert_equal(node.getblockcount(), FINAL_HEIGHT)
        with node.assert_debug_log(expected_msgs=["[snapshot] activation failed - work does not exceed active chainstate"]):
            assert_raises_rpc_error(-32603, "Unable to load UTXO snapshot", node.loadtxoutset, dump_output_path)
        self.restart_node(0, extra_args=self.extra_args[0])

    def test_snapshot_in_a_divergent_chain(self, dump_output_path):
        n0 = self.nodes[0]
        n3 = self.nodes[3]
        assert_equal(n0.getblockcount(), FINAL_HEIGHT)
        assert_equal(n3.getblockcount(), START_HEIGHT)

        self.log.info("Check importing a snapshot where current chain-tip is not an ancestor of the snapshot block but has less work")
        # Generate a divergent chain in n3 up to 298
        self.generate(n3, nblocks=99, sync_fun=self.no_op)
        assert_equal(n3.getblockcount(), SNAPSHOT_BASE_HEIGHT - 1)

        # Try importing the snapshot and assert its success
        loaded = n3.loadtxoutset(dump_output_path)
        assert_equal(loaded['base_height'], SNAPSHOT_BASE_HEIGHT)
        normal, snapshot = n3.getchainstates()["chainstates"]
        assert_equal(normal['blocks'], START_HEIGHT + 99)
        assert_equal(snapshot['blocks'], SNAPSHOT_BASE_HEIGHT)

        # Now lets sync the nodes and wait for the background validation to finish
        self.connect_nodes(0, 3)
        self.sync_blocks(nodes=(n0, n3))
        self.wait_until(lambda: len(n3.getchainstates()['chainstates']) == 1)

    def run_test(self):
        """
        Bring up two (disconnected) nodes, mine some new blocks on the first,
        and generate a UTXO snapshot.

        Load the snapshot into the second, ensure it syncs to tip and completes
        background validation when connected to the first.
        """
        n0 = self.nodes[0]
        n1 = self.nodes[1]
        n2 = self.nodes[2]
        n3 = self.nodes[3]

        self.mini_wallet = MiniWallet(n0)

        # Mock time for a deterministic chain
        for n in self.nodes:
            n.setmocktime(n.getblockheader(n.getbestblockhash())['time'])

        # Generate a series of blocks that `n0` will have in the snapshot,
        # but that n1 and n2 don't yet see.
        assert n0.getblockcount() == START_HEIGHT
        blocks = {START_HEIGHT: Block(n0.getbestblockhash(), 1, START_HEIGHT + 1)}