Статьи

Python — concurrencytest: запуск параллельных тестов

Октябрь 8, 2018

Добавьте параллельное тестирование в вашу среду модульного тестирования.

В моем  предыдущем посте я описал запуск параллельных тестов, используя  нос  в качестве загрузчика и бегуна.

На аналогичном замечании давайте рассмотрим построение параллелизма в вашей собственной тестовой среде , созданной на юнит-тесте Python  .

Посмотрите на этот модуль:  concurrencytest

(Благодаря битам и концепциям, взятым из  testtools  и  bzrlib )

Пример:

Скажем, у вас загружен тестовый пакет TestSuite. Вы можете запустить их с помощью стандартного TextTestRunner, например так:

runner = unittest.TextTestRunner()
runner.run(suite)

Это будет запускать тесты в вашем наборе последовательно в одном процессе.

Добавив   модуль concurrencytest , вы можете вместо этого использовать ConcurrentTestSuite, добавив:

from concurrencytest import ConcurrentTestSuite, fork_for_tests

concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))
runner.run(concurrent_suite)

Это будет запускать те же тесты, разбитые на 4 процесса (рабочие).

Примечание: это зависит от os.fork (), который работает только в системах Unix.

Нет лучшего способа понять это, чем смотреть на некоторые надуманные примеры!

Этот первый пример совершенно нереалистичен, но отлично демонстрирует параллелизм. В тестовых случаях он загружает каждый сон на 0,5 секунды и затем завершает работу.

Код:

#!/usr/bin/env python
#
# Example using `concurrencytest`:
#   https://github.com/cgoldberg/concurrencytest
 
import time
import unittest
 
from concurrencytest import ConcurrentTestSuite, fork_for_tests
 
 
class SampleTestCase(unittest.TestCase):
    def test_it(self):
        time.sleep(0.5)
 
 
if __name__ == '__main__':
 
    # load a TestSuite with 50x TestCases for demo
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    for _ in range(50):
        suite.addTests(loader.loadTestsFromTestCase(SampleTestCase))
    print('Loaded %d test cases...' % suite.countTestCases())
 
    runner = unittest.TextTestRunner()
 
    print('\nRun tests sequentially:')
    runner.run(suite)
 
    print('\nRun same tests across 50 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(50))
    runner.run(concurrent_suite)

Выход:

Loaded 50 test cases...

Run tests sequentially:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 25.031s

OK

Run same tests across 50 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 0.525s

OK

отлично!

Теперь еще один пример, показывающий параллелизм с тестовыми примерами с привязкой к процессору Каждый тестовый набор загружает каждый расчет фибоначчи 31 (рекурсивно!) И затем завершается. Мы можем видеть, как он работает на моем 8-ядерном компьютере (Core2 i7 quad, hyperthreaded).

Код:

#!/usr/bin/env python
#
# Example using `concurrencytest`:
#   https://github.com/cgoldberg/concurrencytest
 
import unittest
 
from concurrencytest import ConcurrentTestSuite, fork_for_tests
 
 
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    return fib(n - 1) + fib(n - 2)
 
 
class FibonacciTestCase(unittest.TestCase):
    def test_fib(self):
        self.assertEqual(fib(31), 1346269)
 
 
if __name__ == '__main__':
 
    # load a TestSuite with 50x TestCases for demo
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    for _ in range(50):
        suite.addTests(loader.loadTestsFromTestCase(FibonacciTestCase))
    print('Loaded %d test cases...' % suite.countTestCases())
 
    runner = unittest.TextTestRunner()
 
    print('\nRun tests sequentially:')
    runner.run(suite)
 
    print('\nRun same tests with 2 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(2))
    runner.run(concurrent_suite)
 
    print('\nRun same tests with 4 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))
    runner.run(concurrent_suite)
 
    print('\nRun same tests with 8 processes:')
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(8))
    runner.run(concurrent_suite)

Выход:

Loaded 50 test cases...

Run tests sequentially:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 21.941s

OK

Run same tests with 2 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 11.081s

OK

Run same tests with 4 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 5.862s

OK

Run same tests with 8 processes:
..................................................
----------------------------------------------------------------------
Ran 50 tests in 4.743s

OK

счастливого взлома.

40
Share: