Differential D9961 Diff 29609 test/functional/feature_notifications.py

Changeset View

Standalone View

test/functional/feature_notifications.py

#!/usr/bin/env python3		#!/usr/bin/env python3
# Copyright (c) 2014-2019 The Bitcoin Core developers		# Copyright (c) 2014-2019 The Bitcoin Core developers
# Copyright (c) 2018 The Bitcoin developers		# Copyright (c) 2018 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying		# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.		# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test the -alertnotify, -blocknotify and -walletnotify options."""		"""Test the -alertnotify, -blocknotify and -walletnotify options."""
import os		import os

from test_framework.address import ADDRESS_ECREG_UNSPENDABLE		from test_framework.address import ADDRESS_ECREG_UNSPENDABLE, keyhash_to_p2pkh
from test_framework.test_framework import BitcoinTestFramework		from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (		from test_framework.util import (
assert_equal,		assert_equal,
connect_nodes,		connect_nodes,
wait_until		wait_until,
		hex_str_to_bytes,
)		)

FORK_WARNING_MESSAGE = "Warning: Large-work fork detected, forking after block {}"		FORK_WARNING_MESSAGE = "Warning: Large-work fork detected, forking after block {}"

# Linux allow all characters other than \x00		# Linux allow all characters other than \x00
# Windows disallow control characters (0-31) and /\?%:\|"<>		# Windows disallow control characters (0-31) and /\?%:\|"<>
FILE_CHAR_START = 32 if os.name == 'nt' else 1		FILE_CHAR_START = 32 if os.name == 'nt' else 1
FILE_CHAR_END = 128		FILE_CHAR_END = 128
▲ Show 20 Lines • Show All 86 Lines • ▼ Show 20 Lines	def run_test(self):

# directory content should equal the generated transaction hashes		# directory content should equal the generated transaction hashes
txids_rpc = list(map(lambda t: notify_outputname(		txids_rpc = list(map(lambda t: notify_outputname(
self.wallet, t['txid']), self.nodes[1].listtransactions("*", block_count)))		self.wallet, t['txid']), self.nodes[1].listtransactions("*", block_count)))
assert_equal(		assert_equal(
sorted(txids_rpc), sorted(		sorted(txids_rpc), sorted(
os.listdir(		os.listdir(
self.walletnotify_dir)))		self.walletnotify_dir)))
		for tx_file in os.listdir(self.walletnotify_dir):
		os.remove(os.path.join(self.walletnotify_dir, tx_file))

		# Transactions tests. Give node 0 same wallet seed as
		# node 1, generate spends from node 0, and check notifications
		# triggered by node 1
		self.log.info("test -walletnotify with conflicting transactions")
		self.nodes[0].sethdseed(
		seed=self.nodes[1].dumpprivkey(
		keyhash_to_p2pkh(
		hex_str_to_bytes(
		self.nodes[1].getwalletinfo()['hdseedid'])[::-1])))
		self.nodes[0].rescanblockchain()
		self.nodes[0].generatetoaddress(100, ADDRESS_ECREG_UNSPENDABLE)

		# Generate transaction on node 0, sync mempools, and check for
		# notification on node 1.
		tx1 = self.nodes[0].sendtoaddress(
		address=ADDRESS_ECREG_UNSPENDABLE, amount=100)
		assert_equal(tx1 in self.nodes[0].getrawmempool(), True)
		self.sync_mempools()
		self.expect_wallet_notify([tx1])

		# Add tx1 transaction to new block, checking for a notification
		# and the correct number of confirmations.
		self.nodes[0].generatetoaddress(1, ADDRESS_ECREG_UNSPENDABLE)
		self.sync_blocks()
		self.expect_wallet_notify([tx1])
		assert_equal(self.nodes[1].gettransaction(tx1)["confirmations"], 1)

# Create an invalid chain and ensure the node warns.		# Create an invalid chain and ensure the node warns.
self.log.info("test -alertnotify for forked chain")		self.log.info("test -alertnotify for forked chain")
fork_block = self.nodes[0].getbestblockhash()		fork_block = self.nodes[0].getbestblockhash()
self.nodes[0].generatetoaddress(1, ADDRESS_ECREG_UNSPENDABLE)		self.nodes[0].generatetoaddress(1, ADDRESS_ECREG_UNSPENDABLE)
invalid_block = self.nodes[0].getbestblockhash()		invalid_block = self.nodes[0].getbestblockhash()
self.nodes[0].generatetoaddress(7, ADDRESS_ECREG_UNSPENDABLE)		self.nodes[0].generatetoaddress(7, ADDRESS_ECREG_UNSPENDABLE)

# Invalidate a large branch, which should trigger an alert.		# Invalidate a large branch, which should trigger an alert.
self.nodes[0].invalidateblock(invalid_block)		self.nodes[0].invalidateblock(invalid_block)

# Give bitcoind 10 seconds to write the alert notification		# Give bitcoind 10 seconds to write the alert notification
wait_until(lambda: len(os.listdir(self.alertnotify_dir)), timeout=10)		wait_until(lambda: len(os.listdir(self.alertnotify_dir)), timeout=10)

# The notification command is unable to properly handle the spaces on		# The notification command is unable to properly handle the spaces on
# windows. Skip the content check in this case.		# windows. Skip the content check in this case.
if os.name != 'nt':		if os.name != 'nt':
assert FORK_WARNING_MESSAGE.format(		assert FORK_WARNING_MESSAGE.format(
fork_block) in os.listdir(self.alertnotify_dir)		fork_block) in os.listdir(self.alertnotify_dir)

for notify_file in os.listdir(self.alertnotify_dir):		for notify_file in os.listdir(self.alertnotify_dir):
os.remove(os.path.join(self.alertnotify_dir, notify_file))		os.remove(os.path.join(self.alertnotify_dir, notify_file))

		def expect_wallet_notify(self, tx_ids):
		wait_until(
		lambda: len(os.listdir(self.walletnotify_dir)) >= len(tx_ids),
		timeout=10)
		assert_equal(
		sorted(notify_outputname(self.wallet, tx_id) for tx_id in tx_ids),
		sorted(os.listdir(self.walletnotify_dir)))
		for tx_file in os.listdir(self.walletnotify_dir):
		os.remove(os.path.join(self.walletnotify_dir, tx_file))


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