← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_test_server.py

  • Committer: v.ladeuil+lp at free
  • Date: 2007-05-15 17:40:32 UTC
  • mto: (2485.8.44 bzr.connection.sharing)
  • mto: This revision was merged to the branch mainline in revision 2646.
  • Revision ID: v.ladeuil+lp@free.fr-20070515174032-qzdkangpv29l9e7g
Add a test that check that init connect only once. It fails.

* __init__.py:
(test_suite): Register the new test class.

* test_init.py: 
(InstrumentedTransport): A transport that can track connections.
(TransportHooks): Transport specific hooks.
(TestInit): Iniit command behavior tests.

* ftp.py:
(FtpTransport.__init__): Mark place that need fixing regarding
transport connection sharing

* builtins.py:
(cmd_init.run): Mark places that need fixing regarding transport
connection sharing.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_test_server.py
1
 
# Copyright (C) 2010, 2011 Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
import errno
18
 
import socket
19
 
import SocketServer
20
 
import threading
21
 
 
22
 
from bzrlib import (
23
 
    osutils,
24
 
    tests,
25
 
    )
26
 
from bzrlib.tests import test_server
27
 
from bzrlib.tests.scenarios import load_tests_apply_scenarios
28
 
 
29
 
 
30
 
load_tests = load_tests_apply_scenarios
31
 
 
32
 
 
33
 
class TCPClient(object):
34
 
 
35
 
    def __init__(self):
36
 
        self.sock = None
37
 
 
38
 
    def connect(self, addr):
39
 
        if self.sock is not None:
40
 
            raise AssertionError('Already connected to %r'
41
 
                                 % (self.sock.getsockname(),))
42
 
        self.sock = osutils.connect_socket(addr)
43
 
 
44
 
    def disconnect(self):
45
 
        if self.sock is not None:
46
 
            try:
47
 
                self.sock.shutdown(socket.SHUT_RDWR)
48
 
                self.sock.close()
49
 
            except socket.error, e:
50
 
                if e[0] in (errno.EBADF, errno.ENOTCONN):
51
 
                    # Right, the socket is already down
52
 
                    pass
53
 
                else:
54
 
                    raise
55
 
            self.sock = None
56
 
 
57
 
    def write(self, s):
58
 
        return self.sock.sendall(s)
59
 
 
60
 
    def read(self, bufsize=4096):
61
 
        return self.sock.recv(bufsize)
62
 
 
63
 
 
64
 
class TCPConnectionHandler(SocketServer.StreamRequestHandler):
65
 
 
66
 
    def handle(self):
67
 
        self.done = False
68
 
        self.handle_connection()
69
 
        while not self.done:
70
 
            self.handle_connection()
71
 
 
72
 
    def handle_connection(self):
73
 
        req = self.rfile.readline()
74
 
        if not req:
75
 
            self.done = True
76
 
        elif req == 'ping\n':
77
 
            self.wfile.write('pong\n')
78
 
        else:
79
 
            raise ValueError('[%s] not understood' % req)
80
 
 
81
 
 
82
 
class TestTCPServerInAThread(tests.TestCase):
83
 
 
84
 
    scenarios = [ 
85
 
        (name, {'server_class': getattr(test_server, name)})
86
 
        for name in
87
 
        ('TestingTCPServer', 'TestingThreadingTCPServer')]
88
 
 
89
 
    # Set by load_tests()
90
 
    server_class = None
91
 
 
92
 
    def get_server(self, server_class=None, connection_handler_class=None):
93
 
        if server_class is not None:
94
 
            self.server_class = server_class
95
 
        if connection_handler_class is None:
96
 
            connection_handler_class = TCPConnectionHandler
97
 
        server =  test_server.TestingTCPServerInAThread(
98
 
            ('localhost', 0), self.server_class, connection_handler_class)
99
 
        server.start_server()
100
 
        self.addCleanup(server.stop_server)
101
 
        return server
102
 
 
103
 
    def get_client(self):
104
 
        client = TCPClient()
105
 
        self.addCleanup(client.disconnect)
106
 
        return client
107
 
 
108
 
    def get_server_connection(self, server, conn_rank):
109
 
        return server.server.clients[conn_rank]
110
 
 
111
 
    def assertClientAddr(self, client, server, conn_rank):
112
 
        conn = self.get_server_connection(server, conn_rank)
