Prototype for a multiprocessing test runner for Django

splicer.py

#!/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