Skip to content

Instantly share code, notes, and snippets.

@batiste batiste/splicer.py
Created Jun 22, 2012

Embed
What would you like to do?
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
You can’t perform that action at this time.