Mock in multithreading context for testing in Python

Question

In a Django project, I would like to write a test for a function that is used in a multiprocessing context (Processing.create_all_files). If I were using a single thread, I would do 'mock' in order to check the parameters used for calling a given function (FileCreator.create in my case).

However, once the function FileCreator.create is called by a multiprocessing.Pool, mock does not work anymore with it.

How should I do my test for create_all_files? Thanks.

test_program.py:

def test_create_all_files(self):
    file_paths = [ (...) ] # Initialize file_paths
    processing = Processing()
        with mock.patch('FileCreator.create', return_value=True) as create:
            with mock.patch('os.path.isfile', return_value=False):
                processing.create_all_files()
                calls = create.call_args_list

        for file_path in file_paths:
            self.assertTrue(((file_path),) in calls)


program.py

def unwrap_self_create_one_file(arg):
    return Processing.process_one_file(*arg)

class Processing:
    (...)

    def create_one_file(self, file_path):
       if os.path.isfile(file_path):
        FileCreator.create(file_path) # CREATE FILE

        def create_all_files(file_paths):
       (...) # define args_lst considering file_paths
       ncpus = 4
           pool = multiprocessing.Pool(ncpus)
           pool.map(unwrap_create_one_file, args_lst, chunksize=1)

Franky Martin · Accepted Answer · 2016-06-17 14:26:21Z

5

After adding multithreading you risk your 'patch' to go out of scope, in other words, when executing FileCreator.create on a different thread chances are the with patch() statement has ended.

edited Jun 17, 2016 at 14:26

answered Jun 16, 2016 at 21:51

Franky Martin

511 silver badge2 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Jerther Over a year ago

Even when using the @patch.object annotation on a test method, the thread could run when the test method has ended so it is necessary to join threads inside the test method.

Michele d'Amico · Accepted Answer · 2015-04-30 13:31:10Z

2

Your issue have nothing to do with multithreading but it is more related on Where to patch.

In your test you use FileCreator.create(file_path) to create your file object so I guess you have something like from mymodule import FileCreator in program.py.

What you should do in these case is patch the FileCreator reference in program by:

with mock.patch('program.FileCreator.create', return_value=True) as create:

edited Apr 30, 2015 at 13:31

answered Apr 30, 2015 at 11:32

Michele d'Amico

24.1k10 gold badges78 silver badges78 bronze badges

1 Comment

eelioss Over a year ago

Ok! Thanks. I misunderstood the point related to 'Where to patch'... now I wonder why it worked before adding multithreading support.

Erik Aronesty · Accepted Answer · 2021-11-09 19:20:56Z

2

I wrote a little library that handles thread safe magic mocking.

It can be problematic to refactor a massive test suite, so this library allows you to patch the base class call to mocking __getattr__ with an RLock.

https://github.com/AtakamaLLC/tsmock

pip install tsmock

You can do this:

from tsmock import MagicMock

Or this:

from tsmock import thread_safe_mocks
thread_safe_mocks()

Either one gets you the correct call count.

answered Nov 9, 2021 at 19:20

Erik Aronesty

13.2k6 gold badges73 silver badges47 bronze badges

Collectives™ on Stack Overflow

Mock in multithreading context for testing in Python

3 Answers 3

1 Comment

1 Comment

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

1 Comment

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Related