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