ROX-Lib2/tests/python/testtasks.py

   1 #!/usr/bin/env python2.3
   2 from __future__ import generators
   3 import unittest
   4 import sys
   5 import os, time
   6 from os.path import dirname, abspath, join
   7
   8 rox_lib = dirname(dirname(dirname(abspath(sys.argv[0]))))
   9 sys.path.insert(0, join(rox_lib, 'python'))
  10
  11 from rox import tasks, g
  12
  13 class TestTasks(unittest.TestCase):
  14         def testIdleBlocker(self):
  15                 def run():
  16                         yield None
  17                         g.main_quit()
  18                 tasks.Task(run())
  19                 g.main()
  20
  21         def testTimeoutBlocker(self):
  22                 def run():
  23                         start = time.time()
  24                         yield tasks.TimeoutBlocker(0.5)
  25                         end = time.time()
  26                         assert end > start + 0.5
  27                         g.main_quit()
  28                 tasks.Task(run())
  29                 g.main()
  30
  31         def testInputBlocker(self):
  32                 readable, writeable = os.pipe()
  33                 def run():
  34                         ib = tasks.InputBlocker(readable)
  35                         tb = tasks.TimeoutBlocker(0.2)
  36                         yield ib, tb
  37                         assert not ib.happened
  38                         assert tb.happened
  39                         os.write(writeable, "!")
  40
  41                         tb = tasks.TimeoutBlocker(0.2)
  42                         yield ib, tb
  43                         assert ib.happened
  44                         assert not tb.happened
  45
  46                         assert os.read(readable, 1) == '!'
  47                         os.close(writeable)
  48                         ib = tasks.InputBlocker(readable)
  49                         yield ib
  50                         assert ib.happened
  51
  52                         g.main_quit()
  53                 tasks.Task(run())
  54                 g.main()
  55
  56         def testOutputBlocker(self):
  57                 readable, writeable = os.pipe()
  58                 def run():
  59                         # Fill the input buffer...
  60                         sent = 0
  61                         while True:
  62                                 ob = tasks.OutputBlocker(writeable)
  63                                 tb = tasks.TimeoutBlocker(0.2)
  64                                 yield ob, tb
  65                                 if ob.happened:
  66                                         sent += os.write(writeable, 'Hello\n')
  67                                 else:
  68                                         assert tb.happened
  69                                         break
  70                         assert sent > 0
  71                         #print "send %d bytes" % sent
  72
  73                         # Read it all back...
  74                         got = 0
  75                         while got < sent:
  76                                 got += len(os.read(readable, sent - got))
  77
  78                         ob = tasks.OutputBlocker(writeable)
  79                         tb = tasks.TimeoutBlocker(0.2)
  80                         yield ob, tb
  81                         assert ob.happened
  82                         assert not tb.happened
  83
  84                         g.main_quit()
  85                 tasks.Task(run())
  86                 g.main()
  87
  88 suite = unittest.makeSuite(TestTasks)
  89 if __name__ == '__main__':
  90         sys.argv.append('-v')
  91         unittest.main()