2 # -*- coding: utf-8 -*-
5 from nepi.util import server
13 class ServerTestCase(unittest.TestCase):
15 self.root_dir = tempfile.mkdtemp()
17 def test_server(self):
18 s = server.Server(self.root_dir)
20 c = server.Client(self.root_dir)
22 reply = c.read_reply()
23 self.assertTrue(reply == "Reply to: Hola")
25 reply = c.read_reply()
26 self.assertTrue(reply == "Stopping server")
28 def test_server_reconnect(self):
29 s = server.Server(self.root_dir)
31 c = server.Client(self.root_dir)
34 reply = c.read_reply()
35 self.assertTrue(reply == "Reply to: Hola")
41 c = server.Client(self.root_dir)
43 reply = c.read_reply()
44 self.assertTrue(reply == "Reply to: Hola")
47 reply = c.read_reply()
48 self.assertTrue(reply == "Stopping server")
50 def test_server_long_message(self):
51 s = server.Server(self.root_dir)
53 c = server.Client(self.root_dir)
56 reply = c.read_reply()
57 self.assertTrue(reply == ("Reply to: "+msg))
59 reply = c.read_reply()
60 self.assertTrue(reply == "Stopping server")
62 def test_ssh_server(self):
63 env = test_util.test_environment()
64 user = getpass.getuser()
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)
73 reply = c.read_reply()
74 self.assertTrue(reply == "Reply to: Hola")
76 reply = c.read_reply()
77 self.assertTrue(reply == "Stopping server")
79 def test_ssh_server_reconnect(self):
80 env = test_util.test_environment()
81 user = getpass.getuser()
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)
88 c = server.Client(self.root_dir, host = "localhost", port = env.port,
89 user = user, agent = True)
92 reply = c.read_reply()
93 self.assertTrue(reply == "Reply to: Hola")
99 c = server.Client(self.root_dir, host = "localhost", port = env.port,
100 user = user, agent = True)
103 reply = c.read_reply()
104 self.assertTrue(reply == "Reply to: Hola")
107 reply = c.read_reply()
108 self.assertTrue(reply == "Stopping server")
111 shutil.rmtree(self.root_dir)
113 if __name__ == '__main__':