docstr: collection schema: [pk, float_vector]
  Steps:
  1. create collection
  2. create index and load collection or not
  3. import data
  4. load collection or not
  5. verify the data entities
  5. verify the index status
  6. verify search and query
code: def test_load_before_or_after_bulk_insert(
      self, loaded_before_bulk_insert, create_index_before_bulk_insert
  ):
      """
      collection schema: [pk, float_vector]
      Steps:
      1. create collection
      2. create index and load collection or not
      3. import data
      4. load collection or not
      5. verify the data entities
      5. verify the index status
      6. verify search and query
      """
      if (
          loaded_before_bulk_insert
          and not create_index_before_bulk_insert
      ):
          pytest.skip("can not load collection if index not created")
      files = prepare_bulk_insert_json_files(
          minio_endpoint=self.minio_endpoint,
          bucket_name=self.bucket_name,
          is_row_based=True,
          rows=500,
          dim=16,
          auto_id=True,
          data_fields=[df.vec_field],
          force=True,
      )
      self._connect()
      c_name = cf.gen_unique_str("bulk_insert")
      fields = [
          cf.gen_int64_field(name=df.pk_field, is_primary=True),
          cf.gen_float_vec_field(name=df.vec_field, dim=16),
      ]
      schema = cf.gen_collection_schema(fields=fields, auto_id=True)
      self.collection_wrap.init_collection(c_name, schema=schema)
      # build index
      if create_index_before_bulk_insert:
          index_params = ct.default_index
          self.collection_wrap.create_index(
              field_name=df.vec_field, index_params=index_params
          )
      if loaded_before_bulk_insert:
          # load collection
          self.collection_wrap.load()
  
      # import data
      t0 = time.time()
      task_id, _ = self.utility_wrap.do_bulk_insert(
          collection_name=c_name, files=files
      )
      logging.info(f"bulk insert task ids:{task_id}")
      (
          success,
          states,
      ) = self.utility_wrap.wait_for_bulk_insert_tasks_completed(
          task_ids=[task_id], timeout=90
      )
      tt = time.time() - t0
      log.info(f"bulk insert state:{success} in {tt}")
      assert success
      if not create_index_before_bulk_insert:
          # build index
          index_params = ct.default_index
          self.collection_wrap.create_index(
              field_name=df.vec_field, index_params=index_params
          )
      if not loaded_before_bulk_insert:
          # load collection
          self.collection_wrap.load()
  
      num_entities = self.collection_wrap.num_entities
      log.info(f"collection entities: {num_entities}")
      assert num_entities == 500
      # verify no index
      success = self.utility_wrap.wait_index_build_completed(c_name)
      assert success
      # verify search and query
      log.info(f"wait for load finished and be ready for search")
      self.collection_wrap.load(_refresh=True)
      nq = 3
      topk = 10
      search_data = cf.gen_vectors(nq, 16)
      search_params = ct.default_search_params
      res, _ = self.collection_wrap.search(
          search_data,
          df.vec_field,
          param=search_params,
          limit=topk,
          check_task=CheckTasks.check_search_results,
          check_items={"nq": nq, "limit": topk},
      )
      for hits in res:
          ids = hits.ids
          expr = f"{df.pk_field} in {ids}"
          expr = expr.replace("'", '"')
          results, _ = self.collection_wrap.query(expr=expr)
          assert len(results) == len(ids)
  
imports: import logging
  import time
  import pytest
  import random
  import numpy
  from pathlib import Path
  from base.client_base import TestcaseBase
  from common import common_func
  from common import common_type
  from common.milvus_sys import MilvusSys
  from common.common_type import CaseLabel, CheckTasks
  from utils.util_k8s import get_pod_ip_name_pairs, get_milvus_instance_name, get_milvus_deploy_tool
  from utils.util_log import test_log
  from common.bulk_insert_data import prepare_bulk_insert_json_files, prepare_bulk_insert_numpy_files, DataField, DataErrorType
label_desc: Use image node_id 7 for calling the TestBulkInsert obj's underlying member method code with example usage: obj.test_load_before_or_after_bulk_insert(loaded_before_bulk_insert, create_index_before_bulk_insert) without return types