tests/python/testprocesses.py

   1 #!/usr/bin/env python2.2
   2 from __future__ import generators
   3 import unittest
   4 import sys
   5 import os, time, gc
   6 from os.path import dirname, abspath, join
   7 from cStringIO import StringIO
   8
   9 rox_lib = dirname(dirname(dirname(abspath(sys.argv[0]))))
  10 sys.path.insert(0, join(rox_lib, 'python'))
  11
  12 from rox import processes, g
  13
  14 def pipe_through_command(command, src, dst):
  15         processes.PipeThroughCommand(command, src, dst).wait()
  16
  17 class TestProcesses(unittest.TestCase):
  18         def testTmp(self):
  19                 tmp_file = processes._Tmp()
  20                 tmp_file.write('Hello')
  21                 print >>tmp_file, ' ',
  22                 tmp_file.flush()
  23                 os.write(tmp_file.fileno(), 'World')
  24
  25                 tmp_file.seek(0)
  26                 assert tmp_file.read() == 'Hello World'
  27
  28         def testInvalidCommand(self):
  29                 try:
  30                         pipe_through_command('bad_command_1234', None, None)
  31                         assert 0
  32                 except processes.ChildError, ex:
  33                         pass
  34                 else:
  35                         assert 0
  36
  37         def testValidCommand(self):
  38                 pipe_through_command('exit 0', None, None)
  39
  40         def testNonFileno(self):
  41                 a = StringIO()
  42                 pipe_through_command('echo Hello', None, a)
  43                 assert a.getvalue() == 'Hello\n'
  44
  45         def testStringIO(self):
  46                 a = StringIO()
  47                 pipe_through_command('echo Hello', None, a)
  48                 tmp_file = processes._Tmp()
  49                 tmp_file.write('Hello World')
  50                 tmp_file.seek(1)
  51                 pipe_through_command('cat', tmp_file, a)
  52                 assert a.getvalue() == 'Hello\nHello World'
  53
  54         def testWriteFileno(self):
  55                 tmp_file = processes._Tmp()
  56                 tmp_file.seek(0)
  57                 tmp_file.truncate(0)
  58                 pipe_through_command('echo Foo', None, tmp_file)
  59                 tmp_file.seek(0)
  60                 assert tmp_file.read() == 'Foo\n'
  61
  62         def testRWfile(self):
  63                 tmp_file = processes._Tmp()
  64                 tmp_file.write('Hello World')
  65                 src = processes._Tmp()
  66                 src.write('123')
  67                 src.seek(0)
  68                 tmp_file.seek(0)
  69                 tmp_file.truncate(0)
  70                 pipe_through_command('cat', src, tmp_file)
  71                 tmp_file.seek(0)
  72                 assert tmp_file.read() == '123'
  73
  74         def testNonZeroExit(self):
  75                 try:
  76                         pipe_through_command('exit 1', None, None)
  77                 except processes.ChildError:
  78                         pass
  79                 else:
  80                         assert 0
  81
  82         def testStderr(self):
  83                 try:
  84                         pipe_through_command('echo one >&2; sleep 2; echo two >&2', None, None)
  85                 except processes.ChildError:
  86                         pass
  87                 else:
  88                         assert 0
  89
  90         def testDelTmp(self):
  91                 tmp_file = processes._Tmp()
  92                 name = tmp_file.name
  93                 assert os.path.exists(name)
  94                 tmp_file = None
  95                 gc.collect()
  96                 assert not os.path.exists(name)
  97
  98         def testKillRunaway(self):
  99                 ptc = processes.PipeThroughCommand('sleep 100; exit 1', None, None)
 100                 def stop():
 101                         ptc.kill()
 102                 g.timeout_add(2000, stop)
 103                 try:
 104                         ptc.wait()
 105                         assert 0
 106                 except processes.ChildKilled:
 107                         pass
 108
 109 suite = unittest.makeSuite(TestProcesses)
 110 if __name__ == '__main__':
 111         sys.argv.append('-v')
 112         unittest.main()