test/util/server.py

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