如何讓wxpython在單元測試中捕獲線程事件？

Question

我構建了一個運行後臺線程來執行一些計算的wx python應用程序。但是，我目前的實現不允許進行單元測試。

我非常緊密地基於我的執行關閉本教程中的第一個例子的：http://wiki.wxpython.org/LongRunningTasks

下面的代碼使用wx.PostEvent()和frame.connect()有註冊的結果的事件，其表示計算已完成主框架。我還展示了一個單元測試代碼片段。

但是，要捕獲線程結果事件，必須啓動wx.App.MainLoop()。但是，我不知道如何在單元測試中模擬這種行爲。

我對GUI單元測試的理解一般是手動模擬事件。然而在這種情況下，我想讓我的後臺線程運行。我應該修改實施嗎？或者爲了單元測試的目的，我可以用其他方式來存儲計算線程嗎？例如，我應該在單元測試代碼中運行線程，然後一旦完成，請調用GUI代碼直接處理此事件？

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

更改您的工作線程的run（）方法：

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

然後在test1的函數替換while循環：

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

沒有事件循環需要。