text: # Concord
  #
  # Copyright (c) 2019 VMware, Inc. All Rights Reserved.
  #
  # This product is licensed to you under the Apache 2.0 license (the "License").
  # You may not use this product except in compliance with the Apache 2.0 License.
  #
  # This product may include a number of subcomponents with separate copyright
  # notices and license terms. Your use of these subcomponents is subject to the
  # terms and conditions of the subcomponent's license, as noted in the LICENSE
  # file.
  
  # Add the pyclient directory to $PYTHONPATH
  
  import sys
  
  import os
  import os.path
  import shutil
  import random
  import subprocess
  from collections import namedtuple, Counter
  import tempfile
  from functools import wraps
  from datetime import datetime
  from functools import partial
  import inspect
  import time
  
  import trio
  
  sys.path.append(os.path.abspath("../../util/pyclient"))
  
  import bft_config
  import bft_client
  import bft_metrics_client
  from enum import Enum
  from util import bft_metrics, eliot_logging as log
  from util.eliot_logging import log_call
  from util import skvbc as kvbc
  from util.bft_test_exceptions import AlreadyRunningError, AlreadyStoppedError, KeyExchangeError, CreError
  DB_FILE_PREFIX = "simpleKVBTests_DB_"
  
  TestConfig = namedtuple('TestConfig', [
      'n',
      'f',
      'c',
      'num_clients',
      'key_file_prefix',
      'start_replica_cmd',
      'stop_replica_cmd',
      'num_ro_replicas'
  ])
  
  class ConsensusPathPrevalentResult(Enum):
     OK = 0
     TOO_FEW_REQUESTS_ON_EXPECTED_PATH = 1
     TOO_MANY_REQUESTS_ON_UNEXPECTED_PATH = 2
  
  KEY_FILE_PREFIX = "replica_keys_"
  # bft_client.py  implement 2 types of clients currently:
  # UdpClient for UDP communication and TcpTlsClient for for TCP/TLS communication
  BFT_CLIENT_TYPE = bft_client.TcpTlsClient if os.environ.get('BUILD_COMM_TCP_TLS', "").lower() == "true" \
                                            else bft_client.UdpClient
  
  # For better performance, we Would like to keep the next constant as minimal as possible.
  # If you need more clients, increase with caution.
  # Reserved clients (RESERVED_CLIENTS_QUOTA) are not part of NUM_CLIENTS
  RESERVED_CLIENTS_QUOTA = 2
  BFT_CONFIGS_NUM_CLIENTS = 10
  NUM_PARTICIPANTS = 5
  
  @log_call(action_type="Test_Configs", include_args=[])
  def interesting_configs(selected=None):
      if selected is None:
          selected=lambda *config: True
  
      bft_configs = [{'n': 6, 'f': 1, 'c': 1, 'num_clients': BFT_CONFIGS_NUM_CLIENTS},
                     {'n': 7, 'f': 2, 'c': 0, 'num_clients': BFT_CONFIGS_NUM_CLIENTS},
                     # {'n': 4, 'f': 1, 'c': 0, 'num_clients': BFT_CONFIGS_NUM_CLIENTS},
                     # {'n': 9, 'f': 2, 'c': 1, 'num_clients': BFT_CONFIGS_NUM_CLIENTS}
                     # {'n': 12, 'f': 3, 'c': 1, 'num_clients': BFT_CONFIGS_NUM_CLIENTS}
                     ]
  
      selected_bft_configs = \
          [conf for conf in bft_configs
           if selected(conf['n'], conf['f'], conf['c'])]
  
      assert len(selected_bft_configs) > 0, "No eligible BFT configs"
  
      for config in selected_bft_configs:
          assert config['n'] == 3 * config['f'] + 2 * config['c'] + 1, \
              "Invariant breached. Expected: n = 3f + 2c + 1"
  
      return selected_bft_configs
  
  def with_trio(async_fn):
      """ Decorator for running a coroutine (async_fn) with trio. """
      @wraps(async_fn)
      def trio_wrapper(*args, **kwargs):
          if "already_in_trio" in kwargs:
              kwargs.pop("already_in_trio")
              return async_fn(*args, **kwargs)
          else:
              return trio.run(async_fn, *args, **kwargs)
  
      return trio_wrapper
  
  def skip_for_tls(async_fn):
      """ Decorator for skipping the test for TCP/TLS. """
      @wraps(async_fn)
      def wrapper(*args, **kwargs):
          if os.environ.get('BUILD_COMM_TCP_TLS', "").lower() != "true":
              return async_fn(*args, **kwargs)
          else:
              pass
  
      return wrapper
  
  def with_constant_load(async_fn):
      """
      Runs the decorated async function in parallel with constant load,
      generated by a dedicated (reserved) client. In order to work
      the wrapper assumes a 'bft_network' will be created and passed
      as a keyword argument. The wrapper also creates a 'skvbc' object
      and a 'nursery' which it passes to the wrapped function.
      """
      @wraps(async_fn)
      async def wrapper(*args, **kwargs):
          if "bft_network" in kwargs:
              bft_network = kwargs.pop("bft_network")
              skvbc = kvbc.SimpleKVBCProtocol(bft_network)
              client = bft_network.new_reserved_client()
  
              async def background_sender(arg1, arg2, *, task_status=trio.TASK_STATUS_IGNORED):
                  with trio.CancelScope() as scope:
                      task_status.started(scope)
                      await skvbc.send_indefinite_write_requests(arg1, arg2)
  
              async with trio.open_nursery() as nursery:
                  constant_load = await nursery.start(background_sender, client, 1)  # send a request every second
                  await async_fn(*args, **kwargs, bft_network=bft_network, skvbc=skvbc, constant_load=constant_load)
                  nursery.cancel_scope.cancel()
      return wrapper
  
  def with_bft_network(start_replica_cmd, selected_configs=None, num_clients=None, num_ro_replicas=0, rotate_keys=False, bft_configs=None, with_cre=False):
      """
      Runs the decorated async function for all selected BFT configs
      start_replica_cmd is a callback which is used to start a replica. It should have the following
      signature:
          def start_replica_cmd(builddir, replica_id)
      or
          def start_replica_cmd(builddir, replica_id, config)
      If you want the bft test network configuration to be passed to your callback you should add
      third parameter named 'config' (the exact name is important!).
      If you don't need this configuration - use two parameters callback with any names you want.
      """
      def decorator(async_fn):
          @wraps(async_fn)
          async def wrapper(*args, **kwargs):
              if "bft_network" in kwargs:
                  bft_network = kwargs.pop("bft_network")
                  bft_network.is_existing = True
                  with log.start_task(action_type=async_fn.__name__):
                      await async_fn(*args, **kwargs, bft_network=bft_network)
              else:
                  configs = bft_configs if bft_configs is not None else interesting_configs(selected_configs)
                  for bft_config in configs:
  
                      config = TestConfig(n=bft_config['n'],
                                          f=bft_config['f'],
                                          c=bft_config['c'],
                                          num_clients=bft_config['num_clients'] \
                                              if num_clients is None \
                                              else num_clients,
                                          key_file_prefix=KEY_FILE_PREFIX,
                                          start_replica_cmd=start_replica_cmd,
                                          stop_replica_cmd=None,
                                          num_ro_replicas=num_ro_replicas)
                      async with trio.open_nursery() as background_nursery:
                          with BftTestNetwork.new(config, background_nursery, with_cre=with_cre) as bft_network:
                              bft_network.current_test = async_fn.__name__ + "_n=" + str(bft_config['n']) \
                                                                           + "_f=" + str(bft_config['f']) \
                                                                           + "_c=" + str(bft_config['c'])
                              with log.start_task(action_type=f"{bft_network.current_test}_num_clients={config.num_clients}"):
                                  if rotate_keys:
                                      await bft_network.check_initital_key_exchange()
                                  bft_network.test_start_time = time.time()
                                  await async_fn(*args, **kwargs, bft_network=bft_network)
          return wrapper
  
      return decorator
  
  MAX_MSG_SIZE = 64*1024 # 64k
  REQ_TIMEOUT_MILLI = 5000
  RETRY_TIMEOUT_MILLI = 250
  METRICS_TIMEOUT_SEC = 5
  
  # TODO: This is not generic, but is required for use by SimpleKVBC. In the
  # future we will likely want to change how we determine the lengths of keys and
  # values, make them parameterizable, or generate keys in the protocols rather
  # than tester. For now, all keys and values must be 21 bytes.
  KV_LEN = 21
  
  class BftTestNetwork:
      """Encapsulates a BFT network instance for testing purposes"""
  
      def __enter__(self):
          """context manager method for 'with' statements"""
          return self
  
      def __exit__(self, etype, value, tb):
          """context manager method for 'with' statements"""
          if not self.is_existing:
              for client in self.clients.values():
                  client.__exit__()
              for client in self.reserved_clients.values():
                  client.__exit__()
              if self.with_cre and self.cre_pid is not None:
                  if os.environ.get('GRACEFUL_SHUTDOWN', "").lower() in set(["true", "on"]):
                      self.cre_pid.terminate()
                  else:
                      self.cre_pid.kill()
                  for fd in self.cre_fds:
                      fd.close()
                  self.cre_pid.wait()
              self.metrics.__exit__()
              self.stop_all_replicas()
              os.chdir(self.origdir)
              # shutil.rmtree(self.testdir, ignore_errors=True)
              shutil.rmtree(self.certdir, ignore_errors=True)
              shutil.rmtree(self.txn_signing_keys_base_path, ignore_errors=True)
              if self.test_dir and self.test_start_time:
                  with open(f"{self.test_dir}test_duration.log", 'w+') as log_file:
                      log_file.write(f"test duration = {time.time() - self.test_start_time} seconds\n")
              if self.perf_proc:
                  self.perf_proc.wait()
  
              self.check_error_logs()
  
      def __init__(self, is_existing, origdir, config, testdir, certdir, builddir, toolsdir,
                   procs, replicas, clients, metrics, client_factory, background_nursery, ro_replicas=[]):
          self.is_existing = is_existing
          self.origdir = origdir
          self.config = config
          self.testdir = testdir
          self.certdir = certdir
          self.builddir = builddir
          self.toolsdir = toolsdir
          self.procs = procs
          self.replicas = replicas
          self.clients = clients
          self.metrics = metrics
          self.reserved_clients = {}
          self.reserved_client_ids_in_use = []
          if client_factory:
              self.client_factory = client_factory
          else:
              self.client_factory = partial(self._create_new_client, BFT_CLIENT_TYPE)
          self.open_fds = {}
          self.current_test = ""
          self.background_nursery = background_nursery
          self.test_dir = None
          self.test_start_time = None
          self.perf_proc = None
          self.ro_replicas = ro_replicas
          self.txn_signing_enabled = (os.environ.get('TXN_SIGNING_ENABLED', "").lower() in ["true", "on"])
          self.with_cre = False
          self.cre_pid = None
          self.cre_fds = None
          self.cre_id = self.config.n + self.config.num_ro_replicas + BFT_CONFIGS_NUM_CLIENTS + RESERVED_CLIENTS_QUOTA
          # Setup transaction signing parameters
          self.setup_txn_signing()
          self._generate_operator_keys()
  
      @classmethod
      def new(cls, config, background_nursery, client_factory=None, with_cre=False):
          builddir = os.path.abspath("../../build")
          toolsdir = os.path.join(builddir, "tools")
          testdir = tempfile.mkdtemp()
          certdir = tempfile.mkdtemp()
          bft_network = cls(
              is_existing=False,
              origdir=os.getcwd(),
              config=config,
              testdir=testdir,
              certdir=certdir,
              builddir=builddir,
              toolsdir=toolsdir,
              procs={},
              replicas=[bft_config.Replica(i, "127.0.0.1",
                                           bft_config.bft_msg_port_from_node_id(i), bft_config.metrics_port_from_node_id(i))
                        for i in range(0, config.n + config.num_ro_replicas)],
              clients = {},
              metrics = None,
              client_factory = client_factory,
              background_nursery = background_nursery,
              ro_replicas=[bft_config.Replica(i, "127.0.0.1",
                                              bft_config.bft_msg_port_from_node_id(i), bft_config.metrics_port_from_node_id(i))
                           for i in range(config.n, config.n + config.num_ro_replicas)]
          )
          bft_network.with_cre = with_cre
          # Copy logging.properties file
          shutil.copy(os.path.abspath("../simpleKVBC/scripts/logging.properties"), testdir)
  
          log.log_message(message_type=f"Running test in {bft_network.testdir}")
  
          os.chdir(bft_network.testdir)
          bft_network._generate_crypto_keys()
          if bft_network.comm_type() == bft_config.COMM_TYPE_TCP_TLS:
              generate_cre = 0 if bft_network.with_cre is False else 1
              # Generate certificates for all replicas, clients, and reserved clients
              bft_network.generate_tls_certs(bft_network.num_total_replicas() + config.num_clients + RESERVED_CLIENTS_QUOTA + generate_cre)
  
          bft_network._init_metrics()
          bft_network._create_clients()
          bft_network._create_reserved_clients()
  
          return bft_network
  
      @classmethod
      def existing(cls, config, replicas, clients, client_factory=None, background_nursery=None):
          certdir = None
          if not client_factory:
              certdir = tempfile.mkdtemp()
              assert background_nursery is not None, "You must transfer a background nursery which lasts for all test duration!"
          bft_network = cls(
              is_existing=True,
              origdir=None,
              config=config,
              testdir=None,
              certdir=certdir,
              builddir=None,
              toolsdir=None,
              procs={r.id: r for r in replicas},
              replicas=replicas,
              clients={i: clients[i] for i in range(len(clients))},
              metrics=None,
              client_factory=client_factory,
              background_nursery = background_nursery
          )
  
          bft_network._init_metrics()
          bft_network._create_reserved_clients()
          return bft_network
  
      def start_cre(self):
          """
          Start the cre if it isn't already started.
          Otherwise raise an AlreadyStoppedError.
          """
          with log.start_action(action_type="start_cre"):
              stdout_file = None
              stderr_file = None
  
              if os.environ.get('KEEP_APOLLO_LOGS', "").lower() in ["true", "on"]:
                  test_name = os.environ.get('TEST_NAME')
  
                  if not test_name:
                      now = datetime.now().strftime("%y-%m-%d_%H:%M:%S")
                      test_name = f"{now}_{self.current_test}"
  
                  self.test_dir = f"{self.builddir}/tests/apollo/logs/{test_name}/{self.current_test}/"
                  test_log = f"{self.test_dir}stdout_cre.log"
  
                  os.makedirs(self.test_dir, exist_ok=True)
  
                  stdout_file = open(test_log, 'w+')
                  stderr_file = open(test_log, 'w+')
  
                  stdout_file.write("############################################\n")
                  stdout_file.flush()
                  stderr_file.write("############################################\n")
                  stderr_file.flush()
  
                  self.cre_fds = (stdout_file, stderr_file)
                  cre_exe = os.path.join(self.builddir, "tests", "simpleKVBC", "TesterCRE", "skvbc_cre")
                  cre_cmd = [cre_exe,
                             "-i", str(self.cre_id),
                             "-f", str(self.config.f),
                             "-c", str(self.config.c),
                             "-r", str(self.config.n),
                             "-k", self.certdir + "/" + str(self.cre_id),
                             "-t", os.path.join(self.txn_signing_keys_base_path, "transaction_signing_keys", str(self.principals_to_participant_map[self.cre_id]), "transaction_signing_priv.pem"),
                             "-o", "1000"]
                  self.cre_pid = subprocess.Popen(
                      cre_cmd,
                      stdout=stdout_file,
                      stderr=stderr_file,
                      close_fds=True)
  
      def stop_cre(self):
          with log.start_action(action_type="stop_cre"):
              p = self.cre_pid
              if os.environ.get('GRACEFUL_SHUTDOWN', "").lower() in set(["true", "on"]):
                  p.terminate()
              else:
                  p.kill()
              for fd in self.cre_fds:
                  fd.close()
              p.wait()
  
      def transfer_db_files(self, source, dests):
          with log.start_action(action_type="transfer db files"):
              source_db_dir = os.path.join(self.testdir, DB_FILE_PREFIX + str(source))
              for r in dests:
                  dest_db_dir = os.path.join(self.testdir, DB_FILE_PREFIX + str(r))
                  res = subprocess.run(['cp', '-rf', source_db_dir, dest_db_dir])
                  log.log_message(message_type=f"copy db files from {source_db_dir} to {dest_db_dir}, result is {res.returncode}")
  
      async def change_configuration(self, config, generate_tls=False):
          self.config = config
          self.replicas = [bft_config.Replica(i, "127.0.0.1",
                                              bft_config.bft_msg_port_from_node_id(i), bft_config.metrics_port_from_node_id(i))
                           for i in range(0, config.n + config.num_ro_replicas)]
  
          self._generate_crypto_keys()
  
          if generate_tls and self.comm_type() == bft_config.COMM_TYPE_TCP_TLS:
              generate_cre = 0 if self.with_cre is False else 1
              # Generate certificates for replicas, clients, and reserved clients
              self.generate_tls_certs(self.num_total_replicas() + config.num_clients + RESERVED_CLIENTS_QUOTA + generate_cre)
  
  
      def restart_clients(self, generate_tx_signing_keys=True, restart_replicas=True):
          # remove all existing clients
          for client in self.clients.values():
              client.__exit__()
          for client in self.reserved_clients.values():
              client.__exit__()
          self.reserved_client_ids_in_use = []
          self.metrics.__exit__()
          self.clients = {}
  
          if generate_tx_signing_keys:
              # remove existing transaction signing keys and generate again
              shutil.rmtree(self.txn_signing_keys_base_path, ignore_errors=True)
              self.setup_txn_signing()
              if restart_replicas:
                  # Now, we must restart the replicas
                  self.stop_all_replicas()
                  self.start_all_replicas()
  
          self._init_metrics()
          self._create_clients()
          self._create_reserved_clients()
  
      def _generate_crypto_keys(self):
          keygen = os.path.join(self.toolsdir, "GenerateConcordKeys")
          args = [keygen, "-n", str(self.config.n), "-f", str(self.config.f)]
          if self.config.num_ro_replicas > 0:
              args.extend(["-r", str(self.config.num_ro_replicas)])
          args.extend(["-o", self.config.key_file_prefix])
          subprocess.run(args, check=True)
  
      def _generate_operator_keys(self):
          if self.builddir is None:
              return 
          with open(self.builddir + "/operator_pub.pem", 'w') as f:
              f.write("-----BEGIN PUBLIC KEY-----\n"
                      "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAENEMHcbJgnnYxfa1zDlIF7lzp/Ioa"
                      "NfwGuJpAg84an5FdPALwZwBp/m/X3d8kwmfZEytqt2PGMNhHMkovIaRI1A==\n"
                      "-----END PUBLIC KEY-----")
          with open(self.builddir + "/operator_priv.pem", 'w') as f:
              f.write("-----BEGIN EC PRIVATE KEY-----\n"
                      "MHcCAQEEIEWf8ZTkCWbdA9WrMSNGCC7GQxvSXiDlU6dlZAi6JaCboAoGCCqGSM49"
                      "AwEHoUQDQgAENEMHcbJgnnYxfa1zDlIF7lzp/IoaNfwGuJpAg84an5FdPALwZwBp"
                      "/m/X3d8kwmfZEytqt2PGMNhHMkovIaRI1A==\n"
                      "-----END EC PRIVATE KEY-----")
  
      def generate_tls_certs(self, num_to_generate, start_index=0):
          """
          Generate 'num_to_generate' certificates and private keys. The certificates are generated on a given range of
          node IDs folders [start_index, start_index+num_to_generate-1] into an output certificate root folder
          self.certdir.
          Since every node might be a potential client and/or server, each node certificate folder consists of the
          subfolders 'client' and 'server'. Each subfolder holds an X.509 certificate client.cert or server.cert and the
          matching private key pk.pem file (PEM format). All certificates are generated using bash script
          create_tls_certs.sh.
          The output is generated into the test folder to a folder called 'certs'.
          """
          certs_gen_script_path = os.path.join(self.builddir, "tests/simpleTest/scripts/create_tls_certs.sh")
          # If not running TLS, just exit here. keep certs_path to pass it by default to any type of replicas
          # We want to save time in non-TLs runs, avoiding certificate generation
          temp_cert_dir = self.certdir + "/tmp"
          os.makedirs(temp_cert_dir, exist_ok=True)
          args = [certs_gen_script_path, str(num_to_generate), temp_cert_dir, str(start_index)]
          subprocess.run(args, check=True, stdout=subprocess.DEVNULL)
          for c in range(num_to_generate):
              comp_cert_dir = self.certdir + "/" + str(c)
              shutil.rmtree(comp_cert_dir, ignore_errors=True)
              shutil.copytree(temp_cert_dir, comp_cert_dir)
          shutil.rmtree(temp_cert_dir, ignore_errors=True)
  
      def copy_certs_from_server_to_clients(self, src):
          src_cert = self.certdir + "/" + str(src)
          for c in range(self.num_total_replicas() , self.num_total_replicas() + self.config.num_clients + RESERVED_CLIENTS_QUOTA):
              comp_cert_dir = self.certdir + "/" + str(c)
              shutil.rmtree(comp_cert_dir, ignore_errors=True)
              shutil.copytree(src_cert, comp_cert_dir)
  
      def _create_clients(self):
          start_id = self.config.n + self.config.num_ro_replicas
          for client_id in range(start_id, start_id + self.config.num_clients):
              self.clients[client_id] = self.client_factory(client_id)
  
      def _create_new_client(self, client_class, client_id):
          config = self._bft_config(client_id)
          ro_replicas = [r.id for r in self.ro_replicas]
          return client_class(config, self.replicas, self.background_nursery, ro_replicas=ro_replicas)
  
      def _create_reserved_clients(self):
          first_id = self.num_total_replicas() + self.config.num_clients
          for reserved_client_id in range(first_id, first_id + RESERVED_CLIENTS_QUOTA):
              self.reserved_clients[reserved_client_id] = self.client_factory(reserved_client_id)
  
      def new_reserved_client(self):
          if len(self.reserved_client_ids_in_use) == RESERVED_CLIENTS_QUOTA:
              raise  NotImplemented("You must increase RESERVED_CLIENTS_QUOTA, see comment above")
          start_id = self.num_total_replicas() + self.config.num_clients
          reserved_client_ids = [ id for id in range(start_id, start_id + RESERVED_CLIENTS_QUOTA) ]
          free_reserved_client_ids = set(reserved_client_ids) - set(self.reserved_client_ids_in_use)
          reserved_client_id = next(iter(free_reserved_client_ids))
          self.reserved_client_ids_in_use.append(reserved_client_id)
          return self.reserved_clients[reserved_client_id]
  
      def _bft_config(self, client_id):
          return bft_config.Config(client_id,
                                   self.config.f,
                                   self.config.c,
                                   MAX_MSG_SIZE,
                                   REQ_TIMEOUT_MILLI,
                                   RETRY_TIMEOUT_MILLI,
                                   self.certdir + "/" + str(client_id),
                                   self.txn_signing_keys_base_path,
                                   self.principals_to_participant_map)
  
      def _init_metrics(self):
          metric_clients = {}
          for r in self.replicas:
              metric_clients[r.id] = bft_metrics_client.MetricsClient(r)
          self.metrics = bft_metrics.BftMetrics(metric_clients)
  
      def random_client(self, without=None):
          if without == None:
              without = set()
  
          return random.choice(list(set(self.clients.values()) - without))
  
      def random_clients(self, max_clients):
          return set(random.choices(list(self.clients.values()), k=max_clients))
  
      def get_all_clients(self):
          return self.clients.values()
  
      def setup_txn_signing(self):
          self.txn_signing_keys_base_path = ""
          self.principals_mapping = ""
          self.principals_to_participant_map = {}
          if self.txn_signing_enabled:
              self.txn_signing_keys_base_path = tempfile.mkdtemp()
              self.principals_mapping, self.principals_to_participant_map = self.create_principals_mapping()
              self.generate_txn_signing_keys(self.txn_signing_keys_base_path)
              
      def generate_txn_signing_keys(self, keys_path):
          """ Generates num_participants number of key pairs """
          script_path = "/concord-bft/scripts/linux/create_concord_clients_transaction_signing_keys.sh"
          args = [script_path, "-n", str(self.num_participants), "-o", keys_path]
          subprocess.run(args, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
  
      def create_principals_mapping(self):
          """
          If client principal ids range from 11-20, for example, this method splits them into groups based on 
          NUM_PARTICIPANTS and self.num_participants.
          Client ids in each group will be space separated, and each group will be semicolon separated. 
          E.g. "11 12;13 14;15 16;17 18;19 20" for 10 client ids divided into 5 participants.
          If there are reserved clients, they are added at the end of the group in round robin manner.
          Thus, if there are 2 reserved client, with ids 21 and 22, the final string would look like:
          "11 12 21;13 14 22;15 16;17 18,19 20".
          This method also returns a principals to participants map, with the key being the principal id
          of the client or reserved client, and the value being the participant it belongs to (numbered 1 onwards).
          If number of clients is not default, but modified from outside, there might be cases where self.num_participants
          will be less than NUM_PARTICIPANTS.
          """
          def split(a, n):
              """ Splits list 'a' into n chunks """
              k, m = divmod(len(a), n)
              return [a[i * k + min(i, m):(i + 1) * k + min(i + 1, m)] for i in range(n)]
  
          start_id = self.config.n + self.config.num_ro_replicas
          client_ids = range(start_id, start_id + self.config.num_clients)
          start_id = self.num_total_replicas() + self.config.num_clients
          reserved_client_ids = range(start_id, start_id + RESERVED_CLIENTS_QUOTA + 1)
  
          principals = ""
          client_ids = sorted(client_ids)
          reserved_client_ids = sorted(reserved_client_ids)
          combined_clients = client_ids + reserved_client_ids
          combined_clients_set = set(combined_clients)
          assert len(combined_clients_set) == len(combined_clients), "Client Ids and Reserved Client Ids must all be unique ids"
          self.num_participants = min(NUM_PARTICIPANTS, len(client_ids))
          client_ids_chunks = split(client_ids, self.num_participants)
          reserved_client_ids_chunks = split(reserved_client_ids, self.num_participants)
          principals_to_participant_map = {}
  
          # iterate number of participants
          for i in range(self.num_participants):
              # add client_ids to principals
              for cid in client_ids_chunks[i]:
                  principals = principals + str(cid) + " "
                  principals_to_participant_map[cid] = i+1
              # add reserved_client_ids to principals
              for rcid in reserved_client_ids_chunks[i]:
                  principals = principals + str(rcid) + " "
                  principals_to_participant_map[rcid] = i+1
              # remove last space
              if principals[-1] == ' ':
                  principals = principals[:-1]
              # add , to separate next set of client_ids
              principals = principals + ";"
  
          # remove last semicolon
          if principals[-1] == ';':
              principals = principals[:-1]
          return principals, principals_to_participant_map
  
      def start_replica_cmd(self, replica_id):
          """
          Returns command line to start replica with the given id
          If the callback accepts three parameters and one of them is named 'config' - pass the network configuration too.
          Append the SSL certificate path. This is needed only for TLS communication.
          """
          with log.start_action(action_type="start_replica_cmd", replica=replica_id):
              start_replica_fn_args = inspect.getfullargspec(self.config.start_replica_cmd).args
              if "config" in start_replica_fn_args and len(start_replica_fn_args) == 3:
                  cmd = self.config.start_replica_cmd(self.builddir, replica_id, self.config)
              else:
                  cmd = self.config.start_replica_cmd(self.builddir, replica_id)
              if self.certdir:
                  cmd.append("-c")
                  cmd.append(self.certdir + "/" + str(replica_id))
              if self.txn_signing_enabled:
                  cmd.append("-p")
                  cmd.append(self.principals_mapping)
                  keys_path = os.path.join(self.txn_signing_keys_base_path, "transaction_signing_keys")
                  cmd.append("-t")
                  cmd.append(keys_path)
              return cmd
  
      def check_error_logs(self):
          """
          Checking FATAL logs in replica logs and writing in a file.
          If the file exists for any testcase, warning is displayed in CI
          """
          with log.start_action(action_type="replica_log_scanner") as action:
              cmd = ['grep', '-R', 'FATAL', '--exclude=ReplicaErrorLogs.txt']
              file_path = f"{self.test_dir}ReplicaErrorLogs.txt"
  
              with open(file_path, 'w+') as outfile:
                  subprocess.run(cmd, stdout=outfile, cwd=self.test_dir)
  
              if os.path.isfile(file_path):
                  if  os.stat(file_path).st_size > 0:
                      action.log(message_type=f'Error in Replica logs')
                  else:
                      os.remove(file_path)
  
      def stop_replica_cmd(self, replica_id):
          """
          Returns command line to stop a replica with the given id
          """
          with log.start_action(action_type="stop_replica_cmd", replica=replica_id):
              return self.config.stop_replica_cmd(replica_id)
  
      def start_all_replicas(self):
          with log.start_action(action_type="start_all_replicas"):
              all_replicas = self.all_replicas()
              for i in all_replicas:
                  try:
                      self.start_replica(i)
                  except AlreadyRunningError:
                      if not self.is_existing:
                          raise
  
              assert len(self.procs) == self.config.n
  
      def stop_all_replicas(self):
          """ Stop all running replicas"""
          [ self.stop_replica(i) for i in self.get_live_replicas() ]
          assert len(self.procs) == 0
  
      def start_replicas(self, replicas):
          """
          Start from list "replicas"
          """
          [ self.start_replica(r) for r in replicas ]
  
      def stop_replicas(self, replicas):
          """
          Start from list "replicas"
          """
          for r in replicas:
              self.stop_replica(r)
  
      def start_replica(self, replica_id):
          """
          Start a replica if it isn't already started.
          Otherwise raise an AlreadyStoppedError.
          """
          with log.start_action(action_type="start_replica", replica=replica_id):
              stdout_file = None
              stderr_file = None
  
              if os.environ.get('KEEP_APOLLO_LOGS', "").lower() in ["true", "on"]:
                  test_name = os.environ.get('TEST_NAME')
  
                  if not test_name:
                      now = datetime.now().strftime("%y-%m-%d_%H:%M:%S")
                      test_name = f"{now}_{self.current_test}"
  
                  self.test_dir = f"{self.builddir}/tests/apollo/logs/{test_name}/{self.current_test}/"
                  test_log = f"{self.test_dir}stdout_{replica_id}.log"
  
                  os.makedirs(self.test_dir, exist_ok=True)
  
                  stdout_file = open(test_log, 'w+')
                  stderr_file = open(test_log, 'w+')
  
                  stdout_file.write("############################################\n")
                  stdout_file.flush()
                  stderr_file.write("############################################\n")
                  stderr_file.flush()
  
                  self.open_fds[replica_id] = (stdout_file, stderr_file)
  
              """
              If the apollo test's CODECOVERAGE flag is set, all raw profile files will be created
              in coverage dir; otherwise, ignore it.
              """
              if os.environ.get('CODECOVERAGE', "").lower() in ["true", "on"] and os.environ.get('GRACEFUL_SHUTDOWN', "").lower() in ["true", "on"]:
                  self.coverage_dir = f"{self.builddir}/tests/apollo/codecoverage/{test_name}/{self.current_test}/"
                  os.makedirs(self.coverage_dir, exist_ok=True)
                  profraw_file_name = f"coverage_{replica_id}_%9m.profraw"
                  profraw_file_path = os.path.join(
                      self.coverage_dir, profraw_file_name)
                  os.environ['LLVM_PROFILE_FILE'] = profraw_file_path
  
              if replica_id in self.procs:
                  raise AlreadyRunningError(replica_id)
  
              if self.is_existing and self.config.stop_replica_cmd is not None:
                  self.procs[replica_id] = self._start_external_replica(replica_id)
              else:
                  self.procs[replica_id] = subprocess.Popen(
                                              self.start_replica_cmd(replica_id),
                                              stdout=stdout_file,
                                              stderr=stderr_file,
                                              close_fds=True)
  
              replica_for_perf = os.environ.get('PERF_REPLICA', "")
              if self.test_start_time and replica_for_perf and int(replica_for_perf) == replica_id:
                  perf_samples = os.environ.get('PERF_SAMPLES', "1000")
                  perf_cmd = ["perf", "record", "-F", perf_samples, "-p", f"{self.procs[replica_id].pid}",
                              "-a", "-g", "-o", f"{self.test_dir}perf.data"]
                  self.perf_proc = subprocess.Popen(perf_cmd, close_fds=True)
  
      def replica_id_from_pid(self, pid):
          """ Return an already-started replica id, according to a given pid """
          for rep_id in self.procs.keys():
              if self.procs[rep_id].pid == pid:
                  return rep_id
          return None
  
      def is_replica_id(self, node_id):
          """ Return True if a node_id is a replica id (includes a read-only replica) """
          return node_id < self.num_total_replicas()
  
      def is_client_id(self, node_id):
          """ Return True if node_id is a client Id - reserved or non-reserved """
          start_id = self.num_total_replicas()
          return start_id <= node_id < (start_id + self.config.num_clients + RESERVED_CLIENTS_QUOTA)
  
      def is_read_only_replica_id(self, node_id):
          """ Return True if node_id is a read-only replica id """
          return self.config.n <= node_id < (self.config.n + self.config.num_ro_replicas)
  
      def is_reserved_client_id(self, node_id):
          """ Return True is node_id is a reserved-client Id """
          start_id = self.num_total_replicas() + self.config.num_clients
          return start_id <= node_id < (start_id + RESERVED_CLIENTS_QUOTA)
  
      def comm_type(self):
          """
          Returns a string representing the communication type.
          Raise NotImplementedError if communication type is not implemented
          """
          if bft_client.TcpTlsClient == BFT_CLIENT_TYPE:
              return bft_config.COMM_TYPE_TCP_TLS
          if bft_client.UdpClient == BFT_CLIENT_TYPE:
              return bft_config.COMM_TYPE_UDP
          raise NotImplementedError(f"{type(self.clients[self.config.n])} is not supported!")
  
      def num_total_replicas(self):
          return self.config.n + self.config.num_ro_replicas
  
      def num_total_clients(self):
          return self.config.num_clients + RESERVED_CLIENTS_QUOTA
  
      def node_id_from_bft_msg_port(self, bft_msg_port):
          assert ((bft_msg_port % 2 == 0) and
                  (bft_msg_port < bft_config.START_METRICS_PORT) and
                  (bft_msg_port >= bft_config.START_BFT_MSG_PORT))
          return (bft_msg_port - bft_config.START_BFT_MSG_PORT) // 2
  
      def _start_external_replica(self, replica_id):
          with log.start_action(action_type="_start_external_replica"):
              subprocess.run(
                  self.start_replica_cmd(replica_id),
                  check=True
              )
  
              return self.replicas[replica_id]
  
      def stop_replica(self, replica_id, force_kill=False):
          """
          Stop a replica if it is running.
          Otherwise raise an AlreadyStoppedError.
          """
          with log.start_action(action_type="stop_replica", replica=replica_id):
              if replica_id not in self.procs.keys():
                  raise AlreadyStoppedError(replica_id)
  
              if self.is_existing and self.config.stop_replica_cmd is not None:
                  self._stop_external_replica(replica_id)
              else:
                  p = self.procs[replica_id]
                  if force_kill:
                      p.kill()
                  else:
                      if os.environ.get('GRACEFUL_SHUTDOWN', "").lower() in set(["true", "on"]):
                          p.terminate()
                      else:
                          p.kill()
                  for fd in self.open_fds.get(replica_id, ()):
                      fd.close()
                  p.wait()
  
              del self.procs[replica_id]
  
      def _stop_external_replica(self, replica_id):
          with log.start_action(action_type="_stop_external_replica", replica=replica_id):
              subprocess.run(
                  self.stop_replica_cmd(replica_id),
                  check=True
              )
  
      def all_replicas(self, without=None):
          """
          Returns a list of all ACTIVE replica IDs excluding the "without" set, and without RO replicas
          """
          if without == None:
              without = set()
          return list(set(range(0, self.config.n)) - without)
  
      def all_client_ids(self, without=None, with_reserved_clients=True):
          """
          Returns a list of all client IDs, excluding the "without" set
          """
          if without == None:
              without = set()
          num_total_clients = self.config.num_clients
          num_total_replicas = self.num_total_replicas()
          if with_reserved_clients:
              num_total_clients += RESERVED_CLIENTS_QUOTA
          return list(set(range(num_total_replicas, num_total_replicas + num_total_clients)) - without)
  
      def random_set_of_replicas(self, size, without=None):
          """ Returns a random list of ACTIVE replica IDs excluding the "without" set, and without any RO replicas """
          if without is None:
              without = set()
          random_replicas = set()
          for _ in range(size):
              exclude_replicas = random_replicas | without
              random_replicas.add(random.choice(self.all_replicas(without=exclude_replicas)))
          return random_replicas
  
      def get_live_replicas(self):
          """
          Returns the id-s of all live replicas
          """
          return list(self.procs.keys())
  
      def get_client(self, id):
          if self.is_client_id(id):
              return self.reserved_clients[id] if self.is_reserved_client_id(id) else self.clients[id]
          return None
  
      async def get_current_primary(self):
          """
          Returns the current primary replica id
          """
          with log.start_action(action_type="get_current_primary") as action:
              current_primary = await self.get_current_view() % self.config.n
              action.add_success_fields(current_primary=current_primary)
              return current_primary
  
      async def get_current_view(self):
          """
          Returns the current view number
          """
          with log.start_action(action_type="get_current_view") as action:
              matching_view = None
              nb_replicas_in_matching_view = 0
  
              async def get_view(replica_id):
                  def expected(_): return True
                  replica_view = await self._wait_for_matching_agreed_view(replica_id, expected)
                  replica_views.append(replica_view)
  
              try:
                  with trio.fail_after(seconds=30):
                      while True:
                          replica_views = []
                          async with trio.open_nursery() as nursery:
                              for r in self.get_live_replicas():
                                  nursery.start_soon(get_view, r)
                          view = Counter(replica_views).most_common(1)[0]
                          # wait for n-f = 2f+2c+1 replicas to have agreed the expected view
                          if view[1] >= 2 * self.config.f + 2 * self.config.c + 1:
                              matching_view = view[0]
                              nb_replicas_in_matching_view = view[1]
                              break
                          await trio.sleep(0.1)
  
                      action.log(
                          message_type=f'Matching view #{matching_view} has been agreed among replicas.')
  
                      action.log(f'View #{matching_view} is active on '
                              f'{nb_replicas_in_matching_view} replicas '
                              f'({nb_replicas_in_matching_view} >= n-f = {self.config.n - self.config.f}).')
                      action.add_success_fields(current_view=matching_view)
  
                      return matching_view
  
              except trio.TooSlowError:
                  assert False, "Could not agree view among replicas."
  
      async def get_metric(self, replica_id, bft_network, mtype, mname, component='replica'):
          with trio.fail_after(seconds=30):
              while True:
                  with trio.move_on_after(seconds=1):
                      try:
                          key = [component, mtype, mname]
                          value = await bft_network.metrics.get(replica_id, *key)
                      except KeyError:
                          # metrics not yet available, continue looping
                          log.log_message(message_type=f"KeyError! '{mname}' not yet available.")
                          await trio.sleep(0.1)
                      else:
                          return value
  
      async def wait_for_view(self, replica_id, expected=None,
                              err_msg="Expected view not reached"):
          """
          Waits for a view that matches the "expected" predicate,
          and returns the corresponding view number.
  
          If the "expected" predicate is not provided,
          returns the current view number.
  
          In case of a timeout, fails with the provided err_msg
          """
          with log.start_action(action_type="wait_for_view", replica=replica_id) as action:
              if expected is None:
                  expected = lambda _: True
  
              matching_view = None
              nb_replicas_in_matching_view = 0
              try:
                  matching_view = await self._wait_for_matching_agreed_view(replica_id, expected)
                  action.log(message_type=f'Matching view #{matching_view} has been agreed among replicas.')
  
                  nb_replicas_in_matching_view = await self._wait_for_active_view(matching_view)
                  action.log(f'View #{matching_view} is active on '
                        f'{nb_replicas_in_matching_view} replicas '
                        f'({nb_replicas_in_matching_view} >= n-f = {self.config.n - self.config.f}).')
  
                  return matching_view
              except trio.TooSlowError:
                  assert False, err_msg + \
                                f'(matchingView={matching_view} ' \
                                f'replicasInMatchingView={nb_replicas_in_matching_view})'
  
      async def _wait_for_matching_agreed_view(self, replica_id, expected):
          """
          Wait for the last agreed view to match the "expected" predicate
          """
          with log.start_action(action_type="_wait_for_matching_agreed_view", replica=replica_id) as action:
              last_agreed_view = None
  
              async def the_last_agreed_view():
                  key = ['replica', 'Gauges', 'lastAgreedView']
                  view = await self.retrieve_metric(replica_id, *key)
  
                  if view is not None and expected(view):
                      return view
                  
              last_agreed_view = await self.wait_for(the_last_agreed_view, 45, 1)
              action.add_success_fields(last_agreed_view=last_agreed_view)
              return last_agreed_view
  
      async def _wait_for_active_view(self, view):
          """
          Wait for a view to become active on enough (n-f) replicas
          """
          with log.start_action(action_type="_wait_for_active_view", view=view):
              with trio.fail_after(seconds=45):
                  while True:
                      nb_replicas_in_view = await self._count_replicas_in_view(view)
  
                      # wait for n-f = 2f+2c+1 replicas to be in the expected view
                      if nb_replicas_in_view >= 2 * self.config.f + 2 * self.config.c + 1:
                          break
                      await trio.sleep(0.1)
              return nb_replicas_in_view
  
      async def _count_replicas_in_view(self, view):
          """
          Count the number of replicas that have activated a given view
          """
          with log.start_action(action_type="_count_replicas_in_view", view=view):
              nb_replicas_in_view = 0
  
              async def count_if_replica_in_view(r, expected_view):
                  """
                  A closure that allows concurrent counting of replicas
                  that have activated a given view.
                  """
                  with log.start_action(action_type="count_if_replica_in_view"):
                      nonlocal nb_replicas_in_view
  
                      key = ['replica', 'Gauges', 'currentActiveView']
  
                      with trio.move_on_after(seconds=5):
                          while True:
                              with trio.move_on_after(seconds=1):
                                  try:
                                      replica_view = await self.metrics.get(r, *key)
                                      if replica_view == expected_view:
                                          nb_replicas_in_view += 1
                                  except KeyError:
                                      # metrics not yet available, continue looping
                                      await trio.sleep(0.25)
                                      continue
                                  else:
                                      break
  
          async with trio.open_nursery() as nursery:
              for r in self.get_live_replicas():
                  nursery.start_soon(
                      count_if_replica_in_view, r, view)
          return nb_replicas_in_view
  
      async def force_quorum_including_replica(self, replica_id):
          """
          Bring down a sufficient number of replicas (excluding the primary),
          so that the remaining replicas form a quorum that includes replica_id
          """
          with log.start_action(action_type="force_quorum_including_replica", replica=replica_id) as action:
              assert len(self.procs) >= 2 * self.config.f + self.config.c + 1
              primary = await self.get_current_primary()
              self.stop_replicas(self.random_set_of_replicas(
                  len(self.procs) - (2 * self.config.f + self.config.c + 1), without={primary, replica_id}))
  
      async def wait_for_fetching_state(self, replica_id):
          """
          Check metrics on fetching replica to see if the replica is in a
          fetching state
  
          Returns the current source replica for state transfer.
          """
          with log.start_action(action_type="wait_for_fetching_state", replica=replica_id) as action:
              async def replica_to_be_in_fetching_state():
                  is_fetching = await self.is_fetching(replica_id)
                  source_replica_id = await self.source_replica(replica_id)
                  if is_fetching:
                      action.add_success_fields(source_replica_id=source_replica_id)
                      return source_replica_id
                  
              return await self.wait_for(replica_to_be_in_fetching_state, 10, .5)
  
      async def is_fetching(self, replica_id):
          """Return whether the current replica is fetching state"""
          key = ['bc_state_transfer', 'Statuses', 'fetching_state']
          state = await self.metrics.get(replica_id, *key)
          return state != "NotFetching"
  
      async def source_replica(self, replica_id):
          """Return whether the current replica has a source replica already set"""
          with log.start_action(action_type="source_replica", replica=replica_id):
              key = ['bc_state_transfer', 'Gauges', 'current_source_replica']
              source_replica_id = await self.metrics.get(replica_id, *key)
              return source_replica_id
  
      async def wait_for_state_transfer_to_start(self):
          """
          Retry checking every .5 seconds until state transfer starts at least one
          node. Stop trying, and fail the test after 30 seconds.
          """
          with log.start_action(action_type="wait_for_state_transfer_to_start"):
              with trio.fail_after(30): # seconds
                  async with trio.open_nursery() as nursery:
                      for replica in self.replicas:
                          nursery.start_soon(self._wait_to_receive_st_msgs,
                                             replica,
                                             nursery.cancel_scope)
  
      async def wait_for_replicas_to_collect_stable_checkpoint(self, replicas, checkpoint, timeout=30):
          with log.start_action(action_type="wait_for_replicas_to_collect_stable_checkpoint", replicas=replicas) as action:
              with trio.fail_after(seconds=timeout):
                  last_stable_seqs = []
                  while True:
                      for replica_id in replicas:
                          last_stable = await self.get_metric(replica_id, self, 'Gauges', "lastStableSeqNum")
                          last_stable_seqs.append(last_stable)
                          action.log(message_type="lastStableSeqNum", replica=replica_id, last_stable=last_stable)
                      if sum(x == 150 * checkpoint for x in last_stable_seqs) == len(replicas):
                          break
                      else:
                          last_stable_seqs.clear()
                          await trio.sleep(seconds=0.1)
  
      async def _wait_to_receive_st_msgs(self, replica, cancel_scope):
          """
          Check metrics to see if state transfer started. If so cancel the
          concurrent coroutines in the request scope.
          """
          with log.start_action(action_type="_wait_to_receive_st_msgs", replica=replica.id) as action:
              while True:
                  with trio.move_on_after(.5): # seconds
                      try:
                          key = ['replica', 'Counters', 'receivedStateTransferMsgs']
                          n = await self.metrics.get(replica.id, *key)
                          if n > 0:
                              action.log(message_type="State transfer has started. Cancelling concurrent coroutines", receivedStateTransferMsgs=n)
                              cancel_scope.cancel()
                      except KeyError:
                          pass # metrics not yet available, continue looping
                      await trio.sleep(0.1)
  
      async def wait_for_state_transfer_to_stop(self, up_to_date_node, stale_node, stop_on_stable_seq_num=False):
          with log.start_action(action_type="wait_for_state_transfer_to_stop", up_to_date_node=up_to_date_node, stale_node=stale_node, stop_on_stable_seq_num=stop_on_stable_seq_num):
              with trio.fail_after(30): # seconds
                  # Get the lastExecutedSeqNumber from a started node
                  if stop_on_stable_seq_num:
                      key = ['replica', 'Gauges', 'lastStableSeqNum']
                  else:
                      key = ['replica', 'Gauges', 'lastExecutedSeqNum']
                  expected_seq_num = await self.metrics.get(up_to_date_node, *key)
                  with log.start_action(action_type='start_polling', key=key[2], expected_seq_num=expected_seq_num) as action:
                      last_n = -1
                      while True:
                          with trio.move_on_after(.5): # seconds
                              metrics = await self.metrics.get_all(stale_node)
                              try:
                                  n = self.metrics.get_local(metrics, *key)
                              except KeyError:
                                  # ignore - the metric will eventually become available
                                  await trio.sleep(0.1)
                              else:
                                  # Debugging
                                  if n != last_n:
                                      last_n = n
                                      last_stored_checkpoint = self.metrics.get_local(metrics,
                                          'bc_state_transfer', 'Gauges', 'last_stored_checkpoint')
                                      on_transferring_complete = self.metrics.get_local(metrics,
                                          'bc_state_transfer', 'Counters', 'on_transferring_complete')
                                      action.log(message_type="Not complete yet",
                                          seq_num=n, last_stored_checkpoint=last_stored_checkpoint, on_transferring_complete=on_transferring_complete)
  
                                  # Exit condition
                                  if n >= expected_seq_num:
                                      action.add_success_fields(n=n, expected_seq_num=expected_seq_num)
                                      return
  
                                  await trio.sleep(0.5)
                                                 
      async def wait_for_replicas_to_checkpoint(self, replica_ids, expected_checkpoint_num=None):
          """
          Wait for every replica in `replicas` to take a checkpoint.
          Check every .5 seconds and give fail after 30 seconds.
          """
          with log.start_action(action_type="wait_for_replicas_to_checkpoint", replica_ids=replica_ids, expected_checkpoint_num=expected_checkpoint_num):
              with trio.fail_after(30): # seconds
                  async with trio.open_nursery() as nursery:
                      for replica_id in replica_ids:
                          nursery.start_soon(self.wait_for_checkpoint, replica_id, expected_checkpoint_num)
  
      async def wait_for_checkpoint(self, replica_id, expected_checkpoint_num=None):
          """
          Wait for a single replica to reach the expected_checkpoint_num.
          If none is provided, return the last stored checkpoint.
          """
          with log.start_action(action_type="wait_for_checkpoint", replica=replica_id, expected_checkpoint_num=expected_checkpoint_num) as action:
              if expected_checkpoint_num is None:
                  expected_checkpoint_num = lambda _: True
  
              async def expected_checkpoint_to_be_reached():
                  key = ['bc_state_transfer', 'Gauges', 'last_stored_checkpoint']
  
                  last_stored_checkpoint = await self.retrieve_metric(replica_id, *key)
                  action.log(message_type=f'[checkpoint] #{last_stored_checkpoint}, replica=#{replica_id}')
      
                  if last_stored_checkpoint is not None and expected_checkpoint_num(last_stored_checkpoint):
                      action.add_success_fields(last_stored_checkpoint=last_stored_checkpoint)
                      return last_stored_checkpoint
                  
          return await self.wait_for(expected_checkpoint_to_be_reached, 30, .5)
  
      async def wait_for_fast_path_to_be_prevalent(self, run_ops, threshold, replica_id=0):
          await self._wait_for_consensus_path_to_be_prevalent(
              fast=True, run_ops=run_ops, threshold=threshold, replica_id=replica_id)
  
      async def wait_for_slow_path_to_be_prevalent(self, run_ops, threshold, replica_id=0):
          await self._wait_for_consensus_path_to_be_prevalent(
              fast=False, run_ops=run_ops, threshold=threshold, replica_id=replica_id)
  
      async def _wait_for_consensus_path_to_be_prevalent(self, fast, run_ops, threshold, replica_id=0, timeout=90):
          """
          Waits until at least threshold operations are being executed in the selected path.
            run_ops: lambda that executes skvbc.run_concurrent_ops or creates requests in some other way
            threshold: minimum number of requests that have to be executed in the correct path
          run_ops should produce at least threshold executions
          """
          with log.start_action(action_type="wait_for_%s_path_to_be_prevalent" % ("fast" if fast else "slow")):
              with trio.fail_after(seconds=timeout):
                  done = False
                  while not done:
                      #initial state to ensure requests are sent
                      res = ConsensusPathPrevalentResult.TOO_MANY_REQUESTS_ON_UNEXPECTED_PATH
                      nb_fast_paths_to_ignore = 0
                      nb_slow_paths_to_ignore = 0
                      with trio.move_on_after(seconds=15):  #retry timeout for not enough right path requests
                          while not done:
                              if res == ConsensusPathPrevalentResult.TOO_MANY_REQUESTS_ON_UNEXPECTED_PATH:
                                  # we need to reset the counters
                                  log.log_message("run_ops")
                                  nb_fast_paths_to_ignore = await self.num_of_fast_path_requests(replica_id)
                                  nb_slow_paths_to_ignore = await self.num_of_slow_path_requests(replica_id)
                                  if run_ops:
                                      await run_ops()
                              res = await self._consensus_path_prevalent(
                                      fast,
                                      nb_fast_paths_to_ignore + ((threshold - 1) if fast else 0),
                                      nb_slow_paths_to_ignore + ((threshold - 1) if not fast else 0),
                                      replica_id)
                              if res == ConsensusPathPrevalentResult.OK:
                                  # the selected path is prevalent - done.
                                  log.log_message("done")
                                  done = True
                                  break
                              if res == ConsensusPathPrevalentResult.TOO_FEW_REQUESTS_ON_EXPECTED_PATH:
                                  # continue polling for enough right path requests
                                  await trio.sleep(seconds=.5)
                                  continue
                              if res == ConsensusPathPrevalentResult.TOO_MANY_REQUESTS_ON_UNEXPECTED_PATH:
                                  # wait a bit before resending new requests
                                  await trio.sleep(seconds=5)
  
      async def wait_for_last_executed_seq_num(self, replica_id=0, expected=0):
          with log.start_action(action_type="wait_for_last_executed_seq_num"):
              async def expected_seq_num_to_be_reached():
                  key = ['replica', 'Gauges', 'lastExecutedSeqNum']
                  last_executed_seq_num = await self.retrieve_metric(replica_id, *key)
  
                  if last_executed_seq_num is not None and last_executed_seq_num >= expected:
                      return last_executed_seq_num
              
              return await self.wait_for(expected_seq_num_to_be_reached, 30, .5)
  
      async def assert_state_transfer_not_started_all_up_nodes(self, up_replica_ids):
          with log.start_action(action_type="assert_state_transfer_not_started_all_up_nodes"):
              with trio.fail_after(METRICS_TIMEOUT_SEC):
                  # Check metrics for all started nodes in parallel
                  async with trio.open_nursery() as nursery:
                      up_replicas = [self.replicas[i] for i in up_replica_ids]
                      for r in up_replicas:
                          nursery.start_soon(self._assert_state_transfer_not_started,
                                             r)
  
      async def assert_fast_path_prevalent(self, nb_fast_paths_to_ignore=0, nb_slow_paths_to_ignore=0, replica_id=0):
          res = await self._consensus_path_prevalent(
              fast=True,
              nb_fast_paths_to_ignore=nb_fast_paths_to_ignore,
              nb_slow_paths_to_ignore=nb_slow_paths_to_ignore,
              replica_id=replica_id)
          assert res == ConsensusPathPrevalentResult.OK, "Fast path is not prevalent"
  
      async def assert_slow_path_prevalent(self, nb_fast_paths_to_ignore=0, nb_slow_paths_to_ignore=0, replica_id=0):
          res = await self._consensus_path_prevalent(
              fast=False,
              nb_fast_paths_to_ignore=nb_fast_paths_to_ignore,
              nb_slow_paths_to_ignore=nb_slow_paths_to_ignore,
              replica_id=replica_id)
          assert res == ConsensusPathPrevalentResult.OK, "Slow path is not prevalent"
  
      async def _consensus_path_prevalent(
              self, fast, nb_fast_paths_to_ignore=0, nb_slow_paths_to_ignore=0, replica_id=0):
          """
          Asserts all executed requests after the ignored number have been processed on the fast or slow path
          depending on the fast parameter value being True/False
          """
          with log.start_action(action_type="assert_%s_path_prevalent" % ("fast" if fast else "slow")):
              total_nb_fast_paths = await self.num_of_fast_path_requests(replica_id)
              total_nb_slow_paths = await self.num_of_slow_path_requests(replica_id)
  
              log.log_message("assert on %s path | slow %d/%d fast %d/%d" % ("fast" if fast else "slow",
                  total_nb_slow_paths, nb_slow_paths_to_ignore, total_nb_fast_paths, nb_fast_paths_to_ignore))
  
              if fast:
                  if total_nb_slow_paths > nb_slow_paths_to_ignore:
                      return ConsensusPathPrevalentResult.TOO_MANY_REQUESTS_ON_UNEXPECTED_PATH  # we can't have any slow
                  if total_nb_fast_paths <= nb_fast_paths_to_ignore:
                      return ConsensusPathPrevalentResult.TOO_FEW_REQUESTS_ON_EXPECTED_PATH  # we don't have enough fast
  
                  return ConsensusPathPrevalentResult.OK
              else:
                  if total_nb_fast_paths > nb_fast_paths_to_ignore:
                      return ConsensusPathPrevalentResult.TOO_MANY_REQUESTS_ON_UNEXPECTED_PATH  # we can't have any fast
                  if total_nb_slow_paths <= nb_slow_paths_to_ignore:
                      return ConsensusPathPrevalentResult.TOO_FEW_REQUESTS_ON_EXPECTED_PATH  # we don't have enough slow
  
                  return ConsensusPathPrevalentResult.OK
  
      async def _assert_state_transfer_not_started(self, replica):
          key = ['replica', 'Counters', 'receivedStateTransferMsgs']
          n = await self.metrics.get(replica.id, *key)
          assert n == 0
  
      async def wait_for(self, task, timeout, interval):
          """
          Wait for the given async task function to return a value. Give up
          waiting for the task's completion after interval (seconds) and retry
          until timeout (seconds) expires. Raise trio.TooSlowError when timeout expires.
  
          Important:
           * The given task function must be async, otherwise a TypeError exception will be thrown.
           * Retries may occur more frequently than interval if the task
             returns None before interval expires. This only matters in that it
             uses more CPU.
          """
          with log.start_action(action_type="wait_for"):
              with trio.fail_after(timeout):
                  while True:
                      with trio.move_on_after(interval):
                          if inspect.iscoroutinefunction(task):
                              result = await task()
                              if result is not None:
                                  return result
                              else:
                                  await trio.sleep(0.1)
                          else:
                              raise TypeError
  
      async def retrieve_metric(self, replica_id, component_name, type_, key):
          try:
              return await self.metrics.get(replica_id, component_name, type_, key)
          except KeyError:
              return None    
  
      async def num_of_fast_path_requests(self, replica_id=0):
          """
          Returns the total number of requests processed on the fast commit path
          """
          with log.start_action(action_type="num_of_fast_path_requests"):
              async def the_number_of_fast_path_requests():
                  metric_key = ['replica', 'Counters', 'totalFastPathRequests']
                  nb_fast_path = await self.retrieve_metric(replica_id, *metric_key)
                  return nb_fast_path
  
          return await self.wait_for(the_number_of_fast_path_requests, 5, 2)
  
      async def num_of_slow_path_requests(self, replica_id=0):
          """
          Returns the total number of requests processed on the slow commit path
          """
          with log.start_action(action_type="num_of_slow_path_requests"):
              async def the_number_of_slow_path_requests():
                  metric_key = ['replica', 'Counters', 'totalSlowPathRequests']
                  nb_slow_path = await self.retrieve_metric(replica_id, *metric_key)
                  return nb_slow_path
              
          return await self.wait_for(the_number_of_slow_path_requests, 5, .5)
          
      async def check_initital_key_exchange(self, stop_replicas=True, full_key_exchange=False, replicas_to_start=[]):
          """
          Performs initial key exchange, starts all replicas, validate the exchange and stops all replicas.
          The stop is done in order for a test who uses this functionality, to proceed without imposing n up replicas.
          """
          with log.start_action(action_type="check_initital_key_exchange"):
              required_exchanges = self.config.n - 1 if full_key_exchange else 2 * self.config.f + self.config.c
              replicas_to_start = [r for r in range(self.config.n)] if replicas_to_start == [] else replicas_to_start
              self.start_replicas(replicas_to_start)
              num_of_exchanged_replicas = 0
              with trio.fail_after(seconds=120):
                  for replica_id in replicas_to_start:
                      while True:
                          with trio.move_on_after(seconds=1):
                              try:
                                  sent_key_exchange_on_start_status = await self.metrics.get(replica_id, *["KeyExchangeManager", "Statuses", "sent_key_exchange_on_start"])
                                  sent_key_exchange_counter = await self.metrics.get(replica_id, *["KeyExchangeManager", "Counters", "sent_key_exchange"])
                                  self_key_exchange_counter = await self.metrics.get(replica_id, *["KeyExchangeManager", "Counters", "self_key_exchange"])
                                  public_key_exchange_for_peer_counter = await self.metrics.get(replica_id, *["KeyExchangeManager", "Counters", "public_key_exchange_for_peer"])
                                  if  sent_key_exchange_on_start_status == 'False' or \
                                      sent_key_exchange_counter < 1 or \
                                      self_key_exchange_counter < 1 or \
                                      public_key_exchange_for_peer_counter < required_exchanges :
                                      await trio.sleep(0.1)
                                      continue
                              except trio.TooSlowError:
                                  print(f"Replica {replica_id} was not able to exchange keys on start")
                                  raise KeyExchangeError
                              else:
                                  assert sent_key_exchange_on_start_status == 'True'
                                  assert sent_key_exchange_counter >= 1
                                  assert self_key_exchange_counter >= 1
                                  assert public_key_exchange_for_peer_counter >= required_exchanges
                                  num_of_exchanged_replicas += 1
                                  break
                      if num_of_exchanged_replicas >= len(replicas_to_start):
                          break
              with trio.fail_after(seconds=5):
                  lastExecutedKey = ['replica', 'Gauges', 'lastExecutedSeqNum']
                  rep = random.choice(replicas_to_start)
                  lastExecutedVal = await self.metrics.get(rep, *lastExecutedKey)
              if stop_replicas:
                  self.stop_replicas(replicas_to_start)
              return lastExecutedVal
  
      async def assert_successful_pre_executions_count(self, replica_id, num_requests):
          try:
              pre_proc_req = 0
              total_pre_exec_requests_executed = 0
              with trio.fail_after(10):
                  while pre_proc_req < num_requests or \
                          total_pre_exec_requests_executed < num_requests:
                      key1 = ["preProcessor", "Counters", "preProcReqCompleted"]
                      pre_proc_req = await self.metrics.get(replica_id, *key1) - self.initial_preexec_sent
                      key2 = ["replica", "Counters", "totalPreExecRequestsExecuted"]
                      total_pre_exec_requests_executed = await self.metrics.get(replica_id, *key2) - self.initial_preexec_executed
                      await trio.sleep(0.1)
          except trio.TooSlowError:
              assert False, "Preprocessor requests " + \
                        f'(expected={num_requests} ' \
                        f'executed={total_pre_exec_requests_executed} ' \
                        f'preexecsent={pre_proc_req})'
          except AttributeError:
              assert False, "Preprocessor counter not initialized"
  
      async def init_preexec_count(self, replica_id=0):
          # read initial preexecutions count, so we can adjust expectations for individual tests
          key1 = ["preProcessor", "Counters", "preProcReqCompleted"]
          self.initial_preexec_sent = await self.metrics.get(replica_id, *key1)
          key2 = ["replica", "Counters", "totalPreExecRequestsExecuted"]
          self.initial_preexec_executed = await self.metrics.get(replica_id, *key2)
  
      async def wait_for_replica_to_ask_for_view_change(self, replica_id, previous_asks_to_leave_view_msg_count = 0):
          """
          Wait for a single replica to send a ReplicaAsksToLeaveViewMsg.
          """
          with log.start_action(action_type="wait_for_replica_asks_to_leave_view_msg", replica=replica_id,
                                previous_asks_to_leave_view_msg_count=previous_asks_to_leave_view_msg_count) as action:
          
              async def the_replica_to_ask_for_view_change(): 
                  key = ['replica', 'Gauges', 'sentReplicaAsksToLeaveViewMsg']
                  replica_asks_to_leave_view_msg_count = await self.retrieve_metric(replica_id, *key)
  
                  if replica_asks_to_leave_view_msg_count is not None and replica_asks_to_leave_view_msg_count > previous_asks_to_leave_view_msg_count:
                                  action.add_success_fields(
                                      replica_asks_to_leave_view_msg_count=replica_asks_to_leave_view_msg_count)
                                  return replica_asks_to_leave_view_msg_count
  
          return await self.wait_for(the_replica_to_ask_for_view_change, 60, .5)
  
      async def wait_for_replicas_to_reach_view(self, replicas_ids, expected_view):
          """
          Wait for all replicas to reach the expected view.
          """
          with log.start_action(action_type="wait_for_replicas_to_reach_view", replicas_ids=replicas_ids, expected_view=expected_view) as action:
              with trio.fail_after(seconds=30):
                  async with trio.open_nursery() as nursery:
                      for replica_id in replicas_ids:
                          nursery.start_soon(self.wait_for_replica_to_reach_view, replica_id, expected_view)
              
      async def wait_for_replica_to_reach_view(self, replica_id, expected_view):
          """
          Wait for a single replica to reach the expected view.
          """
          with log.start_action(action_type="wait_for_replica_to_reach_view", replica=replica_id, expected_view=expected_view) as action:
              async def expected_view_to_be_reached():
                  key = ['replica', 'Gauges', 'view']        
                  replica_view = await self.retrieve_metric(replica_id, *key)
                  
                  if replica_view is not None and replica_view == expected_view:
                      return replica_view
          
          return await self.wait_for(expected_view_to_be_reached, 30, .5)
                  
  