Find Flakey Tests

flakey.py

#!/usr/bin/env python

import requests
import os
import sys
import pprint
import re
from sets        import Set
from collections import defaultdict, Counter
from urlparse    import urlparse

# TODO
# 1. build on command line
# 2. token form env var
# 3. check for files or classes

circle_token = os.environ['CIRCLE_TOKEN']

if not circle_token:
    print('Error: You must specify a CircleCI API token using CIRCLE_TOKEN.')
    print('Visit https://circleci.com/account/api to get a token.')
    sys.exit(1)


if len(sys.argv) < 2:
    print('Usage:')
    print('python flakey.py <Project URL>')
    print('Example: python flakey.py "https://circleci.com/gh/circleci/frontend"')
    sys.exit(1)

url_path = re.search(r'/gh/([^/]*)/([^/]*)', urlparse(sys.argv[1]).path)

project = "%s/%s" % (url_path.group(1), url_path.group(2))

url = "https://circleci.com/api/v1/project/%s" % project

def download_params(page):
    limit = 100
    offset = page * limit
    return {'filter': 'completed',
            'limit':  limit,
            'offset': offset,
            'circle-token': os.environ['CIRCLE_TOKEN']}

headers = {'content-type': 'application/json',
           'Accept':       'application/json'}


pp = pprint.PrettyPrinter()

all_builds = []

print('Downloading results for %s' % project)
for page in range(5):
    sys.stdout.write('.')
    sys.stdout.flush()
    all_builds.extend(requests.get(url, params=download_params(page), headers=headers).json())
print('')

print('Checking results for last %s builds.' % len(all_builds))

results = defaultdict(list)

for build in all_builds:
    results[build['vcs_revision']].append(build)

retried = dict((k,v) for k,v in results.iteritems() if (len(v) > 1))

success_status = Set(['success', 'fixed'])
failed_status = Set(['failed'])

flakey = dict((k,v) for k,v in retried.iteritems() if
          (any (i['status'] in success_status for i in v) and
           any (i['status'] in failed_status  for i in v)))


flakey_builds = sum(flakey.values(), [])
flakey_failed = [build for build in flakey_builds]
flakey_failed = filter(lambda b: b['status'] in failed_status, flakey_failed)


print('flakey failed builds: %s' % len(flakey_failed))

classes = Counter()
files = Counter()
result = Counter()

auth = {'circle-token': os.environ['CIRCLE_TOKEN']}

print('Found %s builds with suspect test failures.' % len(flakey_failed))
for build in flakey_failed:
    sys.stdout.write('.')
    sys.stdout.flush()
    url = "https://circleci.com/api/v1/project/%(username)s/%(reponame)s/%(build_num)s/tests" % build
    tests = requests.get(url, headers=headers, params=auth).json()['tests']
    failures = [test for test in tests if test['result'] != 'success']
    for fail in failures:
        if fail['file']:
            files.update({fail['file']: 1})
        if fail['classname']:
            classes.update({fail['classname']: 1})
        result.update({(fail['file'], fail['classname'], fail['name']) : 1 })
print('')
print('Finished downloading test results')

def print_counts(counter):
    for item, count in counter.most_common():
        print('- %3s: %s' % (count, item))

if not result:
    print('No flaky tests found.')
else:
    print('# Flakiest Tests:')
    for (file_name, class_name, test_name), count in result.most_common():
        print('- %3s: %s %s' % (count, (file_name or class_name), test_name))

    if classes:
        print('# Flakiest classes:')
        print_counts(classes)

    if files:
        print('# Flakiest files:')
        print_counts(files)