def run_step()

in gcpdiag/runbook/__init__.py [0:0]
90 lines of code
10 McCabe index (conditional complexity)

  def run_step(self, step: Step, operator: op.Operator):
    """Executes a single step, handling user decisions for step re-evaluation or termination.

    Args:
      step: The diagnostic step to execute.
      operator: The execution operations object containing the context.
    """
    try:
      user_input = self._run(step, operator=operator)
      while True:
        if user_input is constants.RETEST:
          operator.interface.info(step_type=constants.RETEST,
                                  message='Re-evaluating recent failed step')
          with caching.bypass_cache():
            user_input = self._run(operator.step, operator=operator)
        elif step.type == constants.StepType.END:
          return constants.FINALIZE_INVESTIGATION
        elif user_input is constants.STOP:
          logging.info('Finalize Investigation\n')
          return constants.FINALIZE_INVESTIGATION
        elif step.type == constants.StepType.START and (
          self.interface.rm.reports[operator.run_id]
          .results.get(step.execution_id) is not None and \
          self.interface.rm.reports[operator.run_id]
          .results[step.execution_id].overall_status == 'skipped'):
          logging.info('Start Step was skipped. Can\'t proceed.\n')
          return constants.FINALIZE_INVESTIGATION
        elif user_input is constants.CONTINUE:
          break
        elif (user_input is not constants.RETEST and
              user_input is not constants.CONTINUE and
              user_input is not constants.STOP):
          return user_input
    except Exception as err:  # pylint: disable=broad-exception-caught
      error_msg = str(err)
      end = datetime.now(timezone.utc).isoformat()
      with report_lock:
        self.interface.rm.reports[operator.run_id].results[
            step.execution_id].end_time = end
      if isinstance(err, TemplateNotFound):
        error_msg = (
            f'could not load messages linked to step: {step.id}.'
            'ensure step has a valid template eg: filename::block_prefix')
        logging.error(error_msg)
      elif isinstance(err, exceptions.InvalidStepOperation):
        error_msg = f'invalid step operation: %s: {err}'
        logging.error(error_msg)
      elif isinstance(err, (ValueError, KeyError)):
        error_msg = f'`{step.execution_id}`: {err}'
        logging.error(error_msg)
      elif isinstance(err,
                      (utils.GcpApiError, googleapiclient.errors.HttpError)):
        if isinstance(err, googleapiclient.errors.HttpError):
          err = utils.GcpApiError(err)
        if err.status == 403:
          logging.error(
              ('%s: %s user does not sufficient permissions '
               'to perform operations in step: %s'),
              type(err).__name__,
              err,
              step.execution_id,
          )
          with report_lock:
            self.interface.rm.reports[operator.run_id].results[
                step.execution_id].step_error = err
        elif err.status == 401:
          logging.error(
              '%s: %s request is missing required authentication credential to'
              ' perform operations in step: %s',
              type(err).__name__,
              err,
              step.execution_id,
          )
          with report_lock:
            self.interface.rm.reports[operator.run_id].results[
                step.execution_id].step_error = err
          return
        logging.error(
            '%s: %s while processing step: %s',
            type(err).__name__,
            err,
            step.execution_id,
        )
        with report_lock:
          self.interface.rm.reports[operator.run_id].results[
              step.execution_id].step_error = err
      else:
        logging.error(
            '%s: %s while processing step: %s',
            type(err).__name__,
            err,
            step.execution_id,
        )
      with report_lock:
        self.interface.rm.reports[operator.run_id].results[
            step.execution_id].step_error = error_msg