unix4lyfe git - buftee/blob - tests/test.py

   1 #!/usr/bin/env python
   2 """
   3 Test buftee functionality.
   4 """
   5 __author__ = 'Emil Mikulic'
   6
   7 import unittest
   8 import popen2
   9 import os
  10 import fcntl
  11 import signal
  12 import select
  13
  14 buftee = './buftee'
  15 assert os.path.exists(buftee), 'can\'t find buftee (path was %s)' % repr(buftee)
  16
  17 def wait_for_read(fd, max_wait_sec = 0.1):
  18   r, _, _ = select.select([fd], [], [], max_wait_sec)
  19   return r == [fd]
  20
  21 def unblock(fd):
  22   """Set a file descriptor to be non-blocking.  Don't readline() after this."""
  23   flags = fcntl.fcntl(fd, fcntl.F_GETFL)
  24   fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
  25
  26
  27 class BufteeTestCase(unittest.TestCase):
  28   def popen2(self, cmd, bufsize=-1):
  29     #stdout, stdin, stderr = popen2.popen3(buftee)
  30     self.child_stdout, self.child_stdin = popen2.popen2(cmd, bufsize)
  31     unblock(self.child_stdout)
  32
  33   def write(self, buf):
  34     signal.alarm(1)
  35     ret = os.write(self.child_stdin.fileno(), buf)
  36     signal.alarm(0)
  37     self.assertEqual(ret, len(buf), 'short write!')
  38
  39   def read(self):
  40     signal.alarm(1)
  41     ret = self.child_stdout.read()
  42     signal.alarm(0)
  43     return ret
  44
  45   def assertRead(self, what):
  46     self.assertTrue(wait_for_read(self.child_stdout))
  47     self.assertEqual(what, self.read())
  48
  49   def assertReadEOF(self):
  50     self.assertTrue(wait_for_read(self.child_stdout))
  51     self.assertEqual('', self.read())
  52
  53   def close(self):
  54     self.child_stdin.close()
  55
  56
  57   def testWriteRead(self):
  58     self.popen2(buftee)
  59     msg = 'hello!\n'
  60     self.write(msg)
  61     self.assertRead(msg)
  62     self.close()
  63     self.assertReadEOF()
  64
  65   def testWriteCloseRead(self):
  66     self.popen2(buftee)
  67     msg = 'hello!\n'
  68     self.write(msg)
  69     self.close()
  70     self.assertRead(msg)
  71     self.assertReadEOF()
  72
  73   def testTwoWrites(self):
  74     self.popen2(buftee)
  75     msg1 = 'hello\n'
  76     msg2 = 'world.\n'
  77     self.write(msg1)
  78     self.write(msg2)
  79     self.close()
  80     self.assertRead(msg1+msg2)
  81     self.assertReadEOF()
  82
  83 if __name__ == '__main__':
  84   unittest.main()