wxPython/tests/test_delayedResult.py

   1
   2 import wx
   3 import delayedresult as dr
   4
   5
   6 def testStruct():
   7     ss=dr.Struct(a='a', b='b')
   8     assert ss.a == 'a'
   9     assert ss.b == 'b'
  10
  11
  12 def testHandler():
  13     def handler(b, d, a=None, c=None):
  14         assert a=='a'
  15         assert b=='b'
  16         assert c=='c'
  17         assert d==1
  18     hh=dr.Handler(handler, 1, a='a')
  19     hh('b', c='c')
  20
  21     def handler2(*args, **kwargs):
  22         assert args[0] == 3
  23         assert args[1] == 1
  24         assert kwargs['a'] == 'a'
  25         assert kwargs['b'] == 'b'
  26     hh2 = dr.Handler(handler2, 1, a='a')
  27     args = ()
  28     hh3 = dr.Handler(hh2, b='b', *args)
  29     hh3(3)
  30
  31
  32 def testSender():
  33     triplet = (1,'a',2.34)
  34     b = dr.Struct(called=False, which=1)
  35     assert not b.called
  36     def consumer(result, a, b=None):
  37         assert result.get() == 789 + b.which
  38         assert result.getJobID() == 456
  39         assert a == 'a'
  40         b.called = True
  41     handler = dr.Handler(consumer, 'a', **dict(b=b))
  42     ss = dr.SenderNoWx( handler, jobID=456 )
  43     ss.sendResult(789+1)
  44     assert b.called
  45
  46
  47 def testSendExcept():
  48     def consumer(result):
  49         try:
  50             result.get()
  51             raise RuntimeError('should have raised!')
  52         except AssertionError:
  53             pass
  54     ss = dr.SenderNoWx( dr.Handler(consumer) )
  55     ss.sendException( AssertionError('test') )
  56
  57
  58 def testThread():
  59     expect = dr.Struct(value=1)
  60     def consumer(result):
  61         assert result.getJobID() is None
  62         assert result.get() == expect.value
  63         expect.value += 2
  64     ss = dr.SenderNoWx( dr.Handler(consumer) )
  65     import time
  66     def worker(sender=None):
  67         sender.sendResult(1)
  68         time.sleep(0.1)
  69         sender.sendResult(3)
  70         time.sleep(0.1)
  71         sender.sendResult(5)
  72         time.sleep(0.1)
  73         return 7
  74     tt = dr.Producer(ss, worker, senderArg='sender')
  75     tt.start()
  76     while expect.value < 7:
  77         time.sleep(0.1)
  78         print '.'
  79
  80
  81 def testStartWorker():
  82     print 'Doing worker thread with call-after'
  83     import time
  84     def handleButtonClick():
  85         produce = [123, 456, 789, 012, 345, 678, 901]
  86         expect = dr.Struct(idx=0)
  87         def worker(a, b=None, sender=None):
  88             assert a == 2
  89             assert b == 'b'
  90             for val in produce:
  91                 time.sleep(0.5)
  92                 sender.sendResult(val)
  93         def consumer(result, b, a=None):
  94             assert b == 1
  95             assert a=='a'
  96             result = result.get()
  97             print 'got result', result
  98             if expect.idx < len(produce):
  99                 assert result == produce[ expect.idx ]#, 'Expected %s, got %s' % (
 100             else:
 101                 assert result is None
 102                 app.ExitMainLoop()
 103             expect.idx += 1
 104
 105         dr.startWorker(consumer, worker, cargs=(1,), ckwargs={'a':'a'},
 106             wargs=(2,), wkwargs={'b':'b'}, senderArg='sender')
 107
 108     app = wx.PySimpleApp()
 109     frame = wx.Frame(None) # need this otherwise MainLoop() returns immediately
 110     # pretend user has clicked:
 111     import thread
 112     thread.start_new_thread( wx.CallAfter, (handleButtonClick,))
 113     app.MainLoop()
 114
 115
 116 def testStartWorkerEvent():
 117     print 'Doing same with events'
 118     import time
 119     produce = [123, 456, 789, 012, 345, 678, 901]
 120     expect = dr.Struct(idx=0)
 121     def worker(a, b=None, sender=None):
 122         assert a == 2
 123         assert b == 'b'
 124         for val in produce:
 125             time.sleep(0.5)
 126             sender.sendResult(val)
 127     def consumer(event):
 128         assert event.a=='a'
 129         result = event.result.get()
 130         print 'got result', result
 131         if expect.idx < len(produce):
 132             assert result == produce[ expect.idx ]#, 'Expected %s, got %s' % (
 133         else:
 134             assert result is None
 135             app.ExitMainLoop()
 136         expect.idx += 1
 137     def handleButtonClick():
 138         dr.startWorker(frame, worker,
 139             cargs=(eventClass,), ckwargs={'a':'a','resultAttr':'result'},
 140             wargs=(2,), wkwargs={'b':'b'}, senderArg='sender')
 141
 142     app = wx.PySimpleApp()
 143     frame = wx.Frame(None) # need this otherwise MainLoop() returns immediately
 144     from wx.lib.newevent import NewEvent as wxNewEvent
 145     eventClass, eventBinder = wxNewEvent()
 146     frame.Bind(eventBinder, consumer)
 147     # pretend user has clicked:
 148     import thread
 149     thread.start_new_thread( wx.CallAfter, (handleButtonClick,))
 150     app.MainLoop()
 151
 152
 153 def testAbort():
 154     import threading
 155     abort = dr.AbortEvent()
 156
 157     # create a wx app and a function that will cause
 158     # app to close when abort occurs
 159     app = wx.PySimpleApp()
 160     frame = wx.Frame(None) # need this otherwise MainLoop() returns immediately
 161     def exiter():
 162         abort.wait()
 163         # make sure any events have time to be processed before exit
 164         wx.FutureCall(2000, app.ExitMainLoop)
 165     threading.Thread(target=exiter).start()
 166
 167     # now do the delayed result computation:
 168     def worker():
 169         count = 0
 170         while not abort(1):
 171             print 'Result computation not done, not aborted'
 172         return 'Result computed'
 173     def consumer(dr): # never gets called but as example
 174         print 'Got dr=', dr.get()
 175         app.ExitMainLoop()
 176     dr.startWorker(consumer, worker)
 177
 178     # pretend user doing other stuff
 179     import time
 180     time.sleep(5)
 181     # pretend user aborts now:
 182     print 'Setting abort event'
 183     abort.set()
 184     app.MainLoop()
 185
 186
 187 def testPreProcChain():
 188     # test when no chain
 189     def handler(dr):
 190         assert dr.getJobID() == 123
 191         assert dr.get() == 321
 192     pp=dr.PreProcessChain( handler )
 193     pp( dr.DelayedResult(321, jobID=123) )
 194
 195     # test with chaining
 196     def handlerPP(chainTrav, n, a=None):
 197         print 'In handlerPP'
 198         assert n==1
 199         assert a=='a'
 200         assert chainTrav.getJobID() == 321
 201         res = chainTrav.get()
 202         assert res == 135
 203         print 'Done handlerPP'
 204
 205     def subStart1(handler):
 206         pp=dr.PreProcessChain(handler)
 207         pp.addSub(subEnd1, 1, b='b')
 208         subStart2(pp.clone())
 209     def subEnd1(chainTrav, aa, b=None):
 210         print 'In subEnd1'
 211         assert aa==1
 212         assert b=='b'
 213         assert chainTrav.getJobID() == 321
 214         res = chainTrav.get()
 215         assert res == 246, 'res=%s' % res
 216         print 'Returning from subEnd1'
 217         return res - 111
 218
 219     def subStart2(preProc):
 220         preProc.addSub(subEnd2, 3, c='c')
 221         ss = dr.SenderNoWx(preProc, jobID=321)
 222         ss.sendResult(123)
 223     def subEnd2(chainTrav, a, c=None):
 224         print 'In subEnd2'
 225         assert a==3
 226         assert c=='c'
 227         assert chainTrav.getJobID() == 321
 228         res = chainTrav.get()
 229         assert res == 123
 230         print 'Returning from subEnd2'
 231         return 123*2
 232
 233     subStart1( dr.Handler(handlerPP, 1, a='a') )
 234
 235 testStruct()
 236 testHandler()
 237 testSender()
 238 testSendExcept()
 239 testThread()
 240 testStartWorker()
 241 testStartWorkerEvent()
 242 testAbort()
 243 testPreProcChain()
 244