tools/flakiness/find_flakiness.py
def FindShardingFlakiness(test_path, data_path, supervisor_args):
  """Finds flaky test cases by sharding and running a test for the specified
  number of times. The data file is read at the beginning of each run to find
  the last known counts and is overwritten at the end of each run with the new
  counts. There is an optional sleep interval between each run so the script can
  be killed without losing the data, useful for overnight (or weekend!) runs.
  """

  failed_tests = {}
  # Read a previously written data file.
  if os.path.exists(data_path):
    data_file = open(data_path, 'r')
    num_runs = int(data_file.readline().split(' ')[0])
    num_passes = int(data_file.readline().split(' ')[0])
    for line in data_file:
      if line:
        split_line = line.split(' -> ')
        failed_tests[split_line[0]] = int(split_line[1])
    data_file.close()
  # No data file found.
  else:
    num_runs = 0
    num_passes = 0

  log_lines = False
  args = ['python', '../sharding_supervisor/sharding_supervisor.py']
  args.extend(supervisor_args + [test_path])
  proc = subprocess.Popen(args, stderr=subprocess.PIPE)

  # Shard the test and collect failures.
  while True:
    line = proc.stderr.readline()
    if not line:
      if proc.poll() is not None:
        break
      continue
    print(line.rstrip())
    if log_lines:
      line = line.rstrip()
      if line in failed_tests:
        failed_tests[line] += 1
      else:
        failed_tests[line] = 1
    elif line.find('FAILED TESTS:') >= 0:
      log_lines = True
  num_runs += 1
  if proc.returncode == 0:
    num_passes += 1

  # Write the data file and print results.
  data_file = open(data_path, 'w')
  print('%i runs' % num_runs)
  data_file.write('%i runs\n' % num_runs)
  print('%i passes' % num_passes)
  data_file.write('%i passes\n' % num_passes)
  for (test, count) in failed_tests.iteritems():
    print('%s -> %i' % (test, count))
    data_file.write('%s -> %i\n' % (test, count))
  data_file.close()
Finds flaky test cases by sharding and running a test for the specified
  number of times. The data file is read at the beginning of each run to find
  the last known counts and is overwritten at the end of each run with the new
  counts. There is an optional sleep interval between each run so the script can
  be killed without losing the data, useful for overnight (or weekend!) runs.
  
Finds flaky test cases by sharding and running a test for the specified
number of times. The data file is read at the beginning of each run to find
the last known counts and is overwritten at the end of each run with the new
counts. There is an optional sleep interval between each run so the script can
be killed without losing the data, useful for overnight (or weekend!) runs.
Finds flaky test cases by sharding and running a test for the specified number of times . The data file is read at the beginning of each run to find the last known counts and is overwritten at the end of each run with the new counts . There is an optional sleep interval between each run so the script can be killed without losing the data useful for overnight ( or weekend! ) runs .
def FindShardingFlakiness(test_path, data_path, supervisor_args):
  failed_tests = {}
  if os.path.exists(data_path):
    data_file = open(data_path, 'r')
    num_runs = int(data_file.readline().split(' ')[0])
    num_passes = int(data_file.readline().split(' ')[0])
    for line in data_file:
      if line:
        split_line = line.split(' -> ')
        failed_tests[split_line[0]] = int(split_line[1])
    data_file.close()
  else:
    num_runs = 0
    num_passes = 0
  log_lines = False
  args = ['python', '../sharding_supervisor/sharding_supervisor.py']
  args.extend(supervisor_args + [test_path])
  proc = subprocess.Popen(args, stderr=subprocess.PIPE)
  while True:
    line = proc.stderr.readline()
    if not line:
      if proc.poll() is not None:
        break
      continue
    print(line.rstrip())
    if log_lines:
      line = line.rstrip()
      if line in failed_tests:
        failed_tests[line] += 1
      else:
        failed_tests[line] = 1
    elif line.find('FAILED TESTS:') >= 0:
      log_lines = True
  num_runs += 1
  if proc.returncode == 0:
    num_passes += 1
  data_file = open(data_path, 'w')
  print('%i runs' % num_runs)
  data_file.write('%i runs\n' % num_runs)
  print('%i passes' % num_passes)
  data_file.write('%i passes\n' % num_passes)
  for (test, count) in failed_tests.iteritems():
    print('%s -> %i' % (test, count))
    data_file.write('%s -> %i\n' % (test, count))
  data_file.close()
def FindShardingFlakiness ( test_path , data_path , supervisor_args ) : failed_tests = { } if os . path . exists ( data_path ) : data_file = open ( data_path , 'r' ) num_runs = int ( data_file . readline ( ) . split ( ' ' ) [ 0 ] ) num_passes = int ( data_file . readline ( ) . split ( ' ' ) [ 0 ] ) for line in data_file : if line : split_line = line . split ( ' -> ' ) failed_tests [ split_line [ 0 ] ] = int ( split_line [ 1 ] ) data_file . close ( ) else : num_runs = 0 num_passes = 0 log_lines = False args = [ 'python' , '../sharding_supervisor/sharding_supervisor.py' ] args . extend ( supervisor_args + [ test_path ] ) proc = subprocess . Popen ( args , stderr = subprocess . PIPE ) while True : line = proc . stderr . readline ( ) if not line : if proc . poll ( ) is not None : break continue print ( line . rstrip ( ) ) if log_lines : line = line . rstrip ( ) if line in failed_tests : failed_tests [ line ] += 1 else : failed_tests [ line ] = 1 elif line . find ( 'FAILED TESTS:' ) >= 0 : log_lines = True num_runs += 1 if proc . returncode == 0 : num_passes += 1 data_file = open ( data_path , 'w' ) print ( '%i runs' % num_runs ) data_file . write ( '%i runs\n' % num_runs ) print ( '%i passes' % num_passes ) data_file . write ( '%i passes\n' % num_passes ) for ( test , count ) in failed_tests . iteritems ( ) : print ( '%s -> %i' % ( test , count ) ) data_file . write ( '%s -> %i\n' % ( test , count ) ) data_file . close ( )
Finds flaky test cases by sharding and running a test for the specified
number of times.
Finds flaky test cases by sharding and running a test for the specified number of times .
"""Finds flaky test cases by sharding and running a test for the specified
  number of times. The data file is read at the beginning of each run to find
  the last known counts and is overwritten at the end of each run with the new
  counts. There is an optional sleep interval between each run so the script can
  be killed without losing the data, useful for overnight (or weekend!) runs.
  """ # Read a previously written data file. # No data file found. # Shard the test and collect failures. # Write the data file and print results.
test_path
data_path
supervisor_args