code: def test_operations(self, request_duration, is_check):
      # start the monitor threads to check the milvus ops
      log.info("*********************Test Start**********************")
      log.info(connections.get_connection_addr("default"))
      event_records = EventRecords()
      c_name = None
      event_records.insert("init_health_checkers", "start")
      self.init_health_checkers(collection_name=c_name)
      event_records.insert("init_health_checkers", "finished")
      tasks = cc.start_monitor_threads(self.health_checkers)
      log.info("*********************Load Start**********************")
      # wait request_duration
      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)
          # add an event so that the chaos can start to apply
          if i == 3:
              event_records.insert("init_chaos", "ready")
          for k, v in self.health_checkers.items():
              v.check_result()
      if is_check:
          assert_statistic(
              self.health_checkers, succ_rate_threshold=0.98
          )
          assert_expectations()
      # wait all pod ready
      wait_pods_ready(
          self.milvus_ns,
          f"app.kubernetes.io/instance={self.release_name}",
      )
      time.sleep(60)
      cc.check_thread_status(tasks)
      for k, v in self.health_checkers.items():
          v.pause()
      ra = ResultAnalyzer()
      ra.get_stage_success_rate()
      ra.show_result_table()
      log.info(
          "*********************Chaos Test Completed**********************"
      )
  
file_path: milvus/tests/python_client/chaos/testcases/test_single_request_operation.py
imports: import time
  import pytest
  from time import sleep
  from pymilvus import connections
  from chaos.checker import CollectionCreateChecker, InsertChecker, UpsertChecker, FlushChecker, SearchChecker, QueryChecker, IndexCreateChecker, DeleteChecker, CollectionDropChecker, Op, EventRecords, ResultAnalyzer
  from utils.util_log import test_log
  from utils.util_k8s import wait_pods_ready, get_milvus_instance_name
  from chaos import chaos_commons
  from common.common_type import CaseLabel
  from common.milvus_sys import MilvusSys
  from chaos.chaos_commons import assert_statistic
  from chaos import constants
  from delayed_assert import assert_expectations
label_desc: Use image node_id 3 for calling the TestOperations obj's underlying member method code with example usage: obj.test_operations(request_duration, is_check) without return types