aboutsummaryrefslogtreecommitdiff
path: root/test/functional
diff options
context:
space:
mode:
Diffstat (limited to 'test/functional')
-rw-r--r--test/functional/README.md52
-rwxr-xr-xtest/functional/feature_bip9_softforks.py283
-rwxr-xr-xtest/functional/feature_blocksdir.py3
-rwxr-xr-xtest/functional/feature_config_args.py5
-rwxr-xr-xtest/functional/feature_logging.py6
-rwxr-xr-xtest/functional/feature_uacomment.py5
-rwxr-xr-xtest/functional/interface_rest.py442
-rwxr-xr-xtest/functional/p2p_sendheaders.py9
-rw-r--r--test/functional/test_framework/blockstore.py160
-rwxr-xr-xtest/functional/test_framework/comptool.py397
-rwxr-xr-xtest/functional/test_framework/test_framework.py29
-rwxr-xr-xtest/functional/test_framework/test_node.py16
-rwxr-xr-xtest/functional/test_runner.py1
-rwxr-xr-xtest/functional/wallet_multiwallet.py12
14 files changed, 226 insertions, 1194 deletions
diff --git a/test/functional/README.md b/test/functional/README.md
index 662b4b44d5..21050cc2fa 100644
--- a/test/functional/README.md
+++ b/test/functional/README.md
@@ -89,52 +89,6 @@ thread.)
- Can be used to write tests where specific P2P protocol behavior is tested.
Examples tests are `p2p_unrequested_blocks.py`, `p2p_compactblocks.py`.
-#### Comptool
-
-- Comptool is a Testing framework for writing tests that compare the block/tx acceptance
-behavior of a bitcoind against 1 or more other bitcoind instances. It should not be used
-to write static tests with known outcomes, since that type of test is easier to write and
-maintain using the standard BitcoinTestFramework.
-
-- Set the `num_nodes` variable (defined in `ComparisonTestFramework`) to start up
-1 or more nodes. If using 1 node, then `--testbinary` can be used as a command line
-option to change the bitcoind binary used by the test. If using 2 or more nodes,
-then `--refbinary` can be optionally used to change the bitcoind that will be used
-on nodes 2 and up.
-
-- Implement a (generator) function called `get_tests()` which yields `TestInstance`s.
-Each `TestInstance` consists of:
- - A list of `[object, outcome, hash]` entries
- * `object` is a `CBlock`, `CTransaction`, or
- `CBlockHeader`. `CBlock`'s and `CTransaction`'s are tested for
- acceptance. `CBlockHeader`s can be used so that the test runner can deliver
- complete headers-chains when requested from the bitcoind, to allow writing
- tests where blocks can be delivered out of order but still processed by
- headers-first bitcoind's.
- * `outcome` is `True`, `False`, or `None`. If `True`
- or `False`, the tip is compared with the expected tip -- either the
- block passed in, or the hash specified as the optional 3rd entry. If
- `None` is specified, then the test will compare all the bitcoind's
- being tested to see if they all agree on what the best tip is.
- * `hash` is the block hash of the tip to compare against. Optional to
- specify; if left out then the hash of the block passed in will be used as
- the expected tip. This allows for specifying an expected tip while testing
- the handling of either invalid blocks or blocks delivered out of order,
- which complete a longer chain.
- - `sync_every_block`: `True/False`. If `False`, then all blocks
- are inv'ed together, and the test runner waits until the node receives the
- last one, and tests only the last block for tip acceptance using the
- outcome and specified tip. If `True`, then each block is tested in
- sequence and synced (this is slower when processing many blocks).
- - `sync_every_transaction`: `True/False`. Analogous to
- `sync_every_block`, except if the outcome on the last tx is "None",
- then the contents of the entire mempool are compared across all bitcoind
- connections. If `True` or `False`, then only the last tx's
- acceptance is tested against the given outcome.
-
-- For examples of tests written in this framework, see
- `p2p_invalid_block.py` and `feature_block.py`.
-
### test-framework modules
#### [test_framework/authproxy.py](test_framework/authproxy.py)
@@ -149,15 +103,9 @@ Generally useful functions.
#### [test_framework/mininode.py](test_framework/mininode.py)
Basic code to support P2P connectivity to a bitcoind.
-#### [test_framework/comptool.py](test_framework/comptool.py)
-Framework for comparison-tool style, P2P tests.
-
#### [test_framework/script.py](test_framework/script.py)
Utilities for manipulating transaction scripts (originally from python-bitcoinlib)
-#### [test_framework/blockstore.py](test_framework/blockstore.py)
-Implements disk-backed block and tx storage.
-
#### [test_framework/key.py](test_framework/key.py)
Wrapper around OpenSSL EC_Key (originally from python-bitcoinlib)
diff --git a/test/functional/feature_bip9_softforks.py b/test/functional/feature_bip9_softforks.py
deleted file mode 100755
index ac6176e976..0000000000
--- a/test/functional/feature_bip9_softforks.py
+++ /dev/null
@@ -1,283 +0,0 @@
-#!/usr/bin/env python3
-# Copyright (c) 2015-2017 The Bitcoin Core developers
-# Distributed under the MIT software license, see the accompanying
-# file COPYING or http://www.opensource.org/licenses/mit-license.php.
-"""Test BIP 9 soft forks.
-
-Connect to a single node.
-regtest lock-in with 108/144 block signalling
-activation after a further 144 blocks
-mine 2 block and save coinbases for later use
-mine 141 blocks to transition from DEFINED to STARTED
-mine 100 blocks signalling readiness and 44 not in order to fail to change state this period
-mine 108 blocks signalling readiness and 36 blocks not signalling readiness (STARTED->LOCKED_IN)
-mine a further 143 blocks (LOCKED_IN)
-test that enforcement has not triggered (which triggers ACTIVE)
-test that enforcement has triggered
-"""
-from io import BytesIO
-import shutil
-import time
-import itertools
-
-from test_framework.test_framework import ComparisonTestFramework
-from test_framework.util import *
-from test_framework.mininode import CTransaction, network_thread_start
-from test_framework.blocktools import create_coinbase, create_block
-from test_framework.comptool import TestInstance, TestManager
-from test_framework.script import CScript, OP_1NEGATE, OP_CHECKSEQUENCEVERIFY, OP_DROP
-
-class BIP9SoftForksTest(ComparisonTestFramework):
- def set_test_params(self):
- self.num_nodes = 1
- self.extra_args = [['-whitelist=127.0.0.1']]
- self.setup_clean_chain = True
-
- def run_test(self):
- self.test = TestManager(self, self.options.tmpdir)
- self.test.add_all_connections(self.nodes)
- network_thread_start()
- self.test.run()
-
- def create_transaction(self, node, coinbase, to_address, amount):
- from_txid = node.getblock(coinbase)['tx'][0]
- inputs = [{ "txid" : from_txid, "vout" : 0}]
- outputs = { to_address : amount }
- rawtx = node.createrawtransaction(inputs, outputs)
- tx = CTransaction()
- f = BytesIO(hex_str_to_bytes(rawtx))
- tx.deserialize(f)
- tx.nVersion = 2
- return tx
-
- def sign_transaction(self, node, tx):
- signresult = node.signrawtransactionwithwallet(bytes_to_hex_str(tx.serialize()))
- tx = CTransaction()
- f = BytesIO(hex_str_to_bytes(signresult['hex']))
- tx.deserialize(f)
- return tx
-
- def generate_blocks(self, number, version, test_blocks = []):
- for i in range(number):
- block = create_block(self.tip, create_coinbase(self.height), self.last_block_time + 1)
- block.nVersion = version
- block.rehash()
- block.solve()
- test_blocks.append([block, True])
- self.last_block_time += 1
- self.tip = block.sha256
- self.height += 1
- return test_blocks
-
- def get_bip9_status(self, key):
- info = self.nodes[0].getblockchaininfo()
- return info['bip9_softforks'][key]
-
- def test_BIP(self, bipName, activated_version, invalidate, invalidatePostSignature, bitno):
- assert_equal(self.get_bip9_status(bipName)['status'], 'defined')
- assert_equal(self.get_bip9_status(bipName)['since'], 0)
-
- # generate some coins for later
- self.coinbase_blocks = self.nodes[0].generate(2)
- self.height = 3 # height of the next block to build
- self.tip = int("0x" + self.nodes[0].getbestblockhash(), 0)
- self.nodeaddress = self.nodes[0].getnewaddress()
- self.last_block_time = int(time.time())
-
- assert_equal(self.get_bip9_status(bipName)['status'], 'defined')
- assert_equal(self.get_bip9_status(bipName)['since'], 0)
- tmpl = self.nodes[0].getblocktemplate({})
- assert(bipName not in tmpl['rules'])
- assert(bipName not in tmpl['vbavailable'])
- assert_equal(tmpl['vbrequired'], 0)
- assert_equal(tmpl['version'], 0x20000000)
-
- # Test 1
- # Advance from DEFINED to STARTED
- test_blocks = self.generate_blocks(141, 4)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- assert_equal(self.get_bip9_status(bipName)['status'], 'started')
- assert_equal(self.get_bip9_status(bipName)['since'], 144)
- assert_equal(self.get_bip9_status(bipName)['statistics']['elapsed'], 0)
- assert_equal(self.get_bip9_status(bipName)['statistics']['count'], 0)
- tmpl = self.nodes[0].getblocktemplate({})
- assert(bipName not in tmpl['rules'])
- assert_equal(tmpl['vbavailable'][bipName], bitno)
- assert_equal(tmpl['vbrequired'], 0)
- assert(tmpl['version'] & activated_version)
-
- # Test 1-A
- # check stats after max number of "signalling not" blocks such that LOCKED_IN still possible this period
- test_blocks = self.generate_blocks(36, 4, test_blocks) # 0x00000004 (signalling not)
- test_blocks = self.generate_blocks(10, activated_version) # 0x20000001 (signalling ready)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- assert_equal(self.get_bip9_status(bipName)['statistics']['elapsed'], 46)
- assert_equal(self.get_bip9_status(bipName)['statistics']['count'], 10)
- assert_equal(self.get_bip9_status(bipName)['statistics']['possible'], True)
-
- # Test 1-B
- # check stats after one additional "signalling not" block -- LOCKED_IN no longer possible this period
- test_blocks = self.generate_blocks(1, 4, test_blocks) # 0x00000004 (signalling not)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- assert_equal(self.get_bip9_status(bipName)['statistics']['elapsed'], 47)
- assert_equal(self.get_bip9_status(bipName)['statistics']['count'], 10)
- assert_equal(self.get_bip9_status(bipName)['statistics']['possible'], False)
-
- # Test 1-C
- # finish period with "ready" blocks, but soft fork will still fail to advance to LOCKED_IN
- test_blocks = self.generate_blocks(97, activated_version) # 0x20000001 (signalling ready)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- assert_equal(self.get_bip9_status(bipName)['statistics']['elapsed'], 0)
- assert_equal(self.get_bip9_status(bipName)['statistics']['count'], 0)
- assert_equal(self.get_bip9_status(bipName)['statistics']['possible'], True)
- assert_equal(self.get_bip9_status(bipName)['status'], 'started')
-
- # Test 2
- # Fail to achieve LOCKED_IN 100 out of 144 signal bit 1
- # using a variety of bits to simulate multiple parallel softforks
- test_blocks = self.generate_blocks(50, activated_version) # 0x20000001 (signalling ready)
- test_blocks = self.generate_blocks(20, 4, test_blocks) # 0x00000004 (signalling not)
- test_blocks = self.generate_blocks(50, activated_version, test_blocks) # 0x20000101 (signalling ready)
- test_blocks = self.generate_blocks(24, 4, test_blocks) # 0x20010000 (signalling not)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- assert_equal(self.get_bip9_status(bipName)['status'], 'started')
- assert_equal(self.get_bip9_status(bipName)['since'], 144)
- assert_equal(self.get_bip9_status(bipName)['statistics']['elapsed'], 0)
- assert_equal(self.get_bip9_status(bipName)['statistics']['count'], 0)
- tmpl = self.nodes[0].getblocktemplate({})
- assert(bipName not in tmpl['rules'])
- assert_equal(tmpl['vbavailable'][bipName], bitno)
- assert_equal(tmpl['vbrequired'], 0)
- assert(tmpl['version'] & activated_version)
-
- # Test 3
- # 108 out of 144 signal bit 1 to achieve LOCKED_IN
- # using a variety of bits to simulate multiple parallel softforks
- test_blocks = self.generate_blocks(57, activated_version) # 0x20000001 (signalling ready)
- test_blocks = self.generate_blocks(26, 4, test_blocks) # 0x00000004 (signalling not)
- test_blocks = self.generate_blocks(50, activated_version, test_blocks) # 0x20000101 (signalling ready)
- test_blocks = self.generate_blocks(10, 4, test_blocks) # 0x20010000 (signalling not)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- # check counting stats and "possible" flag before last block of this period achieves LOCKED_IN...
- assert_equal(self.get_bip9_status(bipName)['statistics']['elapsed'], 143)
- assert_equal(self.get_bip9_status(bipName)['statistics']['count'], 107)
- assert_equal(self.get_bip9_status(bipName)['statistics']['possible'], True)
- assert_equal(self.get_bip9_status(bipName)['status'], 'started')
-
- # ...continue with Test 3
- test_blocks = self.generate_blocks(1, activated_version) # 0x20000001 (signalling ready)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- assert_equal(self.get_bip9_status(bipName)['status'], 'locked_in')
- assert_equal(self.get_bip9_status(bipName)['since'], 576)
- tmpl = self.nodes[0].getblocktemplate({})
- assert(bipName not in tmpl['rules'])
-
- # Test 4
- # 143 more version 536870913 blocks (waiting period-1)
- test_blocks = self.generate_blocks(143, 4)
- yield TestInstance(test_blocks, sync_every_block=False)
-
- assert_equal(self.get_bip9_status(bipName)['status'], 'locked_in')
- assert_equal(self.get_bip9_status(bipName)['since'], 576)
- tmpl = self.nodes[0].getblocktemplate({})
- assert(bipName not in tmpl['rules'])
-
- # Test 5
- # Check that the new rule is enforced
- spendtx = self.create_transaction(self.nodes[0],
- self.coinbase_blocks[0], self.nodeaddress, 1.0)
- invalidate(spendtx)
- spendtx = self.sign_transaction(self.nodes[0], spendtx)
- spendtx.rehash()
- invalidatePostSignature(spendtx)
- spendtx.rehash()
- block = create_block(self.tip, create_coinbase(self.height), self.last_block_time + 1)
- block.nVersion = activated_version
- block.vtx.append(spendtx)
- block.hashMerkleRoot = block.calc_merkle_root()
- block.rehash()
- block.solve()
-
- self.last_block_time += 1
- self.tip = block.sha256
- self.height += 1
- yield TestInstance([[block, True]])
-
- assert_equal(self.get_bip9_status(bipName)['status'], 'active')
- assert_equal(self.get_bip9_status(bipName)['since'], 720)
- tmpl = self.nodes[0].getblocktemplate({})
- assert(bipName in tmpl['rules'])
- assert(bipName not in tmpl['vbavailable'])
- assert_equal(tmpl['vbrequired'], 0)
- assert(not (tmpl['version'] & (1 << bitno)))
-
- # Test 6
- # Check that the new sequence lock rules are enforced
- spendtx = self.create_transaction(self.nodes[0],
- self.coinbase_blocks[1], self.nodeaddress, 1.0)
- invalidate(spendtx)
- spendtx = self.sign_transaction(self.nodes[0], spendtx)
- spendtx.rehash()
- invalidatePostSignature(spendtx)
- spendtx.rehash()
-
- block = create_block(self.tip, create_coinbase(self.height), self.last_block_time + 1)
- block.nVersion = 5
- block.vtx.append(spendtx)
- block.hashMerkleRoot = block.calc_merkle_root()
- block.rehash()
- block.solve()
- self.last_block_time += 1
- yield TestInstance([[block, False]])
-
- # Restart all
- self.test.clear_all_connections()
- self.stop_nodes()
- self.nodes = []
- shutil.rmtree(get_datadir_path(self.options.tmpdir, 0))
- self.setup_chain()
- self.setup_network()
- self.test.add_all_connections(self.nodes)
- network_thread_start()
- self.test.p2p_connections[0].wait_for_verack()
-
- def get_tests(self):
- for test in itertools.chain(
- self.test_BIP('csv', 0x20000001, self.sequence_lock_invalidate, self.donothing, 0),
- self.test_BIP('csv', 0x20000001, self.mtp_invalidate, self.donothing, 0),
- self.test_BIP('csv', 0x20000001, self.donothing, self.csv_invalidate, 0)
- ):
- yield test
-
- def donothing(self, tx):
- return
-
- def csv_invalidate(self, tx):
- """Modify the signature in vin 0 of the tx to fail CSV
- Prepends -1 CSV DROP in the scriptSig itself.
- """
- tx.vin[0].scriptSig = CScript([OP_1NEGATE, OP_CHECKSEQUENCEVERIFY, OP_DROP] +
- list(CScript(tx.vin[0].scriptSig)))
-
- def sequence_lock_invalidate(self, tx):
- """Modify the nSequence to make it fails once sequence lock rule is
- activated (high timespan).
- """
- tx.vin[0].nSequence = 0x00FFFFFF
- tx.nLockTime = 0
-
- def mtp_invalidate(self, tx):
- """Modify the nLockTime to make it fails once MTP rule is activated."""
- # Disable Sequence lock, Activate nLockTime
- tx.vin[0].nSequence = 0x90FFFFFF
- tx.nLockTime = self.last_block_time
-
-if __name__ == '__main__':
- BIP9SoftForksTest().main()
diff --git a/test/functional/feature_blocksdir.py b/test/functional/feature_blocksdir.py
index a77014a524..56f91651a8 100755
--- a/test/functional/feature_blocksdir.py
+++ b/test/functional/feature_blocksdir.py
@@ -6,7 +6,6 @@
"""
import os
-import re
import shutil
from test_framework.test_framework import BitcoinTestFramework, initialize_datadir
@@ -23,7 +22,7 @@ class BlocksdirTest(BitcoinTestFramework):
initialize_datadir(self.options.tmpdir, 0)
self.log.info("Starting with non exiting blocksdir ...")
blocksdir_path = os.path.join(self.options.tmpdir, 'blocksdir')
- self.nodes[0].assert_start_raises_init_error(["-blocksdir=" + blocksdir_path], re.escape('Error: Specified blocks directory "{}" does not exist.'.format(blocksdir_path)))
+ self.nodes[0].assert_start_raises_init_error(["-blocksdir=" + blocksdir_path], 'Error: Specified blocks directory "{}" does not exist.'.format(blocksdir_path))
os.mkdir(blocksdir_path)
self.log.info("Starting with exiting blocksdir ...")
self.start_node(0, ["-blocksdir=" + blocksdir_path])
diff --git a/test/functional/feature_config_args.py b/test/functional/feature_config_args.py
index 6b1e473aa2..a1d22191af 100755
--- a/test/functional/feature_config_args.py
+++ b/test/functional/feature_config_args.py
@@ -5,7 +5,6 @@
"""Test various command line arguments and configuration file parameters."""
import os
-import re
from test_framework.test_framework import BitcoinTestFramework
@@ -26,13 +25,13 @@ class ConfArgsTest(BitcoinTestFramework):
# Check that using -datadir argument on non-existent directory fails
self.nodes[0].datadir = new_data_dir
- self.nodes[0].assert_start_raises_init_error(['-datadir=' + new_data_dir], 'Error: Specified data directory "' + re.escape(new_data_dir) + '" does not exist.')
+ self.nodes[0].assert_start_raises_init_error(['-datadir=' + new_data_dir], 'Error: Specified data directory "' + new_data_dir + '" does not exist.')
# Check that using non-existent datadir in conf file fails
conf_file = os.path.join(default_data_dir, "bitcoin.conf")
with open(conf_file, 'a', encoding='utf8') as f:
f.write("datadir=" + new_data_dir + "\n")
- self.nodes[0].assert_start_raises_init_error(['-conf=' + conf_file], 'Error reading configuration file: specified data directory "' + re.escape(new_data_dir) + '" does not exist.')
+ self.nodes[0].assert_start_raises_init_error(['-conf=' + conf_file], 'Error reading configuration file: specified data directory "' + new_data_dir + '" does not exist.')
# Create the directory and ensure the config file now works
os.mkdir(new_data_dir)
diff --git a/test/functional/feature_logging.py b/test/functional/feature_logging.py
index a4ebc7cca3..3c7aecf10a 100755
--- a/test/functional/feature_logging.py
+++ b/test/functional/feature_logging.py
@@ -7,6 +7,8 @@
import os
from test_framework.test_framework import BitcoinTestFramework
+from test_framework.test_node import ErrorMatch
+
class LoggingTest(BitcoinTestFramework):
def set_test_params(self):
@@ -31,7 +33,7 @@ class LoggingTest(BitcoinTestFramework):
invalidname = os.path.join("foo", "foo.log")
self.stop_node(0)
exp_stderr = "Error: Could not open debug log file \S+$"
- self.nodes[0].assert_start_raises_init_error(["-debuglogfile=%s" % (invalidname)], exp_stderr)
+ self.nodes[0].assert_start_raises_init_error(["-debuglogfile=%s" % (invalidname)], exp_stderr, match=ErrorMatch.FULL_REGEX)
assert not os.path.isfile(os.path.join(invdir, "foo.log"))
# check that invalid log (relative) works after path exists
@@ -44,7 +46,7 @@ class LoggingTest(BitcoinTestFramework):
self.stop_node(0)
invdir = os.path.join(self.options.tmpdir, "foo")
invalidname = os.path.join(invdir, "foo.log")
- self.nodes[0].assert_start_raises_init_error(["-debuglogfile=%s" % invalidname], exp_stderr)
+ self.nodes[0].assert_start_raises_init_error(["-debuglogfile=%s" % invalidname], exp_stderr, match=ErrorMatch.FULL_REGEX)
assert not os.path.isfile(os.path.join(invdir, "foo.log"))
# check that invalid log (absolute) works after path exists
diff --git a/test/functional/feature_uacomment.py b/test/functional/feature_uacomment.py
index c73bdcfbb8..80bd7ff29f 100755
--- a/test/functional/feature_uacomment.py
+++ b/test/functional/feature_uacomment.py
@@ -7,6 +7,7 @@
import re
from test_framework.test_framework import BitcoinTestFramework
+from test_framework.test_node import ErrorMatch
from test_framework.util import assert_equal
@@ -27,12 +28,12 @@ class UacommentTest(BitcoinTestFramework):
self.log.info("test -uacomment max length")
self.stop_node(0)
expected = "Error: Total length of network version string \([0-9]+\) exceeds maximum length \(256\). Reduce the number or size of uacomments."
- self.nodes[0].assert_start_raises_init_error(["-uacomment=" + 'a' * 256], expected)
+ self.nodes[0].assert_start_raises_init_error(["-uacomment=" + 'a' * 256], expected, match=ErrorMatch.FULL_REGEX)
self.log.info("test -uacomment unsafe characters")
for unsafe_char in ['/', ':', '(', ')']:
expected = "Error: User Agent comment \(" + re.escape(unsafe_char) + "\) contains unsafe characters."
- self.nodes[0].assert_start_raises_init_error(["-uacomment=" + unsafe_char], expected)
+ self.nodes[0].assert_start_raises_init_error(["-uacomment=" + unsafe_char], expected, match=ErrorMatch.FULL_REGEX)
if __name__ == '__main__':
diff --git a/test/functional/interface_rest.py b/test/functional/interface_rest.py
index 6f585f6825..2ee33aa869 100755
--- a/test/functional/interface_rest.py
+++ b/test/functional/interface_rest.py
@@ -4,351 +4,297 @@
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test the REST API."""
-from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import *
-from struct import *
+import binascii
+from decimal import Decimal
+from enum import Enum
from io import BytesIO
-from codecs import encode
+import json
+from struct import pack, unpack
import http.client
import urllib.parse
-def deser_uint256(f):
- r = 0
- for i in range(8):
- t = unpack(b"<I", f.read(4))[0]
- r += t << (i * 32)
- return r
-
-#allows simple http get calls
-def http_get_call(host, port, path, response_object = 0):
- conn = http.client.HTTPConnection(host, port)
- conn.request('GET', path)
-
- if response_object:
- return conn.getresponse()
-
- return conn.getresponse().read().decode('utf-8')
-
-#allows simple http post calls with a request body
-def http_post_call(host, port, path, requestdata = '', response_object = 0):
- conn = http.client.HTTPConnection(host, port)
- conn.request('POST', path, requestdata)
-
- if response_object:
- return conn.getresponse()
-
- return conn.getresponse().read()
+from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import (
+ assert_equal,
+ assert_greater_than,
+ assert_greater_than_or_equal,
+ hex_str_to_bytes,
+)
+
+class ReqType(Enum):
+ JSON = 1
+ BIN = 2
+ HEX = 3
+
+class RetType(Enum):
+ OBJ = 1
+ BYTES = 2
+ JSON = 3
+
+def filter_output_indices_by_value(vouts, value):
+ for vout in vouts:
+ if vout['value'] == value:
+ yield vout['n']
class RESTTest (BitcoinTestFramework):
- FORMAT_SEPARATOR = "."
-
def set_test_params(self):
self.setup_clean_chain = True
- self.num_nodes = 3
- self.extra_args = [["-rest"]] * self.num_nodes
-
- def setup_network(self, split=False):
- super().setup_network()
- connect_nodes_bi(self.nodes, 0, 2)
+ self.num_nodes = 2
+ self.extra_args = [["-rest"], []]
+
+ def test_rest_request(self, uri, http_method='GET', req_type=ReqType.JSON, body='', status=200, ret_type=RetType.JSON):
+ rest_uri = '/rest' + uri
+ if req_type == ReqType.JSON:
+ rest_uri += '.json'
+ elif req_type == ReqType.BIN:
+ rest_uri += '.bin'
+ elif req_type == ReqType.HEX:
+ rest_uri += '.hex'
+
+ conn = http.client.HTTPConnection(self.url.hostname, self.url.port)
+ self.log.debug('%s %s %s', http_method, rest_uri, body)
+ if http_method == 'GET':
+ conn.request('GET', rest_uri)
+ elif http_method == 'POST':
+ conn.request('POST', rest_uri, body)
+ resp = conn.getresponse()
+
+ assert_equal(resp.status, status)
+
+ if ret_type == RetType.OBJ:
+ return resp
+ elif ret_type == RetType.BYTES:
+ return resp.read()
+ elif ret_type == RetType.JSON:
+ return json.loads(resp.read().decode('utf-8'), parse_float=Decimal)
def run_test(self):
- url = urllib.parse.urlparse(self.nodes[0].url)
- self.log.info("Mining blocks...")
+ self.url = urllib.parse.urlparse(self.nodes[0].url)
+ self.log.info("Mine blocks and send Bitcoin to node 1")
+
+ # Random address so node1's balance doesn't increase
+ not_related_address = "2MxqoHEdNQTyYeX1mHcbrrpzgojbosTpCvJ"
self.nodes[0].generate(1)
self.sync_all()
- self.nodes[2].generate(100)
+ self.nodes[1].generatetoaddress(100, not_related_address)
self.sync_all()
assert_equal(self.nodes[0].getbalance(), 50)
txid = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 0.1)
self.sync_all()
- self.nodes[2].generate(1)
+ self.nodes[1].generatetoaddress(1, not_related_address)
self.sync_all()
bb_hash = self.nodes[0].getbestblockhash()
- assert_equal(self.nodes[1].getbalance(), Decimal("0.1")) #balance now should be 0.1 on node 1
+ assert_equal(self.nodes[1].getbalance(), Decimal("0.1"))
- # load the latest 0.1 tx over the REST API
- json_string = http_get_call(url.hostname, url.port, '/rest/tx/'+txid+self.FORMAT_SEPARATOR+"json")
- json_obj = json.loads(json_string)
- vintx = json_obj['vin'][0]['txid'] # get the vin to later check for utxo (should be spent by then)
+ self.log.info("Load the transaction using the /tx URI")
+
+ json_obj = self.test_rest_request("/tx/{}".format(txid))
+ spent = (json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout']) # get the vin to later check for utxo (should be spent by then)
# get n of 0.1 outpoint
- n = 0
- for vout in json_obj['vout']:
- if vout['value'] == 0.1:
- n = vout['n']
+ n, = filter_output_indices_by_value(json_obj['vout'], Decimal('0.1'))
+ spending = (txid, n)
+ self.log.info("Query an unspent TXO using the /getutxos URI")
- #######################################
- # GETUTXOS: query an unspent outpoint #
- #######################################
- json_request = '/'+txid+'-'+str(n)
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
+ json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spending))
- #check chainTip response
+ # Check chainTip response
assert_equal(json_obj['chaintipHash'], bb_hash)
- #make sure there is one utxo
+ # Make sure there is one utxo
assert_equal(len(json_obj['utxos']), 1)
- assert_equal(json_obj['utxos'][0]['value'], 0.1)
+ assert_equal(json_obj['utxos'][0]['value'], Decimal('0.1'))
+ self.log.info("Query a spent TXO using the /getutxos URI")
- #################################################
- # GETUTXOS: now query an already spent outpoint #
- #################################################
- json_request = '/'+vintx+'-0'
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
+ json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spent))
- #check chainTip response
+ # Check chainTip response
assert_equal(json_obj['chaintipHash'], bb_hash)
- #make sure there is no utxo in the response because this oupoint has been spent
+ # Make sure there is no utxo in the response because this outpoint has been spent
assert_equal(len(json_obj['utxos']), 0)
- #check bitmap
+ # Check bitmap
assert_equal(json_obj['bitmap'], "0")
+ self.log.info("Query two TXOs using the /getutxos URI")
+
+ json_obj = self.test_rest_request("/getutxos/{}-{}/{}-{}".format(*(spending + spent)))
- ##################################################
- # GETUTXOS: now check both with the same request #
- ##################################################
- json_request = '/'+txid+'-'+str(n)+'/'+vintx+'-0'
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
assert_equal(len(json_obj['utxos']), 1)
assert_equal(json_obj['bitmap'], "10")
- #test binary response
- bb_hash = self.nodes[0].getbestblockhash()
-
- binaryRequest = b'\x01\x02'
- binaryRequest += hex_str_to_bytes(txid)
- binaryRequest += pack("i", n)
- binaryRequest += hex_str_to_bytes(vintx)
- binaryRequest += pack("i", 0)
+ self.log.info("Query the TXOs using the /getutxos URI with a binary response")
- bin_response = http_post_call(url.hostname, url.port, '/rest/getutxos'+self.FORMAT_SEPARATOR+'bin', binaryRequest)
- output = BytesIO()
- output.write(bin_response)
- output.seek(0)
- chainHeight = unpack("i", output.read(4))[0]
- hashFromBinResponse = hex(deser_uint256(output))[2:].zfill(64)
+ bin_request = b'\x01\x02'
+ for txid, n in [spending, spent]:
+ bin_request += hex_str_to_bytes(txid)
+ bin_request += pack("i", n)
- assert_equal(bb_hash, hashFromBinResponse) #check if getutxo's chaintip during calculation was fine
- assert_equal(chainHeight, 102) #chain height must be 102
+ bin_response = self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body=bin_request, ret_type=RetType.BYTES)
+ output = BytesIO(bin_response)
+ chain_height, = unpack("i", output.read(4))
+ response_hash = binascii.hexlify(output.read(32)[::-1]).decode('ascii')
+ assert_equal(bb_hash, response_hash) # check if getutxo's chaintip during calculation was fine
+ assert_equal(chain_height, 102) # chain height must be 102
- ############################
- # GETUTXOS: mempool checks #
- ############################
+ self.log.info("Test the /getutxos URI with and without /checkmempool")
+ # Create a transaction, check that it's found with /checkmempool, but
+ # not found without. Then confirm the transaction and check that it's
+ # found with or without /checkmempool.
# do a tx and don't sync
txid = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 0.1)
- json_string = http_get_call(url.hostname, url.port, '/rest/tx/'+txid+self.FORMAT_SEPARATOR+"json")
- json_obj = json.loads(json_string)
+ json_obj = self.test_rest_request("/tx/{}".format(txid))
# get the spent output to later check for utxo (should be spent by then)
- spent = '{}-{}'.format(json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout'])
+ spent = (json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout'])
# get n of 0.1 outpoint
- n = 0
- for vout in json_obj['vout']:
- if vout['value'] == 0.1:
- n = vout['n']
- spending = '{}-{}'.format(txid, n)
-
- json_request = '/'+spending
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
- assert_equal(len(json_obj['utxos']), 0) #there should be no outpoint because it has just added to the mempool
-
- json_request = '/checkmempool/'+spending
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
- assert_equal(len(json_obj['utxos']), 1) #there should be an outpoint because it has just added to the mempool
-
- json_request = '/'+spent
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
- assert_equal(len(json_obj['utxos']), 1) #there should be an outpoint because its spending tx is not confirmed
-
- json_request = '/checkmempool/'+spent
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
- assert_equal(len(json_obj['utxos']), 0) #there should be no outpoint because it has just spent (by mempool tx)
+ n, = filter_output_indices_by_value(json_obj['vout'], Decimal('0.1'))
+ spending = (txid, n)
+
+ json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spending))
+ assert_equal(len(json_obj['utxos']), 0)
+
+ json_obj = self.test_rest_request("/getutxos/checkmempool/{}-{}".format(*spending))
+ assert_equal(len(json_obj['utxos']), 1)
+
+ json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spent))
+ assert_equal(len(json_obj['utxos']), 1)
+
+ json_obj = self.test_rest_request("/getutxos/checkmempool/{}-{}".format(*spent))
+ assert_equal(len(json_obj['utxos']), 0)
self.nodes[0].generate(1)
self.sync_all()
- json_request = '/'+spending
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
- assert_equal(len(json_obj['utxos']), 1) #there should be an outpoint because it was mined
-
- json_request = '/checkmempool/'+spending
- json_string = http_get_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
- assert_equal(len(json_obj['utxos']), 1) #there should be an outpoint because it was mined
-
- #do some invalid requests
- json_request = '{"checkmempool'
- response = http_post_call(url.hostname, url.port, '/rest/getutxos'+self.FORMAT_SEPARATOR+'json', json_request, True)
- assert_equal(response.status, 400) #must be a 400 because we send an invalid json request
-
- json_request = '{"checkmempool'
- response = http_post_call(url.hostname, url.port, '/rest/getutxos'+self.FORMAT_SEPARATOR+'bin', json_request, True)
- assert_equal(response.status, 400) #must be a 400 because we send an invalid bin request
-
- response = http_post_call(url.hostname, url.port, '/rest/getutxos/checkmempool'+self.FORMAT_SEPARATOR+'bin', '', True)
- assert_equal(response.status, 400) #must be a 400 because we send an invalid bin request
-
- #test limits
- json_request = '/checkmempool/'
- for x in range(0, 20):
- json_request += txid+'-'+str(n)+'/'
- json_request = json_request.rstrip("/")
- response = http_post_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json', '', True)
- assert_equal(response.status, 400) #must be a 400 because we exceeding the limits
-
- json_request = '/checkmempool/'
- for x in range(0, 15):
- json_request += txid+'-'+str(n)+'/'
- json_request = json_request.rstrip("/")
- response = http_post_call(url.hostname, url.port, '/rest/getutxos'+json_request+self.FORMAT_SEPARATOR+'json', '', True)
- assert_equal(response.status, 200) #must be a 200 because we are within the limits
-
- self.nodes[0].generate(1) #generate block to not affect upcoming tests
+ json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spending))
+ assert_equal(len(json_obj['utxos']), 1)
+
+ json_obj = self.test_rest_request("/getutxos/checkmempool/{}-{}".format(*spending))
+ assert_equal(len(json_obj['utxos']), 1)
+
+ # Do some invalid requests
+ self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.JSON, body='{"checkmempool', status=400, ret_type=RetType.OBJ)
+ self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body='{"checkmempool', status=400, ret_type=RetType.OBJ)
+ self.test_rest_request("/getutxos/checkmempool", http_method='POST', req_type=ReqType.JSON, status=400, ret_type=RetType.OBJ)
+
+ # Test limits
+ long_uri = '/'.join(["{}-{}".format(txid, n) for n in range(20)])
+ self.test_rest_request("/getutxos/checkmempool/{}".format(long_uri), http_method='POST', status=400, ret_type=RetType.OBJ)
+
+ long_uri = '/'.join(['{}-{}'.format(txid, n) for n in range(15)])
+ self.test_rest_request("/getutxos/checkmempool/{}".format(long_uri), http_method='POST', status=200)
+
+ self.nodes[0].generate(1) # generate block to not affect upcoming tests
self.sync_all()
- ################
- # /rest/block/ #
- ################
+ self.log.info("Test the /block and /headers URIs")
+ bb_hash = self.nodes[0].getbestblockhash()
- # check binary format
- response = http_get_call(url.hostname, url.port, '/rest/block/'+bb_hash+self.FORMAT_SEPARATOR+"bin", True)
- assert_equal(response.status, 200)
+ # Check binary format
+ response = self.test_rest_request("/block/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ)
assert_greater_than(int(response.getheader('content-length')), 80)
- response_str = response.read()
+ response_bytes = response.read()
- # compare with block header
- response_header = http_get_call(url.hostname, url.port, '/rest/headers/1/'+bb_hash+self.FORMAT_SEPARATOR+"bin", True)
- assert_equal(response_header.status, 200)
+ # Compare with block header
+ response_header = self.test_rest_request("/headers/1/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ)
assert_equal(int(response_header.getheader('content-length')), 80)
- response_header_str = response_header.read()
- assert_equal(response_str[0:80], response_header_str)
+ response_header_bytes = response_header.read()
+ assert_equal(response_bytes[:80], response_header_bytes)
- # check block hex format
- response_hex = http_get_call(url.hostname, url.port, '/rest/block/'+bb_hash+self.FORMAT_SEPARATOR+"hex", True)
- assert_equal(response_hex.status, 200)
+ # Check block hex format
+ response_hex = self.test_rest_request("/block/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ)
assert_greater_than(int(response_hex.getheader('content-length')), 160)
- response_hex_str = response_hex.read()
- assert_equal(encode(response_str, "hex_codec")[0:160], response_hex_str[0:160])
+ response_hex_bytes = response_hex.read().strip(b'\n')
+ assert_equal(binascii.hexlify(response_bytes), response_hex_bytes)
- # compare with hex block header
- response_header_hex = http_get_call(url.hostname, url.port, '/rest/headers/1/'+bb_hash+self.FORMAT_SEPARATOR+"hex", True)
- assert_equal(response_header_hex.status, 200)
+ # Compare with hex block header
+ response_header_hex = self.test_rest_request("/headers/1/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ)
assert_greater_than(int(response_header_hex.getheader('content-length')), 160)
- response_header_hex_str = response_header_hex.read()
- assert_equal(response_hex_str[0:160], response_header_hex_str[0:160])
- assert_equal(encode(response_header_str, "hex_codec")[0:160], response_header_hex_str[0:160])
+ response_header_hex_bytes = response_header_hex.read(160)
+ assert_equal(binascii.hexlify(response_bytes[:80]), response_header_hex_bytes)
- # check json format
- block_json_string = http_get_call(url.hostname, url.port, '/rest/block/'+bb_hash+self.FORMAT_SEPARATOR+'json')
- block_json_obj = json.loads(block_json_string)
+ # Check json format
+ block_json_obj = self.test_rest_request("/block/{}".format(bb_hash))
assert_equal(block_json_obj['hash'], bb_hash)
- # compare with json block header
- response_header_json = http_get_call(url.hostname, url.port, '/rest/headers/1/'+bb_hash+self.FORMAT_SEPARATOR+"json", True)
- assert_equal(response_header_json.status, 200)
- response_header_json_str = response_header_json.read().decode('utf-8')
- json_obj = json.loads(response_header_json_str, parse_float=Decimal)
- assert_equal(len(json_obj), 1) #ensure that there is one header in the json response
- assert_equal(json_obj[0]['hash'], bb_hash) #request/response hash should be the same
+ # Compare with json block header
+ json_obj = self.test_rest_request("/headers/1/{}".format(bb_hash))
+ assert_equal(len(json_obj), 1) # ensure that there is one header in the json response
+ assert_equal(json_obj[0]['hash'], bb_hash) # request/response hash should be the same
- #compare with normal RPC block response
+ # Compare with normal RPC block response
rpc_block_json = self.nodes[0].getblock(bb_hash)
- assert_equal(json_obj[0]['hash'], rpc_block_json['hash'])
- assert_equal(json_obj[0]['confirmations'], rpc_block_json['confirmations'])
- assert_equal(json_obj[0]['height'], rpc_block_json['height'])
- assert_equal(json_obj[0]['version'], rpc_block_json['version'])
- assert_equal(json_obj[0]['merkleroot'], rpc_block_json['merkleroot'])
- assert_equal(json_obj[0]['time'], rpc_block_json['time'])
- assert_equal(json_obj[0]['nonce'], rpc_block_json['nonce'])
- assert_equal(json_obj[0]['bits'], rpc_block_json['bits'])
- assert_equal(json_obj[0]['difficulty'], rpc_block_json['difficulty'])
- assert_equal(json_obj[0]['chainwork'], rpc_block_json['chainwork'])
- assert_equal(json_obj[0]['previousblockhash'], rpc_block_json['previousblockhash'])
-
- #see if we can get 5 headers in one response
+ for key in ['hash', 'confirmations', 'height', 'version', 'merkleroot', 'time', 'nonce', 'bits', 'difficulty', 'chainwork', 'previousblockhash']:
+ assert_equal(json_obj[0][key], rpc_block_json[key])
+
+ # See if we can get 5 headers in one response
self.nodes[1].generate(5)
self.sync_all()
- response_header_json = http_get_call(url.hostname, url.port, '/rest/headers/5/'+bb_hash+self.FORMAT_SEPARATOR+"json", True)
- assert_equal(response_header_json.status, 200)
- response_header_json_str = response_header_json.read().decode('utf-8')
- json_obj = json.loads(response_header_json_str)
- assert_equal(len(json_obj), 5) #now we should have 5 header objects
+ json_obj = self.test_rest_request("/headers/5/{}".format(bb_hash))
+ assert_equal(len(json_obj), 5) # now we should have 5 header objects
+
+ self.log.info("Test the /tx URI")
- # do tx test
tx_hash = block_json_obj['tx'][0]['txid']
- json_string = http_get_call(url.hostname, url.port, '/rest/tx/'+tx_hash+self.FORMAT_SEPARATOR+"json")
- json_obj = json.loads(json_string)
+ json_obj = self.test_rest_request("/tx/{}".format(tx_hash))
assert_equal(json_obj['txid'], tx_hash)
- # check hex format response
- hex_string = http_get_call(url.hostname, url.port, '/rest/tx/'+tx_hash+self.FORMAT_SEPARATOR+"hex", True)
- assert_equal(hex_string.status, 200)
- assert_greater_than(int(response.getheader('content-length')), 10)
+ # Check hex format response
+ hex_response = self.test_rest_request("/tx/{}".format(tx_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ)
+ assert_greater_than_or_equal(int(hex_response.getheader('content-length')),
+ json_obj['size']*2)
+ self.log.info("Test tx inclusion in the /mempool and /block URIs")
- # check block tx details
- # let's make 3 tx and mine them on node 1
+ # Make 3 tx and mine them on node 1
txs = []
- txs.append(self.nodes[0].sendtoaddress(self.nodes[2].getnewaddress(), 11))
- txs.append(self.nodes[0].sendtoaddress(self.nodes[2].getnewaddress(), 11))
- txs.append(self.nodes[0].sendtoaddress(self.nodes[2].getnewaddress(), 11))
+ txs.append(self.nodes[0].sendtoaddress(not_related_address, 11))
+ txs.append(self.nodes[0].sendtoaddress(not_related_address, 11))
+ txs.append(self.nodes[0].sendtoaddress(not_related_address, 11))
self.sync_all()
- # check that there are exactly 3 transactions in the TX memory pool before generating the block
- json_string = http_get_call(url.hostname, url.port, '/rest/mempool/info'+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
+ # Check that there are exactly 3 transactions in the TX memory pool before generating the block
+ json_obj = self.test_rest_request("/mempool/info")
assert_equal(json_obj['size'], 3)
# the size of the memory pool should be greater than 3x ~100 bytes
assert_greater_than(json_obj['bytes'], 300)
- # check that there are our submitted transactions in the TX memory pool
- json_string = http_get_call(url.hostname, url.port, '/rest/mempool/contents'+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
+ # Check that there are our submitted transactions in the TX memory pool
+ json_obj = self.test_rest_request("/mempool/contents")
for i, tx in enumerate(txs):
- assert_equal(tx in json_obj, True)
- assert_equal(json_obj[tx]['spentby'], txs[i+1:i+2])
- assert_equal(json_obj[tx]['depends'], txs[i-1:i])
+ assert tx in json_obj
+ assert_equal(json_obj[tx]['spentby'], txs[i + 1:i + 2])
+ assert_equal(json_obj[tx]['depends'], txs[i - 1:i])
- # now mine the transactions
+ # Now mine the transactions
newblockhash = self.nodes[1].generate(1)
self.sync_all()
- #check if the 3 tx show up in the new block
- json_string = http_get_call(url.hostname, url.port, '/rest/block/'+newblockhash[0]+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
- for tx in json_obj['tx']:
- if not 'coinbase' in tx['vin'][0]: #exclude coinbase
- assert_equal(tx['txid'] in txs, True)
+ # Check if the 3 tx show up in the new block
+ json_obj = self.test_rest_request("/block/{}".format(newblockhash[0]))
+ non_coinbase_txs = {tx['txid'] for tx in json_obj['tx']
+ if 'coinbase' not in tx['vin'][0]}
+ assert_equal(non_coinbase_txs, set(txs))
- #check the same but without tx details
- json_string = http_get_call(url.hostname, url.port, '/rest/block/notxdetails/'+newblockhash[0]+self.FORMAT_SEPARATOR+'json')
- json_obj = json.loads(json_string)
+ # Check the same but without tx details
+ json_obj = self.test_rest_request("/block/notxdetails/{}".format(newblockhash[0]))
for tx in txs:
- assert_equal(tx in json_obj['tx'], True)
+ assert tx in json_obj['tx']
+
+ self.log.info("Test the /chaininfo URI")
- #test rest bestblock
bb_hash = self.nodes[0].getbestblockhash()
- json_string = http_get_call(url.hostname, url.port, '/rest/chaininfo.json')
- json_obj = json.loads(json_string)
+ json_obj = self.test_rest_request("/chaininfo")
assert_equal(json_obj['bestblockhash'], bb_hash)
if __name__ == '__main__':
- RESTTest ().main ()
+ RESTTest().main()
diff --git a/test/functional/p2p_sendheaders.py b/test/functional/p2p_sendheaders.py
index 8869aeaaea..4c7d5e65c5 100755
--- a/test/functional/p2p_sendheaders.py
+++ b/test/functional/p2p_sendheaders.py
@@ -411,21 +411,18 @@ class SendHeadersTest(BitcoinTestFramework):
inv_node.check_last_announcement(inv=[tip], headers=[])
test_node.check_last_announcement(inv=[tip], headers=[])
if i == 0:
- # Just get the data -- shouldn't cause headers announcements to resume
+ self.log.debug("Just get the data -- shouldn't cause headers announcements to resume")
test_node.send_get_data([tip])
test_node.wait_for_block(tip)
elif i == 1:
- # Send a getheaders message that shouldn't trigger headers announcements
- # to resume (best header sent will be too old)
+ self.log.debug("Send a getheaders message that shouldn't trigger headers announcements to resume (best header sent will be too old)")
test_node.send_get_headers(locator=[fork_point], hashstop=new_block_hashes[1])
test_node.send_get_data([tip])
test_node.wait_for_block(tip)
elif i == 2:
test_node.send_get_data([tip])
test_node.wait_for_block(tip)
- # This time, try sending either a getheaders to trigger resumption
- # of headers announcements, or mine a new block and inv it, also
- # triggering resumption of headers announcements.
+ self.log.debug("This time, try sending either a getheaders to trigger resumption of headers announcements, or mine a new block and inv it, also triggering resumption of headers announcements.")
if j == 0:
test_node.send_get_headers(locator=[tip], hashstop=0)
test_node.sync_with_ping()
diff --git a/test/functional/test_framework/blockstore.py b/test/functional/test_framework/blockstore.py
deleted file mode 100644
index 6073285a6c..0000000000
--- a/test/functional/test_framework/blockstore.py
+++ /dev/null
@@ -1,160 +0,0 @@
-#!/usr/bin/env python3
-# Copyright (c) 2015-2017 The Bitcoin Core developers
-# Distributed under the MIT software license, see the accompanying
-# file COPYING or http://www.opensource.org/licenses/mit-license.php.
-"""BlockStore and TxStore helper classes."""
-
-from .mininode import *
-from io import BytesIO
-import dbm.dumb as dbmd
-
-logger = logging.getLogger("TestFramework.blockstore")
-
-class BlockStore():
- """BlockStore helper class.
-
- BlockStore keeps a map of blocks and implements helper functions for
- responding to getheaders and getdata, and for constructing a getheaders
- message.
- """
-
- def __init__(self, datadir):
- self.blockDB = dbmd.open(datadir + "/blocks", 'c')
- self.currentBlock = 0
- self.headers_map = dict()
-
- def close(self):
- self.blockDB.close()
-
- def erase(self, blockhash):
- del self.blockDB[repr(blockhash)]
-
- # lookup an entry and return the item as raw bytes
- def get(self, blockhash):
- value = None
- try:
- value = self.blockDB[repr(blockhash)]
- except KeyError:
- return None
- return value
-
- # lookup an entry and return it as a CBlock
- def get_block(self, blockhash):
- ret = None
- serialized_block = self.get(blockhash)
- if serialized_block is not None:
- f = BytesIO(serialized_block)
- ret = CBlock()
- ret.deserialize(f)
- ret.calc_sha256()
- return ret
-
- def get_header(self, blockhash):
- try:
- return self.headers_map[blockhash]
- except KeyError:
- return None
-
- # Note: this pulls full blocks out of the database just to retrieve
- # the headers -- perhaps we could keep a separate data structure
- # to avoid this overhead.
- def headers_for(self, locator, hash_stop, current_tip=None):
- if current_tip is None:
- current_tip = self.currentBlock
- current_block_header = self.get_header(current_tip)
- if current_block_header is None:
- return None
-
- response = msg_headers()
- headersList = [ current_block_header ]
- maxheaders = 2000
- while (headersList[0].sha256 not in locator.vHave):
- prevBlockHash = headersList[0].hashPrevBlock
- prevBlockHeader = self.get_header(prevBlockHash)
- if prevBlockHeader is not None:
- headersList.insert(0, prevBlockHeader)
- else:
- break
- headersList = headersList[:maxheaders] # truncate if we have too many
- hashList = [x.sha256 for x in headersList]
- index = len(headersList)
- if (hash_stop in hashList):
- index = hashList.index(hash_stop)+1
- response.headers = headersList[:index]
- return response
-
- def add_block(self, block):
- block.calc_sha256()
- try:
- self.blockDB[repr(block.sha256)] = bytes(block.serialize())
- except TypeError:
- logger.exception("Unexpected error")
- self.currentBlock = block.sha256
- self.headers_map[block.sha256] = CBlockHeader(block)
-
- def add_header(self, header):
- self.headers_map[header.sha256] = header
-
- # lookup the hashes in "inv", and return p2p messages for delivering
- # blocks found.
- def get_blocks(self, inv):
- responses = []
- for i in inv:
- if (i.type == 2 or i.type == (2 | (1 << 30))): # MSG_BLOCK or MSG_WITNESS_BLOCK
- data = self.get(i.hash)
- if data is not None:
- # Use msg_generic to avoid re-serialization
- responses.append(msg_generic(b"block", data))
- return responses
-
- def get_locator(self, current_tip=None):
- if current_tip is None:
- current_tip = self.currentBlock
- r = []
- counter = 0
- step = 1
- lastBlock = self.get_block(current_tip)
- while lastBlock is not None:
- r.append(lastBlock.hashPrevBlock)
- for i in range(step):
- lastBlock = self.get_block(lastBlock.hashPrevBlock)
- if lastBlock is None:
- break
- counter += 1
- if counter > 10:
- step *= 2
- locator = CBlockLocator()
- locator.vHave = r
- return locator
-
-class TxStore():
- def __init__(self, datadir):
- self.txDB = dbmd.open(datadir + "/transactions", 'c')
-
- def close(self):
- self.txDB.close()
-
- # lookup an entry and return the item as raw bytes
- def get(self, txhash):
- value = None
- try:
- value = self.txDB[repr(txhash)]
- except KeyError:
- return None
- return value
-
- def add_transaction(self, tx):
- tx.calc_sha256()
- try:
- self.txDB[repr(tx.sha256)] = bytes(tx.serialize())
- except TypeError:
- logger.exception("Unexpected error")
-
- def get_transactions(self, inv):
- responses = []
- for i in inv:
- if (i.type == 1 or i.type == (1 | (1 << 30))): # MSG_TX or MSG_WITNESS_TX
- tx = self.get(i.hash)
- if tx is not None:
- responses.append(msg_generic(b"tx", tx))
- return responses
diff --git a/test/functional/test_framework/comptool.py b/test/functional/test_framework/comptool.py
deleted file mode 100755
index e0ca78e5d1..0000000000
--- a/test/functional/test_framework/comptool.py
+++ /dev/null
@@ -1,397 +0,0 @@
-#!/usr/bin/env python3
-# Copyright (c) 2015-2017 The Bitcoin Core developers
-# Distributed under the MIT software license, see the accompanying
-# file COPYING or http://www.opensource.org/licenses/mit-license.php.
-"""Compare two or more bitcoinds to each other.
-
-To use, create a class that implements get_tests(), and pass it in
-as the test generator to TestManager. get_tests() should be a python
-generator that returns TestInstance objects. See below for definition.
-
-TestP2PConn behaves as follows:
- Configure with a BlockStore and TxStore
- on_inv: log the message but don't request
- on_headers: log the chain tip
- on_pong: update ping response map (for synchronization)
- on_getheaders: provide headers via BlockStore
- on_getdata: provide blocks via BlockStore
-"""
-
-from .mininode import *
-from .blockstore import BlockStore, TxStore
-from .util import p2p_port, wait_until
-
-import logging
-
-logger=logging.getLogger("TestFramework.comptool")
-
-global mininode_lock
-
-class RejectResult():
- """Outcome that expects rejection of a transaction or block."""
- def __init__(self, code, reason=b''):
- self.code = code
- self.reason = reason
- def match(self, other):
- if self.code != other.code:
- return False
- return other.reason.startswith(self.reason)
- def __repr__(self):
- return '%i:%s' % (self.code,self.reason or '*')
-
-class TestP2PConn(P2PInterface):
-
- def __init__(self, block_store, tx_store):
- super().__init__()
- self.bestblockhash = None
- self.block_store = block_store
- self.block_request_map = {}
- self.tx_store = tx_store
- self.tx_request_map = {}
- self.block_reject_map = {}
- self.tx_reject_map = {}
-
- # When the pingmap is non-empty we're waiting for
- # a response
- self.pingMap = {}
- self.lastInv = []
- self.closed = False
-
- def on_close(self):
- self.closed = True
-
- def on_headers(self, message):
- if len(message.headers) > 0:
- best_header = message.headers[-1]
- best_header.calc_sha256()
- self.bestblockhash = best_header.sha256
-
- def on_getheaders(self, message):
- response = self.block_store.headers_for(message.locator, message.hashstop)
- if response is not None:
- self.send_message(response)
-
- def on_getdata(self, message):
- [self.send_message(r) for r in self.block_store.get_blocks(message.inv)]
- [self.send_message(r) for r in self.tx_store.get_transactions(message.inv)]
-
- for i in message.inv:
- if i.type == 1 or i.type == 1 | (1 << 30): # MSG_TX or MSG_WITNESS_TX
- self.tx_request_map[i.hash] = True
- elif i.type == 2 or i.type == 2 | (1 << 30): # MSG_BLOCK or MSG_WITNESS_BLOCK
- self.block_request_map[i.hash] = True
-
- def on_inv(self, message):
- self.lastInv = [x.hash for x in message.inv]
-
- def on_pong(self, message):
- try:
- del self.pingMap[message.nonce]
- except KeyError:
- raise AssertionError("Got pong for unknown ping [%s]" % repr(message))
-
- def on_reject(self, message):
- if message.message == b'tx':
- self.tx_reject_map[message.data] = RejectResult(message.code, message.reason)
- if message.message == b'block':
- self.block_reject_map[message.data] = RejectResult(message.code, message.reason)
-
- def send_inv(self, obj):
- mtype = 2 if isinstance(obj, CBlock) else 1
- self.send_message(msg_inv([CInv(mtype, obj.sha256)]))
-
- def send_getheaders(self):
- # We ask for headers from their last tip.
- m = msg_getheaders()
- m.locator = self.block_store.get_locator(self.bestblockhash)
- self.send_message(m)
-
- def send_header(self, header):
- m = msg_headers()
- m.headers.append(header)
- self.send_message(m)
-
- # This assumes BIP31
- def send_ping(self, nonce):
- self.pingMap[nonce] = True
- self.send_message(msg_ping(nonce))
-
- def received_ping_response(self, nonce):
- return nonce not in self.pingMap
-
- def send_mempool(self):
- self.lastInv = []
- self.send_message(msg_mempool())
-
-# TestInstance:
-#
-# Instances of these are generated by the test generator, and fed into the
-# comptool.
-#
-# "blocks_and_transactions" should be an array of
-# [obj, True/False/None, hash/None]:
-# - obj is either a CBlock, CBlockHeader, or a CTransaction, and
-# - the second value indicates whether the object should be accepted
-# into the blockchain or mempool (for tests where we expect a certain
-# answer), or "None" if we don't expect a certain answer and are just
-# comparing the behavior of the nodes being tested.
-# - the third value is the hash to test the tip against (if None or omitted,
-# use the hash of the block)
-# - NOTE: if a block header, no test is performed; instead the header is
-# just added to the block_store. This is to facilitate block delivery
-# when communicating with headers-first clients (when withholding an
-# intermediate block).
-# sync_every_block: if True, then each block will be inv'ed, synced, and
-# nodes will be tested based on the outcome for the block. If False,
-# then inv's accumulate until all blocks are processed (or max inv size
-# is reached) and then sent out in one inv message. Then the final block
-# will be synced across all connections, and the outcome of the final
-# block will be tested.
-# sync_every_tx: analogous to behavior for sync_every_block, except if outcome
-# on the final tx is None, then contents of entire mempool are compared
-# across all connections. (If outcome of final tx is specified as true
-# or false, then only the last tx is tested against outcome.)
-
-class TestInstance():
- def __init__(self, objects=None, sync_every_block=True, sync_every_tx=False):
- self.blocks_and_transactions = objects if objects else []
- self.sync_every_block = sync_every_block
- self.sync_every_tx = sync_every_tx
-
-class TestManager():
-
- def __init__(self, testgen, datadir):
- self.test_generator = testgen
- self.p2p_connections= []
- self.block_store = BlockStore(datadir)
- self.tx_store = TxStore(datadir)
- self.ping_counter = 1
-
- def add_all_connections(self, nodes):
- for i in range(len(nodes)):
- # Create a p2p connection to each node
- node = TestP2PConn(self.block_store, self.tx_store)
- node.peer_connect('127.0.0.1', p2p_port(i))
- self.p2p_connections.append(node)
-
- def clear_all_connections(self):
- self.p2p_connections = []
-
- def wait_for_disconnections(self):
- def disconnected():
- return all(node.closed for node in self.p2p_connections)
- wait_until(disconnected, timeout=10, lock=mininode_lock)
-
- def wait_for_verack(self):
- return all(node.wait_for_verack() for node in self.p2p_connections)
-
- def wait_for_pings(self, counter):
- def received_pongs():
- return all(node.received_ping_response(counter) for node in self.p2p_connections)
- wait_until(received_pongs, lock=mininode_lock)
-
- # sync_blocks: Wait for all connections to request the blockhash given
- # then send get_headers to find out the tip of each node, and synchronize
- # the response by using a ping (and waiting for pong with same nonce).
- def sync_blocks(self, blockhash, num_blocks):
- def blocks_requested():
- return all(
- blockhash in node.block_request_map and node.block_request_map[blockhash]
- for node in self.p2p_connections
- )
-
- # --> error if not requested
- wait_until(blocks_requested, attempts=20*num_blocks, lock=mininode_lock)
-
- # Send getheaders message
- [ c.send_getheaders() for c in self.p2p_connections ]
-
- # Send ping and wait for response -- synchronization hack
- [ c.send_ping(self.ping_counter) for c in self.p2p_connections ]
- self.wait_for_pings(self.ping_counter)
- self.ping_counter += 1
-
- # Analogous to sync_block (see above)
- def sync_transaction(self, txhash, num_events):
- # Wait for nodes to request transaction (50ms sleep * 20 tries * num_events)
- def transaction_requested():
- return all(
- txhash in node.tx_request_map and node.tx_request_map[txhash]
- for node in self.p2p_connections
- )
-
- # --> error if not requested
- wait_until(transaction_requested, attempts=20*num_events, lock=mininode_lock)
-
- # Get the mempool
- [ c.send_mempool() for c in self.p2p_connections ]
-
- # Send ping and wait for response -- synchronization hack
- [ c.send_ping(self.ping_counter) for c in self.p2p_connections ]
- self.wait_for_pings(self.ping_counter)
- self.ping_counter += 1
-
- # Sort inv responses from each node
- with mininode_lock:
- [ c.lastInv.sort() for c in self.p2p_connections ]
-
- # Verify that the tip of each connection all agree with each other, and
- # with the expected outcome (if given)
- def check_results(self, blockhash, outcome):
- with mininode_lock:
- for c in self.p2p_connections:
- if outcome is None:
- if c.bestblockhash != self.p2p_connections[0].bestblockhash:
- return False
- elif isinstance(outcome, RejectResult): # Check that block was rejected w/ code
- if c.bestblockhash == blockhash:
- return False
- if blockhash not in c.block_reject_map:
- logger.error('Block not in reject map: %064x' % (blockhash))
- return False
- if not outcome.match(c.block_reject_map[blockhash]):
- logger.error('Block rejected with %s instead of expected %s: %064x' % (c.block_reject_map[blockhash], outcome, blockhash))
- return False
- elif ((c.bestblockhash == blockhash) != outcome):
- return False
- return True
-
- # Either check that the mempools all agree with each other, or that
- # txhash's presence in the mempool matches the outcome specified.
- # This is somewhat of a strange comparison, in that we're either comparing
- # a particular tx to an outcome, or the entire mempools altogether;
- # perhaps it would be useful to add the ability to check explicitly that
- # a particular tx's existence in the mempool is the same across all nodes.
- def check_mempool(self, txhash, outcome):
- with mininode_lock:
- for c in self.p2p_connections:
- if outcome is None:
- # Make sure the mempools agree with each other
- if c.lastInv != self.p2p_connections[0].lastInv:
- return False
- elif isinstance(outcome, RejectResult): # Check that tx was rejected w/ code
- if txhash in c.lastInv:
- return False
- if txhash not in c.tx_reject_map:
- logger.error('Tx not in reject map: %064x' % (txhash))
- return False
- if not outcome.match(c.tx_reject_map[txhash]):
- logger.error('Tx rejected with %s instead of expected %s: %064x' % (c.tx_reject_map[txhash], outcome, txhash))
- return False
- elif ((txhash in c.lastInv) != outcome):
- return False
- return True
-
- def run(self):
- # Wait until verack is received
- self.wait_for_verack()
-
- test_number = 0
- tests = self.test_generator.get_tests()
- for test_instance in tests:
- test_number += 1
- logger.info("Running test %d: %s line %s" % (test_number, tests.gi_code.co_filename, tests.gi_frame.f_lineno))
- # We use these variables to keep track of the last block
- # and last transaction in the tests, which are used
- # if we're not syncing on every block or every tx.
- [ block, block_outcome, tip ] = [ None, None, None ]
- [ tx, tx_outcome ] = [ None, None ]
- invqueue = []
-
- for test_obj in test_instance.blocks_and_transactions:
- b_or_t = test_obj[0]
- outcome = test_obj[1]
- # Determine if we're dealing with a block or tx
- if isinstance(b_or_t, CBlock): # Block test runner
- block = b_or_t
- block_outcome = outcome
- tip = block.sha256
- # each test_obj can have an optional third argument
- # to specify the tip we should compare with
- # (default is to use the block being tested)
- if len(test_obj) >= 3:
- tip = test_obj[2]
-
- # Add to shared block_store, set as current block
- # If there was an open getdata request for the block
- # previously, and we didn't have an entry in the
- # block_store, then immediately deliver, because the
- # node wouldn't send another getdata request while
- # the earlier one is outstanding.
- first_block_with_hash = True
- if self.block_store.get(block.sha256) is not None:
- first_block_with_hash = False
- with mininode_lock:
- self.block_store.add_block(block)
- for c in self.p2p_connections:
- if first_block_with_hash and block.sha256 in c.block_request_map and c.block_request_map[block.sha256] == True:
- # There was a previous request for this block hash
- # Most likely, we delivered a header for this block
- # but never had the block to respond to the getdata
- c.send_message(msg_block(block))
- else:
- c.block_request_map[block.sha256] = False
- # Either send inv's to each node and sync, or add
- # to invqueue for later inv'ing.
- if (test_instance.sync_every_block):
- # if we expect success, send inv and sync every block
- # if we expect failure, just push the block and see what happens.
- if outcome == True:
- [ c.send_inv(block) for c in self.p2p_connections ]
- self.sync_blocks(block.sha256, 1)
- else:
- [ c.send_message(msg_block(block)) for c in self.p2p_connections ]
- [ c.send_ping(self.ping_counter) for c in self.p2p_connections ]
- self.wait_for_pings(self.ping_counter)
- self.ping_counter += 1
- if (not self.check_results(tip, outcome)):
- raise AssertionError("Test failed at test %d" % test_number)
- else:
- invqueue.append(CInv(2, block.sha256))
- elif isinstance(b_or_t, CBlockHeader):
- block_header = b_or_t
- self.block_store.add_header(block_header)
- [ c.send_header(block_header) for c in self.p2p_connections ]
-
- else: # Tx test runner
- assert(isinstance(b_or_t, CTransaction))
- tx = b_or_t
- tx_outcome = outcome
- # Add to shared tx store and clear map entry
- with mininode_lock:
- self.tx_store.add_transaction(tx)
- for c in self.p2p_connections:
- c.tx_request_map[tx.sha256] = False
- # Again, either inv to all nodes or save for later
- if (test_instance.sync_every_tx):
- [ c.send_inv(tx) for c in self.p2p_connections ]
- self.sync_transaction(tx.sha256, 1)
- if (not self.check_mempool(tx.sha256, outcome)):
- raise AssertionError("Test failed at test %d" % test_number)
- else:
- invqueue.append(CInv(1, tx.sha256))
- # Ensure we're not overflowing the inv queue
- if len(invqueue) == MAX_INV_SZ:
- [ c.send_message(msg_inv(invqueue)) for c in self.p2p_connections ]
- invqueue = []
-
- # Do final sync if we weren't syncing on every block or every tx.
- if (not test_instance.sync_every_block and block is not None):
- if len(invqueue) > 0:
- [ c.send_message(msg_inv(invqueue)) for c in self.p2p_connections ]
- invqueue = []
- self.sync_blocks(block.sha256, len(test_instance.blocks_and_transactions))
- if (not self.check_results(tip, block_outcome)):
- raise AssertionError("Block test failed at test %d" % test_number)
- if (not test_instance.sync_every_tx and tx is not None):
- if len(invqueue) > 0:
- [ c.send_message(msg_inv(invqueue)) for c in self.p2p_connections ]
- invqueue = []
- self.sync_transaction(tx.sha256, len(test_instance.blocks_and_transactions))
- if (not self.check_mempool(tx.sha256, tx_outcome)):
- raise AssertionError("Mempool test failed at test %d" % test_number)
-
- [ c.disconnect_node() for c in self.p2p_connections ]
- self.wait_for_disconnections()
- self.block_store.close()
- self.tx_store.close()
diff --git a/test/functional/test_framework/test_framework.py b/test/functional/test_framework/test_framework.py
index 1a71796e91..1dbd262063 100755
--- a/test/functional/test_framework/test_framework.py
+++ b/test/functional/test_framework/test_framework.py
@@ -434,35 +434,6 @@ class BitcoinTestFramework():
for i in range(self.num_nodes):
initialize_datadir(self.options.tmpdir, i)
-class ComparisonTestFramework(BitcoinTestFramework):
- """Test framework for doing p2p comparison testing
-
- Sets up some bitcoind binaries:
- - 1 binary: test binary
- - 2 binaries: 1 test binary, 1 ref binary
- - n>2 binaries: 1 test binary, n-1 ref binaries"""
-
- def set_test_params(self):
- self.num_nodes = 2
- self.setup_clean_chain = True
-
- def add_options(self, parser):
- parser.add_option("--testbinary", dest="testbinary",
- default=os.getenv("BITCOIND", "bitcoind"),
- help="bitcoind binary to test")
- parser.add_option("--refbinary", dest="refbinary",
- default=os.getenv("BITCOIND", "bitcoind"),
- help="bitcoind binary to use for reference nodes (if any)")
-
- def setup_network(self):
- extra_args = [['-whitelist=127.0.0.1']] * self.num_nodes
- if hasattr(self, "extra_args"):
- extra_args = self.extra_args
- self.add_nodes(self.num_nodes, extra_args,
- binary=[self.options.testbinary] +
- [self.options.refbinary] * (self.num_nodes - 1))
- self.start_nodes()
-
class SkipTest(Exception):
"""This exception is raised to skip a test"""
def __init__(self, message):
diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py
index b3c2fc0a09..966b0463c2 100755
--- a/test/functional/test_framework/test_node.py
+++ b/test/functional/test_framework/test_node.py
@@ -6,6 +6,7 @@
import decimal
import errno
+from enum import Enum
import http.client
import json
import logging
@@ -35,6 +36,12 @@ class FailedToStartError(Exception):
"""Raised when a node fails to start correctly."""
+class ErrorMatch(Enum):
+ FULL_TEXT = 1
+ FULL_REGEX = 2
+ PARTIAL_REGEX = 3
+
+
class TestNode():
"""A class for representing a bitcoind node under test.
@@ -183,7 +190,7 @@ class TestNode():
def wait_until_stopped(self, timeout=BITCOIND_PROC_WAIT_TIMEOUT):
wait_until(self.is_node_stopped, timeout=timeout)
- def assert_start_raises_init_error(self, extra_args=None, expected_msg=None, partial_match=False, *args, **kwargs):
+ def assert_start_raises_init_error(self, extra_args=None, expected_msg=None, match=ErrorMatch.FULL_TEXT, *args, **kwargs):
"""Attempt to start the node and expect it to raise an error.
extra_args: extra arguments to pass through to bitcoind
@@ -205,12 +212,15 @@ class TestNode():
if expected_msg is not None:
log_stderr.seek(0)
stderr = log_stderr.read().decode('utf-8').strip()
- if partial_match:
+ if match == ErrorMatch.PARTIAL_REGEX:
if re.search(expected_msg, stderr, flags=re.MULTILINE) is None:
raise AssertionError('Expected message "{}" does not partially match stderr:\n"{}"'.format(expected_msg, stderr))
- else:
+ elif match == ErrorMatch.FULL_REGEX:
if re.fullmatch(expected_msg, stderr) is None:
raise AssertionError('Expected message "{}" does not fully match stderr:\n"{}"'.format(expected_msg, stderr))
+ elif match == ErrorMatch.FULL_TEXT:
+ if expected_msg != stderr:
+ raise AssertionError('Expected message "{}" does not fully match stderr:\n"{}"'.format(expected_msg, stderr))
else:
if expected_msg is None:
assert_msg = "bitcoind should have exited with an error"
diff --git a/test/functional/test_runner.py b/test/functional/test_runner.py
index 3cae4b1df3..518c16b5f1 100755
--- a/test/functional/test_runner.py
+++ b/test/functional/test_runner.py
@@ -159,7 +159,6 @@ EXTENDED_SCRIPTS = [
'mining_getblocktemplate_longpoll.py',
'p2p_timeouts.py',
# vv Tests less than 60s vv
- 'feature_bip9_softforks.py',
'p2p_feefilter.py',
'rpc_bind.py',
# vv Tests less than 30s vv
diff --git a/test/functional/wallet_multiwallet.py b/test/functional/wallet_multiwallet.py
index 0285263ef9..5ff313997e 100755
--- a/test/functional/wallet_multiwallet.py
+++ b/test/functional/wallet_multiwallet.py
@@ -7,10 +7,10 @@
Verify that a bitcoind node can load multiple wallet files
"""
import os
-import re
import shutil
from test_framework.test_framework import BitcoinTestFramework
+from test_framework.test_node import ErrorMatch
from test_framework.util import (
assert_equal,
assert_raises_rpc_error,
@@ -66,7 +66,7 @@ class MultiWalletTest(BitcoinTestFramework):
# should not initialize if wallet path can't be created
exp_stderr = "boost::filesystem::create_directory: (The system cannot find the path specified|Not a directory):"
- self.nodes[0].assert_start_raises_init_error(['-wallet=wallet.dat/bad'], exp_stderr, partial_match=True)
+ self.nodes[0].assert_start_raises_init_error(['-wallet=wallet.dat/bad'], exp_stderr, match=ErrorMatch.PARTIAL_REGEX)
self.nodes[0].assert_start_raises_init_error(['-walletdir=wallets'], 'Error: Specified -walletdir "wallets" does not exist')
self.nodes[0].assert_start_raises_init_error(['-walletdir=wallets'], 'Error: Specified -walletdir "wallets" is a relative path', cwd=data_dir())
@@ -78,18 +78,18 @@ class MultiWalletTest(BitcoinTestFramework):
# should not initialize if one wallet is a copy of another
shutil.copyfile(wallet_dir('w8'), wallet_dir('w8_copy'))
exp_stderr = "CDB: Can't open database w8_copy \(duplicates fileid \w+ from w8\)"
- self.nodes[0].assert_start_raises_init_error(['-wallet=w8', '-wallet=w8_copy'], exp_stderr, partial_match=True)
+ self.nodes[0].assert_start_raises_init_error(['-wallet=w8', '-wallet=w8_copy'], exp_stderr, match=ErrorMatch.PARTIAL_REGEX)
# should not initialize if wallet file is a symlink
os.symlink('w8', wallet_dir('w8_symlink'))
- self.nodes[0].assert_start_raises_init_error(['-wallet=w8_symlink'], 'Error: Invalid -wallet path \'w8_symlink\'\. .*')
+ self.nodes[0].assert_start_raises_init_error(['-wallet=w8_symlink'], 'Error: Invalid -wallet path \'w8_symlink\'\. .*', match=ErrorMatch.FULL_REGEX)
# should not initialize if the specified walletdir does not exist
self.nodes[0].assert_start_raises_init_error(['-walletdir=bad'], 'Error: Specified -walletdir "bad" does not exist')
# should not initialize if the specified walletdir is not a directory
not_a_dir = wallet_dir('notadir')
open(not_a_dir, 'a').close()
- self.nodes[0].assert_start_raises_init_error(['-walletdir=' + not_a_dir], 'Error: Specified -walletdir "' + re.escape(not_a_dir) + '" is not a directory')
+ self.nodes[0].assert_start_raises_init_error(['-walletdir=' + not_a_dir], 'Error: Specified -walletdir "' + not_a_dir + '" is not a directory')
# if wallets/ doesn't exist, datadir should be the default wallet dir
wallet_dir2 = data_dir('walletdir')
@@ -111,7 +111,7 @@ class MultiWalletTest(BitcoinTestFramework):
os.mkdir(competing_wallet_dir)
self.restart_node(0, ['-walletdir=' + competing_wallet_dir])
exp_stderr = "Error: Error initializing wallet database environment \"\S+competing_walletdir\"!"
- self.nodes[1].assert_start_raises_init_error(['-walletdir=' + competing_wallet_dir], exp_stderr, partial_match=True)
+ self.nodes[1].assert_start_raises_init_error(['-walletdir=' + competing_wallet_dir], exp_stderr, match=ErrorMatch.PARTIAL_REGEX)
self.restart_node(0, extra_args)
generated by cgit v1.2.3 (git 2.26.2) at 2024-11-14 16:24:15 +0000