Prototype for a multiprocessing test runner for Django
#!/usr/bin/env python | |
import multiprocessing | |
import os | |
import sys | |
import unittest | |
from django.test.simple import DjangoTestSuiteRunner | |
def run_test_slice(test_labels, extra_tests, slice_index, number_process): | |
print "Run test slice %s" % (slice_index) | |
runner = DjangoTestSuiteRunnerSlice(slice_index=slice_index, | |
number_process=number_process) | |
return runner.run_tests(test_labels=test_labels, extra_tests=extra_tests) | |
class SliceTestRunner(DjangoTestSuiteRunner): | |
def run_tests(self, test_labels=None, extra_tests=None): | |
pool_size = 6 | |
pool = multiprocessing.Pool(pool_size) | |
results = [] | |
for i in range(0, pool_size): | |
results.append( | |
pool.apply_async(run_test_slice, | |
[test_labels, extra_tests, i, pool_size])) | |
for result in results: | |
result.get(timeout=10) | |
return [] | |
class DjangoTestSuiteRunnerSlice(DjangoTestSuiteRunner): | |
def __init__(self, slice_index, number_process): | |
super(DjangoTestSuiteRunnerSlice, self).__init__() | |
self.slice_index = slice_index | |
self.number_process = number_process | |
def setup_test_environment(self, **kwargs): | |
super(DjangoTestSuiteRunnerSlice, self).setup_test_environment(**kwargs) | |
def build_suite(self, test_labels, extra_tests=None, **kwargs): | |
suite = super(DjangoTestSuiteRunnerSlice, self).build_suite( | |
test_labels, extra_tests, **kwargs) | |
nb_test_cases = suite.countTestCases() | |
share_size = nb_test_cases / self.number_process | |
first_pos = self.slice_index * share_size | |
last_pos = min(nb_test_cases - 1, (self.slice_index + 1) * share_size) | |
test_slice = suite._tests[first_pos:last_pos] | |
suite = unittest.TestSuite() | |
for test in test_slice: | |
suite.addTest(test) | |
return suite | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment