Another aspect of combining testing and concurrent programming is performing tests in a concurrent way. This aspect of testing is more straightforward and intuitive than testing concurrent programs themselves. In this subsection, we will explore a library that can help us facilitate this process, concurrencytest, which can work seamlessly with test cases implemented with the preceding unittest module.

concurrencytest is designed as a testtools extension that implements concurrency in running test suites. It can be installed from PyPI, using pip, as follows:

pip install concurrencytest

Additionally, concurrencytest is dependent on the testtools (pypi.org/project/testtools/) and python-subunit (pypi.org/project/python-subunit/) libraries, which are a test extension framework and a streamlining protocol for test results, respectively. These libraries can also be installed via pip, as follows:

pip install testtools
pip install python-subunit

As always, to verify your installation, try to import the library in a Python interpreter:

>>> import concurrencytest

Receiving no printed errors means that the library and its dependencies were installed successfully. Now, let's look at how this library can help us to achieve better speed for our tests. Navigate to the Chapter19/example6.py file and consider the following code:

# Chapter19/example6.py

import unittest

def fib(i):
    if i in [0, 1]:
        return i

    a, b = 0, 1
    n = 1
    while n < i:
        a, b = b, a + b
        n += 1

    return b

class FibTest(unittest.TestCase):
    def __init__(self, *args, **kwargs):
        super(FibTest, self).__init__(*args, **kwargs)
        self.mod = 10 ** 10

    def test_start_values(self):
        self.assertEqual(fib(0), 0)
        self.assertEqual(fib(1), 1)

    def test_big_value_v1(self):
        self.assertEqual(fib(499990) % self.mod, 9998843695)

    def test_big_value_v2(self):
        self.assertEqual(fib(499995) % self.mod, 1798328130)

    def test_big_value_v3(self):
        self.assertEqual(fib(500000) % self.mod, 9780453125)

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

The main goal of the examples in this section is testing the function that produces numbers in the Fibonacci sequence, specifically numbers with large indices. The fib() function that we have is similar to that of the previous example, although this one performs the calculation iteratively, without using recursion.

In our test case, aside from the two starting values, we are now testing numbers at indices 499,990, 499,995, and 500,000. Since the resulting numbers are significantly large, we are only testing the last ten digits for each number (this is done via the mod attribute of the test class, specified in the initialization method). This testing process will be executed in one process, in a sequential way.

Run the program, and your output should be similar to the following:

> python3 example6.py
....
----------------------------------------------------------------------
Ran 4 tests in 8.809s

OK

Again, the time specified in the output can vary from system to system. With that said, remember the amount of time that the program took, so that you can compare it with the speed of other programs that we will consider later on.

Now, let's look at how we can distribute the testing workload across multiple processes, with concurrencytest. Consider the Chapter19/example7.py file, as follows:

# Chapter19/example7.py

import unittest
from concurrencytest import ConcurrentTestSuite, fork_for_tests

def fib(i):
    if i in [0, 1]:
        return i

    a, b = 0, 1
    n = 1
    while n < i:
        a, b = b, a + b
        n += 1

    return b

class FibTest(unittest.TestCase):
    def __init__(self, *args, **kwargs):
        super(FibTest, self).__init__(*args, **kwargs)
        self.mod = 10 ** 10

    def test_start_values(self):
        self.assertEqual(fib(0), 0)
        self.assertEqual(fib(1), 1)

    def test_big_value_v1(self):
        self.assertEqual(fib(499990) % self.mod, 9998843695)

    def test_big_value_v2(self):
        self.assertEqual(fib(499995) % self.mod, 1798328130)

    def test_big_value_v3(self):
        self.assertEqual(fib(500000) % self.mod, 9780453125)

if __name__ == '__main__':
    suite = unittest.TestLoader().loadTestsFromTestCase(FibTest)
    concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests(4))
    runner.run(concurrent_suite)

This version of the program is examining the same fib() function, using the same test case. However, in the main program, we are initializing an instance of the ConcurrentTestSuite class, from the concurrencytest library. This instance takes in a test suite, which was created using the TestLoader() API from the unittest module, and the fork_for_tests() function, with the parameter 4, to specify that we want to utilize four separate processes to distribute the testing procedure.

Now, let's run this program and compare its speed with that of our previous tests:

> python3 example7.py
....
----------------------------------------------------------------------
Ran 4 tests in 4.363s

OK

You can see that a significant improvement in speed was achieved by this method of multiprocessing. However, this improvement does not fall around perfect scalability (discussed in Chapter 16, Designing Lock-Based and Mutex-Free Concurrent Data Structures); that is because there is significant overhead in creating concurrent test suites that can be executed across multiple processes.

One more thing that we should mention is that it is quite possible to achieve the same multiprocessing setup that we implemented here by using the traditional concurrent programming tools that we discussed in previous chapters, such as concurrent.futures or multiprocessing. With that said, the concurrencytest library, as we have seen, is able to eliminate significant boilerplate code, and thus provides an easy and fast API.