test/util/server.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 from nepi.util import server
   5 from nepi.util.constants import DeploymentConfiguration as DC
   6
   7 import getpass
   8 import os
   9 import shutil
  10 import sys
  11 import tempfile
  12 import test_util
  13 import unittest
  14 import time
  15
  16 class ServerTestCase(unittest.TestCase):
  17     def setUp(self):
  18         self.root_dir = tempfile.mkdtemp()
  19
  20     def test_server(self):
  21         s = server.Server(self.root_dir)
  22         s.run()
  23         c = server.Client(self.root_dir)
  24         c.send_msg("Hola")
  25         reply = c.read_reply()
  26         self.assertEqual(reply, "Reply to: Hola")
  27         c.send_stop()
  28         reply = c.read_reply()
  29         self.assertEqual(reply, "Stopping server")
  30
  31     def test_server_reconnect(self):
  32         s = server.Server(self.root_dir)
  33         s.run()
  34         c = server.Client(self.root_dir)
  35
  36         c.send_msg("Hola")
  37         reply = c.read_reply()
  38         self.assertEqual(reply, "Reply to: Hola")
  39
  40         # disconnect
  41         del c
  42
  43         # reconnect
  44         c = server.Client(self.root_dir)
  45         c.send_msg("Hola")
  46         reply = c.read_reply()
  47         self.assertEqual(reply, "Reply to: Hola")
  48
  49         c.send_stop()
  50         reply = c.read_reply()
  51         self.assertEqual(reply, "Stopping server")
  52
  53     def test_server_auto_reconnect(self):
  54         s = server.Server(self.root_dir)
  55         s.run()
  56         c = server.Client(self.root_dir)
  57
  58         c.send_msg("Hola")
  59         reply = c.read_reply()
  60         self.assertEqual(reply, "Reply to: Hola")
  61
  62         # purposedly break the connection
  63         c._process.stdin.close()
  64         c._process.stdout.close()
  65         c._process.stderr.close()
  66
  67         # assert that the communication works (possible with auto-reconnection)
  68         c.send_msg("Hola")
  69         reply = c.read_reply()
  70         self.assertEqual(reply, "Reply to: Hola")
  71
  72         c.send_stop()
  73         reply = c.read_reply()
  74         self.assertEqual(reply, "Stopping server")
  75
  76     def test_server_long_message(self):
  77         s = server.Server(self.root_dir)
  78         s.run()
  79         c = server.Client(self.root_dir)
  80         msg = "1"*1145
  81         c.send_msg(msg)
  82         reply = c.read_reply()
  83         self.assertEqual(reply, ("Reply to: "+msg))
  84         c.send_stop()
  85         reply = c.read_reply()
  86         self.assertEqual(reply, "Stopping server")
  87
  88     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
  89     def test_sudo_server(self):
  90         env = test_util.test_environment()
  91         user = getpass.getuser()
  92         # launch server
  93         python_code = "from nepi.util import server;s=server.Server('%s');\
  94                 s.run()" % self.root_dir
  95         server.popen_python(python_code,
  96                 sudo = True)
  97         c = server.Client(self.root_dir,
  98                 sudo = True)
  99         c.send_msg("Hola")
 100         reply = c.read_reply()
 101         self.assertEqual(reply, "Reply to: Hola")
 102         c.send_stop()
 103         reply = c.read_reply()
 104         self.assertEqual(reply, "Stopping server")
 105
 106
 107     def test_ssh_server(self):
 108         env = test_util.test_environment()
 109         user = getpass.getuser()
 110         # launch server
 111         python_code = "from nepi.util import server;s=server.Server('%s');\
 112                 s.run()" % self.root_dir
 113         server.popen_python(python_code,
 114                 communication = DC.ACCESS_SSH,
 115                 host = "localhost",
 116                 port = env.port,
 117                 user = user,
 118                 agent = True)
 119         c = server.Client(self.root_dir,
 120                 communication = DC.ACCESS_SSH,
 121                 host = "localhost",
 122                 port = env.port,
 123                 user = user,
 124                 agent = True)
 125         c.send_msg("Hola")
 126         reply = c.read_reply()
 127         self.assertEqual(reply, "Reply to: Hola")
 128         c.send_stop()
 129         reply = c.read_reply()
 130         self.assertEqual(reply, "Stopping server")
 131
 132     def test_ssh_server_reconnect(self):
 133         env = test_util.test_environment()
 134         user = getpass.getuser()
 135         # launch server
 136         python_code = "from nepi.util import server;s=server.Server('%s');\
 137                 s.run()" % self.root_dir
 138         server.popen_python(python_code,
 139                 communication = DC.ACCESS_SSH,
 140                 host = "localhost",
 141                 port = env.port,
 142                 user = user,
 143                 agent = True)
 144
 145         c = server.Client(self.root_dir,
 146                 communication = DC.ACCESS_SSH,
 147                 host = "localhost",
 148                 port = env.port,
 149                 user = user,
 150                 agent = True)
 151
 152         c.send_msg("Hola")
 153         reply = c.read_reply()
 154         self.assertEqual(reply, "Reply to: Hola")
 155
 156         # disconnect
 157         del c
 158
 159         # reconnect
 160         c = server.Client(self.root_dir,
 161                 communication = DC.ACCESS_SSH,
 162                 host = "localhost",
 163                 port = env.port,
 164                 user = user,
 165                 agent = True)
 166
 167         c.send_msg("Hola")
 168         reply = c.read_reply()
 169         self.assertEqual(reply, "Reply to: Hola")
 170
 171         c.send_stop()
 172         reply = c.read_reply()
 173         self.assertEqual(reply, "Stopping server")
 174
 175     def test_ssh_server_auto_reconnect(self):
 176         env = test_util.test_environment()
 177         user = getpass.getuser()
 178         # launch server
 179         python_code = "from nepi.util import server;s=server.Server('%s');\
 180                 s.run()" % self.root_dir
 181         server.popen_python(python_code,
 182                 communication = DC.ACCESS_SSH,
 183                 host = "localhost",
 184                 port = env.port,
 185                 user = user,
 186                 agent = True)
 187
 188         c = server.Client(self.root_dir,
 189                 communication = DC.ACCESS_SSH,
 190                 host = "localhost",
 191                 port = env.port,
 192                 user = user,
 193                 agent = True)
 194
 195         c.send_msg("Hola")
 196         reply = c.read_reply()
 197         self.assertEqual(reply, "Reply to: Hola")
 198
 199         # purposedly break the connection
 200         c._process.stdin.close()
 201         c._process.stdout.close()
 202         c._process.stderr.close()
 203
 204         # assert that the communication works (possible with auto-reconnection)
 205         c.send_msg("Hola")
 206         reply = c.read_reply()
 207         self.assertEqual(reply, "Reply to: Hola")
 208
 209         c.send_stop()
 210         reply = c.read_reply()
 211         self.assertEqual(reply, "Stopping server")
 212
 213     def tearDown(self):
 214         try:
 215             shutil.rmtree(self.root_dir)
 216         except:
 217             # retry
 218             time.sleep(0.1)
 219             shutil.rmtree(self.root_dir)
 220
 221 if __name__ == '__main__':
 222     unittest.main()
 223