jeudi 30 avril 2015

Mock in multithreading context for testing in Python

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)

Aucun commentaire:

Enregistrer un commentaire