Ticket #12: proxy reconnection
[nepi.git] / 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