2 # -*- coding: utf-8 -*-
4 from nepi.util import server
5 from nepi.util.constants import DeploymentConfiguration as DC
16 class ServerTestCase(unittest.TestCase):
18 self.root_dir = tempfile.mkdtemp()
20 def test_server(self):
21 s = server.Server(self.root_dir)
23 c = server.Client(self.root_dir)
25 reply = c.read_reply()
26 self.assertEqual(reply, "Reply to: Hola")
28 reply = c.read_reply()
29 self.assertEqual(reply, "Stopping server")
31 def test_server_reconnect(self):
32 s = server.Server(self.root_dir)
34 c = server.Client(self.root_dir)
37 reply = c.read_reply()
38 self.assertEqual(reply, "Reply to: Hola")
44 c = server.Client(self.root_dir)
46 reply = c.read_reply()
47 self.assertEqual(reply, "Reply to: Hola")
50 reply = c.read_reply()
51 self.assertEqual(reply, "Stopping server")
53 def test_server_auto_reconnect(self):
54 s = server.Server(self.root_dir)
56 c = server.Client(self.root_dir)
59 reply = c.read_reply()
60 self.assertEqual(reply, "Reply to: Hola")
62 # purposedly break the connection
63 c._process.stdin.close()
64 c._process.stdout.close()
65 c._process.stderr.close()
67 # assert that the communication works (possible with auto-reconnection)
69 reply = c.read_reply()
70 self.assertEqual(reply, "Reply to: Hola")
73 reply = c.read_reply()
74 self.assertEqual(reply, "Stopping server")
76 def test_server_long_message(self):
77 s = server.Server(self.root_dir)
79 c = server.Client(self.root_dir)
82 reply = c.read_reply()
83 self.assertEqual(reply, ("Reply to: "+msg))
85 reply = c.read_reply()
86 self.assertEqual(reply, "Stopping server")
88 @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
89 def test_sudo_server(self):
90 env = test_util.test_environment()
91 user = getpass.getuser()
93 python_code = "from nepi.util import server;s=server.Server('%s');\
94 s.run()" % self.root_dir
95 server.popen_python(python_code,
97 c = server.Client(self.root_dir,
100 reply = c.read_reply()
101 self.assertEqual(reply, "Reply to: Hola")
103 reply = c.read_reply()
104 self.assertEqual(reply, "Stopping server")
107 def test_ssh_server(self):
108 env = test_util.test_environment()
109 user = getpass.getuser()
111 python_code = "from nepi.util import server;s=server.Server('%s');\
112 s.run()" % self.root_dir
113 server.popen_python(python_code,
114 communication = DC.ACCESS_SSH,
119 c = server.Client(self.root_dir,
120 communication = DC.ACCESS_SSH,
126 reply = c.read_reply()
127 self.assertEqual(reply, "Reply to: Hola")
129 reply = c.read_reply()
130 self.assertEqual(reply, "Stopping server")
132 def test_ssh_server_reconnect(self):
133 env = test_util.test_environment()
134 user = getpass.getuser()
136 python_code = "from nepi.util import server;s=server.Server('%s');\
137 s.run()" % self.root_dir
138 server.popen_python(python_code,
139 communication = DC.ACCESS_SSH,
145 c = server.Client(self.root_dir,
146 communication = DC.ACCESS_SSH,
153 reply = c.read_reply()
154 self.assertEqual(reply, "Reply to: Hola")
160 c = server.Client(self.root_dir,
161 communication = DC.ACCESS_SSH,
168 reply = c.read_reply()
169 self.assertEqual(reply, "Reply to: Hola")
172 reply = c.read_reply()
173 self.assertEqual(reply, "Stopping server")
175 def test_ssh_server_auto_reconnect(self):
176 env = test_util.test_environment()
177 user = getpass.getuser()
179 python_code = "from nepi.util import server;s=server.Server('%s');\
180 s.run()" % self.root_dir
181 server.popen_python(python_code,
182 communication = DC.ACCESS_SSH,
188 c = server.Client(self.root_dir,
189 communication = DC.ACCESS_SSH,
196 reply = c.read_reply()
197 self.assertEqual(reply, "Reply to: Hola")
199 # purposedly break the connection
200 c._process.stdin.close()
201 c._process.stdout.close()
202 c._process.stderr.close()
204 # assert that the communication works (possible with auto-reconnection)
206 reply = c.read_reply()
207 self.assertEqual(reply, "Reply to: Hola")
210 reply = c.read_reply()
211 self.assertEqual(reply, "Stopping server")
215 shutil.rmtree(self.root_dir)
219 shutil.rmtree(self.root_dir)
221 if __name__ == '__main__':