django_test_parallel.py

run_tests_parallel.py

# coding: utf-8
"""
This module provides
"""
from __future__ import unicode_literals
import os
import sys

from multiprocessing import Process, Queue

from django.apps import apps
from django.conf import settings
from django.test.utils import get_runner
from django.db.backends.creation import BaseDatabaseCreation
from django_nose.management.commands.test import Command as DjangoNoseTestCmd

from django_nose.runner import NoseTestSuiteRunner

TestRunner = get_runner(settings)

if hasattr(TestRunner, 'options'):
    extra_options = TestRunner.options
else:
    extra_options = []


def _get_test_db_name(self):
    TEST_DATABASE_PREFIX = 'test_'

    if os.environ.get('TEST_DB_NAME') is not None:
        return (
            self.connection.settings_dict['NAME'] +
            '_{}'.format(os.environ['TEST_DB_NAME'])
        )

    if self.connection.settings_dict['TEST_NAME']:
        return self.connection.settings_dict['TEST_NAME']
    return TEST_DATABASE_PREFIX + self.connection.settings_dict['NAME']

BaseDatabaseCreation._get_test_db_name = _get_test_db_name


def run_test_for_app(app, options, failures_queue):
    os.environ['TEST_DB_NAME'] = 'sjerlok_test_db_{}'.format(app)
    test_runner = TestRunner(**options)
    failures = test_runner.run_tests([app])

    failures_queue.put(failures)


class Command(DjangoNoseTestCmd):
    """Implement the ``test`` command."""

    def handle(self, *test_labels, **options):
        from django.conf import settings
        from django.test.utils import get_runner

        options['verbosity'] = int(options.get('verbosity'))

        apps = [
            'case', 'category', 'company', 'connector', 'contact',
            'fileupload', 'google_search', 'notification', 'public', 'search',
            'shared', 'sjerlok', 'sjerlok_user'
        ]
        failures_queue = Queue()
        processes = []

        options['interactive'] = False
        for app in apps:
            test_process = Process(
                target=run_test_for_app, args=(app, options, failures_queue))

            test_process.start()
            processes.append(test_process)

        for process in processes:
            process.join()

        failures = []
        while failures_queue.empty() is False:
            failures.append(failures_queue.get())

        print(zip(apps, failures))

        if any(failures):
            sys.exit(any(failures))