/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: 2007-11-30 23:18:15 UTC
  • mto: This revision was merged to the branch mainline in revision 3072.
  • Revision ID: john@arbash-meinel.com-20071130231815-0r6ce70307kmv28r
Use a Graph.heads() check to determine if the ancestries are compatible.
Whether we should do nothing because source is already ahead,
raise an exception because we have diverged,
or move forward because the new revision is a tip revision.

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,
67
65
    hooks = TransportHooks()
68
66
 
69
67
    def __init__(self, base, _from_transport=None):
70
 
        if not base.startswith(_hooked_scheme + '://'):
71
 
            raise ValueError(base)
 
68
        assert base.startswith(_hooked_scheme + '://')
72
69
        # We need to trick the backing transport class about the scheme used
73
70
        # We'll do the reverse when we need to talk to the backing server
74
71
        fake_base = _change_scheme_in(base, _hooked_scheme, _backing_scheme)
100
97
    def setUp(self):
101
98
        register_urlparse_netloc_protocol(_hooked_scheme)
102
99
        register_transport(_hooked_scheme, ConnectionHookedTransport)
103
 
        self.addCleanup(unregister_transport, _hooked_scheme,
104
 
                        ConnectionHookedTransport)
105
 
        self.addCleanup(_unregister_urlparse_netloc_protocol, _hooked_scheme)
 
100
 
 
101
        def unregister():
 
102
            unregister_transport(_hooked_scheme, ConnectionHookedTransport)
 
103
            _unregister_urlparse_netloc_protocol(_hooked_scheme)
 
104
 
 
105
        self.addCleanup(unregister)
106
106
        super(TestCaseWithConnectionHookedTransport, self).setUp()
107
107
        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
108
 
119
109
    def get_url(self, relpath=None):
120
110
        super_self = super(TestCaseWithConnectionHookedTransport, self)
125
115
        return url
126
116
 
127
117
    def start_logging_connections(self):
128
 
        self.overrideAttr(InstrumentedTransport, 'hooks', TransportHooks())
129
 
        # We preserved the hooks class attribute. Now we install our hook.
130
 
        ConnectionHookedTransport.hooks.install_named_hook(
131
 
            '_set_connection', self._collect_connection, None)
 
118
        ConnectionHookedTransport.hooks.install_hook('_set_connection',
 
119
                                                     self._collect_connection)
 
120
        # uninstall our hooks when we are finished
 
121
        self.addCleanup(self.reset_hooks)
 
122
 
 
123
    def reset_hooks(self):
 
124
        InstrumentedTransport.hooks = TransportHooks()
132
125
 
133
126
    def reset_connections(self):
134
127
        self.connections = []