test: add `-alertnotify` test for large work invalid chain warning

diff --git a/test/functional/feature_notifications.py b/test/functional/feature_notifications.py
index de485222..519d80b9 100755
--- a/test/functional/feature_notifications.py
+++ b/test/functional/feature_notifications.py
@@ -7,6 +7,10 @@ import os
 import platform
 
 from test_framework.address import ADDRESS_BCRT1_UNSPENDABLE
+from test_framework.blocktools import (
+    create_block,
+    create_coinbase,
+)
 from test_framework.descriptors import descsum_create
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
@@ -20,6 +24,12 @@ FILE_CHAR_END = 128
 FILE_CHARS_DISALLOWED = '/\\?%*:|"<>' if platform.system() == 'Windows' else '/'
 UNCONFIRMED_HASH_STRING = 'unconfirmed'
 
+LARGE_WORK_INVALID_CHAIN_WARNING = (
+    "Warning: We do not appear to fully agree with our peers "  # Exclamation mark removed by SanitizeString in AlertNotify
+    "You may need to upgrade, or other nodes may need to upgrade."
+)
+
+
 def notify_outputname(walletname, txid):
     return txid if platform.system() == 'Windows' else f'{walletname}_{txid}'
 
@@ -33,6 +43,7 @@ class NotificationsTest(BitcoinTestFramework):
     def setup_network(self):
         self.wallet = ''.join(chr(i) for i in range(FILE_CHAR_START, FILE_CHAR_END) if chr(i) not in FILE_CHARS_DISALLOWED)
         self.alertnotify_dir = os.path.join(self.options.tmpdir, "alertnotify")
+        self.alertnotify_file = os.path.join(self.alertnotify_dir, "alertnotify.txt")
         self.blocknotify_dir = os.path.join(self.options.tmpdir, "blocknotify")
         self.walletnotify_dir = os.path.join(self.options.tmpdir, "walletnotify")
         self.shutdownnotify_dir = os.path.join(self.options.tmpdir, "shutdownnotify")
@@ -44,7 +55,7 @@ class NotificationsTest(BitcoinTestFramework):
 
         # -alertnotify and -blocknotify on node0, walletnotify on node1
         self.extra_args = [[
-            f"-alertnotify=echo > {os.path.join(self.alertnotify_dir, '%s')}",
+            f"-alertnotify=echo %s >> {self.alertnotify_file}",
             f"-blocknotify=echo > {os.path.join(self.blocknotify_dir, '%s')}",
             f"-shutdownnotify=echo > {self.shutdownnotify_file}",
         ], [
@@ -156,12 +167,43 @@ class NotificationsTest(BitcoinTestFramework):
             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 -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
+
     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