ROX-Lib2/tests/python/testprocesses.py

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