- Avoid stop-starting the nodes multiple times during the test if possible. A

stop-start takes several seconds, so doing it several times blows up the

runtime of the test.

- When calling RPCs with lots of arguments, consider using named keyword

arguments instead of positional arguments to make the intent of the call

clear to readers.

__metaclass__ = abc.ABCMeta

# The expected error code given by bitcoind upon submission of the tx.

# Only specified if it differs from mempool acceptance error.

block_reject_reason = ""

"""Override test parameters for your individual test.

This method must be overridden and num_nodes must be explicitly set."""

self.setup_clean_chain = True

self.num_nodes = 3

# Use self.extra_args to change command-line arguments for the nodes

class AsmapTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def test_without_asmap_arg(self):

block 200. node2 will reject block 102 since it's assumed valid, but it

isn't buried by at least two weeks' work.

"""

from test_framework.key import ECKey

from test_framework.messages import (

CBlockHeader,

assert not p2p_conn.is_connected

break

def run_test(self):

p2p0 = self.nodes[0].add_p2p_connection(BaseNode())

# Send blocks to node0. Block 102 will be rejected.

self.send_blocks_until_disconnected(p2p0)

# Send all blocks to node1. All blocks will be accepted.

for i in range(2202):

# Send blocks to node2. Block 102 will be rejected.

self.send_blocks_until_disconnected(p2p2)

if __name__ == '__main__':

hdkeypath = v17_info["hdkeypath"]

pubkey = v17_info["pubkey"]

if __name__ == '__main__':

BackwardsCompatibilityTest().main()

"""Test various command line arguments and configuration file parameters."""

import os

from test_framework.test_framework import BitcoinTestFramework

from test_framework import util

self.start_node(0, extra_args=['-nonetworkactive=1'])

self.stop_node(0)

def run_test(self):

self.stop_node(0)

self.test_log_buffer()

self.test_args_log()

self.test_networkactive()

self.test_config_file_parser()

class ChainstateWriteCrashTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 4

self.rpc_timeout = 480

self.supports_cli = False

# file COPYING or http://www.opensource.org/licenses/mit-license.php.

"""Check that it's not possible to start a second bitcoind instance using the same datadir or wallet."""

import os

from test_framework.test_framework import BitcoinTestFramework

from test_framework.test_node import ErrorMatch

self.nodes[1].assert_start_raises_init_error(extra_args=['-datadir={}'.format(self.nodes[0].datadir), '-noserver'], expected_msg=expected_msg)

if self.is_wallet_compiled():

if __name__ == '__main__':

FilelockTest().main()

class IncludeConfTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def setup_chain(self):

"""Test the -alertnotify, -blocknotify and -walletnotify options."""

import os

from test_framework.test_framework import BitcoinTestFramework

from test_framework.util import (

assert_equal,

)

# Linux allow all characters other than \x00

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_BCRT1_UNSPENDABLE)

for tx_file in os.listdir(self.walletnotify_dir):

os.remove(os.path.join(self.walletnotify_dir, tx_file))

# triggered by node 1

self.log.info("test -walletnotify with conflicting transactions")

self.nodes[0].rescanblockchain()

self.nodes[0].generatetoaddress(100, ADDRESS_BCRT1_UNSPENDABLE)

self.sync_blocks()

RANGE_BEGIN = PORT_MIN + 2 * PORT_RANGE # Start after p2p and rpc ports

NET_IPV4 = "ipv4"

NET_IPV6 = "ipv6"

NET_ONION = "onion"

def random_checksig_style(pubkey):

"""Creates a random CHECKSIG* tapscript that would succeed with only the valid signature on witness stack."""

opcode = random.choice([OP_CHECKSIG, OP_CHECKSIGVERIFY, OP_CHECKSIGADD])

if (opcode == OP_CHECKSIGVERIFY):

ret = CScript([pubkey, opcode, OP_1])

def set_test_params(self):

self.setup_clean_chain = True

self.num_nodes = 1

def skip_test_if_missing_module(self):

self.skip_if_no_cli()

class ZMQSubscriber:

def __init__(self, socket, topic):

