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