~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_server.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-05-28 00:25:32 UTC
  • mfrom: (5264.1.2 command-help-bug-177500)
  • Revision ID: pqm@pqm.ubuntu.com-20100528002532-9bzj1fajyxckd1rg
(lifeless) Stop raising at runtime when a command has no help,
 instead have a test in the test suite that checks all known command objects.
 (Robert Collins)

Show diffs side-by-side

added added

removed removed

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