-
-
Notifications
You must be signed in to change notification settings - Fork 354
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
16 changed files
with
356 additions
and
107 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,104 @@ | ||
#!/usr/bin/env python3 | ||
# Copyright (c) 2014-2016 The Bitcoin Core developers | ||
# Copyright (c) 2020 The Zcoin Core 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 | ||
import time | ||
|
||
from test_framework.test_framework import BitcoinTestFramework | ||
from test_framework.util import assert_equal, connect_nodes_bi, start_node, start_nodes, stop_node | ||
|
||
def wait_until(predicate, timeout): | ||
while timeout > 0: | ||
if predicate(): | ||
return | ||
time.sleep(0.1) | ||
timeout -= 0.1 | ||
|
||
raise Exception("Wait Until Timeout") | ||
|
||
class NotificationsTest(BitcoinTestFramework): | ||
def set_test_params(self): | ||
self.num_nodes = 2 | ||
self.setup_clean_chain = True | ||
|
||
def setup_network(self): | ||
self.alert_filename = os.path.join(self.options.tmpdir, "alert.txt") | ||
self.block_filename = os.path.join(self.options.tmpdir, "blocks.txt") | ||
self.tx_filename = os.path.join(self.options.tmpdir, "transactions.txt") | ||
|
||
# -alertnotify and -blocknotify on node0, walletnotify on node1 | ||
self.extra_args = [[ "-blockversion=2", | ||
"-alertnotify=echo %%s >> %s" % self.alert_filename, | ||
"-blocknotify=echo %%s >> %s" % self.block_filename], | ||
[ "-blockversion=211", | ||
"-rescan", | ||
"-walletnotify=echo %%s >> %s" % self.tx_filename]] | ||
|
||
self.nodes = start_nodes(self.num_nodes, self.options.tmpdir, extra_args = self.extra_args) | ||
connect_nodes_bi(self.nodes, 0, 1) | ||
|
||
self.is_network_split = False | ||
self.sync_all() | ||
|
||
def run_test(self): | ||
self.log.info("test -blocknotify") | ||
block_count = 10 | ||
blocks = self.nodes[1].generate(block_count) | ||
|
||
# wait at most 10 seconds for expected file size before reading the content | ||
wait_until(lambda: os.path.isfile(self.block_filename) and os.stat(self.block_filename).st_size >= (block_count * 65), timeout=10) | ||
|
||
# file content should equal the generated blocks hashes | ||
with open(self.block_filename, 'r') as f: | ||
assert_equal(sorted(blocks), sorted(f.read().splitlines())) | ||
|
||
self.log.info("test -walletnotify") | ||
# wait at most 10 seconds for expected file size before reading the content | ||
wait_until(lambda: os.path.isfile(self.tx_filename) and os.stat(self.tx_filename).st_size >= (block_count * 65), timeout=10) | ||
|
||
# file content should equal the generated transaction hashes | ||
txids_rpc = list(map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count))) | ||
with open(self.tx_filename, 'r') as f: | ||
assert_equal(sorted(txids_rpc), sorted(f.read().splitlines())) | ||
os.remove(self.tx_filename) | ||
|
||
self.log.info("test -walletnotify after rescan") | ||
# restart node to rescan to force wallet notifications | ||
stop_node(self.nodes[1], 1) | ||
self.nodes[1] = start_node(1, self.options.tmpdir, self.extra_args[1]) | ||
|
||
connect_nodes_bi(self.nodes, 0, 1) | ||
|
||
wait_until(lambda: os.path.isfile(self.tx_filename) and os.stat(self.tx_filename).st_size >= (block_count * 65), timeout=10) | ||
|
||
# file content should equal the generated transaction hashes | ||
txids_rpc = list(map(lambda t: t['txid'], self.nodes[1].listtransactions("*", block_count))) | ||
with open(self.tx_filename, 'r') as f: | ||
assert_equal(sorted(txids_rpc), sorted(f.read().splitlines())) | ||
|
||
# Mine another 41 up-version blocks. -alertnotify should trigger on the 51st. | ||
self.log.info("test -alertnotify") | ||
self.nodes[1].generate(41) | ||
self.sync_all() | ||
|
||
# Give bitcoind 10 seconds to write the alert notification | ||
wait_until(lambda: os.path.isfile(self.alert_filename) and os.path.getsize(self.alert_filename), timeout=10) | ||
|
||
with open(self.alert_filename, 'r', encoding='utf8') as f: | ||
alert_text = f.read() | ||
|
||
# Mine more up-version blocks, should not get more alerts: | ||
self.nodes[1].generate(2) | ||
self.sync_all() | ||
|
||
with open(self.alert_filename, 'r', encoding='utf8') as f: | ||
alert_text2 = f.read() | ||
|
||
self.log.info("-alertnotify should not continue notifying for more unknown version blocks") | ||
assert_equal(alert_text, alert_text2) | ||
|
||
if __name__ == '__main__': | ||
NotificationsTest().main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,96 @@ | ||
#!/usr/bin/env python3 | ||
# Copyright (c) 2016 The Bitcoin Core developers | ||
# Copyright (c) 2020 The Zcoin Core developers | ||
# Distributed under the MIT software license, see the accompanying | ||
# file COPYING or http://www.opensource.org/licenses/mit-license.php. | ||
"""Test Wallet encryption""" | ||
|
||
import re | ||
import time | ||
|
||
from test_framework.test_framework import BitcoinTestFramework | ||
from test_framework.util import ( | ||
assert_equal, | ||
assert_raises_jsonrpc, | ||
bitcoind_processes, | ||
start_nodes, | ||
) | ||
|
||
class WalletEncryptionTest(BitcoinTestFramework): | ||
|
||
def __init__(self): | ||
super().__init__() | ||
self.setup_clean_chain = True # lgtm [py/overwritten-inherited-attribute] | ||
self.num_nodes = 1 # lgtm [py/overwritten-inherited-attribute] | ||
|
||
def get_otp(self, address): | ||
# assume test on only 1 node | ||
try: | ||
self.nodes[0].dumpprivkey(address) | ||
except Exception as ex: | ||
found = re.search( | ||
'WARNING! Your one time authorization code is: (.+?)\n', | ||
ex.error['message']) | ||
if found: | ||
return found.group(1) | ||
|
||
raise Exception("Fail to get OTP") | ||
|
||
def run_test(self): | ||
passphrase = "WalletPassphrase" | ||
passphrase2 = "SecondWalletPassphrase" | ||
|
||
# Make sure the wallet isn't encrypted first | ||
address = self.nodes[0].getnewaddress() | ||
privkey = self.nodes[0].dumpprivkey(address, self.get_otp(address)) | ||
|
||
assert_equal(privkey[:1], "c") | ||
assert_equal(len(privkey), 52) | ||
assert_raises_jsonrpc(-15, "Error: running with an unencrypted wallet, but walletpassphrase was called", self.nodes[0].walletpassphrase, 'ff', 1) | ||
assert_raises_jsonrpc(-15, "Error: running with an unencrypted wallet, but walletpassphrasechange was called.", self.nodes[0].walletpassphrasechange, 'ff', 'ff') | ||
|
||
# Encrypt the wallet | ||
assert_raises_jsonrpc(-1, "encryptwallet <passphrase>\nEncrypts the wallet with <passphrase>.", self.nodes[0].encryptwallet, '') | ||
self.nodes[0].encryptwallet(passphrase) | ||
bitcoind_processes[0].wait() | ||
self.nodes = start_nodes(self.num_nodes, self.options.tmpdir) | ||
|
||
# Test that the wallet is encrypted | ||
otp = self.get_otp(address) | ||
assert_raises_jsonrpc(-13, "Please enter the wallet passphrase with walletpassphrase first", self.nodes[0].dumpprivkey, address, otp) | ||
assert_raises_jsonrpc(-15, "Error: running with an encrypted wallet, but encryptwallet was called.", self.nodes[0].encryptwallet, 'ff') | ||
assert_raises_jsonrpc(-1, "walletpassphrase <passphrase> <timeout>\nStores the wallet decryption key in memory for <timeout> seconds.", | ||
self.nodes[0].walletpassphrase, '', 1) | ||
assert_raises_jsonrpc(-1, "walletpassphrasechange <oldpassphrase> <newpassphrase>\nChanges the wallet passphrase from <oldpassphrase> to <newpassphrase>.", | ||
self.nodes[0].walletpassphrasechange, '', 'ff') | ||
|
||
# Check that walletpassphrase works | ||
self.nodes[0].walletpassphrase(passphrase, 2) | ||
otp = self.get_otp(address) | ||
assert_equal(privkey, self.nodes[0].dumpprivkey(address, otp)) | ||
|
||
# Check that the timeout is right | ||
time.sleep(3) | ||
otp = self.get_otp(address) | ||
assert_raises_jsonrpc(-13, "Please enter the wallet passphrase with walletpassphrase first", self.nodes[0].dumpprivkey, address, otp) | ||
|
||
# Test wrong passphrase | ||
assert_raises_jsonrpc(-14, "wallet passphrase entered was incorrect", self.nodes[0].walletpassphrase, passphrase + "wrong", 10) | ||
|
||
# Test walletlock | ||
self.nodes[0].walletpassphrase(passphrase, 84600) | ||
otp = self.get_otp(address) | ||
assert_equal(privkey, self.nodes[0].dumpprivkey(address, otp)) | ||
self.nodes[0].walletlock() | ||
otp = self.get_otp(address) | ||
assert_raises_jsonrpc(-13, "Please enter the wallet passphrase with walletpassphrase first", self.nodes[0].dumpprivkey, address, otp) | ||
|
||
# Test passphrase changes | ||
self.nodes[0].walletpassphrasechange(passphrase, passphrase2) | ||
assert_raises_jsonrpc(-14, "wallet passphrase entered was incorrect", self.nodes[0].walletpassphrase, passphrase, 10) | ||
self.nodes[0].walletpassphrase(passphrase2, 10) | ||
otp = self.get_otp(address) | ||
assert_equal(privkey, self.nodes[0].dumpprivkey(address, otp)) | ||
|
||
if __name__ == '__main__': | ||
WalletEncryptionTest().main() |
Oops, something went wrong.