tests/python/testproxy.py

   1 #!/usr/bin/env python
   2 from __future__ import generators
   3 import unittest
   4 import sys
   5 import os
   6 from os.path import dirname, abspath, join
   7 import tempfile, shutil
   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 proxy, tasks, g, suchild, master_proxy
  13
  14 class Slave(suchild.Slave):
  15         def invoke(self):
  16                 return "Invoked"
  17
  18 class TestProxy(unittest.TestCase):
  19         master = None
  20         slave = None
  21
  22         def setUp(self):
  23                 to_slave = os.pipe()
  24                 from_slave = os.pipe()
  25                 self.master = master_proxy.MasterProxy(to_slave[1], from_slave[0])
  26                 self.slave = proxy.SlaveProxy(from_slave[1], to_slave[0], Slave())
  27
  28         def tearDown(self):
  29                 self.master.finish()
  30                 self.slave.finish()
  31
  32         def testSetup(self):
  33                 pass
  34
  35         def testManual(self):
  36                 response = self.master.root.invoke()
  37                 self.master.write_ready()
  38                 self.slave.read_ready()
  39                 self.slave.write_ready()
  40                 assert not response.happened
  41                 self.master.read_ready()
  42                 assert response.happened
  43                 self.assertEquals('Invoked', response.result)
  44
  45         def testSingle(self):
  46                 blocker = self.master.root.invoke()
  47                 self.master.write_ready()
  48                 self.slave.read_ready()
  49                 self.slave.write_ready()
  50                 self.master.read_ready()
  51                 self.assertEquals('Invoked', blocker.result)
  52
  53         def testMissing(self):
  54                 def run():
  55                         response = self.master.root.missing('foo')
  56                         yield response
  57                         try:
  58                                 response.result
  59                                 assert 0, 'Expected an exception!'
  60                         except AttributeError:
  61                                 pass
  62                         g.mainquit()
  63                 tasks.Task(run())
  64                 g.mainloop()
  65
  66         def testTooSoon(self):
  67                 def run():
  68                         response = self.master.root.invoke()
  69                         try:
  70                                 response.result
  71                                 assert 0, 'Expected an exception!'
  72                         except Exception:
  73                                 pass
  74                         yield response
  75                         response.result
  76                         g.mainquit()
  77                 tasks.Task(run())
  78                 g.mainloop()
  79
  80
  81         # spawnvpe, waitpid, setuid and getuid are tested in testsu.py
  82
  83         def testRmTree(self):
  84                 tmp_dir = tempfile.mkdtemp('-roxlib-test')
  85                 def run():
  86                         assert os.path.isdir(tmp_dir)
  87                         response = self.master.root.rmtree(tmp_dir)
  88                         yield response
  89                         assert response.result is None
  90                         assert not os.path.exists(tmp_dir)
  91                         g.mainquit()
  92                 tasks.Task(run())
  93                 g.mainloop()
  94
  95         def testUnlink(self):
  96                 fd, tmp = tempfile.mkstemp('-roxlib-test')
  97                 os.close(fd)
  98                 def run():
  99                         assert os.path.isfile(tmp)
 100                         response = self.master.root.unlink(tmp)
 101                         yield response
 102                         assert response.result is None
 103                         assert not os.path.exists(tmp)
 104                         g.mainquit()
 105                 tasks.Task(run())
 106                 g.mainloop()
 107
 108         def testFileRead(self):
 109                 tmp_file = tempfile.NamedTemporaryFile(suffix = '-roxlib-test')
 110                 tmp_file.write('Hello\n')
 111                 tmp_file.flush()
 112                 root = self.master.root
 113                 def run():
 114                         response = root.open(tmp_file.name)
 115                         yield response
 116                         stream = response.result
 117
 118                         response = root.read(stream, 5)
 119                         yield response
 120                         assert "Hello" == response.result
 121
 122                         response = root.close(stream)
 123                         yield response
 124                         assert response.result is None
 125
 126                         g.mainquit()
 127                 tasks.Task(run())
 128                 g.mainloop()
 129
 130         def testFileWrite(self):
 131                 tmp_dir = tempfile.mkdtemp('-roxlib-test')
 132                 root = self.master.root
 133                 tmp_file = join(tmp_dir, 'new')
 134                 def run():
 135                         response = root.open(tmp_file, 'w')
 136                         yield response
 137                         stream = response.result
 138
 139                         assert os.path.isfile(tmp_file)
 140
 141                         response = root.write(stream, 'Hello\n')
 142                         yield response
 143                         assert response.result == None
 144
 145                         response = root.close(stream)
 146                         yield response
 147                         assert response.result is None
 148
 149                         assert file(tmp_file).read() == 'Hello\n'
 150
 151                         response = root.close(stream)
 152                         yield response
 153                         try:
 154                                 response.result
 155                                 assert 0, 'Expected an exception!'
 156                         except KeyError:
 157                                 pass
 158
 159                         g.mainquit()
 160                 tasks.Task(run())
 161                 g.mainloop()
 162                 shutil.rmtree(tmp_dir)
 163
 164         def testRename(self):
 165                 tmp_dir = tempfile.mkdtemp('-roxlib-test')
 166                 root = self.master.root
 167                 f = file(join(tmp_dir, 'old'), 'w')
 168                 f.write('Hello\n')
 169                 f.close()
 170
 171                 def run():
 172                         response = root.rename(join(tmp_dir, 'old'),
 173                                             join(tmp_dir, 'new'))
 174                         yield response
 175                         assert response.result == None
 176
 177                         assert file(join(tmp_dir, 'new')).read() == 'Hello\n'
 178
 179                         g.mainquit()
 180                 tasks.Task(run())
 181                 g.mainloop()
 182                 shutil.rmtree(tmp_dir)
 183
 184         def testChmod(self):
 185                 tmp_file = tempfile.NamedTemporaryFile(suffix = '-roxlib-test')
 186                 root = self.master.root
 187                 os.chmod(tmp_file.name, 0700)
 188
 189                 def run():
 190                         assert os.stat(tmp_file.name).st_mode & 0777 == 0700
 191                         response = root.chmod(tmp_file.name, 0655)
 192                         yield response
 193                         response.result
 194                         assert os.stat(tmp_file.name).st_mode & 0777 == 0655
 195                         g.mainquit()
 196                 tasks.Task(run())
 197                 g.mainloop()
 198                 tmp_file = None
 199
 200 sys.argv.append('-v')
 201 unittest.main()