← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_smart_transport.py

  • Committer: Vincent Ladeuil
  • Date: 2007-07-15 11:24:18 UTC
  • mfrom: (2617 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2646.
  • Revision ID: v.ladeuil+lp@free.fr-20070715112418-9nn4n6esxv60ny4b
merge bzr.dev@1617

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_smart_transport.py
834
834
        self._captureVar('BZR_NO_SMART_VFS', None)
835
835
        class FlakyTransport(object):
836
836
            base = 'a_url'
837
 
            def external_url(self):
838
 
                return self.base
839
837
            def get_bytes(self, path):
840
838
                raise Exception("some random exception from inside server")
841
839
        smart_server = server.SmartTCPServer(backing_transport=FlakyTransport())
862
860
    the server is obtained by calling self.setUpServer(readonly=False).
863
861
    """
864
862
 
865
 
    def setUpServer(self, readonly=False, backing_transport=None):
 
863
    def setUpServer(self, readonly=False):
866
864
        """Setup the server.
867
865
 
868
866
        :param readonly: Create a readonly server.
869
867
        """
870
 
        if not backing_transport:
871
 
            self.backing_transport = memory.MemoryTransport()
872
 
        else:
873
 
            self.backing_transport = backing_transport
 
868
        self.backing_transport = memory.MemoryTransport()
874
869
        if readonly:
875
870
            self.real_backing_transport = self.backing_transport
876
871
            self.backing_transport = get_transport("readonly+" + self.backing_transport.abspath('.'))
1004
999
 
1005
1000
class TestServerHooks(SmartTCPTests):
1006
1001
 
1007
 
    def capture_server_call(self, backing_urls, public_url):
 
1002
    def capture_server_call(self, backing_url, public_url):
1008
1003
        """Record a server_started|stopped hook firing."""
1009
 
        self.hook_calls.append((backing_urls, public_url))
1010
 
 
1011
 
    def test_server_started_hook_memory(self):
1012
 
        """The server_started hook fires when the server is started."""
1013
 
        self.hook_calls = []
1014
 
        server.SmartTCPServer.hooks.install_hook('server_started',
1015
 
            self.capture_server_call)
1016
 
        self.setUpServer()
1017
 
        # at this point, the server will be starting a thread up.
1018
 
        # there is no indicator at the moment, so bodge it by doing a request.
1019
 
        self.transport.has('.')
1020
 
        # The default test server uses MemoryTransport and that has no external
1021
 
        # url:
1022
 
        self.assertEqual([([self.backing_transport.base], self.transport.base)],
1023
 
            self.hook_calls)
1024
 
 
1025
 
    def test_server_started_hook_file(self):
1026
 
        """The server_started hook fires when the server is started."""
1027
 
        self.hook_calls = []
1028
 
        server.SmartTCPServer.hooks.install_hook('server_started',
1029
 
            self.capture_server_call)
1030
 
        self.setUpServer(backing_transport=get_transport("."))
1031
 
        # at this point, the server will be starting a thread up.
1032
 
        # there is no indicator at the moment, so bodge it by doing a request.
1033
 
        self.transport.has('.')
1034
 
        # The default test server uses MemoryTransport and that has no external
1035
 
        # url:
1036
 
        self.assertEqual([([
1037
 
            self.backing_transport.base, self.backing_transport.external_url()],
1038
 
             self.transport.base)],
1039
 
            self.hook_calls)
1040
 
 
1041
 
    def test_server_stopped_hook_simple_memory(self):
1042
 
        """The server_stopped hook fires when the server is stopped."""
1043
 
        self.hook_calls = []
1044
 
        server.SmartTCPServer.hooks.install_hook('server_stopped',
1045
 
            self.capture_server_call)
1046
 
        self.setUpServer()
1047
 
        result = [([self.backing_transport.base], self.transport.base)]
1048
 
        # check the stopping message isn't emitted up front.
1049
 
        self.assertEqual([], self.hook_calls)
1050
 
        # nor after a single message
1051
 
        self.transport.has('.')
1052
 
        self.assertEqual([], self.hook_calls)
1053
 
        # clean up the server
1054
 
        self.tearDownServer()
1055
 
        # now it should have fired.
1056
 
        self.assertEqual(result, self.hook_calls)
1057
 
 
1058
 
    def test_server_stopped_hook_simple_file(self):
1059
 
        """The server_stopped hook fires when the server is stopped."""
1060
 
        self.hook_calls = []
1061
 
        server.SmartTCPServer.hooks.install_hook('server_stopped',
1062
 
            self.capture_server_call)
1063
 
        self.setUpServer(backing_transport=get_transport("."))
1064
 
        result = [(
1065
 
            [self.backing_transport.base, self.backing_transport.external_url()]
1066
 
            , self.transport.base)]
 
1004
        self.hook_calls.append((backing_url, public_url))
 
1005
 
 
1006
    def test_server_started_hook(self):
 
1007
        """The server_started hook fires when the server is started."""
 
1008
        self.hook_calls = []
 
1009
        server.SmartTCPServer.hooks.install_hook('server_started',
 
1010
            self.capture_server_call)
 
1011
        self.setUpServer()
 
1012
        # at this point, the server will be starting a thread up.
 
1013
        # there is no indicator at the moment, so bodge it by doing a request.
 
1014
        self.transport.has('.')
 
1015
        self.assertEqual([(self.backing_transport.base, self.transport.base)],
 
1016
            self.hook_calls)
 
1017
 
 
1018
    def test_server_stopped_hook_simple(self):
 
1019
        """The server_stopped hook fires when the server is stopped."""
 
1020
        self.hook_calls = []
 
1021
        server.SmartTCPServer.hooks.install_hook('server_stopped',
 
1022
            self.capture_server_call)
 
1023
        self.setUpServer()
 
1024
        result = [(self.backing_transport.base, self.transport.base)]
1067
1025
        # check the stopping message isn't emitted up front.
1068
1026
        self.assertEqual([], self.hook_calls)
1069
1027
        # nor after a single message