docstr: Method：
  1. prepare a collection with segments
  2. create rgA
  3. transfer 4 nodes to rgA
  4. load 5 replicas in rgA
  verify load fail with msg
  5. load 3 replica in rgA
  verify load succ
  6. release and reload 4 replica in rgA
  verify load succ and each replica occupies 1 query node in rgA
  7. verify load successfully again with no parameters
docstr_tok: ["Method\uff1a","1.","prepare","a","collection","with","segments","2.","create","rgA","3.","transfer","4","nodes","to","rgA","4.","load","5","replicas","in","rgA","verify","load","fail","with","msg","5.","load","3","replica","in","rgA","verify","load","succ","6.","release","and","reload","4","replica","in","rgA","verify","load","succ","and","each","replica","occupies","1","query","node","in","rgA","7.","verify","load","successfully","again","with","no","parameters"]
code: def test_load_with_replicas_and_nodes_num(self, with_growing):
      """
      Method：
      1. prepare a collection with segments
      2. create rgA
      3. transfer 4 nodes to rgA
      4. load 5 replicas in rgA
      verify load fail with msg
      5. load 3 replica in rgA
      verify load succ
      6. release and reload 4 replica in rgA
      verify load succ and each replica occupies 1 query node in rgA
      7. verify load successfully again with no parameters
      """
      # prepare a collection with segments
      dim = 128
      collection_w = self.init_collection_wrap(shards_num=2)
      insert_ids = []
      nb = 3000
      for i in range(5):
          res, _ = collection_w.insert(
              cf.gen_default_list_data(nb=nb, dim=dim, start=i * nb)
          )
          collection_w.flush()
          insert_ids.extend(res.primary_keys)
      collection_w.create_index(
          ct.default_float_vec_field_name, ct.default_index
      )
  
      # make growing
      if with_growing:
          res, _ = collection_w.insert(
              cf.gen_default_list_data(nb, dim=dim, start=6 * nb)
          )
          insert_ids.extend(res.primary_keys)
  
      # create rgA
      rg_name = cf.gen_unique_str("rg")
      self.init_resource_group(name=rg_name)
      # transfer 4 nodes to rgA
      num_nodes_to_rg = 4
      self.utility_wrap.transfer_node(
          source=ct.default_resource_group_name,
          target=rg_name,
          num_node=num_nodes_to_rg,
      )
      rg_info = {
          "name": rg_name,
          "capacity": num_nodes_to_rg,
          "num_available_node": num_nodes_to_rg,
          "num_loaded_replica": {},
          "num_outgoing_node": {},
          "num_incoming_node": {},
      }
      self.utility_wrap.describe_resource_group(
          name=rg_name,
          check_task=CheckTasks.check_rg_property,
          check_items=rg_info,
      )
  
      # load 5 replicas in rgA and verify error msg
      error = {
          ct.err_code: 999,
          ct.err_msg: "failed to load collection, err=failed to spawn replica for collection[nodes not enough]",
      }
      collection_w.load(
          replica_number=num_nodes_to_rg + 1,
          _resource_groups=[rg_name],
          check_task=ct.CheckTasks.err_res,
          check_items=error,
      )
  
      # load 3 replicas in rgA
      replica_number = 3
      collection_w.load(
          replica_number=replica_number, _resource_groups=[rg_name]
      )
      # verify rg state after loaded
      rg_info = {
          "name": rg_name,
          "capacity": num_nodes_to_rg,
          "num_available_node": num_nodes_to_rg,
          "num_loaded_replica": {collection_w.name: replica_number},
          "num_outgoing_node": {},
          "num_incoming_node": {},
      }
      self.utility_wrap.describe_resource_group(
          name=rg_name,
          check_task=CheckTasks.check_rg_property,
          check_items=rg_info,
      )
      # verify replica state
      replicas = collection_w.get_replicas()
      num_nodes_for_replicas = 0
      assert len(replicas[0].groups) == replica_number
      for rep in replicas[0].groups:
          assert rep.resource_group == rg_name
          assert rep.num_outbound_node == {}
          num_nodes_for_replicas += len(rep.group_nodes)
      assert num_nodes_for_replicas == num_nodes_to_rg
  
      # make growing
      if with_growing:
          res, _ = collection_w.insert(
              cf.gen_default_list_data(nb, dim=dim, start=7 * nb)
          )
          insert_ids.extend(res.primary_keys)
  
      # verify search succ
      nq = 5
      vectors = [
          [random.random() for _ in range(dim)] for _ in range(nq)
      ]
      collection_w.search(
          vectors[:nq],
          ct.default_float_vec_field_name,
          ct.default_search_params,
          ct.default_limit,
          check_task=CheckTasks.check_search_results,
          check_items={
              "nq": nq,
              "ids": insert_ids.copy(),
              "limit": ct.default_limit,
          },
      )
  
      # release
      collection_w.release()
      # verify rg state after release
      rg_info = {
          "name": rg_name,
          "capacity": num_nodes_to_rg,
          "num_available_node": num_nodes_to_rg,
          "num_loaded_replica": {},
          "num_outgoing_node": {},
          "num_incoming_node": {},
      }
      self.utility_wrap.describe_resource_group(
          name=rg_name,
          check_task=CheckTasks.check_rg_property,
          check_items=rg_info,
      )
      # reload 4 replica in rgA
      replica_number = 4
      collection_w.load(
          replica_number=replica_number, _resource_groups=[rg_name]
      )
      # verify rg state after reload
      rg_info = {
          "name": rg_name,
          "capacity": num_nodes_to_rg,
          "num_available_node": num_nodes_to_rg,
          "num_loaded_replica": {collection_w.name: replica_number},
          "num_outgoing_node": {},
          "num_incoming_node": {},
      }
      self.utility_wrap.describe_resource_group(
          name=rg_name,
          check_task=CheckTasks.check_rg_property,
          check_items=rg_info,
      )
  
      # verify replica state
      replicas = collection_w.get_replicas()
      assert len(replicas[0].groups) == replica_number
      for rep in replicas[0].groups:
          assert rep.resource_group == rg_name
          assert rep.num_outbound_node == {}
          assert len(rep.group_nodes) == 1  # one replica for each node
  
      # make growing
      if with_growing:
          res, _ = collection_w.insert(
              cf.gen_default_list_data(nb, dim=dim, start=8 * nb)
          )
          insert_ids.extend(res.primary_keys)
  
      # verify search succ
      collection_w.search(
          vectors[:nq],
          ct.default_float_vec_field_name,
          ct.default_search_params,
          ct.default_limit,
          check_task=CheckTasks.check_search_results,
          check_items={
              "nq": nq,
              "ids": insert_ids.copy(),
              "limit": ct.default_limit,
          },
      )
      # verify load successfully again with no parameters
      collection_w.load(replica_number=replica_number)
      collection_w.search(
          vectors[:nq],
          ct.default_float_vec_field_name,
          ct.default_search_params,
          ct.default_limit,
          check_task=CheckTasks.check_search_results,
          check_items={
              "nq": nq,
              "ids": insert_ids.copy(),
              "limit": ct.default_limit,
          },
      )
  
imports: import pytest
  from base.client_base import TestcaseBase
  from common import common_func
  from common import common_type
  from common.common_type import CaseLabel, CheckTasks
  from utils.util_pymilvus import *
label_desc: Use image node_id 2 for calling the TestResourceGroupMultiNodes obj's underlying member method code with example usage: obj.test_load_with_replicas_and_nodes_num(with_growing) without return types