Python Unittest and Multiprocessing fail when run sequentially

user_4301113 0
Viewed 2492

I'm attempting to create unittests for my application which uses multiple processes, but have been having strange issues when attempting to run all the tests together. Basically when running tests individually they pass without issue but when run sequentially, such as when running all tests in the file, some tests will fail.

What I'm seeing is that many python processes are being created but they aren't closing when the test is reported as passed. For example if 2 tests are run that each generate 5 proceses, then 10 python processes show up in the system monitor.

I've tried using terminate and join but neither work. Is there a way to force a test to correctly close all processes that it generated before running the next test?

I'm running Python 2.7 in Ubuntu 16.04.

Edit: It's a fairly large code base so here a simplified example.

from multiprocessing import Pipe, Process

class BaseDevice:
    # Various methods
    pass

class BaseInstr(BaseDevice, Process):
    def __init__(self, pipe):
        Process.__init__(self)
        self.pipe = pipe

    def run(self):
        # Do stuff and wait for terminate message on pipe

    # Various other higher level methods

class BaseCompountInstrument(BaseInstr):
    def __init__(self, pipe):
        # Create multiple instruments, usually done with config file but simplified here
        BaseInstr.__init__(self, pipe)
        instrlist = list()
        for _ in range(5):
            masterpipe, slavepipe = Pipe()
            instrlist.append([BaseInstr(slavepipe), masterpipe])

    def run(self):
        pass
        # Listen for message from pipe, send messages to sub-instruments

    def shutdown(self):
        # When shutdown message received, send to all sub-instruments
        pass

class test(unittest.TestCase):
    def setUp(self):
        # Load up a configuration file from the sample configs so that they're updated
        self.parentConn, self.childConn = Pipe()
        self.instr = BaseCompountInstrument( self.childConn)
        self.instr.start()

    def tearDown(self):
        self.parentConn.send("shutdown") # Propagates to all sub-instruments

    def test1(self):
        pass

    def test2(self):
        pass
python-2.7 unit-testing linux python
1 Answers

After struggling a while (2 days actually) with this, I found a solution with it is not technically wrong, but removes all the parallel code you can have (Only in tests, only in tests...)

I use this package mock to mock functions (which I realize now it's part of the unittest module since Python 3.3 xD), you can suppose the execution of certain function worked well, fix a certain return value, or change the function itself.

So I did the last option: Change the function itself.

In my case I used a list of Process (because Pool didn't work in my case) and Manager's list to share data between the processes.

My original code would be something like this:

import multiprocessing as mp


manager = mp.Manager()
list_data = manager.list()
list_return = manager.list()

def parallel_function(list_data, list_return)

    while len(list_data) > 0:
        # Do things and make sure to "pop" the data in list_data
        list_return.append(return_data)
    return None

# Create as many processes as images or cpus, the lesser number
processes = [mp.Process(target=parallel_function,
                        args=(list_data, list_return))
             for num_p in range(mp.cpu_count())]
for p in processes:
    p.start()
for p in processes:
    p.join(10)

So in my test I mock the function Process._init_ from the multiprocessing module to do my parallel_function instead create a new process.

In the test file, before any test you should define the same function you try to parallelize:

def fake_process(self, list_data, list_return):
    while len(list_data) > 0:
        # Do things and make sure to "pop" the data in list_data
        list_return.append(return_data)
    return None

And before the definition of any method which is going to execute this part of the code you have to define its decorators to overwrite the Process._init_ function.

@patch('multiprocessing.Process.__init__', new=fake_process)
@patch('multiprocessing.Process.start', new=lambda x: None)
@patch('multiprocessing.Process.join', new=lambda x, y: None)
def test_from_the_hell(self):
    # Do things

If you use Manager data structures there is no need of use Locks or anything to control the access to the data, because those structures are thread safe.

I hope this will help any other lost soul who is trying to test multiprocessing code.

user_93040670
Related