Comment faire pour simuler un module intégré dans un thread

Question

La question est comme dans le titre, comment puis-je simuler select.select par exemple pour tester ma fonction d'exécution de thread. la fonction de test échoue avec

ready = select.select([self.sock], [], [], 5) 
TypeError: fileno() returned a non-integer 

et l'impression de type donne

type « builtin_function_or_method »

clairement select.select ne se moque pas dans le périmètre du fil alors que dans le test, il est .. . (affirmer isinstance)

import select 
import threading 

RECEIVE_BYTES = 256 


class Bar(threading.Thread): 
    def __init__(self, sock): 
     threading.Thread.__init__(self) 
     self.sock = sock 

    def run(self): 
     print type(select.select) 
     ready = select.select([self.sock],[],[],5) 
     if ready[0]: 
      print self.sock.recv(RECEIVE_BYTES) 

l'essai est la suivante dans un autre module

def test_run(self): 
    with patch("select.select"): 
     select.select.return_value = [True] 
     mock_sock = MagicMock() 
     foo = Bar(mock_sock) 
     assert isinstance(select.select, MagicMock) 
     foo.start() 

essais

sont exécutés par le nez

Answer 1

La réponse courte est d'appeler foo.join() attendre le fil pour terminer avant de quitter le bloc with patch(...). L'erreur a été provoquée par la suppression du correctif avant la fin du thread. À propos, il est beaucoup plus facile pour les gens de vous aider si vous postez un exemple qui peut être exécuté. Votre exemple était incomplet et comportait des erreurs de syntaxe.

Voici le test corrigé. J'ai ajouté la boucle pour faciliter la reproduction de l'erreur.

import select 
from mock import patch, MagicMock 
from time import sleep 

from scratch import Bar 

IS_FIXED = True 

def test_run(): 
    for _ in range(20): 
     with patch("select.select"): 
      select.select.return_value = [True] 
      mock_sock = MagicMock() 
      foo = Bar(mock_sock) 
      assert isinstance(select.select, MagicMock) 
      foo.start() 
      if IS_FIXED: 
       foo.join() 
     sleep(0.1) 

Et voici la classe Bar avec quelques corrections de syntaxe.

import select 
import threading 

RECEIVE_BYTES = 256 


class Bar(threading.Thread): 
    def __init__(self, sock): 
     threading.Thread.__init__(self) 
     self.sock = sock 

    def run(self): 
     print type(select.select) 
     ready = select.select([self.sock],[],[],5) 
     if ready[0]: 
      print self.sock.recv(RECEIVE_BYTES)