self.socket = socket

self.topic = topic

self.socket.setsockopt(zmq.SUBSCRIBE, self.topic)

topic, body, seq = self.socket.recv_multipart()

# Topic should match the subscriber topic.

assert_equal(topic, self.topic)

# Sequence should be incremental.

self.sequence += 1

return body

def receive_sequence(self):

hash = body[:32].hex()

label = chr(body[32])

mempool_sequence = None if len(body) != 32+1+8 else struct.unpack("<Q", body[32+1:])[0]

class ZMQTest (BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 2

def skip_test_if_missing_module(self):

self.skip_if_no_py3_zmq()

# Restart node with the specified zmq notifications enabled, subscribe to

# all of them and return the corresponding ZMQSubscriber objects.

subscribers = []

for topic, address in services:

socket = self.ctx.socket(zmq.SUB)

subscribers.append(ZMQSubscriber(socket, topic.encode()))

for i, sub in enumerate(subscribers):

sub.socket.connect(services[i][1])

return subscribers

self.restart_node(0, ["-zmqpubrawtx=foo", "-zmqpubhashtx=bar"])

address = 'tcp://127.0.0.1:28332'

hashblock = subs[0]

hashtx = subs[1]

hashblock, hashtx = self.setup_zmq_test(

[(topic, address) for topic in ["hashblock", "hashtx"]],

recv_timeout=2) # 2 second timeout to check end of notifications

# Generate 1 block in nodes[0] with 1 mempool tx and receive all notifications

payment_txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1.0)

