python 3.4 le multitraitement ne fonctionne pas avec unittest

Question

J'ai un unittest qui utilise le multitraitement. Après la mise à niveau de Python 3.2 vers Python 3.4, je reçois une erreur de suivi après la mise à niveau de Python 3.2 vers Python 3.4. Je ne trouve pas d'indice, ce qui a été changé dans Python et ce que je dois changer, pour faire fonctionner mon code.

Merci d'avance.

Traceback (most recent call last): 
    File "<string>", line 1, in <module> 
    File "C:\Python341_64\lib\multiprocessing\spawn.py", line 106, in spawn_main 
    exitcode = _main(fd) 
    File "C:\Python341_64\lib\multiprocessing\spawn.py", line 116, in _main 
    self = pickle.load(from_parent) 
EOFError: Ran out of input 

Error 
Traceback (most recent call last): 
    File "D:\test_multiproc.py", line 46, in testSmallWorkflow 
    p.start() 
    File "C:\Python341_64\lib\multiprocessing\process.py", line 105, in start 
    self._popen = self._Popen(self) 
    File "C:\Python341_64\lib\multiprocessing\context.py", line 212, in _Popen 
    return _default_context.get_context().Process._Popen(process_obj) 
    File "C:\Python341_64\lib\multiprocessing\context.py", line 313, in _Popen 
    return Popen(process_obj) 
    File "C:\Python341_64\lib\multiprocessing\popen_spawn_win32.py", line 66, in __init__ 
    reduction.dump(process_obj, to_child) 
    File "C:\Python341_64\lib\multiprocessing\reduction.py", line 59, in dump 
    ForkingPickler(file, protocol).dump(obj) 
TypeError: cannot serialize '_io.TextIOWrapper' object 

Après un exemple de code, comment je peux reproduire l'erreur:

import shutil 
import traceback 
import unittest 
import time 
from multiprocessing import Process 
import os 


class MyTest(unittest.TestCase): 

    #--------------------------------------------------------------------------- 
    def setUp(self): 
     self.working_dir = os.path.join(os.environ["TEMP"], "Testing") 
     os.mkdir(self.working_dir) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def tearDown(self): 
     try: 
      time.sleep(5) 
      shutil.rmtree(self.working_dir, ignore_errors=True) 
     except OSError as err: 
      traceback.print_tb(err) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def info(self, title): 
     print(title) 
     print('module name:', __name__) 
     if hasattr(os, 'getppid'): # only available on Unix 
      print('parent process:', os.getppid()) 
     print('process id:', os.getpid()) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def f(self, name): 
     self.info('function f') 
     print('hello', name) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def testSmallWorkflow(self): 
     self.info('main line') 
     p = Process(target=self.f, args=('bob',)) 
     p.start() 
     p.join() 
    #--------------------------------------------------------------------------- 

Answer 1

Le problème est que lui-même n'est classe unittest.TestCase plus pickleable, et vous devez le décaper pour décaper un de ses méthodes liées (self.f). Une solution facile serait de créer une catégorie distincte pour les méthodes dont vous avez besoin de faire appel dans le processus de l'enfant:

class Tester: 
    def info(self, title=None): 
     print("title {}".format(title)) 
     print('module name:', __name__) 
     if hasattr(os, 'getppid'): # only available on Unix 
      print('parent process:', os.getppid()) 
     print('process id:', os.getpid()) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def f(self, name): 
     self.info('function f') 
     print('hello', name) 
    #------------------------------- 


class MyTest(unittest.TestCase): 

    #--------------------------------------------------------------------------- 
    def setUp(self): 
     self.working_dir = os.path.join(os.environ["TEMP"], "Testing") 
     os.mkdir(self.working_dir) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def tearDown(self): 
     try: 
      time.sleep(5) 
      shutil.rmtree(self.working_dir, ignore_errors=True) 
     except OSError as err: 
      traceback.print_tb(err) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def testSmallWorkflow(self): 
     t = Tester() 
     self.info('main line') 
     p = Process(target=t.f, args=('bob',)) 
     p.start() 
     p.join() 

Alternativement, vous utilisez __setstate__/__getstate__ pour supprimer l'objet de la TestCase qui est unpickleable. Dans ce cas, il s'agit d'une classe interne appelée _Outcome. Nous ne nous en soucions pas chez l'enfant, donc nous pouvons simplement le supprimer de l'état décapé:

class MyTest(unittest.TestCase): 

    #--------------------------------------------------------------------------- 
    def setUp(self): 
     self.working_dir = os.path.join(os.environ["TEMP"], "Testing") 
     os.mkdir(self.working_dir) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def tearDown(self): 
     try: 
      time.sleep(2) 
      shutil.rmtree(self.working_dir, ignore_errors=True) 
     except OSError as err: 
      traceback.print_tb(err) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def info(self, title=None): 
     print("title {}".format(title)) 
     print('module name:', __name__) 
     if hasattr(os, 'getppid'): # only available on Unix 
      print('parent process:', os.getppid()) 
     print('process id:', os.getpid()) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def f(self, name): 
     self.info('function f') 
     print('hello', name) 
    #--------------------------------------------------------------------------- 

    #--------------------------------------------------------------------------- 
    def testSmallWorkflow(self): 
     t = Tester() 
     self.info('main line') 
     p = Process(target=self.f, args=('bob',)) 
     p.start() 
     p.join() 

    def __getstate__(self): 
     self_dict = self.__dict__.copy() 
     del self_dict['_outcome'] 
     return self_dict 

    def __setstate(self, state): 
     self.__dict__.update(self_dict)