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_long_message(self):
  29         s = server.Server(self.root_dir)
  30         s.run()
  31         c = server.Client(self.root_dir)
  32         c.send_msg("1111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111")
  33         reply = c.read_reply()
  34         self.assertTrue(reply == "Reply to: 1111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111")
  35         c.send_stop()
  36         reply = c.read_reply()
  37         self.assertTrue(reply == "Stopping server")
  38
  39     def test_ssh_server(self):
  40         env = test_util.test_environment()
  41         user = getpass.getuser()
  42         # launch server
  43         python_code = "from nepi.util import server;s=server.Server('%s');\
  44                 s.run()" % self.root_dir
  45         server.popen_ssh_subprocess(python_code, host = "localhost",
  46                 port = env.port, user = user, agent = True)
  47         c = server.Client(self.root_dir, host = "localhost", port = env.port,
  48                 user = user, agent = True)
  49         c.send_msg("Hola")
  50         reply = c.read_reply()
  51         self.assertTrue(reply == "Reply to: Hola")
  52         c.send_stop()
  53         reply = c.read_reply()
  54         self.assertTrue(reply == "Stopping server")
  55
  56     def tearDown(self):
  57         shutil.rmtree(self.root_dir)
  58
  59 if __name__ == '__main__':
  60     unittest.main()
  61