"""

self.log.info("Testing 'sequence' publisher")

[seq] = self.setup_zmq_test([("sequence", "tcp://127.0.0.1:28333")])

# Mempool sequence number starts at 1

seq_num = 1

return

self.log.info("Testing 'mempool sync' usage of sequence notifier")

# In-memory counter, should always start at 1

next_mempool_seq = self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"]

def test_multiple_interfaces(self):

# Set up two subscribers with different addresses

subscribers = self.setup_zmq_test([

("hashblock", "tcp://127.0.0.1:28334"),

("hashblock", "tcp://127.0.0.1:28335"),

# Generate 1 block in nodes[0] and receive all notifications

self.nodes[0].generatetoaddress(1, ADDRESS_BCRT1_UNSPENDABLE)

assert_equal(len(self.nodes[0].getrawmempool()), 5)

assert_equal(len(self.nodes[1].getrawmempool()), 5)

self.log.debug("Prioritize a transaction on node0")

fees = self.nodes[0].getmempoolentry(txid=last_txid)['fees']

assert_equal(fees['base'], fees['modified'])

fees = self.nodes[0].getmempoolentry(txid=last_txid)['fees']

assert_equal(fees['base'] + Decimal('0.00001000'), fees['modified'])

tx_creation_time = self.nodes[0].getmempoolentry(txid=last_txid)['time']

assert_greater_than_or_equal(tx_creation_time, tx_creation_time_lower)

assert_greater_than_or_equal(tx_creation_time_higher, tx_creation_time)

class P2PAddConnections(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 2

def setup_network(self):

class AddrTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def run_test(self):

class P2PBlocksOnly(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

self.extra_args = [["-blocksonly"]]

class FilterTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

self.extra_args = [[

'-peerbloomfilters',

class AddrTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def run_test(self):

class InvalidLocatorTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def run_test(self):

node = self.nodes[0] # convenience reference to the node

no_version_node.send_message(msg_ping())

expected_timeout_logs = [

]

with self.nodes[0].assert_debug_log(expected_msgs=expected_timeout_logs):

class TxDownloadTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 2

def test_tx_requests(self):

res5 = node.gettxoutsetinfo(hash_type='none')

assert 'hash_serialized_2' not in res5

def _test_getblockheader(self):

node = self.nodes[0]

class EstimateFeeTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def run_test(self):

from test_framework.test_framework import BitcoinTestFramework

from test_framework.util import assert_equal, assert_raises_rpc_error

import os

class HelpRpcTest(BitcoinTestFramework):

self.supports_cli = False

def run_test(self):

self.test_categories()

self.dump_help()

if self.is_wallet_compiled():

self.wallet_help()

def test_categories(self):

node = self.nodes[0]

assert_raises_rpc_error,

p2p_port,

)

def assert_net_servicesnames(servicesflag, servicenames):

self.supports_cli = False

def run_test(self):

# Get out of IBD for the minfeefilter and getpeerinfo tests.

self.nodes[0].generate(101)

def test_getpeerinfo(self):

self.log.info("Test getpeerinfo")

# Create a few getpeerinfo last_block/last_transaction values.

self.nodes[1].generate(1)

self.sync_all()

time_now = int(time.time())

assert_equal(peer_info[1][0]['connection_type'], 'manual')

assert_equal(peer_info[1][1]['connection_type'], 'inbound')

def test_getnettotals(self):

self.log.info("Test getnettotals")

# Test getnettotals and getpeerinfo by doing a ping. The bytes

for info in network_info:

assert_net_servicesnames(int(info["localservices"], 0x10), info["localservicesnames"])

def test_getaddednodeinfo(self):

self.log.info("Test getaddednodeinfo")

assert_equal(self.nodes[0].getaddednodeinfo(), [])

class PSBTTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 3

self.extra_args = [

["-walletrbf=1"],

import time

from test_framework.test_framework import BitcoinTestFramework

class UptimeTest(BitcoinTestFramework):

self.setup_clean_chain = True

def run_test(self):

self._test_uptime()

def _test_uptime(self):

wait_time = 10

self.nodes[0].setmocktime(int(time.time() + wait_time))

| `num_nodes` | `1` | Sets the number of initialized bitcoind processes. |

| `perf` | False | Profiles running nodes with `perf` for the duration of the test if set to `True`. |

| `rpc_timeout` | `60` | Sets the RPC server timeout for the underlying bitcoind processes. |

| `randomseed` | Random Integer | `TestShell.options.randomseed` is a member of `TestShell` which can be accessed during a test to seed a random generator. User can override default with a constant value for reproducible test runs. |

| `supports_cli` | `False` | Whether the bitcoin-cli utility is compiled and available for the test. |

| `tmpdir` | `"/var/folders/.../"` | Sets directory for test logs. Will be deleted upon a successful test run unless `nocleanup` is set to `True` |

page_info['pgno'] = pgno

page_info['prev_pgno'] = prev_pgno

page_info['next_pgno'] = next_pgno

page_info['hf_offset'] = hf_offset

page_info['level'] = level

page_info['pg_type'] = pg_type

ss += data

return hashlib.sha256(ss).digest()

def jacobi_symbol(n, k):

"""Compute the Jacobi symbol of n modulo k

if pubkey is not None:

keys[privkey] = pubkey

for msg in byte_arrays: # test every combination of message, signing key, verification key

sig = sign_schnorr(sign_privkey, msg)

for verify_privkey, verify_pubkey in keys.items():

if verify_privkey == sign_privkey:

# for cases where a user needs tighter control over what is sent over the wire

# note that the user must supply the name of the msgtype, and the data

class msg_generic:

def __init__(self, msgtype, data=None):

self.msgtype = msgtype

LOCKTIME_THRESHOLD = 500000000

ANNEX_TAG = 0x50

LEAF_VERSION_TAPSCRIPT = 0xc0

def hash160(s):

# Serialize to bytes

return encoded_v.to_bytes(n_bytes, 'little')

class CScriptOp(int):

"""A single script opcode"""

__slots__ = ()

_opcode_instances.append(super().__new__(cls, n))

return _opcode_instances[n]

# Populate opcode instance table

for n in range(0xff + 1):

CScriptOp(n)

import tempfile

import time

from .authproxy import JSONRPCException

from . import coverage

from .p2p import NetworkThread

