← 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-25 00:52:21 UTC
  • mfrom: (2650 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2651.
  • Revision ID: robertc@robertcollins.net-20070725005221-0ysm6il5mqnme3wz
Merge bzr.dev.

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('.'))
948
953
        # we create a real connection not a loopback one, but it will use the
949
954
        # same server and pipes
950
955
        conn2 = self.transport.clone('.')
951
 
        self.assertIs(self.transport._medium, conn2._medium)
 
956
        self.assertIs(self.transport.get_smart_medium(),
 
957
                      conn2.get_smart_medium())
952
958
 
953
959
    def test__remote_path(self):
954
960
        self.assertEquals('/foo/bar',
998
1004
 
999
1005
class TestServerHooks(SmartTCPTests):
1000
1006
 
1001
 
    def capture_server_call(self, backing_url, public_url):
 
1007
    def capture_server_call(self, backing_urls, public_url):
1002
1008
        """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)]
 
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)]
1024
1067
        # check the stopping message isn't emitted up front.
1025
1068
        self.assertEqual([], self.hook_calls)
1026
1069
        # nor after a single message