test/functional/notifications.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2014-2017 The Bitcoin Core developers
   3 # Distributed under the MIT software license, see the accompanying
   4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5 """Test the -alertnotify, -blocknotify and -walletnotify options."""
   6 import os
   7
   8 from test_framework.test_framework import BitcoinTestFramework
   9 from test_framework.util import assert_equal, wait_until, connect_nodes_bi
  10
  11 class NotificationsTest(BitcoinTestFramework):
  12     def set_test_params(self):
  13         self.num_nodes = 2
  14         self.setup_clean_chain = True
  15
  16     def setup_network(self):
  17         self.alert_filename = os.path.join(self.options.tmpdir, "alert.txt")
  18         self.block_filename = os.path.join(self.options.tmpdir, "blocks.txt")
  19         self.tx_filename = os.path.join(self.options.tmpdir, "transactions.txt")
  20
  21         # -alertnotify and -blocknotify on node0, walletnotify on node1
  22         self.extra_args = [["-blockversion=2",
  23                             "-alertnotify=echo %%s >> %s" % self.alert_filename,
  24                             "-blocknotify=echo %%s >> %s" % self.block_filename],
  25                            ["-blockversion=211",
  26                             "-rescan",
  27                             "-walletnotify=echo %%s >> %s" % self.tx_filename]]
  28         super().setup_network()
  29
  30     def run_test(self):
  31         self.log.info("test -blocknotify")
  32         block_count = 10
  33         blocks = self.nodes[1].generate(block_count)
  34
  35         # wait at most 10 seconds for expected file size before reading the content
  36         wait_until(lambda: os.path.isfile(self.block_filename) and os.stat(self.block_filename).st_size >= (block_count * 65), timeout=10)
  37
  38         # file content should equal the generated blocks hashes
  39         with open(self.block_filename, 'r') as f:
  40             assert_equal(sorted(blocks), sorted(f.read().splitlines()))
  41
  42         self.log.info("test -walletnotify")
  43         # wait at most 10 seconds for expected file size before reading the content
  44         wait_until(lambda: os.path.isfile(self.tx_filename) and os.stat(self.tx_filename).st_size >= (block_count * 65), timeout=10)
  45
  46         # file content should equal the generated transaction hashes
  47         txids_rpc = list(map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count)))
  48         with open(self.tx_filename, 'r') as f:
  49             assert_equal(sorted(txids_rpc), sorted(f.read().splitlines()))
  50         os.remove(self.tx_filename)
  51
  52         self.log.info("test -walletnotify after rescan")
  53         # restart node to rescan to force wallet notifications
  54         self.restart_node(1)
  55         connect_nodes_bi(self.nodes, 0, 1)
  56
  57         wait_until(lambda: os.path.isfile(self.tx_filename) and os.stat(self.tx_filename).st_size >= (block_count * 65), timeout=10)
  58
  59         # file content should equal the generated transaction hashes
  60         txids_rpc = list(map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count)))
  61         with open(self.tx_filename, 'r') as f:
  62             assert_equal(sorted(txids_rpc), sorted(f.read().splitlines()))
  63
  64         # Mine another 41 up-version blocks. -alertnotify should trigger on the 51st.
  65         self.log.info("test -alertnotify")
  66         self.nodes[1].generate(41)
  67         self.sync_all()
  68
  69         # Give bitcoind 10 seconds to write the alert notification
  70         wait_until(lambda: os.path.isfile(self.alert_filename) and os.path.getsize(self.alert_filename), timeout=10)
  71
  72         with open(self.alert_filename, 'r', encoding='utf8') as f:
  73             alert_text = f.read()
  74
  75         # Mine more up-version blocks, should not get more alerts:
  76         self.nodes[1].generate(2)
  77         self.sync_all()
  78
  79         with open(self.alert_filename, 'r', encoding='utf8') as f:
  80             alert_text2 = f.read()
  81
  82         self.log.info("-alertnotify should not continue notifying for more unknown version blocks")
  83         assert_equal(alert_text, alert_text2)
  84
  85 if __name__ == '__main__':
  86     NotificationsTest().main()