This class also contains various public and private helper methods."""

def __init__(self):

"""Sets test framework defaults. Do not override this method. Instead, override the set_test_params() method"""

self.network_thread = None

self.rpc_timeout = 60 # Wait for up to 60 seconds for the RPC server to respond

self.supports_cli = True

# skipped. If list is truncated, wallet creation is skipped and keys

# are not imported.

self.wallet_names = None

self.set_test_params()

assert self.wallet_names is None or len(self.wallet_names) <= self.num_nodes

if self.options.timeout_factor == 0 :

parser.add_argument('--timeout-factor', dest="timeout_factor", type=float, default=1.0, help='adjust test timeouts by a factor. Setting it to 0 disables all timeouts')

group = parser.add_mutually_exclusive_group()

help="Run test using a descriptor wallet", dest='descriptors')

help="Run test using legacy wallets", dest='descriptors')

self.add_options(parser)

self.options = parser.parse_args()

self.options.previous_releases_path = previous_releases_path

def setup(self):

"""Call this method to start up the test framework object with options set."""

self.options.cachedir = os.path.abspath(self.options.cachedir)

fname_bitcoind = os.path.join(

config["environment"]["BUILDDIR"],

"src",

extra_args = self.extra_args

self.add_nodes(self.num_nodes, extra_args)

self.start_nodes()

self.import_deterministic_coinbase_privkeys()

if not self.setup_clean_chain:

for n in self.nodes:

def skip_if_no_wallet(self):

"""Skip the running test if wallet has not been compiled."""

if not self.is_wallet_compiled():

raise SkipTest("wallet has not been compiled.")

if self.options.descriptors:

self.skip_if_no_sqlite()

def skip_if_no_sqlite(self):

"""Skip the running test if sqlite has not been compiled."""

BASE_SCRIPTS = [

# Scripts that are run by default.

# Longest test should go first, to favor running tests in parallel

'wallet_hd.py --descriptors',

'wallet_backup.py --descriptors',

# vv Tests less than 5m vv

'mining_getblocktemplate_longpoll.py',

'feature_maxuploadtarget.py',

'feature_block.py',

'rpc_fundrawtransaction.py --descriptors',

'p2p_compactblocks.py',

'feature_segwit.py --legacy-wallet',

# vv Tests less than 2m vv

'wallet_basic.py --descriptors',

'wallet_labels.py --descriptors',

'p2p_segwit.py',

'p2p_timeouts.py',

'p2p_tx_download.py',

'mempool_updatefromblock.py',

'wallet_dump.py --legacy-wallet',

'wallet_listtransactions.py --descriptors',

'feature_taproot.py',

# vv Tests less than 60s vv

'wallet_importmulti.py --legacy-wallet',

'mempool_limit.py',

'rpc_txoutproof.py',

'wallet_listreceivedby.py --descriptors',

'wallet_abandonconflict.py --descriptors',

'feature_csv_activation.py',

'rpc_rawtransaction.py --descriptors',

'wallet_address_types.py --descriptors',

'feature_bip68_sequence.py',

'p2p_feefilter.py',

'feature_reindex.py',

'feature_abortnode.py',

# vv Tests less than 30s vv

'wallet_keypool_topup.py --descriptors',

'feature_fee_estimation.py',

'interface_zmq.py',

'interface_bitcoin_cli.py',

'mempool_resurrect.py',

'wallet_txn_doublespend.py --mineblock',

'tool_wallet.py --descriptors',

'wallet_txn_clone.py',

'wallet_txn_clone.py --segwit',

'rpc_misc.py',

'interface_rest.py',

'mempool_spend_coinbase.py',

'wallet_avoidreuse.py --descriptors',

'mempool_reorg.py',

'mempool_persist.py',

'wallet_multiwallet.py --descriptors',

'wallet_multiwallet.py --usecli',

'wallet_createwallet.py --usecli',

'wallet_createwallet.py --descriptors',

'wallet_watchonly.py --legacy-wallet',

'wallet_reorgsrestore.py',

'interface_http.py',

'interface_rpc.py',

'rpc_psbt.py --descriptors',

'rpc_users.py',

'rpc_whitelist.py',

'feature_proxy.py',

'rpc_signrawtransaction.py --descriptors',

'p2p_addrv2_relay.py',

'wallet_groups.py --descriptors',

'p2p_disconnect_ban.py',

'rpc_decodescript.py',

'rpc_blockchain.py',

'rpc_deprecated.py',

'wallet_disable.py --descriptors',

'p2p_addr_relay.py',

'p2p_getaddr_caching.py',

'p2p_getdata.py',

'rpc_net.py',

'wallet_keypool.py --descriptors',

'wallet_descriptor.py --descriptors',

'p2p_nobloomfilter_messages.py',

'p2p_invalid_tx.py',

'feature_assumevalid.py',

'example_test.py',

'wallet_txn_doublespend.py --descriptors',

'feature_backwards_compatibility.py --descriptors',

'wallet_txn_clone.py --mineblock',

'feature_notifications.py',

'rpc_getblockfilter.py',

'rpc_invalidateblock.py',

'feature_rbf.py',

'mempool_packages.py',

'mempool_package_onemore.py',

'rpc_createmultisig.py --descriptors',

'feature_versionbits_warning.py',

'rpc_preciousblock.py',

'wallet_importprunedfunds.py --descriptors',

'p2p_leak_tx.py',

'p2p_eviction.py',

'rpc_signmessage.py',

'rpc_generateblock.py',

'rpc_generate.py',

'wallet_balance.py --descriptors',

'feature_nulldummy.py --descriptors',

'mempool_accept.py',

'mempool_expiry.py',

'wallet_import_rescan.py --legacy-wallet',

'wallet_import_with_label.py --legacy-wallet',

'wallet_importdescriptors.py --descriptors',

'rpc_bind.py --ipv4',

'rpc_bind.py --ipv6',

'rpc_bind.py --nonloopback',

'mining_basic.py',

'feature_signet.py',

'wallet_bumpfee.py --descriptors',

'wallet_implicitsegwit.py --legacy-wallet',

'rpc_named_arguments.py',

'wallet_listsinceblock.py --descriptors',

'p2p_leak.py',

'wallet_encryption.py --descriptors',

'feature_dersig.py',

'feature_cltv.py',

'rpc_uptime.py',

'wallet_resendwallettransactions.py --descriptors',

'wallet_fallbackfee.py --descriptors',

'rpc_dumptxoutset.py',

'feature_minchainwork.py',

'rpc_estimatefee.py',

'rpc_getblockstats.py',

'wallet_create_tx.py --descriptors',

'p2p_fingerprint.py',

'feature_uacomment.py',

'wallet_coinbase_category.py --descriptors',

'feature_filelock.py',

'feature_loadblock.py',

'p2p_add_connections.py',

'p2p_unrequested_blocks.py',

'p2p_blockfilters.py',

'feature_includeconf.py',

'feature_asmap.py',

'mempool_unbroadcast.py',

def test_invalid_tool_commands_and_args(self):

self.log.info('Testing that various invalid commands raise with specific error messages')

# `bitcoin-wallet help` raises an error. Use `bitcoin-wallet -help`.

self.assert_raises_tool_error('Error parsing command line arguments: Invalid parameter -foo', '-foo')

locked_dir = os.path.join(self.options.tmpdir, "node0", "regtest", "wallets")

error = 'Error initializing wallet database environment "{}"!'.format(locked_dir)

if self.options.descriptors:

self.log.info('Checking createfromdump')

self.do_tool_createfromdump("load", "wallet.dump")

if self.is_sqlite_compiled():

self.do_tool_createfromdump("load-sqlite", "wallet.dump", "sqlite")

class AvoidReuseTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 2

# This test isn't testing txn relay/timing, so set whitelist on the

# peers for instant txn relay. This speeds up the test run time 2-3x.

assert_equal(total_txs, len(self.nodes[0].listtransactions("*", 99999)))

# Test getaddressinfo on external address. Note that these addresses are taken from disablewallet.py

address_info = self.nodes[0].getaddressinfo("mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ")

assert_equal(address_info['address'], "mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ")

assert_equal(address_info["scriptPubKey"], "76a9144e3854046c7bd1594ac904e4793b6a45b36dea0988ac")

class CreateWalletTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def skip_test_if_missing_module(self):

self.skip_if_no_sqlite()

def run_test(self):

# Make a descriptor wallet

self.log.info("Making a descriptor wallet")

if self.options.descriptors:

assert_raises_rpc_error(-4, "Wallet file verification failed. SQLiteDatabase: Unable to obtain an exclusive lock on the database, is it being used by another bitcoind?", self.nodes[0].loadwallet, wallet_names[0])

else:

# This tests the default wallet that BDB makes, so SQLite wallet doesn't need to test this

# Fail to load duplicate wallets by different ways (directory and filepath)

path = os.path.join(self.options.tmpdir, "node0", "regtest", "wallets", "wallet.dat")

# Only BDB doesn't open duplicate wallet files. SQLite does not have this limitation. While this may be desired in the future, it is not necessary

# Fail to load if one wallet is a copy of another

from itertools import product

from test_framework.authproxy import JSONRPCException

from test_framework.test_framework import BitcoinTestFramework

from test_framework.util import (

assert_equal,

self.nodes[1].createwallet(wallet_name="w1")

w1 = self.nodes[1].get_wallet_rpc("w1")

# w2 contains the private keys for w3

w2 = self.nodes[1].get_wallet_rpc("w2")

# w3 is a watch-only wallet, based on w2

self.nodes[1].createwallet(wallet_name="w3", disable_private_keys=True)

w3 = self.nodes[1].get_wallet_rpc("w3")

"timestamp": "now",

"keypool": True,

"watchonly": True

},{

"timestamp": "now",

"keypool": True,

"internal": True,

"watchonly": True

}])

assert_equal(res, [{"success": True}, {"success": True}])

for _ in range(3):

a2_receive = w2.getnewaddress()

self.nodes[0].generate(1)

self.sync_blocks()

self.log.info("Send to address...")

self.test_send(from_wallet=w0, to_wallet=w1, amount=1)

self.test_send(from_wallet=w0, to_wallet=w1, amount=1, add_to_wallet=True)

res = w2.walletprocesspsbt(res["psbt"])

assert res["complete"]

self.log.info("Create OP_RETURN...")

self.test_send(from_wallet=w0, to_wallet=w1, amount=1)

def skip_test_if_missing_module(self):

self.skip_if_no_wallet()

self.skip_if_no_previous_releases()

def setup_network(self):

class CreateWalletWatchonlyTest(BitcoinTestFramework):

def set_test_params(self):

self.num_nodes = 1

def skip_test_if_missing_module(self):

import sys

def main():

parser = argparse.ArgumentParser(

formatter_class=argparse.ArgumentDefaultsHelpFormatter,

os.path.join(config["environment"]["BUILDDIR"], 'src', 'test', 'fuzz', 'fuzz'),

'-help=1',

],

timeout=20,

check=True,

stderr=subprocess.PIPE,

if args.generate:

return generate_corpus_seeds(

fuzz_pool=fuzz_pool,

build_dir=config["environment"]["BUILDDIR"],

seed_dir=args.seed_dir,

targets=test_list_selection,

fuzz_pool=fuzz_pool,

corpus=args.seed_dir,

test_list=test_list_selection,

build_dir=config["environment"]["BUILDDIR"],

merge_dir=args.m_dir,

)

fuzz_pool=fuzz_pool,

corpus=args.seed_dir,

test_list=test_list_selection,

build_dir=config["environment"]["BUILDDIR"],

use_valgrind=args.valgrind,

)

"""Generates new corpus seeds.

