-
Notifications
You must be signed in to change notification settings - Fork 791
/
feature_notifications.py
executable file
·127 lines (106 loc) · 4.96 KB
/
feature_notifications.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
#!/usr/bin/env python3
# Copyright (c) 2014-2019 The Bitcoin Core developers
# Copyright (c) 2018 The Bitcoin developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test the -alertnotify, -blocknotify and -walletnotify options."""
import os
from test_framework.address import ADDRESS_BCHREG_UNSPENDABLE
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
assert_equal,
connect_nodes,
wait_until
)
FORK_WARNING_MESSAGE = "Warning: Large-work fork detected, forking after block {}"
class NotificationsTest(BitcoinTestFramework):
def set_test_params(self):
self.num_nodes = 2
self.setup_clean_chain = True
def setup_network(self):
self.alertnotify_dir = os.path.join(self.options.tmpdir, "alertnotify")
self.blocknotify_dir = os.path.join(self.options.tmpdir, "blocknotify")
self.walletnotify_dir = os.path.join(
self.options.tmpdir, "walletnotify")
os.mkdir(self.alertnotify_dir)
os.mkdir(self.blocknotify_dir)
os.mkdir(self.walletnotify_dir)
# -alertnotify and -blocknotify on node0, walletnotify on node1
self.extra_args = [["-alertnotify=echo > {}".format(
os.path.join(self.alertnotify_dir, '%s')),
"-blocknotify=echo > {}".format(os.path.join(self.blocknotify_dir, '%s'))],
["-blockversion=211",
"-rescan",
"-walletnotify=echo > {}".format(os.path.join(self.walletnotify_dir, '%s'))]]
super().setup_network()
def run_test(self):
self.log.info("test -blocknotify")
block_count = 10
blocks = self.nodes[1].generatetoaddress(
block_count,
self.nodes[1].getnewaddress() if self.is_wallet_compiled()
else ADDRESS_BCHREG_UNSPENDABLE
)
# wait at most 10 seconds for expected number of files before reading
# the content
wait_until(
lambda: len(
os.listdir(
self.blocknotify_dir)) == block_count,
timeout=10)
# directory content should equal the generated blocks hashes
assert_equal(sorted(blocks), sorted(os.listdir(self.blocknotify_dir)))
if self.is_wallet_compiled():
self.log.info("test -walletnotify")
# wait at most 10 seconds for expected number of files before reading
# the content
wait_until(
lambda: len(
os.listdir(
self.walletnotify_dir)) == block_count,
timeout=10)
# directory content should equal the generated transaction hashes
txids_rpc = list(
map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count)))
assert_equal(
sorted(txids_rpc), sorted(
os.listdir(
self.walletnotify_dir)))
self.stop_node(1)
for tx_file in os.listdir(self.walletnotify_dir):
os.remove(os.path.join(self.walletnotify_dir, tx_file))
self.log.info("test -walletnotify after rescan")
# restart node to rescan to force wallet notifications
self.start_node(1)
connect_nodes(self.nodes[0], self.nodes[1])
wait_until(
lambda: len(
os.listdir(
self.walletnotify_dir)) == block_count,
timeout=10)
# directory content should equal the generated transaction hashes
txids_rpc = list(
map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count)))
assert_equal(
sorted(txids_rpc), sorted(
os.listdir(
self.walletnotify_dir)))
# Create an invalid chain and ensure the node warns.
self.log.info("test -alertnotify for forked chain")
fork_block = self.nodes[0].getbestblockhash()
self.nodes[0].generatetoaddress(1, ADDRESS_BCHREG_UNSPENDABLE)
invalid_block = self.nodes[0].getbestblockhash()
self.nodes[0].generatetoaddress(7, ADDRESS_BCHREG_UNSPENDABLE)
# Invalidate a large branch, which should trigger an alert.
self.nodes[0].invalidateblock(invalid_block)
# Give bitcoind 10 seconds to write the alert notification
wait_until(lambda: len(os.listdir(self.alertnotify_dir)), timeout=10)
# The notification command is unable to properly handle the spaces on
# windows. Skip the content check in this case.
if os.name != 'nt':
assert FORK_WARNING_MESSAGE.format(
fork_block) 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))
if __name__ == '__main__':
NotificationsTest().main()