Jak zdobyć zdarzenia wątku wxpython w teście jednostkowym?

Question

Zbudowałem aplikację python wx, która uruchamia wątek tła, aby wykonać pewne obliczenia. Jednak moja obecna implementacja nie pozwala na testowanie jednostkowe.

ja bardzo ściśle oparte moje wykonanie off z pierwszego przykładu w tym tutorialu: http://wiki.wxpython.org/LongRunningTasks

Poniższy kod wykorzystuje wx.PostEvent() i frame.connect() mieć zarejestrować zdarzenie wynik do ramy głównej, który wskazuje, że obliczenia zostały zakończone. Pokazałem także fragment kodu testu jednostkowego.

Jednak, aby możliwe było przechwycenie zdarzenia wyniku wątku, należy uruchomić wx.App.MainLoop(). Jednak nie wiem, jak symulować takie zachowanie w teście jednostkowym.

Moje rozumienie testów jednostkowych GUI ogólnie polega na symulacji zdarzeń ręcznie. Jednak w tym przypadku chciałbym, aby mój wątek tła działał. Czy powinienem zmodyfikować implementację? Lub dla celów testowania jednostkowego mogę odciąć wątek obliczeniowy w inny sposób? Na przykład, czy powinienem uruchomić wątek w kodzie testu jednostkowego, a po jego zakończeniu wywołać kod GUI, aby obsłużyć to zdarzenie bezpośrednio?

import time 
from threading import * 
import unittest 
import wx 

# Button definitions 
ID_START = wx.NewId() 
ID_STOP = wx.NewId() 

# Define notification event for thread completion 
EVT_RESULT_ID = wx.NewId() 

def EVT_RESULT(win, func): 
    """Define Result Event.""" 
    win.Connect(-1, -1, EVT_RESULT_ID, func) 

class ResultEvent(wx.PyEvent): 
    """Simple event to carry arbitrary result data.""" 
    def __init__(self, data): 
     """Init Result Event.""" 
     wx.PyEvent.__init__(self) 
     self.SetEventType(EVT_RESULT_ID) 
     self.data = data 
     print "inside result event" 

class WorkerThread(Thread): 
    """Worker Thread Class.""" 
    def __init__(self, notify_window, delay): 
     """Init Worker Thread Class.""" 
     Thread.__init__(self) 
     self._notify_window = notify_window 
     self._want_abort = 0 
     self._delay = delay 
     self.start() 

    def run(self): 
     """Run Worker Thread.""" 
     for i in range(self._delay): 
      print "thread running..." 
      time.sleep(1) 
      if self._want_abort: 
       # Use a result of None to acknowledge the abort (of 
       # course you can use whatever you'd like or even 
       # a separate event type) 
       wx.PostEvent(self._notify_window, ResultEvent(None)) 
       return 
     wx.PostEvent(self._notify_window, ResultEvent("My result")) 

    def abort(self): 
     """abort worker thread.""" 
     self._want_abort = 1 

class MainFrame(wx.Frame): 
    """Class MainFrame.""" 
    def __init__(self, parent, id): 
     """Create the MainFrame.""" 
     wx.Frame.__init__(self, parent, id, 'Thread Test') 

     # Dumb sample frame with two buttons 
     wx.Button(self, ID_START, 'Start', pos=(0,0)) 
     wx.Button(self, ID_STOP, 'Stop', pos=(0,50)) 
     self.status = wx.StaticText(self, -1, '', pos=(0,100)) 

     self.Bind(wx.EVT_BUTTON, self.OnStart, id=ID_START) 
     self.Bind(wx.EVT_BUTTON, self.OnStop, id=ID_STOP) 

     # Set up event handler for any worker thread results 
     EVT_RESULT(self,self.OnResult) 

     self.worker = None 
     self.thread_running = False 

    def OnStart(self, event): 
     """Start Computation.""" 
     print "OnStart" 
     self.thread_running = True 
     if not self.worker: 
      self.status.SetLabel('Starting computation') 
      self.worker = WorkerThread(self, 3) 

    def OnStop(self, event): 
     """Stop Computation.""" 
     print "OnStop" 
     if self.worker: 
      self.status.SetLabel('Trying to abort computation') 
      self.worker.abort() 
     else: 
      print "no worker" 

    def OnResult(self, event): 
     """Show Result status.""" 
     # NEVER GETS INSIDE HERE! 
     print "ON RESULT" 
     self.thread_running = False 
     if event.data is None: 
      self.status.SetLabel('Computation aborted') 
     else: 
      self.status.SetLabel('Computation Result: %s' % event.data) 
     self.worker = None 

class MainApp(wx.App): 
    """Class Main App.""" 
    def OnInit(self): 
     """Init Main App.""" 
     self.frame = MainFrame(None, -1) 
     self.frame.Show(True) 
     self.SetTopWindow(self.frame) 
     return True 

class TestGui(unittest.TestCase): 
    def setUp(self): 
     print "set up" 
     self.app = MainApp(0) 
     self.frame = self.app.frame 
     # Need MainLoop to capture ResultEvent, but how to test? 
     # self.app.MainLoop() 

    def tearDown(self): 
     print "tear down" 
     self.frame.Destroy() 

    def test1(self): 
     self.assertTrue(self.frame.worker is None) 
     self.assertEquals(self.frame.thread_running, False) 
     self.frame.OnStart(None) 
     self.assertTrue(self.frame.worker is not None) 
     self.assertEquals(self.frame.thread_running, True) 
     while self.frame.thread_running: 
      print 'waiting...' 
      time.sleep(.5) 
      # NEVER EXITS! 
     self.assertTrue(self.frame.worker is None) 

def suite(): 
    suite = unittest.makeSuite(TestGui, 'test') 
    return suite 

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

Answer 1

zmienić metodę run() Twojego wątku roboczego do:

def run(self): 
    for i in range(self._delay): 
     print "thread running..." 
     time.sleep(1) 
     if self._want_abort: 
      self.work_done(None) 
      return 
    self.work_done("My result") 

def work_done(self, result): 
    self.result = result 
    wx.PostEvent(self._notify_window, ResultEvent(result)) 

potem w swojej funkcji test1 zastąpić pętli while z:

frame.worker.join() 
frame.OnResult(ResultEvent(frame.worker.result)) 

Nie pętla zdarzenie potrzebne.