Inside Bitcoin

501        if self.options.coveragedir is not None:
502            coverage.write_all_rpc_commands(self.options.coveragedir, node._rpc)
503
504    def start_nodes(self, extra_args=None, *args, **kwargs):
505        """Start multiple bitcoinds"""
506
507        if extra_args is None:
508            extra_args = [None] * self.num_nodes
509        assert_equal(len(extra_args), self.num_nodes)
510        for i, node in enumerate(self.nodes):
511            node.start(extra_args[i], *args, **kwargs)
512        for node in self.nodes:
513            node.wait_for_rpc_connection()
514
515        if self.options.coveragedir is not None:
516            for node in self.nodes:
517                coverage.write_all_rpc_commands(self.options.coveragedir, node._rpc)
518
519    def stop_node(self, i, expected_stderr='', wait=0):
520        """Stop a bitcoind test node"""
521        self.nodes[i].stop_node(expected_stderr, wait=wait)
522
523    def stop_nodes(self, wait=0):
524        """Stop multiple bitcoind test nodes"""
525        for node in self.nodes:
526            # Issue RPC to stop nodes
527            node.stop_node(wait=wait, wait_until_stopped=False)
528
529        for node in self.nodes:
530            # Wait for nodes to stop
531            node.wait_until_stopped()
532
533    def cleanup_partially_started_nodes(self):
534        """Tear down nodes left running after a failed start_nodes().
535
536        After start_nodes() raises (e.g. FailedToStartError), some nodes may be
537        RPC-connected, some may have a live process without RPC, and some may
538        already have exited. Stop the connected ones cleanly and force-kill the
539        rest so the framework's teardown can proceed.
540        """
541        for node in self.nodes:
542            if not node.running:
543                continue
544            if node.rpc_connected:
545                node.stop_node(wait=node.rpc_timeout)
546            else:
547                node.process.kill()
548                node.process.wait(timeout=node.rpc_timeout)
549                node.process = None
550                node.stdout.close()
551                node.stderr.close()
552                node.running = False
553
554    def restart_node(self, i, extra_args=None, clear_addrman=False, *, expected_stderr=''):
555        """Stop and start a test node"""
556        self.stop_node(i, expected_stderr=expected_stderr)
557        if clear_addrman:
558            peers_dat = self.nodes[i].chain_path / "peers.dat"
559            os.remove(peers_dat)
560            with self.nodes[i].assert_debug_log(expected_msgs=[f'Creating peers.dat because the file was not found ("{peers_dat}")']):
561                self.start_node(i, extra_args)
562        else:
563            self.start_node(i, extra_args)
564
565    def wait_for_node_exit(self, i, timeout):
566        self.nodes[i].process.wait(timeout)
567
568    def connect_nodes(self, a, b, *, peer_advertises_v2=None):
569        from_connection = self.nodes[a]
570        to_connection = self.nodes[b]
571
572        # Use subversion as peer id. Test nodes have their node number appended to the user agent string
573        from_connection_subver = from_connection.getnetworkinfo()['subversion']
574        to_connection_subver = to_connection.getnetworkinfo()['subversion']
575
576        def find_conn(node, peer_subversion, inbound):
577            return next(filter(lambda peer: peer['subver'] == peer_subversion and peer['inbound'] == inbound, node.getpeerinfo()), None)
578
579        self.wait_until(lambda: not find_conn(from_connection, to_connection_subver, inbound=False))
580        self.wait_until(lambda: not find_conn(to_connection, from_connection_subver, inbound=True))
581
582        ip_port = "127.0.0.1:" + str(p2p_port(b))
583
584        if peer_advertises_v2 is None:
585            peer_advertises_v2 = from_connection.use_v2transport
586
587        if peer_advertises_v2 != from_connection.use_v2transport:
588            from_connection.addnode(node=ip_port, command="onetry", v2transport=peer_advertises_v2)
589        else:
590            # skip the optional third argument if it matches the default, for
591            # compatibility with older clients
592            from_connection.addnode(ip_port, "onetry")
593
594        self.wait_until(lambda: find_conn(from_connection, to_connection_subver, inbound=False) is not None)
595        self.wait_until(lambda: find_conn(to_connection, from_connection_subver, inbound=True) is not None)
596
597        def check_bytesrecv(peer, msg_type, min_bytes_recv):
598            assert peer is not None, "Error: peer disconnected"
599            return peer['bytesrecv_per_msg'].pop(msg_type, 0) >= min_bytes_recv
600
601        # Poll until version handshake (fSuccessfullyConnected) is complete to
602        # avoid race conditions, because some message types are blocked from
603        # being sent or received before fSuccessfullyConnected.
604        #
605        # As the flag fSuccessfullyConnected is not exposed, check it by
606        # waiting for a pong, which can only happen after the flag was set.
607        self.wait_until(lambda: check_bytesrecv(find_conn(from_connection, to_connection_subver, inbound=False), 'pong', 29))
608        self.wait_until(lambda: check_bytesrecv(find_conn(to_connection, from_connection_subver, inbound=True), 'pong', 29))
609
610    def disconnect_nodes(self, a, b):
611        def disconnect_nodes_helper(node_a, node_b):
612            def get_peer_ids(from_connection, node_num):
613                result = []
614                for peer in from_connection.getpeerinfo():
615                    if "testnode{}".format(node_num) in peer['subver']:
616                        result.append(peer['id'])
617                return result
618
619            peer_ids = get_peer_ids(node_a, node_b.index)
620            if not peer_ids:
621                self.log.warning("disconnect_nodes: {} and {} were not connected".format(
622                    node_a.index,
623                    node_b.index,
624                ))
625                return
626            for peer_id in peer_ids:
627                try:
628                    node_a.disconnectnode(nodeid=peer_id)
629                except JSONRPCException as e:
630                    # If this node is disconnected between calculating the peer id
631                    # and issuing the disconnect, don't worry about it.
632                    # This avoids a race condition if we're mass-disconnecting peers.
633                    if e.error['code'] != -29:  # RPC_CLIENT_NODE_NOT_CONNECTED
634                        raise
635
636            # wait to disconnect
637            self.wait_until(lambda: not get_peer_ids(node_a, node_b.index), timeout=5)
638            self.wait_until(lambda: not get_peer_ids(node_b, node_a.index), timeout=5)
639
640        disconnect_nodes_helper(self.nodes[a], self.nodes[b])
641
642    def split_network(self):
643        """
644        Split the network of four nodes into nodes 0/1 and 2/3.
645        """
646        self.disconnect_nodes(1, 2)
647        self.sync_all(self.nodes[:2])
648        self.sync_all(self.nodes[2:])
649
650    def join_network(self):
651        """
652        Join the (previously split) network halves together.
653        """
654        self.connect_nodes(1, 2)
655        self.sync_all()
656
657    def no_op(self):
658        pass
659
660    def generate(self, generator, *args, sync_fun=None, **kwargs):
661        blocks = generator.generate(*args, called_by_framework=True, **kwargs)
662        sync_fun() if sync_fun else self.sync_all()
663        return blocks
664
665    def generateblock(self, generator, *args, sync_fun=None, **kwargs):
666        blocks = generator.generateblock(*args, called_by_framework=True, **kwargs)
667        sync_fun() if sync_fun else self.sync_all()
668        return blocks
669
670    def generatetoaddress(self, generator, *args, sync_fun=None, **kwargs):
671        blocks = generator.generatetoaddress(*args, called_by_framework=True, **kwargs)
672        sync_fun() if sync_fun else self.sync_all()
673        return blocks
674
675    def generatetodescriptor(self, generator, *args, sync_fun=None, **kwargs):
676        blocks = generator.generatetodescriptor(*args, called_by_framework=True, **kwargs)
677        sync_fun() if sync_fun else self.sync_all()
678        return blocks
679
680    def create_outpoints(self, node, *, outputs):
681        """Send funds to a given list of `{address: amount}` targets using the bitcoind
682        wallet and return the corresponding outpoints as a list of dictionaries
683        `[{"txid": txid, "vout": vout1}, {"txid": txid, "vout": vout2}, ...]`.
684        The result can be used to specify inputs for RPCs like `createrawtransaction`,
685        `createpsbt`, `lockunspent` etc."""
686        for output in outputs:
687            assert_equal(len(output.keys()), 1)
688        send_res = node.send(outputs)
689        assert send_res["complete"]
690        utxos = []
691        for output in outputs:
692            address = list(output.keys())[0]
693            vout = find_vout_for_address(node, send_res["txid"], address)
694            utxos.append({"txid": send_res["txid"], "vout": vout})
695        return utxos
696
697    def sync_blocks(self, nodes=None, wait=1, timeout=60):
698        """
699        Wait until everybody has the same tip.
700        sync_blocks needs to be called with an rpc_connections set that has least
701        one node already synced to the latest, stable tip, otherwise there's a
702        chance it might return before all nodes are stably synced.
703        """
704        rpc_connections = nodes or self.nodes
705        timeout = int(timeout * self.options.timeout_factor)
706        stop_time = time.time() + timeout
707        while time.time() <= stop_time:
708            best_hash = [x.getbestblockhash() for x in rpc_connections]
709            if best_hash.count(best_hash[0]) == len(rpc_connections):
710                return
711            # Check that each peer has at least one connection
712            assert (all([len(x.getpeerinfo()) for x in rpc_connections]))
713            time.sleep(wait)
714        raise AssertionError("Block sync timed out after {}s:{}".format(
715            timeout,
716            "".join("\n  {!r}".format(b) for b in best_hash),
717        ))
718
719    def sync_mempools(self, nodes=None, wait=1, timeout=60, flush_scheduler=True):
720        """
721        Wait until everybody has the same transactions in their memory
722        pools
723        """
724        rpc_connections = nodes or self.nodes
725        timeout = int(timeout * self.options.timeout_factor)
726        stop_time = time.time() + timeout
727        while time.time() <= stop_time:
728            pool = [set(r.getrawmempool()) for r in rpc_connections]
729            if pool.count(pool[0]) == len(rpc_connections):
730                if flush_scheduler:
731                    for r in rpc_connections:
732                        r.syncwithvalidationinterfacequeue()
733                return
734            # Check that each peer has at least one connection
735            assert (all([len(x.getpeerinfo()) for x in rpc_connections]))
736            time.sleep(wait)
737        raise AssertionError("Mempool sync timed out after {}s:{}".format(
738            timeout,
739            "".join("\n  {!r}".format(m) for m in pool),
740        ))
741
742    def sync_all(self, nodes=None):
743        self.sync_blocks(nodes)
744        self.sync_mempools(nodes)
745
746    def wait_until(self, test_function, timeout=60, check_interval=0.05):
747        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.options.timeout_factor, check_interval=check_interval)
748
749    def fill_node_addrman(self, *, node_index, address_types_to_add):
750        ADDRESSES = {