Python multi-threaded unittesting

Question

Używam projektu wielowątkowego (nie miałem wyboru), ale większość mojego kodu znajduje się w jednym wątku, gdzie wszystkie zdarzenia w nim są zarządzane przez queue. W ten sposób większość mojego kodu zachowuje się tak, jakby była pojedynczym gwintem i nie muszę się martwić o zamki, semafory i co nie.

Niestety, doszedłem do punktu, w którym muszę dokonać ujęcia mojego kodu (proszę, nie przejmuj się, jeśli nie jest to TDDing), a ja jestem w stanie zagubienia - jak testujesz coś w innym wątku ?

Załóżmy, że mam następujące klasy:

class MyClass(): 
    def __init__(self): 
     self.a=0 
     # register event to self.on_event 

    def on_some_event(self, b): 
     self.a += b 

    def get(self): 
     return self.a 

i chcę testu:

import unittest 
from queued_thread import ThreadedQueueHandler 

class TestMyClass(unittest.TestCase): 
    def setUp(self): 
     # create the queued thread and assign the queue to self.queue 

    def test_MyClass(self): 
     mc = MyClass() 
     self.queue.put({'event_name':'some_event', 'val':1}) 
     self.queue.put({'event_name':'some_event', 'val':2}) 
     self.queue.put({'event_name':'some_event', 'val':3}) 
     self.assertEqual(mc.get(),6) 

if __name__ == '__main__': 
    unittest.main() 

MyClass.get() działa dobrze na wszystko wewnątrz kolejce wątku, ale będzie się nazywać asynchronicznie w głównym wątku przez test, więc wynik może nie być poprawny!

Answer 1

Jeśli twój projekt zakłada, że ​​wszystko musi przejść przez kolejkę, nie walcz z tym - spraw, aby wszystko przez to przechodziło!

Dodaj wydarzenie on_call do swojej kolejce obsługi zdarzeń i zarejestrować się do niego następującą funkcję:

def on_call(self, callback): 
    callback() 

następnie zmodyfikować swój test:

def test_MyClass(self): 
    def threaded_test(): 
     self.assertEqual(mc.get(),6) 

    mc = MyClass() 
    self.queue.put(1) 
    self.queue.put(2) 
    self.queue.put(3) 
    self.queue.put({'event_name':'call','val':threaded_test}) 

Answer 2

Można spojrzeć na test_threading.py w testy stdlib, które robią coś podobnego do tego, co próbujesz zrobić. Podstawową ideą jest ochrona wykonania wątku za pomocą muteksa i semafora, tak aby wykonanie było zakończone przed potwierdzeniem warunku testu.