Created
June 13, 2015 17:37
-
-
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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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