Run {targets} without input, and outputs the generated corpus seeds to

' '.join(command),

subprocess.run(

command,

check=True,

stderr=subprocess.PIPE,

universal_newlines=True,

future.result()

logging.info("Merge the inputs from the passed dir into the seed_dir. Passed dir {}".format(merge_dir))

jobs = []

for t in test_list:

args = [

os.path.join(build_dir, 'src', 'test', 'fuzz', 'fuzz'),

'-merge=1',

'-use_value_profile=1', # Also done by oss-fuzz https://github.com/google/oss-fuzz/issues/1406#issuecomment-387790487

os.path.join(corpus, t),

os.path.join(merge_dir, t),

output = 'Run {} with args {}\n'.format(t, " ".join(args))

output += subprocess.run(

args,

check=True,

stderr=subprocess.PIPE,

universal_newlines=True,

future.result()

jobs = []

for t in test_list:

corpus_path = os.path.join(corpus, t)

def job(t, args):

output = 'Run {} with args {}'.format(t, args)

output += result.stderr

return output, result

"txmempool -> validation -> txmempool"

"wallet/fees -> wallet/wallet -> wallet/fees"

"wallet/wallet -> wallet/walletdb -> wallet/wallet"

)

EXIT_CODE=0

boost/multi_index/ordered_index.hpp

boost/multi_index/sequenced_index.hpp

boost/multi_index_container.hpp

boost/process.hpp

boost/signals2/connection.hpp

boost/signals2/optional_last_value.hpp

boost/signals2/signal.hpp

boost/test/unit_test.hpp

)

for BOOST_INCLUDE in $(git grep '^#include <boost/' -- "*.cpp" "*.h" | cut -f2 -d: | cut -f2 -d'<' | cut -f1 -d'>' | sort -u); do

race:BerkeleyDatabase

race:DatabaseBatch

race:leveldb::DBImpl::DeleteObsoleteFiles

race:zmq::*

race:bitcoin-qt

# -fsanitize=integer suppressions

# ===============================

# Unsigned integer overflow occurs when the result of an unsigned integer

# contains files in which we expect unsigned integer overflows to occur. The

# list is used to suppress -fsanitize=integer warnings when running our CI UBSan

# job.

unsigned-integer-overflow:arith_uint256.h

unsigned-integer-overflow:basic_string.h

unsigned-integer-overflow:bench/bench.h

unsigned-integer-overflow:chain.cpp

unsigned-integer-overflow:chain.h

unsigned-integer-overflow:coded_stream.h

unsigned-integer-overflow:core_write.cpp

unsigned-integer-overflow:hash.cpp

unsigned-integer-overflow:policy/fees.cpp

unsigned-integer-overflow:prevector.h

unsigned-integer-overflow:script/interpreter.cpp

unsigned-integer-overflow:stl_bvector.h

unsigned-integer-overflow:txmempool.cpp

unsigned-integer-overflow:util/strencodings.cpp

unsigned-integer-overflow:validation.cpp

implicit-integer-sign-change:*/new_allocator.h

implicit-integer-sign-change:arith_uint256.cpp

implicit-integer-sign-change:bech32.cpp

implicit-integer-sign-change:bloom.cpp

implicit-integer-sign-change:coins.h

implicit-integer-sign-change:compat/stdin.cpp

implicit-integer-sign-change:compressor.h

implicit-integer-sign-change:key.cpp

implicit-integer-sign-change:noui.cpp

implicit-integer-sign-change:prevector.h

implicit-integer-sign-change:protocol.cpp

implicit-integer-sign-change:script/bitcoinconsensus.cpp

implicit-integer-sign-change:test/pow_tests.cpp

implicit-integer-sign-change:test/prevector_tests.cpp

implicit-integer-sign-change:test/sighash_tests.cpp

implicit-integer-sign-change:test/streams_tests.cpp

implicit-integer-sign-change:test/transaction_tests.cpp

implicit-integer-sign-change:txmempool.cpp

implicit-integer-sign-change:validation.cpp

implicit-integer-sign-change:zmq/zmqpublishnotifier.cpp

implicit-signed-integer-truncation,implicit-integer-sign-change:chain.h

implicit-signed-integer-truncation,implicit-integer-sign-change:test/skiplist_tests.cpp

implicit-signed-integer-truncation:chain.h

implicit-signed-integer-truncation:cuckoocache.h

implicit-signed-integer-truncation:streams.h

implicit-signed-integer-truncation:test/arith_uint256_tests.cpp

implicit-signed-integer-truncation:test/skiplist_tests.cpp

implicit-signed-integer-truncation:torcontrol.cpp

# std::variant warning fixed in https://github.com/gcc-mirror/gcc/commit/074436cf8cdd2a9ce75cadd36deb8301f00e55b9

implicit-unsigned-integer-truncation:std::__detail::__variant::_Variant_storage