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

Code before:
    def test_acceptstalefeeestimates_option(self):
        # Get the initial fee rate while node is running
        fee_rate = self.nodes[0].estimatesmartfee(1)["feerate"]

        self.stop_node(0)

        fee_dat = self.nodes[0].chain_path / "fee_estimates.dat"

        # Stop the node and backdate the fee_estimates.dat file more than MAX_FILE_AGE
        last_modified_time = time.time() - (MAX_FILE_AGE + 1) * SECONDS_PER_HOUR
        os.utime(fee_dat, (last_modified_time, last_modified_time))

        # Restart node with -acceptstalefeeestimates option to ensure fee_estimate.dat file is read
        self.start_node(0,extra_args=["-acceptstalefeeestimates"])
        assert_equal(self.nodes[0].estimatesmartfee(1)["feerate"], fee_rate)


    def run_test(self):
        self.log.info("This test is time consuming, please be patient")
        self.log.info("Splitting inputs so we can generate tx's")

        # Split two coinbases into many small utxos
        self.start_node(0)
        self.wallet = MiniWallet(self.nodes[0])
        self.initial_split(self.nodes[0])
        self.log.info("Finished splitting")

        # Now we can connect the other nodes, didn't want to connect them earlier
        # so the estimates would not be affected by the splitting transactions
        self.start_node(1)
        self.start_node(2)
        self.connect_nodes(1, 0)
        self.connect_nodes(0, 2)
        self.connect_nodes(2, 1)
        self.sync_all()

        self.log.info("Testing estimates with single transactions.")
        self.sanity_check_estimates_range()

        self.log.info("Test fee_estimates.dat is flushed periodically")
        self.test_estimate_dat_is_flushed_periodically()

        # check that the effective feerate is greater than or equal to the mempoolminfee even for high mempoolminfee
        self.log.info(
            "Test fee rate estimation after restarting node with high MempoolMinFee"
        )
        self.test_feerate_mempoolminfee()

        self.log.info("Test acceptstalefeeestimates option")
        self.test_acceptstalefeeestimates_option()


Review comment (suggestion):
nit: if you retouch `Pathlib` lets you use the `/` operator to create child paths:

```suggestion
        fee_dat = self.nodes[0].chain_path / "fee_estimates.dat"
```

Code after:

        # Generate blocks before shutdown and verify that the fee estimates are not the same
        self.generate(self.nodes[0], 5, sync_fun=self.no_op)
        self.restart_node(0)
        fee_dat_current_content = open(fee_dat, "rb").read()
        assert fee_dat_current_content != fee_dat_initial_content


    def test_acceptstalefeeestimates_option(self):
        # Get the initial fee rate while node is running
        fee_rate = self.nodes[0].estimatesmartfee(1)["feerate"]

        self.stop_node(0)

        fee_dat = self.nodes[0].chain_path / "fee_estimates.dat"

        # Stop the node and backdate the fee_estimates.dat file more than MAX_FILE_AGE
        last_modified_time = time.time() - (MAX_FILE_AGE + 1) * SECONDS_PER_HOUR
        os.utime(fee_dat, (last_modified_time, last_modified_time))

        # Restart node with -acceptstalefeeestimates option to ensure fee_estimate.dat file is read
        self.start_node(0,extra_args=["-acceptstalefeeestimates"])
        assert_equal(self.nodes[0].estimatesmartfee(1)["feerate"], fee_rate)

    def clear_estimates(self):
        self.log.info("Restarting node with fresh estimation")
        self.stop_node(0)
        fee_dat = self.nodes[0].chain_path / "fee_estimates.dat"
        os.remove(fee_dat)
        self.start_node(0)
        self.connect_nodes(0, 1)
        self.connect_nodes(0, 2)
        assert_equal(self.nodes[0].estimatesmartfee(1)["errors"], ["Insufficient data or no feerate found"])

    def broadcast_and_mine(self, broadcaster, miner, feerate, count):
        """Broadcast and mine some number of transactions with a specified fee rate."""
        for _ in range(count):
            self.wallet.send_self_transfer(from_node=broadcaster, fee_rate=feerate)
        self.sync_mempools()
        self.generate(miner, 1)

    def test_estimation_modes(self):
        low_feerate = Decimal("0.001")
        high_feerate = Decimal("0.005")
        tx_count = 24
        # Broadcast and mine high fee transactions for the first 12 blocks.
        for _ in range(12):
            self.broadcast_and_mine(self.nodes[1], self.nodes[2], high_feerate, tx_count)
        check_fee_estimates_btw_modes(self.nodes[0], high_feerate, high_feerate)

        # We now track 12 blocks; short horizon stats will start decaying.