/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/transport_util.py

  • Committer: John Arbash Meinel
  • Date: 2008-07-09 21:42:24 UTC
  • mto: This revision was merged to the branch mainline in revision 3543.
  • Revision ID: john@arbash-meinel.com-20080709214224-r75k87r6a01pfc3h
Restore a real weave merge to 'bzr merge --weave'.

To do so efficiently, we only add the simple LCAs to the final weave
object, unless we run into complexities with the merge graph.
This gives the same effective result as adding all the texts,
with the advantage of not having to extract all of them.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/transport_util.py
1
 
# Copyright (C) 2007-2010 Canonical Ltd
 
1
# Copyright (C) 2007 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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
import bzrlib.hooks
18
 
from bzrlib.tests import features
19
18
 
20
19
# SFTPTransport offers better performances but relies on paramiko, if paramiko
21
20
# is not available, we fallback to FtpTransport
22
 
if features.paramiko.available():
23
 
    from bzrlib.tests import test_sftp_transport
 
21
from bzrlib.tests import test_sftp_transport
 
22
if test_sftp_transport.paramiko_loaded:
24
23
    from bzrlib.transport import sftp
25
24
    _backing_scheme = 'sftp'
26
25
    _backing_transport_class = sftp.SFTPTransport
34
33
 
35
34
from bzrlib.transport import (
36
35
    ConnectedTransport,
37
 
    get_transport,
38
36
    register_transport,
39
37
    register_urlparse_netloc_protocol,
40
38
    unregister_transport,
100
98
    def setUp(self):
101
99
        register_urlparse_netloc_protocol(_hooked_scheme)
102
100
        register_transport(_hooked_scheme, ConnectionHookedTransport)
103
 
        self.addCleanup(unregister_transport, _hooked_scheme,
104
 
                        ConnectionHookedTransport)
105
 
        self.addCleanup(_unregister_urlparse_netloc_protocol, _hooked_scheme)
 
101
 
 
102
        def unregister():
 
103
            unregister_transport(_hooked_scheme, ConnectionHookedTransport)
 
104
            _unregister_urlparse_netloc_protocol(_hooked_scheme)
 
105
 
 
106
        self.addCleanup(unregister)
106
107
        super(TestCaseWithConnectionHookedTransport, self).setUp()
107
108
        self.reset_connections()
108
 
        # Add the 'hooked' url to the permitted url list.
109
 
        # XXX: See TestCase.start_server. This whole module shouldn't need to
110
 
        # exist - a bug has been filed on that. once its cleanedup/removed, the
111
 
        # standard test support code will work and permit the server url
112
 
        # correctly.
113
 
        url = self.get_url()
114
 
        t = get_transport(url)
115
 
        if t.base.endswith('work/'):
116
 
            t = t.clone('../..')
117
 
        self.permit_url(t.base)
118
109
 
119
110
    def get_url(self, relpath=None):
120
111
        super_self = super(TestCaseWithConnectionHookedTransport, self)
125
116
        return url
126
117
 
127
118
    def start_logging_connections(self):
128
 
        self.overrideAttr(InstrumentedTransport, 'hooks', TransportHooks())
129
 
        # We preserved the hooks class attribute. Now we install our hook.
130
119
        ConnectionHookedTransport.hooks.install_named_hook(
131
120
            '_set_connection', self._collect_connection, None)
 
121
        # uninstall our hooks when we are finished
 
122
        self.addCleanup(self.reset_hooks)
 
123
 
 
124
    def reset_hooks(self):
 
125
        InstrumentedTransport.hooks = TransportHooks()
132
126
 
133
127
    def reset_connections(self):
134
128
        self.connections = []