← 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: Robert Collins
  • Date: 2007-07-20 03:20:20 UTC
  • mfrom: (2592 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2635.
  • Revision ID: robertc@robertcollins.net-20070720032020-xiftpb5gqeebo861
(robertc) Reinstate the accidentally backed out external_url patch.

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