Skip to content

Instantly share code, notes, and snippets.

@bryanhelmig
Created June 13, 2015 17:37
Show Gist options
  • Save bryanhelmig/5b24b024d68d2014e372 to your computer and use it in GitHub Desktop.
Save bryanhelmig/5b24b024d68d2014e372 to your computer and use it in GitHub Desktop.
Really simple parallel Django test runner. Primarily for use with something like https://pypi.python.org/pypi/django-jux in Jenkins. BSD 3 clause licensed.
import importlib
from multiprocessing import Pool
from django.test.simple import DjangoTestSuiteRunner
from django.conf import settings
__doc__ = """
In settings.py do this:
JUXD_FILENAME = '/Code/project/junit.xml'
JUXD_FILENAME_TEMPLATE = '/Code/project/junit-{}.xml'
TEST_PROCESS_COUNT = 2
TEST_RUNNER_BACKEND = 'juxd.JUXDTestSuiteRunner'
TEST_RUNNER = 'path.to.ParallelTestRunner'
Run tests like normal.
"""
TEST_PROCESS_COUNT = getattr(settings, 'TEST_PROCESS_COUNT', 4)
TEST_RUNNER_BACKEND = getattr(settings, 'TEST_RUNNER_BACKEND', 'django.test.simple.DjangoTestSuiteRunner')
def get_class(path):
module, key = path.rsplit('.', 1)
try:
return getattr(importlib.import_module(module), key)
except (ImportError, AttributeError):
return path
def make_buckets(items, count):
"""
Split a list into N buckets.
"""
buckets = {x: [] for x in range(count)}
for index, item in enumerate(items):
buckets[index % count].append(item)
return buckets.values()
def run_tests_async(test_labels, **kwargs):
"""
This is ran under children processes.
Respect settings.TEST_RUNNER_BACKEND, If JUXD_FILENAME_TEMPLATE we'll
use that as well and override JUXD_FILENAME.
"""
try:
from south.management.commands import patch_for_test_db_setup
patch_for_test_db_setup()
except ImportError:
pass
if hasattr(settings, 'JUXD_FILENAME_TEMPLATE'):
settings.JUXD_FILENAME = settings.JUXD_FILENAME_TEMPLATE.format('-'.join(test_labels))
SuiteRunner = get_class(TEST_RUNNER_BACKEND)
return SuiteRunner(**kwargs).run_tests(test_labels)
test_pool = Pool(processes=TEST_PROCESS_COUNT)
class ParallelTestRunner(DjangoTestSuiteRunner):
"""
Run tests in a process pool. All test suite arguments are bucketed
into N process count buckets. So, for example, if process count is
set to 2 and you do `manage.py test app1 app2 app3` then we will
basically do:
process 1: manage.py test app1 app3
process 2: manage.py test app2
Respects settings.TEST_RUNNER_BACKEND for the test class behind the scenes.
"""
def run_tests(self, test_labels, **kwargs):
results = []
for labels in make_buckets(test_labels, TEST_PROCESS_COUNT):
if not labels:
continue
ar = (labels,)
kw = dict(
verbosity=self.verbosity,
interactive=self.interactive,
failfast=self.failfast
)
result = test_pool.apply_async(run_tests_async, ar, kw)
results.append(result)
test_pool.close()
return sum([r.get() for r in results])
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment