code: def test_multi_replicas_with_only_one_group_available(
      self,
      chaos_type,
      failed_node_type,
      failed_group_scope,
      is_streaming,
  ):
      # start the monitor threads to check the milvus ops
      log.info(
          "*********************Chaos Test Start**********************"
      )
      log.info("Test config")
      log.info(cc.gen_experiment_config(config_file_name))
      # log.info(f"chaos_yaml: {chaos_yaml}")
      log.info(connections.get_connection_addr("default"))
      if is_streaming is False:
          del self.health_checkers[Op.insert]
      cc.start_monitor_threads(self.health_checkers)
      # get replicas info
      release_name = self.instance_name
      querynode_id_pod_pair = get_querynode_info(release_name)
      log.info(querynode_id_pod_pair)
      group_list = []
      shard_leader_list = []
      replicas_info, _ = self.health_checkers[
          Op.search
      ].c_wrap.get_replicas()
      for g in replicas_info.groups:
          group_list.append(list(g.group_nodes))
          for shard in g.shards:
              shard_leader_list.append(shard.shard_leader)
      # keep only one group in healthy status, other groups will be unhealthy by injecting pod failure chaos,
      # In the effected groups, each group has one pod is in pod failure status
      target_pod_list = []
      target_group = []
      group_list = sorted(group_list, key=lambda x: -len(x))
      if failed_group_scope == "one":
          target_group = random.sample(group_list, 1)
      if failed_group_scope == "except_one":
          target_group = random.sample(group_list, len(group_list) - 1)
      if failed_group_scope == "all":
          target_group = group_list[:]
      for g in target_group:
          target_nodes = []
          if failed_node_type == "shard_leader":
              target_nodes = list(set(g) & set(shard_leader_list))
          if failed_node_type == "non_shard_leader":
              target_nodes = list(set(g) - set(shard_leader_list))
          if len(target_nodes) == 0:
              log.info("there is no node satisfied, chose one randomly")
              target_nodes = [random.choice(g)]
          for target_node in target_nodes:
              pod = querynode_id_pod_pair[target_node]
              target_pod_list.append(pod)
      log.info(f"target_pod_list: {target_pod_list}")
      chaos_config = cc.gen_experiment_config(
          f"{str(Path(__file__).absolute().parent)}/chaos_objects/template/{chaos_type}-by-pod-list.yaml"
      )
      chaos_config["metadata"][
          "name"
      ] = f"test-multi-replicase-{int(time.time())}"
      meta_name = chaos_config.get("metadata", None).get("name", None)
      chaos_config["spec"]["selector"]["pods"][
          "chaos-testing"
      ] = target_pod_list
      self._chaos_config = (
          chaos_config  # cache the chaos config for tear down
      )
  
      log.info(f"chaos_config: {chaos_config}")
      # wait 20s
      sleep(constants.WAIT_PER_OP * 2)
      # replicas info
      replicas_info, _ = self.health_checkers[
          Op.search
      ].c_wrap.get_replicas()
      log.info(
          f"replicas_info for search collection {self.health_checkers[Op.search].c_wrap.name}: {replicas_info}"
      )
  
      # assert statistic:all ops 100% succ
      log.info("******1st assert before chaos: ")
      assert_statistic(self.health_checkers)
      # apply chaos object
      chaos_res = CusResource(
          kind=chaos_config["kind"],
          group=constants.CHAOS_GROUP,
          version=constants.CHAOS_VERSION,
          namespace=constants.CHAOS_NAMESPACE,
      )
  
      chaos_res.create(chaos_config)
      log.info("chaos injected")
      sleep(constants.WAIT_PER_OP * 2)
      # reset counting
      cc.reset_counting(self.health_checkers)
  
      # wait 120s
      sleep(constants.CHAOS_DURATION)
      log.info(f"Alive threads: {threading.enumerate()}")
      # node info
      querynode_id_pod_pair = get_querynode_info(release_name)
      log.info(querynode_id_pod_pair)
      # replicas info
      replicas_info, _ = self.health_checkers[
          Op.search
      ].c_wrap.get_replicas()
      log.info(
          f"replicas_info for search collection {self.health_checkers[Op.search].c_wrap.name}: {replicas_info}"
      )
  
      replicas_info, _ = self.health_checkers[
          Op.query
      ].c_wrap.get_replicas()
      log.info(
          f"replicas_info for query collection {self.health_checkers[Op.query].c_wrap.name}: {replicas_info}"
      )
      # assert statistic
      log.info("******2nd assert after chaos injected: ")
      expectations = {
          Op.search: constants.SUCC,
          Op.query: constants.SUCC,
      }
      if failed_group_scope == "all":
          expectations = {
              Op.search: constants.FAIL,
              Op.query: constants.FAIL,
          }
      assert_statistic(self.health_checkers, expectations=expectations)
      # delete chaos
      chaos_res.delete(meta_name)
      log.info("chaos deleted")
      sleep(2)
      # wait all pods ready
      log.info(
          f"wait for pods in namespace {constants.CHAOS_NAMESPACE} with label app.kubernetes.io/instance={release_name}"
      )
      ready_1 = wait_pods_ready(
          constants.CHAOS_NAMESPACE,
          f"app.kubernetes.io/instance={release_name}",
      )
      log.info(
          f"wait for pods in namespace {constants.CHAOS_NAMESPACE} with label release={release_name}"
      )
      ready_2 = wait_pods_ready(
          constants.CHAOS_NAMESPACE, f"release={release_name}"
      )
      if ready_1 and ready_2:
          log.info("all pods are ready")
      # reconnect if needed
      sleep(constants.WAIT_PER_OP * 2)
      # cc.reconnect(connections, alias='default')
      # reset counting again
      cc.reset_counting(self.health_checkers)
      # wait 50s (varies by feature)
      sleep(constants.WAIT_PER_OP * 5)
      # node info
      querynode_id_pod_pair = get_querynode_info(release_name)
      log.info(querynode_id_pod_pair)
      sleep(30)
      # replicas info
      replicas_info, _ = self.health_checkers[
          Op.search
      ].c_wrap.get_replicas()
      log.info(
          f"replicas_info for collection {self.health_checkers[Op.search].c_wrap.name}: {replicas_info}"
      )
      replicas_info, _ = self.health_checkers[
          Op.query
      ].c_wrap.get_replicas()
      log.info(
          f"replicas_info for collection {self.health_checkers[Op.query].c_wrap.name}: {replicas_info}"
      )
      # assert statistic: all ops success again
      log.info("******3rd assert after chaos deleted: ")
      assert_statistic(self.health_checkers)
      # assert all expectations
      assert_expectations()
  
      log.info(
          "*********************Chaos Test Completed**********************"
      )
  
imports: import threading
  import pytest
  import time
  import random
  from pathlib import Path
  from time import sleep
  from pymilvus import connections
  from chaos.checker import InsertFlushChecker, SearchChecker, QueryChecker, Op
  from common.cus_resource_opts import CustomResourceOperations
  from common.milvus_sys import MilvusSys
  from utils.util_log import test_log
  from utils.util_k8s import wait_pods_ready, get_pod_ip_name_pairs, get_milvus_instance_name
  from chaos import chaos_commons
  from common.common_type import CaseLabel
  from common import common_func
  from chaos import constants
  from delayed_assert import expect, assert_expectations
label_desc: Use image node_id 4 for calling the TestChaos obj's underlying member method code with example usage: obj.test_multi_replicas_with_only_one_group_available(chaos_type, failed_node_type, failed_group_scope, is_streaming) without return types