A simple, working example for testtools ConcurrentStreamTestSuite

Question

I'm looking for a simple example demonstrating how to use testtool's concurrent tests. I've found one example, from here:

import unittest
import testtools

class MyTester(unittest.TestCase):
    # Tests...

suite = unittest.TestLoader().loadTestsFromTestCase(MyTester)
concurrent_suite = testtools.ConcurrentStreamTestSuite(lambda: ((case, None) for case in suite))
concurrent_suite.run(testtools.StreamResult()) 

However, it suffers from a rather large problem. It doesn't tell you if your tests passed or failed. Not only is the StreamResult object temporary, StreamResult's methods don't do anything.

Answer 1

According ot the StreamResult documentation, the status method is called to handle event, and default implementation does nothing.

To make it print something you need to inherit it to do what you want. For example, to print what's going on:

import unittest
import testtools

class MyTester(unittest.TestCase):
    def test_foo(self):
        self.assertEqual(1+2, 2)
    def test_foo2(self):
        self.assertEqual(1+2, 3)

class TracingStreamResult(testtools.StreamResult):
    def status(self, *args, **kwargs):
        print('{0[test_id]}: {0[test_status]}'.format(kwargs))

suite = unittest.TestLoader().loadTestsFromTestCase(MyTester)
concurrent_suite = testtools.ConcurrentStreamTestSuite(lambda: ((case, None) for case in suite))
result = TracingStreamResult()
result.startTestRun()
concurrent_suite.run(result)
result.stopTestRun()

But, there are already multiple classes that inherit the StreamResult. So you maybe not needed to define your own version. See Extensions to TestResult.

Answer 2

Have a look at concurrencytest library which already implemented testtools classes