code: def test_operations(
      self,
      request_duration,
      is_check,
      collection_name,
      collection_num,
      db_name,
  ):
      # start the monitor threads to check the milvus ops
      log.info("*********************Test Start**********************")
      log.info(connections.get_connection_addr("default"))
      all_checkers = []
  
      def worker(c_name):
          log.info(f"start checker for collection name: {c_name}")
          op_checker = self.init_health_checkers(collection_name=c_name)
          all_checkers.append(op_checker)
          # insert data in init stage
          try:
              num_entities = op_checker[Op.insert].c_wrap.num_entities
              if num_entities < 200000:
                  nb = 5000
                  num_to_insert = 200000 - num_entities
                  for i in range(num_to_insert // nb):
                      op_checker[Op.insert].insert_data(nb=nb)
              else:
                  log.info(
                      f"collection {c_name} has enough data {num_entities}, skip insert data"
                  )
          except Exception as e:
              log.error(f"insert data error: {e}")
  
      threads = []
      for i in range(collection_num):
          c_name = (
              collection_name
              if collection_name
              else f"DB_{db_name}_Collection_{i}_Checker"
          )
          thread = threading.Thread(target=worker, args=(c_name,))
          threads.append(thread)
          thread.start()
      for thread in threads:
          thread.join()
  
      for checker in all_checkers:
          cc.start_monitor_threads(checker)
  
      log.info("*********************Load Start**********************")
      request_duration = (
          request_duration.replace("h", "*3600+")
          .replace("m", "*60+")
          .replace("s", "")
      )
      if request_duration[-1] == "+":
          request_duration = request_duration[:-1]
      request_duration = eval(request_duration)
      for i in range(10):
          sleep(request_duration // 10)
          for checker in all_checkers:
              for k, v in checker.items():
                  v.check_result()
      try:
          ra = ResultAnalyzer()
          ra.get_stage_success_rate()
          ra.show_result_table()
      except Exception as e:
          log.error(f"get stage success rate error: {e}")
      log.info(
          "*********************Chaos Test Completed**********************"
      )
  
file_path: milvus/tests/python_client/chaos/testcases/test_concurrent_operation_for_multi_tenancy.py
imports: import time
  import pytest
  import threading
  import json
  from time import sleep
  from pymilvus import connections, db
  from chaos.checker import InsertChecker, UpsertChecker, SearchChecker, QueryChecker, DeleteChecker, Op, ResultAnalyzer
  from utils.util_log import test_log
  from chaos import chaos_commons
  from common.common_type import CaseLabel
  from chaos import constants
label_desc: Use image node_id 4 for calling the TestOperations obj's underlying member method code with example usage: obj.test_operations(request_duration, is_check, collection_name, collection_num, db_name) without return types