1 #!/usr/bin/env python3
2 # Copyright (c) 2014-2022 The Limenka developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """Helpful routines for regression testing."""
6 7 from base64 import b64encode
8 from decimal import Decimal
9 from subprocess import CalledProcessError
10 import hashlib
11 import inspect
12 import json
13 import logging
14 import os
15 import pathlib
16 import platform
17 import random
18 import re
19 import time
20 21 from . import coverage
22 from .authproxy import AuthServiceProxy, JSONRPCException
23 from collections.abc import Callable
24 from typing import Optional, Union
25 26 SATOSHI_PRECISION = Decimal('0.00000001')
27 28 logger = logging.getLogger("TestFramework.utils")
29 30 # Assert functions
31 ##################
32 33 34 def assert_approx(v, vexp, vspan=0.00001):
35 """Assert that `v` is within `vspan` of `vexp`"""
36 if isinstance(v, Decimal) or isinstance(vexp, Decimal):
37 v=Decimal(v)
38 vexp=Decimal(vexp)
39 vspan=Decimal(vspan)
40 if v < vexp - vspan:
41 raise AssertionError("%s < [%s..%s]" % (str(v), str(vexp - vspan), str(vexp + vspan)))
42 if v > vexp + vspan:
43 raise AssertionError("%s > [%s..%s]" % (str(v), str(vexp - vspan), str(vexp + vspan)))
44 45 46 def assert_fee_amount(fee, tx_size, feerate_BTC_kvB):
47 """Assert the fee is in range."""
48 assert isinstance(tx_size, int)
49 target_fee = get_fee(tx_size, feerate_BTC_kvB)
50 if fee < target_fee:
51 raise AssertionError("Fee of %s BTC too low! (Should be %s BTC)" % (str(fee), str(target_fee)))
52 # allow the wallet's estimation to be at most 2 bytes off
53 high_fee = get_fee(tx_size + 2, feerate_BTC_kvB)
54 if fee > high_fee:
55 raise AssertionError("Fee of %s BTC too high! (Should be %s BTC)" % (str(fee), str(target_fee)))
56 57 58 def summarise_dict_differences(thing1, thing2):
59 if not isinstance(thing1, dict) or not isinstance(thing2, dict):
60 return thing1, thing2
61 d1, d2 = {}, {}
62 for k in sorted(thing1.keys()):
63 if k not in thing2:
64 d1[k] = thing1[k]
65 elif thing1[k] != thing2[k]:
66 d1[k], d2[k] = summarise_dict_differences(thing1[k], thing2[k])
67 for k in sorted(thing2.keys()):
68 if k not in thing1:
69 d2[k] = thing2[k]
70 return d1, d2
71 72 def assert_equal(thing1, thing2, *args):
73 if thing1 != thing2 and not args and isinstance(thing1, dict) and isinstance(thing2, dict):
74 d1,d2 = summarise_dict_differences(thing1, thing2)
75 raise AssertionError("not(%s == %s)\n in particular not(%s == %s)" % (thing1, thing2, d1, d2))
76 if thing1 != thing2 or any(thing1 != arg for arg in args):
77 raise AssertionError("not(%s)" % " == ".join(str(arg) for arg in (thing1, thing2) + args))
78 79 80 def assert_equal_without_usage(actual, expected):
81 """
82 Assert that testmempoolaccept results match expected values, ignoring the 'usage' field.
83 This helper is for tests that were written before the 'usage' field was added.
84 """
85 if isinstance(actual, list) and isinstance(expected, list):
86 assert_equal(len(actual), len(expected))
87 for act, exp in zip(actual, expected):
88 assert_equal_without_usage(act, exp)
89 elif isinstance(actual, dict) and isinstance(expected, dict):
90 # Check that all expected keys match
91 for key in expected:
92 assert key in actual, f"Expected key '{key}' not in actual result"
93 if key != 'usage': # Skip usage comparison
94 assert_equal(actual[key], expected[key])
95 # Verify usage exists and is positive if transaction was validated
96 if 'usage' in actual:
97 assert isinstance(actual['usage'], int), "usage should be an integer"
98 assert actual['usage'] > 0, "usage should be positive"
99 else:
100 assert_equal(actual, expected)
101 102 103 def assert_greater_than(thing1, thing2):
104 if thing1 <= thing2:
105 raise AssertionError("%s <= %s" % (str(thing1), str(thing2)))
106 107 108 def assert_greater_than_or_equal(thing1, thing2):
109 if thing1 < thing2:
110 raise AssertionError("%s < %s" % (str(thing1), str(thing2)))
111 112 113 def assert_raises(exc, fun, *args, **kwds):
114 assert_raises_message(exc, None, fun, *args, **kwds)
115 116 117 def assert_raises_message(exc, message, fun, *args, **kwds):
118 try:
119 fun(*args, **kwds)
120 except JSONRPCException:
121 raise AssertionError("Use assert_raises_rpc_error() to test RPC failures")
122 except exc as e:
123 if message is not None and message not in e.error['message']:
124 raise AssertionError(
125 "Expected substring not found in error message:\nsubstring: '{}'\nerror message: '{}'.".format(
126 message, e.error['message']))
127 except Exception as e:
128 raise AssertionError("Unexpected exception raised: " + type(e).__name__)
129 else:
130 raise AssertionError("No exception raised")
131 132 133 def assert_raises_process_error(returncode: int, output: str, fun: Callable, *args, **kwds):
134 """Execute a process and asserts the process return code and output.
135 136 Calls function `fun` with arguments `args` and `kwds`. Catches a CalledProcessError
137 and verifies that the return code and output are as expected. Throws AssertionError if
138 no CalledProcessError was raised or if the return code and output are not as expected.
139 140 Args:
141 returncode: the process return code.
142 output: [a substring of] the process output.
143 fun: the function to call. This should execute a process.
144 args*: positional arguments for the function.
145 kwds**: named arguments for the function.
146 """
147 try:
148 fun(*args, **kwds)
149 except CalledProcessError as e:
150 if returncode != e.returncode:
151 raise AssertionError("Unexpected returncode %i" % e.returncode)
152 if output not in e.output:
153 raise AssertionError("Expected substring not found:" + e.output)
154 else:
155 raise AssertionError("No exception raised")
156 157 158 def assert_raises_rpc_error(code: Optional[int], message: Optional[str], fun: Callable, *args, **kwds):
159 """Run an RPC and verify that a specific JSONRPC exception code and message is raised.
160 161 Calls function `fun` with arguments `args` and `kwds`. Catches a JSONRPCException
162 and verifies that the error code and message are as expected. Throws AssertionError if
163 no JSONRPCException was raised or if the error code/message are not as expected.
164 165 Args:
166 code: the error code returned by the RPC call (defined in src/rpc/protocol.h).
167 Set to None if checking the error code is not required.
168 message: [a substring of] the error string returned by the RPC call.
169 Set to None if checking the error string is not required.
170 fun: the function to call. This should be the name of an RPC.
171 args*: positional arguments for the function.
172 kwds**: named arguments for the function.
173 """
174 assert try_rpc(code, message, fun, *args, **kwds), "No exception raised"
175 176 177 def try_rpc(code, message, fun, *args, **kwds):
178 """Tries to run an rpc command.
179 180 Test against error code and message if the rpc fails.
181 Returns whether a JSONRPCException was raised."""
182 try:
183 fun(*args, **kwds)
184 except JSONRPCException as e:
185 # JSONRPCException was thrown as expected. Check the code and message values are correct.
186 if (code is not None) and (code != e.error["code"]):
187 raise AssertionError("Unexpected JSONRPC error code %i" % e.error["code"])
188 if (message is not None) and (message not in e.error['message']):
189 raise AssertionError(
190 "Expected substring not found in error message:\nsubstring: '{}'\nerror message: '{}'.".format(
191 message, e.error['message']))
192 return True
193 except Exception as e:
194 raise AssertionError("Unexpected exception raised: " + type(e).__name__)
195 else:
196 return False
197 198 199 def assert_is_hex_string(string):
200 try:
201 int(string, 16)
202 except Exception as e:
203 raise AssertionError("Couldn't interpret %r as hexadecimal; raised: %s" % (string, e))
204 205 206 def assert_is_hash_string(string, length=64):
207 if not isinstance(string, str):
208 raise AssertionError("Expected a string, got type %r" % type(string))
209 elif length and len(string) != length:
210 raise AssertionError("String of length %d expected; got %d" % (length, len(string)))
211 elif not re.match('[abcdef0-9]+$', string):
212 raise AssertionError("String %r contains invalid characters for a hash." % string)
213 214 215 def assert_array_result(object_array, to_match, expected, should_not_find=False):
216 """
217 Pass in array of JSON objects, a dictionary with key/value pairs
218 to match against, and another dictionary with expected key/value
219 pairs.
220 If the should_not_find flag is true, to_match should not be found
221 in object_array
222 """
223 if should_not_find:
224 assert_equal(expected, {})
225 num_matched = 0
226 for item in object_array:
227 all_match = True
228 for key, value in to_match.items():
229 if item[key] != value:
230 all_match = False
231 if not all_match:
232 continue
233 elif should_not_find:
234 num_matched = num_matched + 1
235 for key, value in expected.items():
236 if item[key] != value:
237 raise AssertionError("%s : expected %s=%s" % (str(item), str(key), str(value)))
238 num_matched = num_matched + 1
239 if num_matched == 0 and not should_not_find:
240 raise AssertionError("No objects matched %s" % (str(to_match)))
241 if num_matched > 0 and should_not_find:
242 raise AssertionError("Objects were found %s" % (str(to_match)))
243 244 def assert_scale(number, expected_scale=8):
245 """Assert number has expected scale, e.g. fractional digits; number of
246 digits after the decimal. The default of 8 corresponds to a Limenka amount."""
247 number = str(number)
248 mantissa = number.split('.')[-1].upper()
249 if mantissa[:3] == '0E-':
250 assert_equal(mantissa, '0E-{}'.format(expected_scale)) # zeros in exponent notation
251 elif mantissa == number:
252 assert_equal(0, expected_scale) # no mantissa, ergo, expected scale must be 0
253 else:
254 assert_equal(len(mantissa), expected_scale)
255 256 257 # Utility functions
258 ###################
259 260 261 def check_json_precision():
262 """Make sure json library being used does not lose precision converting BTC values"""
263 n = Decimal("20000000.00000003")
264 satoshis = int(json.loads(json.dumps(float(n))) * 1.0e8)
265 if satoshis != 2000000000000003:
266 raise RuntimeError("JSON encode/decode loses precision")
267 268 269 def count_bytes(hex_string):
270 return len(bytearray.fromhex(hex_string))
271 272 273 def str_to_b64str(string):
274 return b64encode(string.encode('utf-8')).decode('ascii')
275 276 277 def ceildiv(a, b):
278 """
279 Divide 2 ints and round up to next int rather than round down
280 Implementation requires python integers, which have a // operator that does floor division.
281 Other types like decimal.Decimal whose // operator truncates towards 0 will not work.
282 """
283 assert isinstance(a, int)
284 assert isinstance(b, int)
285 return -(-a // b)
286 287 288 def random_bitflip(data):
289 data = list(data)
290 data[random.randrange(len(data))] ^= (1 << (random.randrange(8)))
291 return bytes(data)
292 293 294 def get_fee(tx_size, feerate_btc_kvb):
295 """Calculate the fee in BTC given a feerate is BTC/kvB. Reflects CFeeRate::GetFee"""
296 feerate_sat_kvb = int(feerate_btc_kvb * Decimal(1e8)) # Fee in sat/kvb as an int to avoid float precision errors
297 target_fee_sat = ceildiv(feerate_sat_kvb * tx_size, 1000) # Round calculated fee up to nearest sat
298 return target_fee_sat / Decimal(1e8) # Return result in BTC
299 300 301 def satoshi_round(amount: Union[int, float, str], *, rounding: str) -> Decimal:
302 """Rounds a Decimal amount to the nearest satoshi using the specified rounding mode."""
303 return Decimal(amount).quantize(SATOSHI_PRECISION, rounding=rounding)
304 305 306 def ensure_for(*, duration, f, check_interval=0.2):
307 """Check if the predicate keeps returning True for duration.
308 309 check_interval can be used to configure the wait time between checks.
310 Setting check_interval to 0 will allow to have two checks: one in the
311 beginning and one after duration.
312 """
313 # If check_interval is 0 or negative or larger than duration, we fall back
314 # to checking once in the beginning and once at the end of duration
315 if check_interval <= 0 or check_interval > duration:
316 check_interval = duration
317 time_end = time.time() + duration
318 predicate_source = "''''\n" + inspect.getsource(f) + "'''"
319 while True:
320 if not f():
321 raise AssertionError(f"Predicate {predicate_source} became false within {duration} seconds")
322 if time.time() > time_end:
323 return
324 time.sleep(check_interval)
325 326 327 def wait_until_helper_internal(predicate, *, timeout=60, lock=None, timeout_factor=1.0, check_interval=0.05):
328 """Sleep until the predicate resolves to be True.
329 330 Warning: Note that this method is not recommended to be used in tests as it is
331 not aware of the context of the test framework. Using the `wait_until()` members
332 from `LimenkaTestFramework` or `P2PInterface` class ensures the timeout is
333 properly scaled. Furthermore, `wait_until()` from `P2PInterface` class in
334 `p2p.py` has a preset lock.
335 """
336 timeout = timeout * timeout_factor
337 time_end = time.time() + timeout
338 339 while time.time() < time_end:
340 if lock:
341 with lock:
342 if predicate():
343 return
344 else:
345 if predicate():
346 return
347 time.sleep(check_interval)
348 349 # Print the cause of the timeout
350 predicate_source = "''''\n" + inspect.getsource(predicate) + "'''"
351 logger.error("wait_until() failed. Predicate: {}".format(predicate_source))
352 raise AssertionError("Predicate {} not true after {} seconds".format(predicate_source, timeout))
353 354 355 def bpf_cflags():
356 return [
357 "-Wno-error=implicit-function-declaration",
358 "-Wno-duplicate-decl-specifier", # https://github.com/limenka/limenka/issues/32322
359 ]
360 361 362 def sha256sum_file(filename):
363 h = hashlib.sha256()
364 with open(filename, 'rb') as f:
365 d = f.read(4096)
366 while len(d) > 0:
367 h.update(d)
368 d = f.read(4096)
369 return h.digest()
370 371 372 def util_xor(data, key, *, offset):
373 data = bytearray(data)
374 for i in range(len(data)):
375 data[i] ^= key[(i + offset) % len(key)]
376 return bytes(data)
377 378 379 # RPC/P2P connection constants and functions
380 ############################################
381 382 # The maximum number of nodes a single test can spawn
383 MAX_NODES = 12
384 # Don't assign p2p, rpc or tor ports lower than this
385 PORT_MIN = int(os.getenv('TEST_RUNNER_PORT_MIN', default=11000))
386 # The number of ports to "reserve" for p2p, rpc and tor, each
387 PORT_RANGE = 5000
388 389 390 class PortSeed:
391 # Must be initialized with a unique integer for each process
392 n = None
393 394 395 def get_rpc_proxy(url: str, node_number: int, *, timeout: Optional[int]=None, coveragedir: Optional[str]=None) -> coverage.AuthServiceProxyWrapper:
396 """
397 Args:
398 url: URL of the RPC server to call
399 node_number: the node number (or id) that this calls to
400 401 Kwargs:
402 timeout: HTTP timeout in seconds
403 coveragedir: Directory
404 405 Returns:
406 AuthServiceProxy. convenience object for making RPC calls.
407 408 """
409 proxy_kwargs = {}
410 if timeout is not None:
411 proxy_kwargs['timeout'] = int(timeout)
412 413 proxy = AuthServiceProxy(url, **proxy_kwargs)
414 415 coverage_logfile = coverage.get_filename(coveragedir, node_number) if coveragedir else None
416 417 return coverage.AuthServiceProxyWrapper(proxy, url, coverage_logfile)
418 419 420 def p2p_port(n):
421 assert n <= MAX_NODES
422 return PORT_MIN + n + (MAX_NODES * PortSeed.n) % (PORT_RANGE - 1 - MAX_NODES)
423 424 425 def rpc_port(n):
426 return p2p_port(n) + PORT_RANGE
427 428 429 def tor_port(n):
430 return p2p_port(n) + PORT_RANGE * 2
431 432 433 def rpc_url(datadir, i, chain, rpchost):
434 rpc_u, rpc_p = get_auth_cookie(datadir, chain)
435 host = '127.0.0.1'
436 port = rpc_port(i)
437 if rpchost:
438 parts = rpchost.split(':')
439 if len(parts) == 2:
440 host, port = parts
441 else:
442 host = rpchost
443 return "http://%s:%s@%s:%d" % (rpc_u, rpc_p, host, int(port))
444 445 446 # Node functions
447 ################
448 449 450 def initialize_datadir(dirname, n, chain, disable_autoconnect=True):
451 datadir = get_datadir_path(dirname, n)
452 if not os.path.isdir(datadir):
453 os.makedirs(datadir)
454 write_config(os.path.join(datadir, "limenka.conf"), n=n, chain=chain, disable_autoconnect=disable_autoconnect)
455 os.makedirs(os.path.join(datadir, 'stderr'), exist_ok=True)
456 os.makedirs(os.path.join(datadir, 'stdout'), exist_ok=True)
457 return datadir
458 459 460 def write_config(config_path, *, n, chain, extra_config="", disable_autoconnect=True):
461 # Translate chain subdirectory name to config name
462 if chain == 'testnet3':
463 chain_name_conf_arg = 'testnet'
464 chain_name_conf_section = 'test'
465 elif chain == 'limenka':
466 chain_name_conf_arg = 'chain=limenka'
467 chain_name_conf_section = 'limenka'
468 else:
469 chain_name_conf_arg = chain
470 chain_name_conf_section = chain
471 with open(config_path, 'w', encoding='utf8') as f:
472 if chain == 'limenka':
473 f.write("chain=limenka\n")
474 f.write("[limenka]\n")
475 elif chain_name_conf_arg:
476 f.write("{}=1\n".format(chain_name_conf_arg))
477 if chain_name_conf_section:
478 f.write("[{}]\n".format(chain_name_conf_section))
479 f.write("port=" + str(p2p_port(n)) + "\n")
480 f.write("rpcport=" + str(rpc_port(n)) + "\n")
481 # Disable server-side timeouts to avoid intermittent issues
482 f.write("rpcservertimeout=99000\n")
483 f.write("rpcdoccheck=1\n")
484 f.write("fallbackfee=0.0002\n")
485 f.write("server=1\n")
486 f.write("keypool=1\n")
487 f.write("discover=0\n")
488 f.write("dnsseed=0\n")
489 f.write("fixedseeds=0\n")
490 f.write("listenonion=0\n")
491 # Increase peertimeout to avoid disconnects while using mocktime.
492 # peertimeout is measured in mock time, so setting it large enough to
493 # cover any duration in mock time is sufficient. It can be overridden
494 # in tests.
495 f.write("peertimeout=999999999\n")
496 f.write("printtoconsole=0\n")
497 f.write("upnp=0\n")
498 f.write("natpmp=0\n")
499 f.write("shrinkdebugfile=0\n")
500 f.write("deprecatedrpc=create_bdb\n") # Required to run the tests
501 # To improve SQLite wallet performance so that the tests don't timeout, use -unsafesqlitesync
502 f.write("unsafesqlitesync=1\n")
503 if disable_autoconnect:
504 f.write("connect=0\n")
505 # Limit max connections to mitigate test failures on some systems caused by the warning:
506 # "Warning: Reducing -maxconnections from <...> to <...> due to system limitations".
507 # The value is calculated as follows:
508 # available_fds = 256 // Same as FD_SETSIZE on NetBSD.
509 # MIN_CORE_FDS = 151 // Number of file descriptors required for core functionality.
510 # MAX_ADDNODE_CONNECTIONS = 8 // Maximum number of -addnode outgoing nodes.
511 # nBind == 3 // Maximum number of bound interfaces used in a test.
512 #
513 # min_required_fds = MIN_CORE_FDS + MAX_ADDNODE_CONNECTIONS + nBind = 151 + 8 + 3 = 162;
514 # nMaxConnections = available_fds - min_required_fds = 256 - 161 = 94;
515 f.write("maxconnections=94\n")
516 f.write(extra_config)
517 518 519 def get_datadir_path(dirname, n):
520 return pathlib.Path(dirname) / f"node{n}"
521 522 523 def get_temp_default_datadir(temp_dir: pathlib.Path) -> tuple[dict, pathlib.Path]:
524 """Return os-specific environment variables that can be set to make the
525 GetDefaultDataDir() function return a datadir path under the provided
526 temp_dir, as well as the complete path it would return."""
527 if platform.system() == "Windows":
528 env = dict(APPDATA=str(temp_dir))
529 datadir = temp_dir / "Limenka"
530 else:
531 env = dict(HOME=str(temp_dir))
532 if platform.system() == "Darwin":
533 datadir = temp_dir / "Library/Application Support/Limenka"
534 else:
535 datadir = temp_dir / ".limenka"
536 return env, datadir
537 538 539 def append_config(datadir, options):
540 with open(os.path.join(datadir, "limenka.conf"), 'a', encoding='utf8') as f:
541 for option in options:
542 f.write(option + "\n")
543 544 545 def get_auth_cookie(datadir, chain):
546 user = None
547 password = None
548 if os.path.isfile(os.path.join(datadir, "limenka.conf")):
549 with open(os.path.join(datadir, "limenka.conf"), 'r', encoding='utf8') as f:
550 for line in f:
551 if line.startswith("rpcuser="):
552 assert user is None # Ensure that there is only one rpcuser line
553 user = line.split("=")[1].strip("\n")
554 if line.startswith("rpcpassword="):
555 assert password is None # Ensure that there is only one rpcpassword line
556 password = line.split("=")[1].strip("\n")
557 cookie_chain = 'fork' if chain == 'limenka' else chain
558 try:
559 with open(os.path.join(datadir, cookie_chain, ".cookie"), 'r', encoding="ascii") as f:
560 userpass = f.read()
561 split_userpass = userpass.split(':')
562 user = split_userpass[0]
563 password = split_userpass[1]
564 except OSError:
565 pass
566 if user is None or password is None:
567 raise ValueError("No RPC credentials")
568 return user, password
569 570 571 # If a cookie file exists in the given datadir, delete it.
572 def delete_cookie_file(datadir, chain):
573 if os.path.isfile(os.path.join(datadir, chain, ".cookie")):
574 logger.debug("Deleting leftover cookie file")
575 os.remove(os.path.join(datadir, chain, ".cookie"))
576 577 578 def softfork_active(node, key):
579 """Return whether a softfork is active."""
580 return node.getdeploymentinfo()['deployments'][key]['active']
581 582 583 def set_node_times(nodes, t):
584 for node in nodes:
585 node.setmocktime(t)
586 587 588 def check_node_connections(*, node, num_in, num_out):
589 info = node.getnetworkinfo()
590 assert_equal(info["connections_in"], num_in)
591 assert_equal(info["connections_out"], num_out)
592 593 594 # Transaction/Block functions
595 #############################
596 597 598 # Create large OP_RETURN txouts that can be appended to a transaction
599 # to make it large (helper for constructing large transactions). The
600 # total serialized size of the txouts is about 66k vbytes.
601 def gen_return_txouts():
602 from .messages import CTxOut
603 from .script import CScript, OP_RETURN
604 txouts = [CTxOut(nValue=0, scriptPubKey=CScript([OP_RETURN, b'\x01'*80]))] * 733
605 txouts.append(CTxOut(nValue=0, scriptPubKey=CScript([OP_RETURN, b'\x01'*9])))
606 assert_equal(sum([len(txout.serialize()) for txout in txouts]), 67456)
607 return txouts
608 609 610 # Create a spend of each passed-in utxo, splicing in "txouts" to each raw
611 # transaction to make it large. See gen_return_txouts() above.
612 def create_lots_of_big_transactions(mini_wallet, node, fee, tx_batch_size, txouts, utxos=None):
613 txids = []
614 use_internal_utxos = utxos is None
615 for _ in range(tx_batch_size):
616 tx = mini_wallet.create_self_transfer(
617 utxo_to_spend=None if use_internal_utxos else utxos.pop(),
618 fee=fee,
619 )["tx"]
620 tx.vout.extend(txouts)
621 res = node.testmempoolaccept([tx.serialize().hex()])[0]
622 assert_equal(res['fees']['base'], fee)
623 txids.append(node.sendrawtransaction(tx.serialize().hex()))
624 return txids
625 626 627 def mine_large_block(test_framework, mini_wallet, node):
628 # generate a 66k transaction,
629 # and 14 of them is close to the 1MB block limit
630 txouts = gen_return_txouts()
631 fee = 100 * node.getnetworkinfo()["relayfee"]
632 create_lots_of_big_transactions(mini_wallet, node, fee, 14, txouts)
633 test_framework.generate(node, 1)
634 635 636 def find_vout_for_address(node, txid, addr):
637 """
638 Locate the vout index of the given transaction sending to the
639 given address. Raises runtime error exception if not found.
640 """
641 tx = node.getrawtransaction(txid, True)
642 for i in range(len(tx["vout"])):
643 if addr == tx["vout"][i]["scriptPubKey"]["address"]:
644 return i
645 raise RuntimeError("Vout not found for address: txid=%s, addr=%s" % (txid, addr))
646 647 def is_dir_writable(dir_path: pathlib.Path) -> bool:
648 """Return True if we can create a file in the directory, False otherwise"""
649 try:
650 tmp = dir_path / f".tmp_{random.randrange(1 << 32)}"
651 tmp.touch()
652 tmp.unlink()
653 return True
654 except OSError:
655 return False
656