def main()

in marketplace/deployer_util/run_tester.py [0:0]

• 60 lines of code
• 10 McCabe index (conditional complexity)

def main():
  parser = ArgumentParser(description=_PROG_HELP)
  parser.add_argument('--namespace')
  parser.add_argument('--manifest')
  parser.add_argument('--timeout', type=int, default=300)
  args = parser.parse_args()

  try:
    Command(
        '''
        kubectl apply
        --namespace="{}"
        --filename="{}"
        '''.format(args.namespace, args.manifest),
        print_call=True)
  except CommandException as ex:
    log.error("Failed to apply tester job. Reason: {}", ex.message)
    return

  resources = load_resources_yaml(args.manifest)

  test_failed = False
  for resource_def in resources:
    full_name = "{}/{}".format(resource_def['kind'],
                               deep_get(resource_def, 'metadata', 'name'))

    if resource_def['kind'] != 'Pod':
      log.info("Skip '{}'", full_name)
      continue

    start_time = time.time()
    poll_interval = 4
    tester_timeout = args.timeout

    while True:
      try:
        resource = Command(
            '''
          kubectl get "{}"
          --namespace="{}"
          -o=json
          '''.format(full_name, args.namespace),
            print_call=True).json()
      except CommandException as ex:
        log.info(str(ex))
        log.info("retrying")
        time.sleep(poll_interval)
        continue

      result = deep_get(resource, 'status', 'phase')

      if result == "Failed":
        print_tester_logs(full_name, args.namespace)
        log.error("Tester '{}' failed.", full_name)
        test_failed = True
        break

      if result == "Succeeded":
        print_tester_logs(full_name, args.namespace)
        log.info("Tester '{}' succeeded.", full_name)
        break

      if time.time() - start_time > tester_timeout:
        print_tester_logs(full_name, args.namespace)
        log.error("Tester '{}' timeout.", full_name)
        test_failed = True
        break

      time.sleep(poll_interval)

  if test_failed:
    sys.exit("At least 1 test failed or timed out.")