/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_test_server.py

  • Committer: John Arbash Meinel
  • Date: 2011-10-03 08:15:39 UTC
  • mto: This revision was merged to the branch mainline in revision 6186.
  • Revision ID: john@arbash-meinel.com-20111003081539-ondt7hnvoec1jg8o
Suppress ConnectionTimeout as a server-side exception.

Make sure that if we get a 'shutdown' request while we are waiting on an idle
connection, this won't generate an exception as part of handling the request.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_test_server.py
31
31
load_tests = load_tests_apply_scenarios
32
32
 
33
33
 
 
34
def portable_socket_pair():
 
35
    """Return a pair of TCP sockets connected to each other.
 
36
 
 
37
    Unlike socket.socketpair, this should work on Windows.
 
38
    """
 
39
    listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
40
    listen_sock.bind(('127.0.0.1', 0))
 
41
    listen_sock.listen(1)
 
42
    client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
43
    client_sock.connect(listen_sock.getsockname())
 
44
    server_sock, addr = listen_sock.accept()
 
45
    listen_sock.close()
 
46
    return server_sock, client_sock
 
47
 
 
48
 
34
49
class TCPClient(object):
35
50
 
36
51
    def __init__(self):
255
270
        h = server._make_handler(sock)
256
271
        self.assertEqual(test_server._DEFAULT_TESTING_CLIENT_TIMEOUT,
257
272
                         h._client_timeout)
 
273
 
 
274
 
 
275
class FakeServer(object):
 
276
    """Minimal implementation to pass to TestingSmartConnectionHandler"""
 
277
    backing_transport = None
 
278
    root_client_path = '/'
 
279
 
 
280
 
 
281
class TestTestingSmartConnectionHandler(tests.TestCase):
 
282
 
 
283
    def test_connection_timeout_suppressed(self):
 
284
        self.overrideAttr(test_server, '_DEFAULT_TESTING_CLIENT_TIMEOUT', 0.01)
 
285
        s = FakeServer()
 
286
        server_sock, client_sock = portable_socket_pair()
 
287
        # This should timeout quickly, but not generate an exception.
 
288
        handler = test_server.TestingSmartConnectionHandler(server_sock,
 
289
            server_sock.getpeername(), s)
 
290
 
 
291
    def test_connection_shutdown_while_serving_no_error(self):
 
292
        s = FakeServer()
 
293
        server_sock, client_sock = portable_socket_pair()
 
294
        class ShutdownConnectionHandler(
 
295
            test_server.TestingSmartConnectionHandler):
 
296
 
 
297
            def _build_protocol(self):
 
298
                self.finished = True
 
299
                return super(ShutdownConnectionHandler, self)._build_protocol()
 
300
        # This should trigger shutdown after the entering _build_protocol, and
 
301
        # we should exit cleanly, without raising an exception.
 
302
        handler = ShutdownConnectionHandler(server_sock,
 
303
            server_sock.getpeername(), s)