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.assertTrue(reply == "Reply to: Hola")
  24         c.send_stop()
  25         reply = c.read_reply()
  26         self.assertTrue(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.assertTrue(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.assertTrue(reply == "Reply to: Hola")
  45
  46         c.send_stop()
  47         reply = c.read_reply()
  48         self.assertTrue(reply == "Stopping server")
  49
  50     def test_server_long_message(self):
  51         s = server.Server(self.root_dir)
  52         s.run()
  53         c = server.Client(self.root_dir)
  54         msg = "1"*1145
  55         c.send_msg(msg)
  56         reply = c.read_reply()
  57         self.assertTrue(reply == ("Reply to: "+msg))
  58         c.send_stop()
  59         reply = c.read_reply()
  60         self.assertTrue(reply == "Stopping server")
  61
  62     def test_ssh_server(self):
  63         env = test_util.test_environment()
  64         user = getpass.getuser()
  65         # launch server
  66         python_code = "from nepi.util import server;s=server.Server('%s');\
  67                 s.run()" % self.root_dir
  68         server.popen_ssh_subprocess(python_code, host = "localhost",
  69                 port = env.port, user = user, agent = True)
  70         c = server.Client(self.root_dir, host = "localhost", port = env.port,
  71                 user = user, agent = True)
  72         c.send_msg("Hola")
  73         reply = c.read_reply()
  74         self.assertTrue(reply == "Reply to: Hola")
  75         c.send_stop()
  76         reply = c.read_reply()
  77         self.assertTrue(reply == "Stopping server")
  78
  79     def test_ssh_server_reconnect(self):
  80         env = test_util.test_environment()
  81         user = getpass.getuser()
  82         # launch server
  83         python_code = "from nepi.util import server;s=server.Server('%s');\
  84                 s.run()" % self.root_dir
  85         server.popen_ssh_subprocess(python_code, host = "localhost",
  86                 port = env.port, user = user, agent = True)
  87
  88         c = server.Client(self.root_dir, host = "localhost", port = env.port,
  89                 user = user, agent = True)
  90
  91         c.send_msg("Hola")
  92         reply = c.read_reply()
  93         self.assertTrue(reply == "Reply to: Hola")
  94
  95         # disconnect
  96         del c
  97
  98         # reconnect
  99         c = server.Client(self.root_dir, host = "localhost", port = env.port,
 100                 user = user, agent = True)
 101
 102         c.send_msg("Hola")
 103         reply = c.read_reply()
 104         self.assertTrue(reply == "Reply to: Hola")
 105
 106         c.send_stop()
 107         reply = c.read_reply()
 108         self.assertTrue(reply == "Stopping server")
 109
 110     def tearDown(self):
 111         shutil.rmtree(self.root_dir)
 112
 113 if __name__ == '__main__':
 114     unittest.main()
 115