Come avere gli eventi thread di cattura wxpython in un test unitario?

Question

Ho costruito un'app python wx che esegue un thread in background per eseguire alcuni calcoli. Tuttavia, la mia attuale implementazione non consente il test dell'unità.

ho molto strettamente basato la mia implementazione iniziale del primo esempio di questo manuale: http://wiki.wxpython.org/LongRunningTasks

Il seguente codice utilizza wx.PostEvent() e frame.connect() avere il registrare un evento risultato al telaio principale che indica che il calcolo è completata. Ho anche mostrato uno snippet di codice di test unitario.

Tuttavia, per l'evento del risultato del thread da catturare, è necessario avviare wx.App.MainLoop(). Tuttavia, non so come simulare tale comportamento in un test unitario.

La mia comprensione dei test dell'unità GUI in generale è di simulare manualmente gli eventi. Tuttavia, in questo caso, mi piacerebbe avere il mio thread in background. Dovrei modificare l'implementazione? Oppure, ai fini del test unitario, posso escludere il thread di calcolo in qualche altro modo? Ad esempio, dovrei eseguire il thread nel codice di test dell'unità e una volta completato, chiamare il codice GUI per gestire direttamente questo evento?

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

Modificare il metodo run() del thread di lavoro per:

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)) 

quindi nella funzione test1 sostituire il vostro ciclo while con:

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

Nessun ciclo di eventi necessari.