1 #!/usr/bin/env python3
2 # Copyright (c) 2017-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 """Class for limenkad node under test"""
6 7 import contextlib
8 import decimal
9 import errno
10 from enum import Enum
11 import http.client
12 import json
13 import logging
14 import os
15 import platform
16 import re
17 import subprocess
18 import tempfile
19 import time
20 import urllib.parse
21 import collections
22 import shlex
23 import sys
24 from pathlib import Path
25 26 from .authproxy import (
27 JSONRPCException,
28 serialization_fallback,
29 )
30 from .descriptors import descsum_create
31 from .messages import MAX_OP_RETURN_RELAY
32 from .messages import NODE_P2P_V2
33 from .p2p import P2P_SERVICES, P2P_SUBVERSION
34 from .util import (
35 MAX_NODES,
36 assert_equal,
37 append_config,
38 delete_cookie_file,
39 get_auth_cookie,
40 get_rpc_proxy,
41 rpc_url,
42 wait_until_helper_internal,
43 p2p_port,
44 tor_port,
45 )
46 47 LIMENKAD_PROC_WAIT_TIMEOUT = 60
48 # The size of the blocks xor key
49 # from InitBlocksdirXorKey::xor_key.size()
50 NUM_XOR_BYTES = 8
51 # The null blocks key (all 0s)
52 NULL_BLK_XOR_KEY = bytes([0] * NUM_XOR_BYTES)
53 LIMENKA_PID_FILENAME_DEFAULT = "limenkad.pid"
54 55 56 class FailedToStartError(Exception):
57 """Raised when a node fails to start correctly."""
58 59 60 class ErrorMatch(Enum):
61 FULL_TEXT = 1
62 FULL_REGEX = 2
63 PARTIAL_REGEX = 3
64 65 66 class TestNode():
67 """A class for representing a limenkad node under test.
68 69 This class contains:
70 71 - state about the node (whether it's running, etc)
72 - a Python subprocess.Popen object representing the running process
73 - an RPC connection to the node
74 - one or more P2P connections to the node
75 76 77 To make things easier for the test writer, any unrecognised messages will
78 be dispatched to the RPC connection."""
79 80 def __init__(self, i, datadir_path, *, chain, rpchost, timewait, timeout_factor, limenkad, limenka_cli, coverage_dir, cwd, extra_conf=None, extra_args=None, use_cli=False, start_perf=False, use_valgrind=False, version=None, descriptors=False, v2transport=False):
81 """
82 Kwargs:
83 start_perf (bool): If True, begin profiling the node with `perf` as soon as
84 the node starts.
85 """
86 87 self.index = i
88 self.p2p_conn_index = 1
89 self.datadir_path = datadir_path
90 self.limenkaconf = self.datadir_path / "limenka.conf"
91 self.stdout_dir = self.datadir_path / "stdout"
92 self.stderr_dir = self.datadir_path / "stderr"
93 self.chain = chain
94 self.rpchost = rpchost
95 self.rpc_timeout = timewait # Already multiplied by timeout_factor
96 self.timeout_factor = timeout_factor
97 self.binary = limenkad
98 self.coverage_dir = coverage_dir
99 self.cwd = cwd
100 self.descriptors = descriptors
101 self.has_explicit_bind = False
102 if extra_conf is not None:
103 append_config(self.datadir_path, extra_conf)
104 # Remember if there is bind=... in the config file.
105 self.has_explicit_bind = any(e.startswith("bind=") for e in extra_conf)
106 # Most callers will just need to add extra args to the standard list below.
107 # For those callers that need more flexibility, they can just set the args property directly.
108 # Note that common args are set in the config file (see initialize_datadir)
109 self.extra_args = extra_args
110 self.version = version
111 # Configuration for logging is set as command-line args rather than in the limenka.conf file.
112 # This means that starting a limenkad using the temp dir to debug a failed test won't
113 # spam debug.log.
114 self.args = [
115 self.binary,
116 f"-datadir={self.datadir_path}",
117 "-logtimemicros",
118 "-debug",
119 "-debugexclude=libevent",
120 "-debugexclude=leveldb",
121 "-debugexclude=rand",
122 "-uacomment=testnode%d" % i, # required for subversion uniqueness across peers
123 ]
124 if self.descriptors is None:
125 self.args.append("-disablewallet")
126 127 # Use valgrind, expect for previous release binaries
128 if use_valgrind and version is None:
129 default_suppressions_file = Path(__file__).parents[3] / "contrib" / "valgrind.supp"
130 suppressions_file = os.getenv("VALGRIND_SUPPRESSIONS_FILE",
131 default_suppressions_file)
132 self.args = ["valgrind", "--suppressions={}".format(suppressions_file),
133 "--gen-suppressions=all", "--exit-on-first-error=yes",
134 "--error-exitcode=1", "--quiet"] + self.args
135 136 if self.version is None:
137 self.args += [
138 "-corepolicy",
139 "-walletimplicitsegwit",
140 ]
141 142 if self.version_is_at_least(190000):
143 self.args.append("-logthreadnames")
144 if self.version_is_at_least(219900):
145 self.args.append("-logsourcelocations")
146 if self.version_is_at_least(239000):
147 self.args.append("-loglevel=trace")
148 if self.version_is_at_least(290100):
149 self.args.append("-nologratelimit")
150 151 # Default behavior from global -v2transport flag is added to args to persist it over restarts.
152 # May be overwritten in individual tests, using extra_args.
153 self.default_to_v2 = v2transport
154 if self.version_is_at_least(260000):
155 # 26.0 and later support v2transport
156 if v2transport:
157 self.args.append("-v2transport=1")
158 else:
159 self.args.append("-v2transport=0")
160 # if v2transport is requested via global flag but not supported for node version, ignore it
161 162 self.cli = TestNodeCLI(
163 limenka_cli,
164 self.datadir_path,
165 self.rpc_timeout // 2, # timeout identical to the one used in self._rpc
166 )
167 self.use_cli = use_cli
168 self.start_perf = start_perf
169 170 self.running = False
171 self.process = None
172 self.rpc_connected = False
173 self.rpc = None
174 self.url = None
175 self.log = logging.getLogger('TestFramework.node%d' % i)
176 # Cache perf subprocesses here by their data output filename.
177 self.perf_subprocesses = {}
178 179 self.p2ps = []
180 181 self.mocktime = None
182 183 AddressKeyPair = collections.namedtuple('AddressKeyPair', ['address', 'key'])
184 PRIV_KEYS = [
185 # address , privkey
186 AddressKeyPair('mjTkW3DjgyZck4KbiRusZsqTgaYTxdSz6z', 'cVpF924EspNh8KjYsfhgY96mmxvT6DgdWiTYMtMjuM74hJaU5psW'),
187 AddressKeyPair('msX6jQXvxiNhx3Q62PKeLPrhrqZQdSimTg', 'cUxsWyKyZ9MAQTaAhUQWJmBbSvHMwSmuv59KgxQV7oZQU3PXN3KE'),
188 AddressKeyPair('mnonCMyH9TmAsSj3M59DsbH8H63U3RKoFP', 'cTrh7dkEAeJd6b3MRX9bZK8eRmNqVCMH3LSUkE3dSFDyzjU38QxK'),
189 AddressKeyPair('mqJupas8Dt2uestQDvV2NH3RU8uZh2dqQR', 'cVuKKa7gbehEQvVq717hYcbE9Dqmq7KEBKqWgWrYBa2CKKrhtRim'),
190 AddressKeyPair('msYac7Rvd5ywm6pEmkjyxhbCDKqWsVeYws', 'cQDCBuKcjanpXDpCqacNSjYfxeQj8G6CAtH1Dsk3cXyqLNC4RPuh'),
191 AddressKeyPair('n2rnuUnwLgXqf9kk2kjvVm8R5BZK1yxQBi', 'cQakmfPSLSqKHyMFGwAqKHgWUiofJCagVGhiB4KCainaeCSxeyYq'),
192 AddressKeyPair('myzuPxRwsf3vvGzEuzPfK9Nf2RfwauwYe6', 'cQMpDLJwA8DBe9NcQbdoSb1BhmFxVjWD5gRyrLZCtpuF9Zi3a9RK'),
193 AddressKeyPair('mumwTaMtbxEPUswmLBBN3vM9oGRtGBrys8', 'cSXmRKXVcoouhNNVpcNKFfxsTsToY5pvB9DVsFksF1ENunTzRKsy'),
194 AddressKeyPair('mpV7aGShMkJCZgbW7F6iZgrvuPHjZjH9qg', 'cSoXt6tm3pqy43UMabY6eUTmR3eSUYFtB2iNQDGgb3VUnRsQys2k'),
195 AddressKeyPair('mq4fBNdckGtvY2mijd9am7DRsbRB4KjUkf', 'cN55daf1HotwBAgAKWVgDcoppmUNDtQSfb7XLutTLeAgVc3u8hik'),
196 AddressKeyPair('mpFAHDjX7KregM3rVotdXzQmkbwtbQEnZ6', 'cT7qK7g1wkYEMvKowd2ZrX1E5f6JQ7TM246UfqbCiyF7kZhorpX3'),
197 AddressKeyPair('mzRe8QZMfGi58KyWCse2exxEFry2sfF2Y7', 'cPiRWE8KMjTRxH1MWkPerhfoHFn5iHPWVK5aPqjW8NxmdwenFinJ'),
198 ]
199 200 def get_deterministic_priv_key(self):
201 """Return a deterministic priv key in base58, that only depends on the node's index"""
202 assert len(self.PRIV_KEYS) == MAX_NODES
203 return self.PRIV_KEYS[self.index]
204 205 def _node_msg(self, msg: str) -> str:
206 """Return a modified msg that identifies this node by its index as a debugging aid."""
207 return "[node %d] %s" % (self.index, msg)
208 209 def _raise_assertion_error(self, msg: str):
210 """Raise an AssertionError with msg modified to identify this node."""
211 raise AssertionError(self._node_msg(msg))
212 213 def __del__(self):
214 # Ensure that we don't leave any limenkad processes lying around after
215 # the test ends
216 if self.process:
217 # Should only happen on test failure
218 # Avoid using logger, as that may have already been shutdown when
219 # this destructor is called.
220 print(self._node_msg("Cleaning up leftover process"), file=sys.stderr)
221 self.process.kill()
222 223 def __getattr__(self, name):
224 """Dispatches any unrecognised messages to the RPC connection or a CLI instance."""
225 if self.use_cli:
226 return getattr(RPCOverloadWrapper(self.cli, True, self.descriptors), name)
227 else:
228 assert self.rpc_connected and self.rpc is not None, self._node_msg("Error: no RPC connection")
229 return getattr(RPCOverloadWrapper(self.rpc, descriptors=self.descriptors), name)
230 231 def start(self, extra_args=None, *, cwd=None, stdout=None, stderr=None, env=None, **kwargs):
232 """Start the node."""
233 if extra_args is None:
234 extra_args = self.extra_args
235 236 # If listening and no -bind is given, then limenkad would bind P2P ports on
237 # 0.0.0.0:P and 127.0.0.1:P+1 (for incoming Tor connections), where P is
238 # a unique port chosen by the test framework and configured as port=P in
239 # limenka.conf. To avoid collisions, change it to 127.0.0.1:tor_port().
240 will_listen = all(e != "-nolisten" and e != "-listen=0" for e in extra_args)
241 has_explicit_bind = self.has_explicit_bind or any(e.startswith("-bind=") for e in extra_args)
242 if will_listen and not has_explicit_bind:
243 extra_args.append(f"-bind=0.0.0.0:{p2p_port(self.index)}")
244 extra_args.append(f"-bind=127.0.0.1:{tor_port(self.index)}=onion")
245 246 self.use_v2transport = "-v2transport=1" in extra_args or (self.default_to_v2 and "-v2transport=0" not in extra_args)
247 248 # Add a new stdout and stderr file each time limenkad is started
249 if stderr is None:
250 stderr = tempfile.NamedTemporaryFile(dir=self.stderr_dir, delete=False)
251 if stdout is None:
252 stdout = tempfile.NamedTemporaryFile(dir=self.stdout_dir, delete=False)
253 self.stderr = stderr
254 self.stdout = stdout
255 256 if cwd is None:
257 cwd = self.cwd
258 259 # Delete any existing cookie file -- if such a file exists (eg due to
260 # unclean shutdown), it will get overwritten anyway by limenkad, and
261 # potentially interfere with our attempt to authenticate
262 delete_cookie_file(self.datadir_path, self.chain)
263 264 # add environment variable LIBC_FATAL_STDERR_=1 so that libc errors are written to stderr and not the terminal
265 subp_env = dict(os.environ, LIBC_FATAL_STDERR_="1")
266 if env is not None:
267 subp_env.update(env)
268 269 for arg in extra_args:
270 if arg.startswith('-datacarriersize=') and int(arg[17:]) > MAX_OP_RETURN_RELAY:
271 extra_args = list(extra_args)
272 extra_args.append('-acceptnonstdtxn=1')
273 274 self.process = subprocess.Popen(self.args + extra_args, env=subp_env, stdout=stdout, stderr=stderr, cwd=cwd, **kwargs)
275 276 self.running = True
277 self.log.debug("limenkad started, waiting for RPC to come up")
278 279 if self.start_perf:
280 self._start_perf()
281 282 def wait_for_rpc_connection(self, *, wait_for_import=True):
283 """Sets up an RPC connection to the limenkad process. Returns False if unable to connect."""
284 # Poll at a rate of four times per second
285 poll_per_s = 4
286 for _ in range(poll_per_s * self.rpc_timeout):
287 if self.process.poll() is not None:
288 # Attach abrupt shutdown error/s to the exception message
289 self.stderr.seek(0)
290 str_error = ''.join(line.decode('utf-8') for line in self.stderr)
291 str_error += "************************\n" if str_error else ''
292 293 raise FailedToStartError(self._node_msg(
294 f'limenkad exited with status {self.process.returncode} during initialization. {str_error}'))
295 try:
296 rpc = get_rpc_proxy(
297 rpc_url(self.datadir_path, self.index, self.chain, self.rpchost),
298 self.index,
299 timeout=self.rpc_timeout // 2, # Shorter timeout to allow for one retry in case of ETIMEDOUT
300 coveragedir=self.coverage_dir,
301 )
302 rpc.getblockcount()
303 # If the call to getblockcount() succeeds then the RPC connection is up
304 if self.version_is_at_least(190000) and wait_for_import:
305 # getmempoolinfo.loaded is available since commit
306 # bb8ae2c (version 0.19.0)
307 self.wait_until(lambda: rpc.getmempoolinfo()['loaded'])
308 # Wait for the node to finish reindex, block import, and
309 # loading the mempool. Usually importing happens fast or
310 # even "immediate" when the node is started. However, there
311 # is no guarantee and sometimes ImportBlocks might finish
312 # later. This is going to cause intermittent test failures,
313 # because generally the tests assume the node is fully
314 # ready after being started.
315 #
316 # For example, the node will reject block messages from p2p
317 # when it is still importing with the error "Unexpected
318 # block message received"
319 #
320 # The wait is done here to make tests as robust as possible
321 # and prevent racy tests and intermittent failures as much
322 # as possible. Some tests might not need this, but the
323 # overhead is trivial, and the added guarantees are worth
324 # the minimal performance cost.
325 self.log.debug("RPC successfully started")
326 if self.use_cli:
327 return
328 self.rpc = rpc
329 self.rpc_connected = True
330 self.url = self.rpc.rpc_url
331 return
332 except JSONRPCException as e: # Initialization phase
333 # -28 RPC in warmup
334 # -342 Service unavailable, RPC server started but is shutting down due to error
335 if e.error['code'] != -28 and e.error['code'] != -342:
336 raise # unknown JSON RPC exception
337 except ConnectionResetError:
338 # This might happen when the RPC server is in warmup, but shut down before the call to getblockcount
339 # succeeds. Try again to properly raise the FailedToStartError
340 pass
341 except OSError as e:
342 if e.errno == errno.ETIMEDOUT:
343 pass # Treat identical to ConnectionResetError
344 elif e.errno == errno.ECONNREFUSED:
345 pass # Port not yet open?
346 else:
347 raise # unknown OS error
348 except ValueError as e: # cookie file not found and no rpcuser or rpcpassword; limenkad is still starting
349 if "No RPC credentials" not in str(e):
350 raise
351 time.sleep(1.0 / poll_per_s)
352 self._raise_assertion_error("Unable to connect to limenkad after {}s".format(self.rpc_timeout))
353 354 def wait_for_cookie_credentials(self):
355 """Ensures auth cookie credentials can be read, e.g. for testing CLI with -rpcwait before RPC connection is up."""
356 self.log.debug("Waiting for cookie credentials")
357 # Poll at a rate of four times per second.
358 poll_per_s = 4
359 for _ in range(poll_per_s * self.rpc_timeout):
360 try:
361 get_auth_cookie(self.datadir_path, self.chain)
362 self.log.debug("Cookie credentials successfully retrieved")
363 return
364 except ValueError: # cookie file not found and no rpcuser or rpcpassword; limenkad is still starting
365 pass # so we continue polling until RPC credentials are retrieved
366 time.sleep(1.0 / poll_per_s)
367 self._raise_assertion_error("Unable to retrieve cookie credentials after {}s".format(self.rpc_timeout))
368 369 def generate(self, nblocks, maxtries=1000000, **kwargs):
370 self.log.debug("TestNode.generate() dispatches `generate` call to `generatetoaddress`")
371 return self.generatetoaddress(nblocks=nblocks, address=self.get_deterministic_priv_key().address, maxtries=maxtries, **kwargs)
372 373 def generateblock(self, *args, called_by_framework, **kwargs):
374 assert called_by_framework, "Direct call of this mining RPC is discouraged. Please use one of the self.generate* methods on the test framework, which sync the nodes to avoid intermittent test issues. You may use sync_fun=self.no_op to disable the sync explicitly."
375 return self.__getattr__('generateblock')(*args, **kwargs)
376 377 def generatetoaddress(self, *args, called_by_framework, **kwargs):
378 assert called_by_framework, "Direct call of this mining RPC is discouraged. Please use one of the self.generate* methods on the test framework, which sync the nodes to avoid intermittent test issues. You may use sync_fun=self.no_op to disable the sync explicitly."
379 return self.__getattr__('generatetoaddress')(*args, **kwargs)
380 381 def generatetodescriptor(self, *args, called_by_framework, **kwargs):
382 assert called_by_framework, "Direct call of this mining RPC is discouraged. Please use one of the self.generate* methods on the test framework, which sync the nodes to avoid intermittent test issues. You may use sync_fun=self.no_op to disable the sync explicitly."
383 return self.__getattr__('generatetodescriptor')(*args, **kwargs)
384 385 def getprioritisedtransactions(self, *args, **kwargs):
386 res = self.__getattr__('getprioritisedtransactions')(*args, **kwargs)
387 assert not (args or kwargs)
388 for res_val in res.values():
389 if res_val['priority_delta'] == 0:
390 del res_val['priority_delta']
391 return res
392 393 def setmocktime(self, timestamp):
394 """Wrapper for setmocktime RPC, sets self.mocktime"""
395 if timestamp == 0:
396 # setmocktime(0) resets to system time.
397 self.mocktime = None
398 else:
399 self.mocktime = timestamp
400 return self.__getattr__('setmocktime')(timestamp)
401 402 def get_wallet_rpc(self, wallet_name):
403 if self.use_cli:
404 return RPCOverloadWrapper(self.cli("-rpcwallet={}".format(wallet_name)), True, self.descriptors)
405 else:
406 assert self.rpc_connected and self.rpc, self._node_msg("RPC not connected")
407 wallet_path = "wallet/{}".format(urllib.parse.quote(wallet_name))
408 return RPCOverloadWrapper(self.rpc / wallet_path, descriptors=self.descriptors)
409 410 def version_is_at_least(self, ver):
411 return self.version is None or self.version >= ver
412 413 def stop_node(self, expected_stderr='', *, wait=0, wait_until_stopped=True):
414 """Stop the node."""
415 if not self.running:
416 return
417 self.log.debug("Stopping node")
418 try:
419 # Do not use wait argument when testing older nodes, e.g. in wallet_backwards_compatibility.py
420 if self.version_is_at_least(180000):
421 self.stop(wait=wait)
422 else:
423 self.stop()
424 except http.client.CannotSendRequest:
425 self.log.exception("Unable to stop node.")
426 427 # If there are any running perf processes, stop them.
428 for profile_name in tuple(self.perf_subprocesses.keys()):
429 self._stop_perf(profile_name)
430 431 del self.p2ps[:]
432 433 assert (not expected_stderr) or wait_until_stopped # Must wait to check stderr
434 if wait_until_stopped:
435 self.wait_until_stopped(expected_stderr=expected_stderr)
436 437 def is_node_stopped(self, *, expected_stderr="", expected_ret_code=0):
438 """Checks whether the node has stopped.
439 440 Returns True if the node has stopped. False otherwise.
441 This method is responsible for freeing resources (self.process)."""
442 if not self.running:
443 return True
444 return_code = self.process.poll()
445 if return_code is None:
446 return False
447 448 # process has stopped. Assert that it didn't return an error code.
449 assert return_code == expected_ret_code, self._node_msg(
450 f"Node returned unexpected exit code ({return_code}) vs ({expected_ret_code}) when stopping")
451 # Check that stderr is as expected
452 self.stderr.seek(0)
453 stderr = self.stderr.read().decode('utf-8').strip()
454 # Suppress the fork chain's "[limenka] is not recognized" config
455 # warning (harmless; the node starts fine with the section present;
456 # the framework's write_config wrote it for the fork's bind/port).
457 suppressed = [l for l in stderr.splitlines() if 'Section [limenka] is not recognized' not in l]
458 suppressed_stderr = "\n".join(suppressed)
459 if suppressed_stderr != expected_stderr:
460 raise AssertionError("Unexpected stderr {} != {}".format(stderr, expected_stderr))
461 462 self.stdout.close()
463 self.stderr.close()
464 465 self.running = False
466 self.process = None
467 self.rpc_connected = False
468 self.rpc = None
469 self.log.debug("Node stopped")
470 return True
471 472 def wait_until_stopped(self, *, timeout=LIMENKAD_PROC_WAIT_TIMEOUT, expect_error=False, **kwargs):
473 if "expected_ret_code" not in kwargs:
474 kwargs["expected_ret_code"] = 1 if expect_error else 0 # Whether node shutdown return EXIT_FAILURE or EXIT_SUCCESS
475 self.wait_until(lambda: self.is_node_stopped(**kwargs), timeout=timeout)
476 477 def kill_process(self):
478 self.process.kill()
479 self.wait_until_stopped(expected_ret_code=1 if platform.system() == "Windows" else -9)
480 assert self.is_node_stopped()
481 482 def replace_in_config(self, replacements):
483 """
484 Perform replacements in the configuration file.
485 The substitutions are passed as a list of search-replace-tuples, e.g.
486 [("old", "new"), ("foo", "bar"), ...]
487 """
488 with open(self.limenkaconf, 'r', encoding='utf8') as conf:
489 conf_data = conf.read()
490 for replacement in replacements:
491 assert_equal(len(replacement), 2)
492 old, new = replacement[0], replacement[1]
493 conf_data = conf_data.replace(old, new)
494 with open(self.limenkaconf, 'w', encoding='utf8') as conf:
495 conf.write(conf_data)
496 497 @property
498 def chain_path(self) -> Path:
499 return self.datadir_path / self.chain
500 501 @property
502 def debug_log_path(self) -> Path:
503 return self.chain_path / 'debug.log'
504 505 @property
506 def blocks_path(self) -> Path:
507 return self.chain_path / "blocks"
508 509 @property
510 def blocks_key_path(self) -> Path:
511 return self.blocks_path / "xor.dat"
512 513 def read_xor_key(self) -> bytes:
514 with open(self.blocks_key_path, "rb") as xor_f:
515 return xor_f.read(NUM_XOR_BYTES)
516 517 @property
518 def wallets_path(self) -> Path:
519 return self.chain_path / "wallets"
520 521 def debug_log_size(self, **kwargs) -> int:
522 with open(self.debug_log_path, **kwargs) as dl:
523 dl.seek(0, 2)
524 return dl.tell()
525 526 @contextlib.contextmanager
527 def assert_debug_log(self, expected_msgs, unexpected_msgs=None, timeout=2):
528 if unexpected_msgs is None:
529 unexpected_msgs = []
530 assert_equal(type(expected_msgs), list)
531 assert_equal(type(unexpected_msgs), list)
532 533 if not self.debug_log_path.exists():
534 # File must exist for this to work
535 os.makedirs(self.debug_log_path.parent, exist_ok=True)
536 with open(self.debug_log_path, mode='a', encoding='utf-8'):
537 pass
538 539 time_end = time.time() + timeout * self.timeout_factor
540 prev_size = self.debug_log_size(encoding="utf-8") # Must use same encoding that is used to read() below
541 542 yield
543 544 while True:
545 found = True
546 with open(self.debug_log_path, encoding="utf-8", errors="replace") as dl:
547 dl.seek(prev_size)
548 log = dl.read()
549 print_log = " - " + "\n - ".join(log.splitlines())
550 for unexpected_msg in unexpected_msgs:
551 if re.search(re.escape(unexpected_msg), log, flags=re.MULTILINE):
552 self._raise_assertion_error('Unexpected message "{}" partially matches log:\n\n{}\n\n'.format(unexpected_msg, print_log))
553 for expected_msg in expected_msgs:
554 if re.search(re.escape(expected_msg), log, flags=re.MULTILINE) is None:
555 found = False
556 if found:
557 return
558 if time.time() >= time_end:
559 break
560 time.sleep(0.05)
561 self._raise_assertion_error('Expected messages "{}" does not partially match log:\n\n{}\n\n'.format(str(expected_msgs), print_log))
562 563 @contextlib.contextmanager
564 def busy_wait_for_debug_log(self, expected_msgs, timeout=60, *, forbid_msgs=()):
565 """
566 Block until we see a particular debug log message fragment or until we exceed the timeout.
567 Return:
568 the number of log lines we encountered when matching
569 """
570 time_end = time.time() + timeout * self.timeout_factor
571 prev_size = self.debug_log_size(mode="rb") # Must use same mode that is used to read() below
572 573 yield
574 575 while True:
576 found = True
577 with open(self.debug_log_path, "rb") as dl:
578 dl.seek(prev_size)
579 log = dl.read()
580 581 for msg in forbid_msgs:
582 if msg in log:
583 print_log = " - " + "\n - ".join(log.decode("utf8", errors="replace").splitlines())
584 self._raise_assertion_error(
585 'Forbidden message "{}" partially matched log:\n\n{}\n\n'.format(
586 str(msg), print_log))
587 588 for expected_msg in expected_msgs:
589 if expected_msg not in log:
590 found = False
591 592 if found:
593 return
594 595 if time.time() >= time_end:
596 print_log = " - " + "\n - ".join(log.decode("utf8", errors="replace").splitlines())
597 break
598 599 # No sleep here because we want to detect the message fragment as fast as
600 # possible.
601 602 self._raise_assertion_error(
603 'Expected messages "{}" does not partially match log:\n\n{}\n\n'.format(
604 str(expected_msgs), print_log))
605 606 @contextlib.contextmanager
607 def wait_for_new_peer(self, timeout=5):
608 """
609 Wait until the node is connected to at least one new peer. We detect this
610 by watching for an increased highest peer id, using the `getpeerinfo` RPC call.
611 Note that the simpler approach of only accounting for the number of peers
612 suffers from race conditions, as disconnects from unrelated previous peers
613 could happen anytime in-between.
614 """
615 def get_highest_peer_id():
616 peer_info = self.getpeerinfo()
617 return peer_info[-1]["id"] if peer_info else -1
618 619 initial_peer_id = get_highest_peer_id()
620 yield
621 self.wait_until(lambda: get_highest_peer_id() > initial_peer_id, timeout=timeout)
622 623 @contextlib.contextmanager
624 def profile_with_perf(self, profile_name: str):
625 """
626 Context manager that allows easy profiling of node activity using `perf`.
627 628 See `test/functional/README.md` for details on perf usage.
629 630 Args:
631 profile_name: This string will be appended to the
632 profile data filename generated by perf.
633 """
634 subp = self._start_perf(profile_name)
635 636 yield
637 638 if subp:
639 self._stop_perf(profile_name)
640 641 def _start_perf(self, profile_name=None):
642 """Start a perf process to profile this node.
643 644 Returns the subprocess running perf."""
645 subp = None
646 647 def test_success(cmd):
648 return subprocess.call(
649 # shell=True required for pipe use below
650 cmd, shell=True,
651 stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL) == 0
652 653 if platform.system() != 'Linux':
654 self.log.warning("Can't profile with perf; only available on Linux platforms")
655 return None
656 657 if not test_success('which perf'):
658 self.log.warning("Can't profile with perf; must install perf-tools")
659 return None
660 661 if not test_success('readelf -S {} | grep .debug_str'.format(shlex.quote(self.binary))):
662 self.log.warning(
663 "perf output won't be very useful without debug symbols compiled into limenkad")
664 665 output_path = tempfile.NamedTemporaryFile(
666 dir=self.datadir_path,
667 prefix="{}.perf.data.".format(profile_name or 'test'),
668 delete=False,
669 ).name
670 671 cmd = [
672 'perf', 'record',
673 '-g', # Record the callgraph.
674 '--call-graph', 'dwarf', # Compatibility for gcc's --fomit-frame-pointer.
675 '-F', '101', # Sampling frequency in Hz.
676 '-p', str(self.process.pid),
677 '-o', output_path,
678 ]
679 subp = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
680 self.perf_subprocesses[profile_name] = subp
681 682 return subp
683 684 def _stop_perf(self, profile_name):
685 """Stop (and pop) a perf subprocess."""
686 subp = self.perf_subprocesses.pop(profile_name)
687 output_path = subp.args[subp.args.index('-o') + 1]
688 689 subp.terminate()
690 subp.wait(timeout=10)
691 692 stderr = subp.stderr.read().decode()
693 if 'Consider tweaking /proc/sys/kernel/perf_event_paranoid' in stderr:
694 self.log.warning(
695 "perf couldn't collect data! Try "
696 "'sudo sysctl -w kernel.perf_event_paranoid=-1'")
697 else:
698 report_cmd = "perf report -i {}".format(output_path)
699 self.log.info("See perf output by running '{}'".format(report_cmd))
700 701 def assert_start_raises_init_error(self, extra_args=None, expected_msg=None, match=ErrorMatch.FULL_TEXT, *args, **kwargs):
702 """Attempt to start the node and expect it to raise an error.
703 704 extra_args: extra arguments to pass through to limenkad
705 expected_msg: regex that stderr should match when limenkad fails
706 707 Will throw if limenkad starts without an error.
708 Will throw if an expected_msg is provided and it does not match limenkad's stdout."""
709 assert not self.running
710 with tempfile.NamedTemporaryFile(dir=self.stderr_dir, delete=False) as log_stderr, \
711 tempfile.NamedTemporaryFile(dir=self.stdout_dir, delete=False) as log_stdout:
712 try:
713 self.start(extra_args, stdout=log_stdout, stderr=log_stderr, *args, **kwargs)
714 ret = self.process.wait(timeout=self.rpc_timeout)
715 self.log.debug(self._node_msg(f'limenkad exited with status {ret} during initialization'))
716 assert ret != 0 # Exit code must indicate failure
717 self.running = False
718 self.process = None
719 # Check stderr for expected message
720 if expected_msg is not None:
721 log_stderr.seek(0)
722 stderr = log_stderr.read().decode('utf-8').strip()
723 if match == ErrorMatch.PARTIAL_REGEX:
724 if re.search(expected_msg, stderr, flags=re.MULTILINE) is None:
725 self._raise_assertion_error(
726 'Expected message "{}" does not partially match stderr:\n"{}"'.format(expected_msg, stderr))
727 elif match == ErrorMatch.FULL_REGEX:
728 if re.fullmatch(expected_msg, stderr) is None:
729 self._raise_assertion_error(
730 'Expected message "{}" does not fully match stderr:\n"{}"'.format(expected_msg, stderr))
731 elif match == ErrorMatch.FULL_TEXT:
732 if expected_msg != stderr:
733 self._raise_assertion_error(
734 'Expected message "{}" does not fully match stderr:\n"{}"'.format(expected_msg, stderr))
735 except subprocess.TimeoutExpired:
736 self.process.kill()
737 self.running = False
738 self.process = None
739 assert_msg = f'limenkad should have exited within {self.rpc_timeout}s '
740 if expected_msg is None:
741 assert_msg += "with an error"
742 else:
743 assert_msg += "with expected error " + expected_msg
744 self._raise_assertion_error(assert_msg)
745 746 def add_p2p_connection(self, p2p_conn, *, wait_for_verack=True, send_version=True, supports_v2_p2p=None, wait_for_v2_handshake=True, expect_success=True, **kwargs):
747 """Add an inbound p2p connection to the node.
748 749 This method adds the p2p connection to the self.p2ps list and also
750 returns the connection to the caller.
751 752 When self.use_v2transport is True, TestNode advertises NODE_P2P_V2 service flag
753 754 An inbound connection is made from TestNode <------ P2PConnection
755 - if TestNode doesn't advertise NODE_P2P_V2 service, P2PConnection sends version message and v1 P2P is followed
756 - if TestNode advertises NODE_P2P_V2 service, (and if P2PConnections supports v2 P2P)
757 P2PConnection sends ellswift bytes and v2 P2P is followed
758 """
759 if 'dstport' not in kwargs:
760 kwargs['dstport'] = p2p_port(self.index)
761 if 'dstaddr' not in kwargs:
762 kwargs['dstaddr'] = '127.0.0.1'
763 if supports_v2_p2p is None:
764 supports_v2_p2p = self.use_v2transport
765 766 if self.use_v2transport:
767 kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
768 supports_v2_p2p = self.use_v2transport and supports_v2_p2p
769 p2p_conn.peer_connect(**kwargs, send_version=send_version, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p)()
770 771 self.p2ps.append(p2p_conn)
772 if not expect_success:
773 return p2p_conn
774 p2p_conn.wait_until(lambda: p2p_conn.is_connected, check_connected=False)
775 if supports_v2_p2p and wait_for_v2_handshake:
776 p2p_conn.wait_until(lambda: p2p_conn.v2_state.tried_v2_handshake)
777 if send_version:
778 p2p_conn.wait_until(lambda: not p2p_conn.on_connection_send_msg)
779 if wait_for_verack:
780 # Wait for the node to send us the version and verack
781 p2p_conn.wait_for_verack()
782 # At this point we have sent our version message and received the version and verack, however the full node
783 # has not yet received the verack from us (in reply to their version). So, the connection is not yet fully
784 # established (fSuccessfullyConnected).
785 #
786 # This shouldn't lead to any issues when sending messages, since the verack will be in-flight before the
787 # message we send. However, it might lead to races where we are expecting to receive a message. E.g. a
788 # transaction that will be added to the mempool as soon as we return here.
789 #
790 # So syncing here is redundant when we only want to send a message, but the cost is low (a few milliseconds)
791 # in comparison to the upside of making tests less fragile and unexpected intermittent errors less likely.
792 p2p_conn.sync_with_ping()
793 794 # Consistency check that the node received our user agent string.
795 # Find our connection in getpeerinfo by our address:port and theirs, as this combination is unique.
796 sockname = p2p_conn._transport.get_extra_info("socket").getsockname()
797 our_addr_and_port = f"{sockname[0]}:{sockname[1]}"
798 dst_addr_and_port = f"{p2p_conn.dstaddr}:{p2p_conn.dstport}"
799 info = [peer for peer in self.getpeerinfo() if peer["addr"] == our_addr_and_port and peer["addrbind"] == dst_addr_and_port]
800 assert_equal(len(info), 1)
801 assert_equal(info[0]["subver"], P2P_SUBVERSION)
802 803 return p2p_conn
804 805 def add_outbound_p2p_connection(self, p2p_conn, *, wait_for_verack=True, wait_for_disconnect=False, p2p_idx, connection_type="outbound-full-relay", supports_v2_p2p=None, advertise_v2_p2p=None, **kwargs):
806 """Add an outbound p2p connection from node. Must be an
807 "outbound-full-relay", "block-relay-only", "addr-fetch" or "feeler" connection.
808 809 This method adds the p2p connection to the self.p2ps list and returns
810 the connection to the caller.
811 812 p2p_idx must be different for simultaneously connected peers. When reusing it for the next peer
813 after disconnecting the previous one, it is necessary to wait for the disconnect to finish to avoid
814 a race condition.
815 816 Parameters:
817 supports_v2_p2p: whether p2p_conn supports v2 P2P or not
818 advertise_v2_p2p: whether p2p_conn is advertised to support v2 P2P or not
819 820 An outbound connection is made from TestNode -------> P2PConnection
821 - if P2PConnection doesn't advertise_v2_p2p, TestNode sends version message and v1 P2P is followed
822 - if P2PConnection both supports_v2_p2p and advertise_v2_p2p, TestNode sends ellswift bytes and v2 P2P is followed
823 - if P2PConnection doesn't supports_v2_p2p but advertise_v2_p2p,
824 TestNode sends ellswift bytes and P2PConnection disconnects,
825 TestNode reconnects by sending version message and v1 P2P is followed
826 """
827 828 def addconnection_callback(address, port):
829 self.log.debug("Connecting to %s:%d %s" % (address, port, connection_type))
830 self.addconnection('%s:%d' % (address, port), connection_type, advertise_v2_p2p)
831 832 if supports_v2_p2p is None:
833 supports_v2_p2p = self.use_v2transport
834 if advertise_v2_p2p is None:
835 advertise_v2_p2p = self.use_v2transport
836 837 if advertise_v2_p2p:
838 kwargs['services'] = kwargs.get('services', P2P_SERVICES) | NODE_P2P_V2
839 assert self.use_v2transport # only a v2 TestNode could make a v2 outbound connection
840 841 # if P2PConnection is advertised to support v2 P2P when it doesn't actually support v2 P2P,
842 # reconnection needs to be attempted using v1 P2P by sending version message
843 reconnect = advertise_v2_p2p and not supports_v2_p2p
844 # P2PConnection needs to be advertised to support v2 P2P so that ellswift bytes are sent instead of msg_version
845 supports_v2_p2p = supports_v2_p2p and advertise_v2_p2p
846 p2p_conn.peer_accept_connection(connect_cb=addconnection_callback, connect_id=p2p_idx + 1, net=self.chain, timeout_factor=self.timeout_factor, supports_v2_p2p=supports_v2_p2p, reconnect=reconnect, **kwargs)()
847 848 if reconnect:
849 p2p_conn.wait_for_reconnect()
850 851 if connection_type == "feeler" or wait_for_disconnect:
852 # feeler connections are closed as soon as the node receives a `version` message
853 p2p_conn.wait_until(lambda: p2p_conn.message_count["version"] == 1, check_connected=False)
854 p2p_conn.wait_until(lambda: not p2p_conn.is_connected, check_connected=False)
855 else:
856 p2p_conn.wait_for_connect()
857 self.p2ps.append(p2p_conn)
858 859 if supports_v2_p2p:
860 p2p_conn.wait_until(lambda: p2p_conn.v2_state.tried_v2_handshake)
861 p2p_conn.wait_until(lambda: not p2p_conn.on_connection_send_msg)
862 if wait_for_verack:
863 p2p_conn.wait_for_verack()
864 p2p_conn.sync_with_ping()
865 866 return p2p_conn
867 868 def num_test_p2p_connections(self):
869 """Return number of test framework p2p connections to the node."""
870 return len([peer for peer in self.getpeerinfo() if peer['subver'] == P2P_SUBVERSION])
871 872 def disconnect_p2ps(self):
873 """Close all p2p connections to the node.
874 Use only after each p2p has sent a version message to ensure the wait works."""
875 for p in self.p2ps:
876 p.peer_disconnect()
877 del self.p2ps[:]
878 879 self.wait_until(lambda: self.num_test_p2p_connections() == 0)
880 881 def bumpmocktime(self, seconds):
882 """Fast forward using setmocktime to self.mocktime + seconds. Requires setmocktime to have
883 been called at some point in the past."""
884 assert self.mocktime
885 self.mocktime += seconds
886 self.setmocktime(self.mocktime)
887 888 def wait_until(self, test_function, timeout=60, check_interval=0.05):
889 return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor, check_interval=check_interval)
890 891 892 class TestNodeCLIAttr:
893 def __init__(self, cli, command):
894 self.cli = cli
895 self.command = command
896 897 def __call__(self, *args, **kwargs):
898 return self.cli.send_cli(self.command, *args, **kwargs)
899 900 def get_request(self, *args, **kwargs):
901 return lambda: self(*args, **kwargs)
902 903 904 def arg_to_cli(arg):
905 if isinstance(arg, bool):
906 return str(arg).lower()
907 elif arg is None:
908 return 'null'
909 elif isinstance(arg, dict) or isinstance(arg, list):
910 return json.dumps(arg, default=serialization_fallback)
911 else:
912 return str(arg)
913 914 915 class TestNodeCLI():
916 """Interface to limenka-cli for an individual node"""
917 def __init__(self, binary, datadir, rpc_timeout):
918 self.options = []
919 self.binary = binary
920 self.datadir = datadir
921 self.rpc_timeout = rpc_timeout
922 self.input = None
923 self.log = logging.getLogger('TestFramework.limenkacli')
924 925 def __call__(self, *options, input=None):
926 # TestNodeCLI is callable with limenka-cli command-line options
927 cli = TestNodeCLI(self.binary, self.datadir, self.rpc_timeout)
928 cli.options = [str(o) for o in options]
929 cli.input = input
930 return cli
931 932 def __getattr__(self, command):
933 return TestNodeCLIAttr(self, command)
934 935 def batch(self, requests):
936 results = []
937 for request in requests:
938 try:
939 results.append(dict(result=request()))
940 except JSONRPCException as e:
941 results.append(dict(error=e))
942 return results
943 944 def send_cli(self, clicommand=None, *args, **kwargs):
945 """Run limenka-cli command. Deserializes returned string as python object."""
946 pos_args = [arg_to_cli(arg) for arg in args]
947 named_args = [str(key) + "=" + arg_to_cli(value) for (key, value) in kwargs.items()]
948 p_args = [
949 self.binary,
950 f"-datadir={self.datadir}",
951 f"-rpcclienttimeout={int(self.rpc_timeout)}",
952 ] + self.options
953 if named_args:
954 p_args += ["-named"]
955 if clicommand is not None:
956 p_args += [clicommand]
957 p_args += pos_args + named_args
958 self.log.debug("Running limenka-cli {}".format(p_args[2:]))
959 process = subprocess.Popen(p_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
960 cli_stdout, cli_stderr = process.communicate(input=self.input)
961 returncode = process.poll()
962 if returncode:
963 match = re.match(r'error code: ([-0-9]+)\nerror message:\n(.*)', cli_stderr)
964 if match:
965 code, message = match.groups()
966 raise JSONRPCException(dict(code=int(code), message=message))
967 # Ignore cli_stdout, raise with cli_stderr
968 raise subprocess.CalledProcessError(returncode, self.binary, output=cli_stderr)
969 try:
970 if not cli_stdout.strip():
971 return None
972 return json.loads(cli_stdout, parse_float=decimal.Decimal)
973 except (json.JSONDecodeError, decimal.InvalidOperation):
974 return cli_stdout.rstrip("\n")
975 976 class RPCOverloadWrapper():
977 def __init__(self, rpc, cli=False, descriptors=False):
978 self.rpc = rpc
979 self.is_cli = cli
980 self.descriptors = descriptors
981 982 def __getattr__(self, name):
983 return getattr(self.rpc, name)
984 985 def createwallet_passthrough(self, *args, **kwargs):
986 return self.__getattr__("createwallet")(*args, **kwargs)
987 988 def createwallet(self, wallet_name, disable_private_keys=None, blank=None, passphrase='', avoid_reuse=None, descriptors=None, load_on_startup=None, external_signer=None):
989 if descriptors is None:
990 descriptors = self.descriptors
991 return self.__getattr__('createwallet')(wallet_name, disable_private_keys, blank, passphrase, avoid_reuse, descriptors, load_on_startup, external_signer)
992 993 def importprivkey(self, privkey, label=None, rescan=None):
994 wallet_info = self.getwalletinfo()
995 if 'descriptors' not in wallet_info or ('descriptors' in wallet_info and not wallet_info['descriptors']):
996 return self.__getattr__('importprivkey')(privkey, label, rescan)
997 desc = descsum_create('combo(' + privkey + ')')
998 req = [{
999 'desc': desc,
1000 'timestamp': 0 if rescan else 'now',
1001 'label': label if label else ''
1002 }]
1003 import_res = self.importdescriptors(req)
1004 if not import_res[0]['success']:
1005 raise JSONRPCException(import_res[0]['error'])
1006 1007 def addmultisigaddress(self, nrequired, keys, label=None, address_type=None):
1008 wallet_info = self.getwalletinfo()
1009 if 'descriptors' not in wallet_info or ('descriptors' in wallet_info and not wallet_info['descriptors']):
1010 return self.__getattr__('addmultisigaddress')(nrequired, keys, label, address_type)
1011 if isinstance(label, dict):
1012 options = dict(label) # copy, so we can pop and check for emptiness
1013 assert address_type is None
1014 address_type = options.pop('address_type', None)
1015 label = options.pop('label', None)
1016 assert not options
1017 cms = self.createmultisig(nrequired, keys, address_type)
1018 req = [{
1019 'desc': cms['descriptor'],
1020 'timestamp': 0,
1021 'label': label if label else ''
1022 }]
1023 import_res = self.importdescriptors(req)
1024 if not import_res[0]['success']:
1025 raise JSONRPCException(import_res[0]['error'])
1026 return cms
1027 1028 def importpubkey(self, pubkey, label=None, rescan=None):
1029 wallet_info = self.getwalletinfo()
1030 if 'descriptors' not in wallet_info or ('descriptors' in wallet_info and not wallet_info['descriptors']):
1031 return self.__getattr__('importpubkey')(pubkey, label, rescan)
1032 desc = descsum_create('combo(' + pubkey + ')')
1033 req = [{
1034 'desc': desc,
1035 'timestamp': 0 if rescan else 'now',
1036 'label': label if label else ''
1037 }]
1038 import_res = self.importdescriptors(req)
1039 if not import_res[0]['success']:
1040 raise JSONRPCException(import_res[0]['error'])
1041 1042 def _deleted_importaddress(self, address, label=None, rescan=None, p2sh=None):
1043 wallet_info = self.getwalletinfo()
1044 if 'descriptors' not in wallet_info or ('descriptors' in wallet_info and not wallet_info['descriptors']):
1045 return self.__getattr__('importaddress')(address, label, rescan, p2sh)
1046 is_hex = False
1047 try:
1048 int(address ,16)
1049 is_hex = True
1050 desc = descsum_create('raw(' + address + ')')
1051 except Exception:
1052 desc = descsum_create('addr(' + address + ')')
1053 reqs = [{
1054 'desc': desc,
1055 'timestamp': 0 if rescan else 'now',
1056 'label': label if label else ''
1057 }]
1058 if is_hex and p2sh:
1059 reqs.append({
1060 'desc': descsum_create('p2sh(raw(' + address + '))'),
1061 'timestamp': 0 if rescan else 'now',
1062 'label': label if label else ''
1063 })
1064 import_res = self.importdescriptors(reqs)
1065 for res in import_res:
1066 if not res['success']:
1067 raise JSONRPCException(res['error'])
1068