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()
21 self.stderr = sys.stderr
22 sys.stderr = open("/dev/null","r+b")
25 sys.stderr = self.stderr
27 shutil.rmtree(self.root_dir)
31 shutil.rmtree(self.root_dir)
33 def test_server(self):
34 s = server.Server(self.root_dir)
36 c = server.Client(self.root_dir)
38 reply = c.read_reply()
39 self.assertEqual(reply, "Reply to: Hola")
41 reply = c.read_reply()
42 self.assertEqual(reply, "Stopping server")
44 def test_server_reconnect(self):
45 s = server.Server(self.root_dir)
47 c = server.Client(self.root_dir)
50 reply = c.read_reply()
51 self.assertEqual(reply, "Reply to: Hola")
57 c = server.Client(self.root_dir)
59 reply = c.read_reply()
60 self.assertEqual(reply, "Reply to: Hola")
63 reply = c.read_reply()
64 self.assertEqual(reply, "Stopping server")
66 def test_server_auto_reconnect(self):
67 s = server.Server(self.root_dir)
69 c = server.Client(self.root_dir)
72 reply = c.read_reply()
73 self.assertEqual(reply, "Reply to: Hola")
75 # purposedly break the connection
76 c._process.stdin.close()
77 c._process.stdout.close()
78 c._process.stderr.close()
80 # assert that the communication works (possible with auto-reconnection)
82 reply = c.read_reply()
83 self.assertEqual(reply, "Reply to: Hola")
86 reply = c.read_reply()
87 self.assertEqual(reply, "Stopping server")
89 def test_server_long_message(self):
90 s = server.Server(self.root_dir)
92 c = server.Client(self.root_dir)
95 reply = c.read_reply()
96 self.assertEqual(reply, ("Reply to: "+msg))
98 reply = c.read_reply()
99 self.assertEqual(reply, "Stopping server")
101 @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
102 def test_sudo_server(self):
103 env = test_util.test_environment()
104 user = getpass.getuser()
106 python_code = "from nepi.util import server;s=server.Server('%s');\
107 s.run()" % self.root_dir
108 server.popen_python(python_code,
110 c = server.Client(self.root_dir,
113 reply = c.read_reply()
114 self.assertEqual(reply, "Reply to: Hola")
116 reply = c.read_reply()
117 self.assertEqual(reply, "Stopping server")
120 def test_ssh_server(self):
121 env = test_util.test_environment()
122 user = getpass.getuser()
124 python_code = "from nepi.util import server;s=server.Server('%s');\
125 s.run()" % self.root_dir
126 server.popen_python(python_code,
127 communication = DC.ACCESS_SSH,
132 c = server.Client(self.root_dir,
133 communication = DC.ACCESS_SSH,
139 reply = c.read_reply()
140 self.assertEqual(reply, "Reply to: Hola")
142 reply = c.read_reply()
143 self.assertEqual(reply, "Stopping server")
145 def test_ssh_server_reconnect(self):
146 env = test_util.test_environment()
147 user = getpass.getuser()
149 python_code = "from nepi.util import server;s=server.Server('%s');\
150 s.run()" % self.root_dir
151 server.popen_python(python_code,
152 communication = DC.ACCESS_SSH,
158 c = server.Client(self.root_dir,
159 communication = DC.ACCESS_SSH,
166 reply = c.read_reply()
167 self.assertEqual(reply, "Reply to: Hola")
173 c = server.Client(self.root_dir,
174 communication = DC.ACCESS_SSH,
181 reply = c.read_reply()
182 self.assertEqual(reply, "Reply to: Hola")
185 reply = c.read_reply()
186 self.assertEqual(reply, "Stopping server")
188 def test_ssh_server_auto_reconnect(self):
189 env = test_util.test_environment()
190 user = getpass.getuser()
192 python_code = "from nepi.util import server;s=server.Server('%s');\
193 s.run()" % self.root_dir
194 server.popen_python(python_code,
195 communication = DC.ACCESS_SSH,
201 c = server.Client(self.root_dir,
202 communication = DC.ACCESS_SSH,
209 reply = c.read_reply()
210 self.assertEqual(reply, "Reply to: Hola")
212 # purposedly break the connection
213 c._process.stdin.close()
214 c._process.stdout.close()
215 c._process.stderr.close()
217 # assert that the communication works (possible with auto-reconnection)
219 reply = c.read_reply()
220 self.assertEqual(reply, "Reply to: Hola")
223 reply = c.read_reply()
224 self.assertEqual(reply, "Stopping server")
226 if __name__ == '__main__':