~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_server.py

  • Committer: Jelmer Vernooij
  • Date: 2011-12-05 14:12:23 UTC
  • mto: This revision was merged to the branch mainline in revision 6348.
  • Revision ID: jelmer@samba.org-20111205141223-8qxae4h37satlzgq
Move more functionality to vf_search.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2010 Canonical Ltd
 
1
# Copyright (C) 2010, 2011 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
 
17
import errno
 
18
import socket
 
19
import SocketServer
 
20
import sys
 
21
import threading
 
22
import traceback
 
23
 
 
24
 
17
25
from bzrlib import (
 
26
    cethread,
 
27
    errors,
 
28
    osutils,
18
29
    transport,
19
30
    urlutils,
20
31
    )
22
33
    chroot,
23
34
    pathfilter,
24
35
    )
25
 
from bzrlib.smart import server
 
36
from bzrlib.smart import (
 
37
    medium,
 
38
    server,
 
39
    )
 
40
 
 
41
 
 
42
def debug_threads():
 
43
    # FIXME: There is a dependency loop between bzrlib.tests and
 
44
    # bzrlib.tests.test_server that needs to be fixed. In the mean time
 
45
    # defining this function is enough for our needs. -- vila 20100611
 
46
    from bzrlib import tests
 
47
    return 'threads' in tests.selftest_debug_flags
26
48
 
27
49
 
28
50
class TestServer(transport.Server):
192
214
    def start_server(self, backing_server=None):
193
215
        """Setup the Chroot on backing_server."""
194
216
        if backing_server is not None:
