← 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: Canonical.com Patch Queue Manager
  • Date: 2007-07-13 04:12:12 UTC
  • mfrom: (2484.1.25 knit_index_pyrex)
  • Revision ID: pqm@pqm.ubuntu.com-20070713041212-ar46c24wgu0jhtm5
(John Arbash Meinel) Implement _KnitIndex._load_data in pyrex.

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('.'))
1003
998
 
1004
999
class TestServerHooks(SmartTCPTests):
1005
1000
 
1006
 
    def capture_server_call(self, backing_urls, public_url):
 
1001
    def capture_server_call(self, backing_url, public_url):
1007
1002
        """Record a server_started|stopped hook firing."""
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)]
 
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)]
1066
1024
        # check the stopping message isn't emitted up front.
1067
1025
        self.assertEqual([], self.hook_calls)
1068
1026
        # nor after a single message