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

Code before:
            self.sync_blocks()
            self.expect_wallet_notify([(bump2, blockheight2, blockhash2), (tx2, -1, UNCONFIRMED_HASH_STRING)])
            assert_equal(self.nodes[1].gettransaction(bump2)["confirmations"], 1)

        # TODO: add test for `-alertnotify` large fork notifications

        self.log.info("test -shutdownnotify")
        self.stop_nodes()
        self.wait_until(lambda: os.path.isfile(self.shutdownnotify_file), timeout=10)

    def expect_wallet_notify(self, tx_details):
        self.wait_until(lambda: len(os.listdir(self.walletnotify_dir)) >= len(tx_details), timeout=10)
        # Should have no more and no less files than expected
        assert_equal(sorted(notify_outputname(self.wallet, tx_id) for tx_id, _, _ in tx_details), sorted(os.listdir(self.walletnotify_dir)))
        # Should now verify contents of each file
        for tx_id, blockheight, blockhash in tx_details:
            fname = os.path.join(self.walletnotify_dir, notify_outputname(self.wallet, tx_id))
            # Wait for the cached writes to hit storage
            self.wait_until(lambda: os.path.getsize(fname) > 0, timeout=10)
            with open(fname, 'rt', encoding='utf-8') as f:
                text = f.read()
                # Universal newline ensures '\n' on 'nt'
                assert_equal(text[-1], '\n')
                text = text[:-1]
                if platform.system() == 'Windows':
                    # On Windows, echo as above will append a whitespace
                    assert_equal(text[-1], ' ')
                    text = text[:-1]
                expected = str(blockheight) + '_' + blockhash
                assert_equal(text, expected)

        for tx_file in os.listdir(self.walletnotify_dir):
            os.remove(os.path.join(self.walletnotify_dir, tx_file))


if __name__ == '__main__':
    NotificationsTest(__file__).main()


Review comment (suggestion):
Re: https://github.com/bitcoin/bitcoin/pull/33893#issuecomment-3549085496

> Another alternative could be to create and submit one invalid block, then submit a longer chain forking enough blocks before the invalid block

I was tinkering with this portion and found that by making only the first block invalid while keeping the rest of the blocks valid, and submitting the invalid block at the end also causes the warning.

```diff
--- a/test/functional/feature_notifications.py
+++ b/test/functional/feature_notifications.py
@@ -174,10 +174,11 @@ class NotificationsTest(BitcoinTestFramework):
         block_time = self.nodes[0].getblock(tip)['time'] + 1
 
         invalid_blocks = []
-        for _ in range(7):  # invalid chain must be longer than 6 blocks to trigger warning
+        for i in range(7):  # invalid chain must be longer than 6 blocks to trigger warning
             block = create_block(int(tip, 16), create_coinbase(height), block_time)
             # make block invalid by exceeding block subsidy
-            block.vtx[0].vout[0].nValue += 1
+            if i == 0:
+                block.vtx[0].vout[0].nValue += 1
             block.hashMerkleRoot = block.calc_merkle_root()
             block.solve()
             invalid_blocks.append(block)
```

Code after:

            # Bump tx2 as bump2 and generate a block on node 0 while
            # disconnected, then reconnect and check for notifications on node 1
            # about newly confirmed bump2 and newly conflicted tx2.
            self.disconnect_nodes(0, 1)
            bump2 = self.nodes[0].bumpfee(tx2)["txid"]
            blockhash2 = self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE, sync_fun=self.no_op)[0]
            blockheight2 = self.nodes[0].getblockcount()
            assert_equal(self.nodes[0].gettransaction(bump2)["confirmations"], 1)
            assert_equal(tx2 in self.nodes[1].getrawmempool(), True)
            self.connect_nodes(0, 1)
            self.sync_blocks()
            self.expect_wallet_notify([(bump2, blockheight2, blockhash2), (tx2, -1, UNCONFIRMED_HASH_STRING)])
            assert_equal(self.nodes[1].gettransaction(bump2)["confirmations"], 1)

        self.log.info("test -alertnotify with large work invalid chain")
        # create a bunch of invalid blocks
        tip = self.nodes[0].getbestblockhash()
        height = self.nodes[0].getblockcount() + 1
        block_time = self.nodes[0].getblock(tip)['time'] + 1

        invalid_blocks = []
        for _ in range(7):  # invalid chain must be longer than 6 blocks to trigger warning
            block = create_block(int(tip, 16), create_coinbase(height), block_time)
            # make block invalid by exceeding block subsidy
            block.vtx[0].vout[0].nValue += 1
            block.hashMerkleRoot = block.calc_merkle_root()
            block.solve()
            invalid_blocks.append(block)
            tip = block.hash_hex
            height += 1
            block_time += 1

        # submit headers of invalid blocks
        for invalid_block in invalid_blocks:
            self.nodes[0].submitheader(invalid_block.serialize().hex())
        # submit invalid blocks in reverse order (tip first, to set m_best_invalid)
        for invalid_block in reversed(invalid_blocks):
            self.nodes[0].submitblock(invalid_block.serialize().hex())

        self.wait_until(lambda: os.path.isfile(self.alertnotify_file), timeout=10)
        self.wait_until(self.large_work_invalid_chain_warning_in_alert_file, timeout=10)

        self.log.info("test -shutdownnotify")
        self.stop_nodes()
        self.wait_until(lambda: os.path.isfile(self.shutdownnotify_file), timeout=10)

    def large_work_invalid_chain_warning_in_alert_file(self):
        with open(self.alertnotify_file, 'r', encoding='utf8') as f:
            alert_text = f.read()
        return LARGE_WORK_INVALID_CHAIN_WARNING in alert_text