195
 
            self.backing_transport = transport.get_transport(
 
217
            self.backing_transport = transport.get_transport_from_url(
196
218
                backing_server.get_url())
197
219
        else:
198
 
            self.backing_transport = transport.get_transport('.')
 
220
            self.backing_transport = transport.get_transport_from_path('.')
199
221
        self.backing_transport.clone('added-by-filter').ensure_base()
200
222
        self.filter_func = lambda x: 'added-by-filter/' + x
201
223
        super(TestingPathFilteringServer, self).start_server()
213
235
    def start_server(self, backing_server=None):
214
236
        """Setup the Chroot on backing_server."""
215
237
        if backing_server is not None:
216
 
            self.backing_transport = transport.get_transport(
 
238
            self.backing_transport = transport.get_transport_from_url(
217
239
                backing_server.get_url())
218
240
        else:
219
 
            self.backing_transport = transport.get_transport('.')
 
241
            self.backing_transport = transport.get_transport_from_path('.')
220
242
        super(TestingChrootServer, self).start_server()
221
243
 
222
244
    def get_bogus_url(self):
223
245
        raise NotImplementedError
224
246
 
225
247
 
226
 
class SmartTCPServer_for_testing(server.SmartTCPServer):
 
248
class TestThread(cethread.CatchingExceptionThread):
 
249
 
 
250
    def join(self, timeout=5):
 
251
        """Overrides to use a default timeout.
 
252
 
 
253
        The default timeout is set to 5 and should expire only when a thread
 
254
        serving a client connection is hung.
 
255
        """
 
256
        super(TestThread, self).join(timeout)
 
257
        if timeout and self.isAlive():
 
258
            # The timeout expired without joining the thread, the thread is
 
259
            # therefore stucked and that's a failure as far as the test is
 
260
            # concerned. We used to hang here.
 
261
 
 
262
            # FIXME: we need to kill the thread, but as far as the test is
 
263
            # concerned, raising an assertion is too strong. On most of the
 
264
            # platforms, this doesn't occur, so just mentioning the problem is
 
265
            # enough for now -- vila 2010824
 
266
            sys.stderr.write('thread %s hung\n' % (self.name,))
 
267
            #raise AssertionError('thread %s hung' % (self.name,))
 
268
 
 
269
 
 
270
class TestingTCPServerMixin(object):
 
271
    """Mixin to support running SocketServer.TCPServer in a thread.
 
272
 
 
273
    Tests are connecting from the main thread, the server has to be run in a
 
274
    separate thread.
 
275
    """
 
276
 
 
277
    def __init__(self):
 
278
        self.started = threading.Event()
 
279
        self.serving = None
 
280
        self.stopped = threading.Event()
 
281
        # We collect the resources used by the clients so we can release them
 
282
        # when shutting down
 
283
        self.clients = []
 
284
        self.ignored_exceptions = None
 
285
 
 
286
    def server_bind(self):
 
287
        self.socket.bind(self.server_address)
 
288
        self.server_address = self.socket.getsockname()
 
289
 
 
290
    def serve(self):
 
291
        self.serving = True
 
292
        # We are listening and ready to accept connections
 
293
        self.started.set()
 
294
        try:
 
295
            while self.serving:
 
296
                # Really a connection but the python framework is generic and
 
297
                # call them requests
 
298
                self.handle_request()
 
299
            # Let's close the listening socket
 
300
            self.server_close()
 
301
        finally:
 
302
            self.stopped.set()
 
303
 
 
304
    def handle_request(self):
 
305
        """Handle one request.
 
306
 
 
307
        The python version swallows some socket exceptions and we don't use
 
308
        timeout, so we override it to better control the server behavior.
 
309
        """
 
310
        request, client_address = self.get_request()
 
311
        if self.verify_request(request, client_address):
 
312
            try:
 
313
                self.process_request(request, client_address)
 
314
            except:
 
315
                self.handle_error(request, client_address)
 
316
        else:
 
317
            self.close_request(request)
 
318
 
 
319
    def get_request(self):
 
320
        return self.socket.accept()
 
321
 
 
322
    def verify_request(self, request, client_address):
 
323
        """Verify the request.
 
324
 
 
325
        Return True if we should proceed with this request, False if we should
 
326
        not even touch a single byte in the socket ! This is useful when we
 
327
        stop the server with a dummy last connection.
 
328
        """
 
329
        return self.serving
 
330
 
 
331
    def handle_error(self, request, client_address):
 
332
        # Stop serving and re-raise the last exception seen
 
333
        self.serving = False
 
334
        # The following can be used for debugging purposes, it will display the
 
335
        # exception and the traceback just when it occurs instead of waiting
 
336
        # for the thread to be joined.
 
337
        # SocketServer.BaseServer.handle_error(self, request, client_address)
 
338
 
 
339
        # We call close_request manually, because we are going to raise an
 
340
        # exception. The SocketServer implementation calls:
 
341
        #   handle_error(...)
 
342
        #   close_request(...)
 
343
        # But because we raise the exception, close_request will never be
 
344
        # triggered. This helps client not block waiting for a response when
 
345
        # the server gets an exception.
 
346
        self.close_request(request)
 
347
        raise
 
348
 
 
349
    def ignored_exceptions_during_shutdown(self, e):
 
350
        if sys.platform == 'win32':
 
351
            accepted_errnos = [errno.EBADF,
 
352
                               errno.EPIPE,
 
353
                               errno.WSAEBADF,
 
354
                               errno.WSAECONNRESET,
 
355
                               errno.WSAENOTCONN,
 
356
                               errno.WSAESHUTDOWN,
 
357
                               ]
 
358
        else:
 
359
            accepted_errnos = [errno.EBADF,
 
360
                               errno.ECONNRESET,
 
361
                               errno.ENOTCONN,
 
362
                               errno.EPIPE,
 
363
                               ]
 
364
        if isinstance(e, socket.error) and e[0] in accepted_errnos:
 
365
            return True
 
366
        return False
 
367
 
 
368
    # The following methods are called by the main thread
 
369
 
 
370
    def stop_client_connections(self):
 
371
        while self.clients:
 
372
            c = self.clients.pop()
 
373
            self.shutdown_client(c)
 
374
 
 
375
    def shutdown_socket(self, sock):
 
376
        """Properly shutdown a socket.
 
377
 
 
378
        This should be called only when no other thread is trying to use the
 
379
        socket.
 
380
        """
 
381
        try:
 
382
            sock.shutdown(socket.SHUT_RDWR)
 
383
            sock.close()
 
384
        except Exception, e:
 
385
            if self.ignored_exceptions(e):
 
386
                pass
 
387
            else:
 
388
                raise
 
389
 
 
390
    # The following methods are called by the main thread
 
391
 
 
392
    def set_ignored_exceptions(self, thread, ignored_exceptions):
 
393
        self.ignored_exceptions = ignored_exceptions
 
394
        thread.set_ignored_exceptions(self.ignored_exceptions)
 
395
 
 
396
    def _pending_exception(self, thread):
 
397
        """Raise server uncaught exception.
 
398
 
 
399
        Daughter classes can override this if they use daughter threads.
 
400
        """
 
401
        thread.pending_exception()
 
402
 
 
403
 
 
404
class TestingTCPServer(TestingTCPServerMixin, SocketServer.TCPServer):
 
405
 
 
406
    def __init__(self, server_address, request_handler_class):
 
407
        TestingTCPServerMixin.__init__(self)
 
408
        SocketServer.TCPServer.__init__(self, server_address,
 
409
                                        request_handler_class)
 
410
 
 
411
    def get_request(self):
 
412
        """Get the request and client address from the socket."""
 
413
        sock, addr = TestingTCPServerMixin.get_request(self)
 
414
        self.clients.append((sock, addr))
 
415
        return sock, addr
 
416
 
 
417
    # The following methods are called by the main thread
 
418
 
 
419
    def shutdown_client(self, client):
 
420
        sock, addr = client
 
421
        self.shutdown_socket(sock)
 
422
 
 
423
 
 
424
class TestingThreadingTCPServer(TestingTCPServerMixin,
 
425
                                SocketServer.ThreadingTCPServer):
 
426
 
 
427
    def __init__(self, server_address, request_handler_class):
 
428
        TestingTCPServerMixin.__init__(self)
 
429
        SocketServer.ThreadingTCPServer.__init__(self, server_address,
 
430
                                                 request_handler_class)
 
431
 
 
432
    def get_request(self):
 
433
        """Get the request and client address from the socket."""
 
434
        sock, addr = TestingTCPServerMixin.get_request(self)
 
435
        # The thread is not created yet, it will be updated in process_request
 
436
        self.clients.append((sock, addr, None))
 
437
        return sock, addr
 
438
 
 
439
    def process_request_thread(self, started, detached, stopped,
 
440
                               request, client_address):
 
441
        started.set()
 
442
        # We will be on our own once the server tells us we're detached
 
443
        detached.wait()
 
444
        SocketServer.ThreadingTCPServer.process_request_thread(
 
445
            self, request, client_address)
 
446
        self.close_request(request)
 
447
        stopped.set()
 
448
 
 
449
    def process_request(self, request, client_address):
 
450
        """Start a new thread to process the request."""
 
451
        started = threading.Event()
 
452
        detached = threading.Event()
 
453
        stopped = threading.Event()
 
454
        t = TestThread(
 
455
            sync_event=stopped,
 
456
            name='%s -> %s' % (client_address, self.server_address),
 
457
            target = self.process_request_thread,
 
458
            args = (started, detached, stopped, request, client_address))
 
459
        # Update the client description
 
460
        self.clients.pop()
 
461
        self.clients.append((request, client_address, t))
 
462
        # Propagate the exception handler since we must use the same one as
 
463
        # TestingTCPServer for connections running in their own threads.
 
464
        t.set_ignored_exceptions(self.ignored_exceptions)
 
465
        t.start()
 
466
        started.wait()
 
467
        # If an exception occured during the thread start, it will get raised.
 
468
        t.pending_exception()
 
469
        if debug_threads():
 
470
            sys.stderr.write('Client thread %s started\n' % (t.name,))
 
471
        # Tell the thread, it's now on its own for exception handling.
 
472
        detached.set()
 
473
 
 
474
    # The following methods are called by the main thread
 
475
 
 
476
    def shutdown_client(self, client):
 
477
        sock, addr, connection_thread = client
 
478
        self.shutdown_socket(sock)
 
479
        if connection_thread is not None:
 
480
            # The thread has been created only if the request is processed but
 
481
            # after the connection is inited. This could happen during server
 
482
            # shutdown. If an exception occurred in the thread it will be
 
483
            # re-raised
 
484
            if debug_threads():
 
485
                sys.stderr.write('Client thread %s will be joined\n'
 
486
                                 % (connection_thread.name,))
 
487
            connection_thread.join()
 
488
 
 
489
    def set_ignored_exceptions(self, thread, ignored_exceptions):
 
490
        TestingTCPServerMixin.set_ignored_exceptions(self, thread,
 
491
                                                     ignored_exceptions)
 
492
        for sock, addr, connection_thread in self.clients:
 
493
            if connection_thread is not None:
 
494
                connection_thread.set_ignored_exceptions(
 
495
                    self.ignored_exceptions)
 
496
 
 
497
    def _pending_exception(self, thread):
 
498
        for sock, addr, connection_thread in self.clients:
 
499
            if connection_thread is not None:
 
500
                connection_thread.pending_exception()
 
501
        TestingTCPServerMixin._pending_exception(self, thread)
 
502
 
 
503
 
 
504
class TestingTCPServerInAThread(transport.Server):
 
505
    """A server in a thread that re-raise thread exceptions."""
 
506
 
 
507
    def __init__(self, server_address, server_class, request_handler_class):
 
508
        self.server_class = server_class
 
509
        self.request_handler_class = request_handler_class
 
510
        self.host, self.port = server_address
 
511
        self.server = None
 
512
        self._server_thread = None
 
513
 
 
514
    def __repr__(self):
 
515
        return "%s(%s:%s)" % (self.__class__.__name__, self.host, self.port)
 
516
 
 
517
    def create_server(self):
 
518
        return self.server_class((self.host, self.port),
 
519
                                 self.request_handler_class)
 
520
 
 
521
    def start_server(self):
 
522
        self.server = self.create_server()
 
523
        self._server_thread = TestThread(
 
524
            sync_event=self.server.started,
 
525
            target=self.run_server)
 
526
        self._server_thread.start()
 
527
        # Wait for the server thread to start (i.e. release the lock)
 
528
        self.server.started.wait()
 
529
        # Get the real address, especially the port
 
530
        self.host, self.port = self.server.server_address
 
531
        self._server_thread.name = self.server.server_address
 
532
        if debug_threads():
 
533
            sys.stderr.write('Server thread %s started\n'
 
534
                             % (self._server_thread.name,))
 
535
        # If an exception occured during the server start, it will get raised,
 
536
        # otherwise, the server is blocked on its accept() call.
 
537
        self._server_thread.pending_exception()
 
538
        # From now on, we'll use a different event to ensure the server can set
 
539
        # its exception
 
540
        self._server_thread.set_sync_event(self.server.stopped)
 
541
 
 
542
    def run_server(self):
 
543
        self.server.serve()
 
544
 
 
545
    def stop_server(self):
 
546
        if self.server is None:
 
547
            return
 
548
        try:
 
549
            # The server has been started successfully, shut it down now.  As
 
550
            # soon as we stop serving, no more connection are accepted except
 
551
            # one to get out of the blocking listen.
 
552
            self.set_ignored_exceptions(
 
553
                self.server.ignored_exceptions_during_shutdown)
 
554
            self.server.serving = False
 
555
            if debug_threads():
 
556
                sys.stderr.write('Server thread %s will be joined\n'
 
557
                                 % (self._server_thread.name,))
 
558
            # The server is listening for a last connection, let's give it:
 
559
            last_conn = None
 
560
            try:
 
561
                last_conn = osutils.connect_socket((self.host, self.port))
 
562
            except socket.error, e:
 
563
                # But ignore connection errors as the point is to unblock the
 
564
                # server thread, it may happen that it's not blocked or even
 
565
                # not started.
 
566
                pass
 
567
            # We start shutting down the clients while the server itself is
 
568
            # shutting down.
 
569
            self.server.stop_client_connections()
 
570
            # Now we wait for the thread running self.server.serve() to finish
 
571
            self.server.stopped.wait()
 
572
            if last_conn is not None:
 
573
                # Close the last connection without trying to use it. The
 
574
                # server will not process a single byte on that socket to avoid
 
575
                # complications (SSL starts with a handshake for example).
 
576
                last_conn.close()
 
577
            # Check for any exception that could have occurred in the server
 
578
            # thread
 
579
            try:
 
580
                self._server_thread.join()
 
581
            except Exception, e:
 
582
                if self.server.ignored_exceptions(e):
 
583
                    pass
 
584
                else:
 
585
                    raise
 
586
        finally:
 
587
            # Make sure we can be called twice safely, note that this means
 
588
            # that we will raise a single exception even if several occurred in
 
589
            # the various threads involved.
 
590
            self.server = None
 
591
 
 
592
    def set_ignored_exceptions(self, ignored_exceptions):
 
593
        """Install an exception handler for the server."""
 
594
        self.server.set_ignored_exceptions(self._server_thread,
 
595
                                           ignored_exceptions)
 
596
 
 
597
    def pending_exception(self):
 
598
        """Raise uncaught exception in the server."""
 
599
        self.server._pending_exception(self._server_thread)
 
600
 
 
601
 
 
602
class TestingSmartConnectionHandler(SocketServer.BaseRequestHandler,
 
603
                                    medium.SmartServerSocketStreamMedium):
 
604
 
 
605
    def __init__(self, request, client_address, server):
 
606
        medium.SmartServerSocketStreamMedium.__init__(
 
607
            self, request, server.backing_transport,
 
608
            server.root_client_path,
 
609
            timeout=_DEFAULT_TESTING_CLIENT_TIMEOUT)
 
610
        request.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 
611
        SocketServer.BaseRequestHandler.__init__(self, request, client_address,
 
612
                                                 server)
 
613
 
 
614
    def handle(self):
 
615
        try:
 
616
            while not self.finished:
 
617
                server_protocol = self._build_protocol()
 
618
                self._serve_one_request(server_protocol)
 
619
        except errors.ConnectionTimeout:
 
620
            # idle connections aren't considered a failure of the server
 
621
            return
 
622
 
 
623
 
 
624
_DEFAULT_TESTING_CLIENT_TIMEOUT = 60.0
 
625
 
 
626
class TestingSmartServer(TestingThreadingTCPServer, server.SmartTCPServer):
 
627
 
 
628
    def __init__(self, server_address, request_handler_class,
 
629
                 backing_transport, root_client_path):
 
630
        TestingThreadingTCPServer.__init__(self, server_address,
 
631
                                           request_handler_class)
 
632
        server.SmartTCPServer.__init__(self, backing_transport,
 
633
            root_client_path, client_timeout=_DEFAULT_TESTING_CLIENT_TIMEOUT)
 
634
 
 
635
    def serve(self):
 
636
        self.run_server_started_hooks()
 
637
        try:
 
638
            TestingThreadingTCPServer.serve(self)
 
639
        finally:
 
640
            self.run_server_stopped_hooks()
 
641
 
 
642
    def get_url(self):
 
643
        """Return the url of the server"""
 
644
        return "bzr://%s:%d/" % self.server_address
 
645
 
 
646
 
 
647
class SmartTCPServer_for_testing(TestingTCPServerInAThread):
227
648
    """Server suitable for use by transport tests.
228
649
 
229
650
    This server is backed by the process's cwd.
230
651
    """
231
 
 
232
652
    def __init__(self, thread_name_suffix=''):
233
 
        super(SmartTCPServer_for_testing, self).__init__(None)
234
653
        self.client_path_extra = None
235
654
        self.thread_name_suffix = thread_name_suffix
236
 
 
237
 
    def get_backing_transport(self, backing_transport_server):
238
 
        """Get a backing transport from a server we are decorating."""
239
 
        return transport.get_transport(backing_transport_server.get_url())
 
655
        self.host = '127.0.0.1'
 
656
        self.port = 0
 
657
        super(SmartTCPServer_for_testing, self).__init__(
 
658
                (self.host, self.port),
 
659
                TestingSmartServer,
 
660
                TestingSmartConnectionHandler)
 
661
 
 
662
    def create_server(self):
 
663
        return self.server_class((self.host, self.port),
 
664
                                 self.request_handler_class,
 
665
                                 self.backing_transport,
 
666
                                 self.root_client_path)
 
667
 
240
668
 
241
669
    def start_server(self, backing_transport_server=None,
242
 
              client_path_extra='/extra/'):
 
670
                     client_path_extra='/extra/'):
243
671
        """Set up server for testing.
244
672
 
245
673
        :param backing_transport_server: backing server to use.  If not
254
682
        """
255
683
        if not client_path_extra.startswith('/'):
256
684
            raise ValueError(client_path_extra)
 
685
        self.root_client_path = self.client_path_extra = client_path_extra
257
686
        from bzrlib.transport.chroot import ChrootServer
258
687
        if backing_transport_server is None:
259
688
            backing_transport_server = LocalURLServer()
260
689
        self.chroot_server = ChrootServer(
261
690
            self.get_backing_transport(backing_transport_server))
262
691
        self.chroot_server.start_server()
263
 
        self.backing_transport = transport.get_transport(
 
692
        self.backing_transport = transport.get_transport_from_url(
264
693
            self.chroot_server.get_url())
265
 
        self.root_client_path = self.client_path_extra = client_path_extra
266
 
        self.start_background_thread(self.thread_name_suffix)
 
694
        super(SmartTCPServer_for_testing, self).start_server()
267
695
 
268
696
    def stop_server(self):
269
 
        self.stop_background_thread()
270
 
        self.chroot_server.stop_server()
 
697
        try:
 
698
            super(SmartTCPServer_for_testing, self).stop_server()
 
699
        finally:
 
700
            self.chroot_server.stop_server()
 
701
 
 
702
    def get_backing_transport(self, backing_transport_server):
 
703
        """Get a backing transport from a server we are decorating."""
 
704
        return transport.get_transport_from_url(
 
705
            backing_transport_server.get_url())
271
706
 
272
707
    def get_url(self):
273
 
        url = super(SmartTCPServer_for_testing, self).get_url()
 
708
        url = self.server.get_url()
274
709
        return url[:-1] + self.client_path_extra
275
710
 
276
711
    def get_bogus_url(self):
284
719
    def get_backing_transport(self, backing_transport_server):
285
720
        """Get a backing transport from a server we are decorating."""
286
721
        url = 'readonly+' + backing_transport_server.get_url()
287
 
        return transport.get_transport(url)
 
722
        return transport.get_transport_from_url(url)
288
723
 
289
724
 
290
725
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
305
740
    def get_backing_transport(self, backing_transport_server):
306
741
        """Get a backing transport from a server we are decorating."""
307
742
        url = 'readonly+' + backing_transport_server.get_url()
308
 
        return transport.get_transport(url)
309
 
 
310
 
 
311
 
 
312
 
 
 
743
        return transport.get_transport_from_url(url)