113
 
        self.assertEquals(client.sock.getsockname(), conn[1])
114
 
 
115
 
    def test_start_stop(self):
116
 
        server = self.get_server()
117
 
        client = self.get_client()
118
 
        server.stop_server()
119
 
        # since the server doesn't accept connections anymore attempting to
120
 
        # connect should fail
121
 
        client = self.get_client()
122
 
        self.assertRaises(socket.error,
123
 
                          client.connect, (server.host, server.port))
124
 
 
125
 
    def test_client_talks_server_respond(self):
126
 
        server = self.get_server()
127
 
        client = self.get_client()
128
 
        client.connect((server.host, server.port))
129
 
        self.assertIs(None, client.write('ping\n'))
130
 
        resp = client.read()
131
 
        self.assertClientAddr(client, server, 0)
132
 
        self.assertEquals('pong\n', resp)
133
 
 
134
 
    def test_server_fails_to_start(self):
135
 
        class CantStart(Exception):
136
 
            pass
137
 
 
138
 
        class CantStartServer(test_server.TestingTCPServer):
139
 
 
140
 
            def server_bind(self):
141
 
                raise CantStart()
142
 
 
143
 
        # The exception is raised in the main thread
144
 
        self.assertRaises(CantStart,
145
 
                          self.get_server, server_class=CantStartServer)
146
 
 
147
 
    def test_server_fails_while_serving_or_stopping(self):
148
 
        class CantConnect(Exception):
149
 
            pass
150
 
 
151
 
        class FailingConnectionHandler(TCPConnectionHandler):
152
 
 
153
 
            def handle(self):
154
 
                raise CantConnect()
155
 
 
156
 
        server = self.get_server(
157
 
            connection_handler_class=FailingConnectionHandler)
158
 
        # The server won't fail until a client connect
159
 
        client = self.get_client()
160
 
        client.connect((server.host, server.port))
161
 
        try:
162
 
            # Now we must force the server to answer by sending the request and
163
 
            # waiting for some answer. But since we don't control when the
164
 
            # server thread will be given cycles, we don't control either
165
 
            # whether our reads or writes may hang.
166
 
            client.sock.settimeout(0.1)
167
 
            client.write('ping\n')
168
 
            client.read()
169
 
        except socket.error:
170
 
            pass
171
 
        # Now the server has raised the exception in its own thread
172
 
        self.assertRaises(CantConnect, server.stop_server)
173
 
 
174
 
    def test_server_crash_while_responding(self):
175
 
        sync = threading.Event()
176
 
        sync.clear()
177
 
        class FailToRespond(Exception):
178
 
            pass
179
 
 
180
 
        class FailingDuringResponseHandler(TCPConnectionHandler):
181
 
 
182
 
            def handle_connection(self):
183
 
                req = self.rfile.readline()
184
 
                threading.currentThread().set_sync_event(sync)
185
 
                raise FailToRespond()
186
 
 
187
 
        server = self.get_server(
188
 
            connection_handler_class=FailingDuringResponseHandler)
189
 
        client = self.get_client()
190
 
        client.connect((server.host, server.port))
191
 
        client.write('ping\n')
192
 
        sync.wait()
193
 
        self.assertRaises(FailToRespond, server.pending_exception)
194
 
 
195
 
    def test_exception_swallowed_while_serving(self):
196
 
        sync = threading.Event()
197
 
        sync.clear()
198
 
        class CantServe(Exception):
199
 
            pass
200
 
 
201
 
        class FailingWhileServingConnectionHandler(TCPConnectionHandler):
202
 
 
203
 
            def handle(self):
204
 
                # We want to sync with the thread that is serving the
205
 
                # connection.
206
 
                threading.currentThread().set_sync_event(sync)
207
 
                raise CantServe()
208
 
 
209
 
        server = self.get_server(
210
 
            connection_handler_class=FailingWhileServingConnectionHandler)
211
 
        # Install the exception swallower
212
 
        server.set_ignored_exceptions(CantServe)
213
 
        client = self.get_client()
214
 
        # Connect to the server so the exception is raised there
215
 
        client.connect((server.host, server.port))
216
 
        # Wait for the exception to propagate.
217
 
        sync.wait()
218
 
        # The connection wasn't served properly but the exception should have
219
 
        # been swallowed.
220
 
        server.pending_exception()