1

For example I have two tests:

class Test(unittest.TestCase):
    def setUp(self):
    self.driver = webdriver.Firefox()
    self.driver.get("http://google.com")

def tearDown(self):
    self.driver.quit()

def test_selenium_1(self):
    search_field = self.driver.find_element_by_id("lst-ib")
    search_field.send_keys("Test 1. Number 1")
    search_field.submit()
    time.sleep(2)

def test_selenium_2(self):
    search_field = self.driver.find_element_by_id("lst-ib")
    search_field.send_keys("Test 1. Number 2")
    search_field.submit()
    time.sleep(2)

if __name__ == '__main__':
    unittest.main()

How can I execute these two tests in parallel with concurrent.futures.Executor?

Is it possible?

5
  • nose.readthedocs.io/en/latest/doc_tests/test_multiprocess/… Commented Jul 4, 2016 at 18:42
  • Are there any ways to do it without nose, pytest, etc. I want to find ways to do it with unittest. Commented Jul 4, 2016 at 19:47
  • nose multiprocess code is open source (they list it all on their site) and could provide a lot of insight in how to implement parallel tests Commented Jul 4, 2016 at 19:53
  • With this specific example, it looks like you'd have a race condition if those tests executed in parallel... Commented Jul 5, 2016 at 15:59
  • I executed about 40 tests and it was ok. Why did you decide it? Maybe I should fix something Commented Jul 5, 2016 at 19:50

2 Answers 2

3

I created Runner for these purposes.

And now I can execute tests in parallel by modules, classes and methods.

import unittest
from concurrent.futures import ThreadPoolExecutor

class Runner():

    def parallel_execution(self, *name, options='by_module'):

        """
        name - name of the class with tests or module with classes that contain tests
        modules - name of the module with tests or with class that contains tests
        options:
            by_method - gather all tests methods in the class/classes and execute in parallel
            by_module - gather all tests from modules in parallel
            by_class - will execute all classes (with tests) in parallel
        """

        suite = unittest.TestSuite()

        if (options=='by_method'):
            for object in name:
                for method in dir(object):
                    if (method.startswith('test')):
                        suite.addTest(object(method))
        elif (options=='by_class'):
            for object in name:
                suite.addTest(unittest.TestLoader().loadTestsFromTestCase(object))

        elif (options=='by_module'):
            for module in name:
                suite.addTest(unittest.TestLoader().loadTestsFromModule(module))
        else:
            raise ValueError("Parameter 'options' is incorrect."
                             "Available options: 'by_method', 'by_class', 'by_module'")

        with ThreadPoolExecutor(max_workers=10) as executor:
            list_of_suites = list(suite)
            for test in range(len(list_of_suites)):
                test_name = str(list_of_suites[test])
                executor.submit(unittest.TextTestRunner().run, list_of_suites[test])

EXAMPLES:

#by_methods
Runner().parallel_execution(Test1.Test1, Test2.Test22, Test2.Test33, options='by_method')

#by_class
Runner().parallel_execution(Test1.Test1, Test2.Test22, Test2.Test33, options='by_class')

#by_modules
Runner().parallel_execution(Test1, Test2)
Sign up to request clarification or add additional context in comments.

2 Comments

I am facing the same problem you solution is working fine for the test cases but i am also using before and after method and they are not getting called, only test cases are getting executed
Did you solve the issue with running the setup/teardown on tests with this solution?
0

I rewrote @vadim-kovrizhkin's solution as a separate runner file that one can use to add separate test cases (importing each test case from their respective py files). This will run setUp, the unit tests in your test cases concurrently, and then tearDown.

from unittest import TestLoader, TestSuite, TextTestRunner
from concurrent.futures import ThreadPoolExecutor
from YourTestFolder.SubFolder.test_foo import Test_Foo

def suite():
    loader = TestLoader()
    suite = TestSuite(loader.loadTestsFromTestCase(Test_Foo))
    return suite

if __name__ == "__main__":

    suite = suite()
    with ThreadPoolExecutor() as executor:
        list_of_suites = list(suite)
        for test in range(len(list_of_suites)):
            test_name = str(list_of_suites[test])
            executor.submit(TextTestRunner().run, list_of_suites[test])

Thanks again to Vadim for their elegant solution!

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.