Issue
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)
Solution
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:
Answered By - Michele d'Amico
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.