~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransport.py

Exclude more files from dumb-rsync upload

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
2
 
#
 
1
# Copyright (C) 2004, 2005 by Canonical Ltd
 
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
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
7
 
#
 
7
 
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
11
# GNU General Public License for more details.
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
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
 
18
18
import os
19
 
import sys
20
 
import stat
21
19
from cStringIO import StringIO
22
20
 
23
 
import bzrlib
24
 
from bzrlib import (
25
 
    errors,
26
 
    urlutils,
27
 
    )
28
 
from bzrlib.errors import (ConnectionError,
29
 
                           DependencyNotPresent,
30
 
                           FileExists,
31
 
                           InvalidURLJoin,
32
 
                           NoSuchFile,
33
 
                           PathNotChild,
34
 
                           TransportNotPossible,
35
 
                           ConnectionError,
36
 
                           DependencyNotPresent,
37
 
                           ReadError,
38
 
                           UnsupportedProtocol,
39
 
                           )
40
 
from bzrlib.tests import TestCase, TestCaseInTempDir
41
 
from bzrlib.transport import (_CoalescedOffset,
42
 
                              ConnectedTransport,
43
 
                              _get_protocol_handlers,
44
 
                              _set_protocol_handlers,
45
 
                              _get_transport_modules,
46
 
                              get_transport,
47
 
                              LateReadError,
48
 
                              register_lazy_transport,
49
 
                              register_transport_proto,
50
 
                              _clear_protocol_handlers,
51
 
                              Transport,
52
 
                              )
53
 
from bzrlib.transport.chroot import ChrootServer
54
 
from bzrlib.transport.memory import MemoryTransport
55
 
from bzrlib.transport.local import (LocalTransport,
56
 
                                    EmulatedWin32LocalTransport)
57
 
 
58
 
 
59
 
# TODO: Should possibly split transport-specific tests into their own files.
60
 
 
 
21
from bzrlib.errors import (NoSuchFile, FileExists, TransportNotPossible,
 
22
                           ConnectionError)
 
23
from bzrlib.selftest import TestCase, TestCaseInTempDir
 
24
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
 
25
from bzrlib.transport import memory, urlescape
 
26
 
 
27
 
 
28
def _append(fn, txt):
 
29
    """Append the given text (file-like object) to the supplied filename."""
 
30
    f = open(fn, 'ab')
 
31
    f.write(txt)
 
32
    f.flush()
 
33
    f.close()
 
34
    del f
61
35
 
62
36
class TestTransport(TestCase):
63
37
    """Test the non transport-concrete class functionality."""
64
38
 
65
 
    def test__get_set_protocol_handlers(self):
66
 
        handlers = _get_protocol_handlers()
67
 
        self.assertNotEqual([], handlers.keys( ))
68
 
        try:
69
 
            _clear_protocol_handlers()
70
 
            self.assertEqual([], _get_protocol_handlers().keys())
71
 
        finally:
72
 
            _set_protocol_handlers(handlers)
73
 
 
74
 
    def test_get_transport_modules(self):
75
 
        handlers = _get_protocol_handlers()
76
 
        class SampleHandler(object):
77
 
            """I exist, isnt that enough?"""
78
 
        try:
79
 
            _clear_protocol_handlers()
80
 
            register_transport_proto('foo')
81
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
82
 
            register_transport_proto('bar')
83
 
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
84
 
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
85
 
                             _get_transport_modules())
86
 
        finally:
87
 
            _set_protocol_handlers(handlers)
88
 
 
89
 
    def test_transport_dependency(self):
90
 
        """Transport with missing dependency causes no error"""
91
 
        saved_handlers = _get_protocol_handlers()
92
 
        try:
93
 
            register_transport_proto('foo')
94
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
95
 
                    'BadTransportHandler')
96
 
            try:
97
 
                get_transport('foo://fooserver/foo')
98
 
            except UnsupportedProtocol, e:
99
 
                e_str = str(e)
100
 
                self.assertEquals('Unsupported protocol'
101
 
                                  ' for url "foo://fooserver/foo":'
102
 
                                  ' Unable to import library "some_lib":'
103
 
                                  ' testing missing dependency', str(e))
104
 
            else:
105
 
                self.fail('Did not raise UnsupportedProtocol')
106
 
        finally:
107
 
            # restore original values
108
 
            _set_protocol_handlers(saved_handlers)
109
 
            
110
 
    def test_transport_fallback(self):
111
 
        """Transport with missing dependency causes no error"""
112
 
        saved_handlers = _get_protocol_handlers()
113
 
        try:
114
 
            _clear_protocol_handlers()
115
 
            register_transport_proto('foo')
116
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
117
 
                    'BackupTransportHandler')
118
 
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
119
 
                    'BadTransportHandler')
120
 
            t = get_transport('foo://fooserver/foo')
121
 
            self.assertTrue(isinstance(t, BackupTransportHandler))
122
 
        finally:
123
 
            _set_protocol_handlers(saved_handlers)
124
 
 
125
 
    def test_LateReadError(self):
126
 
        """The LateReadError helper should raise on read()."""
127
 
        a_file = LateReadError('a path')
128
 
        try:
129
 
            a_file.read()
130
 
        except ReadError, error:
131
 
            self.assertEqual('a path', error.path)
132
 
        self.assertRaises(ReadError, a_file.read, 40)
133
 
        a_file.close()
134
 
 
135
 
    def test__combine_paths(self):
136
 
        t = Transport('/')
137
 
        self.assertEqual('/home/sarah/project/foo',
138
 
                         t._combine_paths('/home/sarah', 'project/foo'))
139
 
        self.assertEqual('/etc',
140
 
                         t._combine_paths('/home/sarah', '../../etc'))
141
 
        self.assertEqual('/etc',
142
 
                         t._combine_paths('/home/sarah', '../../../etc'))
143
 
        self.assertEqual('/etc',
144
 
                         t._combine_paths('/home/sarah', '/etc'))
145
 
 
146
 
    def test_local_abspath_non_local_transport(self):
147
 
        # the base implementation should throw
148
 
        t = MemoryTransport()
149
 
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
150
 
        self.assertEqual('memory:///t is not a local path.', str(e))
151
 
 
152
 
 
153
 
class TestCoalesceOffsets(TestCase):
154
 
    
155
 
    def check(self, expected, offsets, limit=0, fudge=0):
156
 
        coalesce = Transport._coalesce_offsets
157
 
        exp = [_CoalescedOffset(*x) for x in expected]
158
 
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
159
 
        self.assertEqual(exp, out)
160
 
 
161
 
    def test_coalesce_empty(self):
162
 
        self.check([], [])
163
 
 
164
 
    def test_coalesce_simple(self):
165
 
        self.check([(0, 10, [(0, 10)])], [(0, 10)])
166
 
 
167
 
    def test_coalesce_unrelated(self):
168
 
        self.check([(0, 10, [(0, 10)]),
169
 
                    (20, 10, [(0, 10)]),
170
 
                   ], [(0, 10), (20, 10)])
171
 
            
172
 
    def test_coalesce_unsorted(self):
173
 
        self.check([(20, 10, [(0, 10)]),
174
 
                    (0, 10, [(0, 10)]),
175
 
                   ], [(20, 10), (0, 10)])
176
 
 
177
 
    def test_coalesce_nearby(self):
178
 
        self.check([(0, 20, [(0, 10), (10, 10)])],
179
 
                   [(0, 10), (10, 10)])
180
 
 
181
 
    def test_coalesce_overlapped(self):
182
 
        self.check([(0, 15, [(0, 10), (5, 10)])],
183
 
                   [(0, 10), (5, 10)])
184
 
 
185
 
    def test_coalesce_limit(self):
186
 
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
187
 
                              (30, 10), (40, 10)]),
188
 
                    (60, 50, [(0, 10), (10, 10), (20, 10),
189
 
                              (30, 10), (40, 10)]),
190
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
191
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
192
 
                       (90, 10), (100, 10)],
193
 
                    limit=5)
194
 
 
195
 
    def test_coalesce_no_limit(self):
196
 
        self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
197
 
                               (30, 10), (40, 10), (50, 10),
198
 
                               (60, 10), (70, 10), (80, 10),
199
 
                               (90, 10)]),
200
 
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
201
 
                       (50, 10), (60, 10), (70, 10), (80, 10),
202
 
                       (90, 10), (100, 10)])
203
 
 
204
 
    def test_coalesce_fudge(self):
205
 
        self.check([(10, 30, [(0, 10), (20, 10)]),
206
 
                    (100, 10, [(0, 10),]),
207
 
                   ], [(10, 10), (30, 10), (100, 10)],
208
 
                   fudge=10
209
 
                  )
 
39
    def test_urlescape(self):
 
40
        self.assertEqual('%25', urlescape('%'))
 
41
 
 
42
 
 
43
class TestTransportMixIn(object):
 
44
    """Subclass this, and it will provide a series of tests for a Transport.
 
45
    It assumes that the Transport object is connected to the 
 
46
    current working directory.  So that whatever is done 
 
47
    through the transport, should show up in the working 
 
48
    directory, and vice-versa.
 
49
 
 
50
    This also tests to make sure that the functions work with both
 
51
    generators and lists (assuming iter(list) is effectively a generator)
 
52
    """
 
53
    readonly = False
 
54
    def get_transport(self):
 
55
        """Children should override this to return the Transport object.
 
56
        """
 
57
        raise NotImplementedError
 
58
 
 
59
    def test_has(self):
 
60
        t = self.get_transport()
 
61
 
 
62
        files = ['a', 'b', 'e', 'g', '%']
 
63
        self.build_tree(files)
 
64
        self.assertEqual(t.has('a'), True)
 
65
        self.assertEqual(t.has('c'), False)
 
66
        self.assertEqual(t.has(urlescape('%')), True)
 
67
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
 
68
                [True, True, False, False, True, False, True, False])
 
69
        self.assertEqual(t.has_any(['a', 'b', 'c']), True)
 
70
        self.assertEqual(t.has_any(['c', 'd', 'f', urlescape('%%')]), False)
 
71
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
 
72
                [True, True, False, False, True, False, True, False])
 
73
        self.assertEqual(t.has_any(['c', 'c', 'c']), False)
 
74
        self.assertEqual(t.has_any(['b', 'b', 'b']), True)
 
75
 
 
76
    def test_get(self):
 
77
        t = self.get_transport()
 
78
 
 
79
        files = ['a', 'b', 'e', 'g']
 
80
        self.build_tree(files)
 
81
        self.assertEqual(t.get('a').read(), open('a').read())
 
82
        content_f = t.get_multi(files)
 
83
        for path,f in zip(files, content_f):
 
84
            self.assertEqual(open(path).read(), f.read())
 
85
 
 
86
        content_f = t.get_multi(iter(files))
 
87
        for path,f in zip(files, content_f):
 
88
            self.assertEqual(open(path).read(), f.read())
 
89
 
 
90
        self.assertRaises(NoSuchFile, t.get, 'c')
 
91
        try:
 
92
            files = list(t.get_multi(['a', 'b', 'c']))
 
93
        except NoSuchFile:
 
94
            pass
 
95
        else:
 
96
            self.fail('Failed to raise NoSuchFile for missing file in get_multi')
 
97
        try:
 
98
            files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
 
99
        except NoSuchFile:
 
100
            pass
 
101
        else:
 
102
            self.fail('Failed to raise NoSuchFile for missing file in get_multi')
 
103
 
 
104
    def test_put(self):
 
105
        t = self.get_transport()
 
106
 
 
107
        if self.readonly:
 
108
            self.assertRaises(TransportNotPossible,
 
109
                    t.put, 'a', 'some text for a\n')
 
110
            open('a', 'wb').write('some text for a\n')
 
111
        else:
 
112
            t.put('a', 'some text for a\n')
 
113
        self.assert_(os.path.exists('a'))
 
114
        self.check_file_contents('a', 'some text for a\n')
 
115
        self.assertEqual(t.get('a').read(), 'some text for a\n')
 
116
        # Make sure 'has' is updated
 
117
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
 
118
                [True, False, False, False, False])
 
119
        if self.readonly:
 
120
            self.assertRaises(TransportNotPossible,
 
121
                    t.put_multi,
 
122
                    [('a', 'new\ncontents for\na\n'),
 
123
                        ('d', 'contents\nfor d\n')])
 
124
            open('a', 'wb').write('new\ncontents for\na\n')
 
125
            open('d', 'wb').write('contents\nfor d\n')
 
126
        else:
 
127
            # Put also replaces contents
 
128
            self.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
 
129
                                          ('d', 'contents\nfor d\n')]),
 
130
                             2)
 
131
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e'])),
 
132
                [True, False, False, True, False])
 
133
        self.check_file_contents('a', 'new\ncontents for\na\n')
 
134
        self.check_file_contents('d', 'contents\nfor d\n')
 
135
 
 
136
        if self.readonly:
 
137
            self.assertRaises(TransportNotPossible,
 
138
                t.put_multi, iter([('a', 'diff\ncontents for\na\n'),
 
139
                                  ('d', 'another contents\nfor d\n')]))
 
140
            open('a', 'wb').write('diff\ncontents for\na\n')
 
141
            open('d', 'wb').write('another contents\nfor d\n')
 
142
        else:
 
143
            self.assertEqual(
 
144
                t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
 
145
                                  ('d', 'another contents\nfor d\n')]))
 
146
                             , 2)
 
147
        self.check_file_contents('a', 'diff\ncontents for\na\n')
 
148
        self.check_file_contents('d', 'another contents\nfor d\n')
 
149
 
 
150
        if self.readonly:
 
151
            self.assertRaises(TransportNotPossible,
 
152
                    t.put, 'path/doesnt/exist/c', 'contents')
 
153
        else:
 
154
            self.assertRaises(NoSuchFile,
 
155
                    t.put, 'path/doesnt/exist/c', 'contents')
 
156
 
 
157
    def test_put_file(self):
 
158
        t = self.get_transport()
 
159
 
 
160
        # Test that StringIO can be used as a file-like object with put
 
161
        f1 = StringIO('this is a string\nand some more stuff\n')
 
162
        if self.readonly:
 
163
            open('f1', 'wb').write(f1.read())
 
164
        else:
 
165
            t.put('f1', f1)
 
166
 
 
167
        del f1
 
168
 
 
169
        self.check_file_contents('f1', 
 
170
                'this is a string\nand some more stuff\n')
 
171
 
 
172
        f2 = StringIO('here is some text\nand a bit more\n')
 
173
        f3 = StringIO('some text for the\nthird file created\n')
 
174
 
 
175
        if self.readonly:
 
176
            open('f2', 'wb').write(f2.read())
 
177
            open('f3', 'wb').write(f3.read())
 
178
        else:
 
179
            t.put_multi([('f2', f2), ('f3', f3)])
 
180
 
 
181
        del f2, f3
 
182
 
 
183
        self.check_file_contents('f2', 'here is some text\nand a bit more\n')
 
184
        self.check_file_contents('f3', 'some text for the\nthird file created\n')
 
185
 
 
186
        # Test that an actual file object can be used with put
 
187
        f4 = open('f1', 'rb')
 
188
        if self.readonly:
 
189
            open('f4', 'wb').write(f4.read())
 
190
        else:
 
191
            t.put('f4', f4)
 
192
 
 
193
        del f4
 
194
 
 
195
        self.check_file_contents('f4', 
 
196
                'this is a string\nand some more stuff\n')
 
197
 
 
198
        f5 = open('f2', 'rb')
 
199
        f6 = open('f3', 'rb')
 
200
        if self.readonly:
 
201
            open('f5', 'wb').write(f5.read())
 
202
            open('f6', 'wb').write(f6.read())
 
203
        else:
 
204
            t.put_multi([('f5', f5), ('f6', f6)])
 
205
 
 
206
        del f5, f6
 
207
 
 
208
        self.check_file_contents('f5', 'here is some text\nand a bit more\n')
 
209
        self.check_file_contents('f6', 'some text for the\nthird file created\n')
 
210
 
 
211
 
 
212
 
 
213
    def test_mkdir(self):
 
214
        t = self.get_transport()
 
215
 
 
216
        # Test mkdir
 
217
        os.mkdir('dir_a')
 
218
        self.assertEqual(t.has('dir_a'), True)
 
219
        self.assertEqual(t.has('dir_b'), False)
 
220
 
 
221
        if self.readonly:
 
222
            self.assertRaises(TransportNotPossible,
 
223
                    t.mkdir, 'dir_b')
 
224
            os.mkdir('dir_b')
 
225
        else:
 
226
            t.mkdir('dir_b')
 
227
        self.assertEqual(t.has('dir_b'), True)
 
228
        self.assert_(os.path.isdir('dir_b'))
 
229
 
 
230
        if self.readonly:
 
231
            self.assertRaises(TransportNotPossible,
 
232
                    t.mkdir_multi, ['dir_c', 'dir_d'])
 
233
            os.mkdir('dir_c')
 
234
            os.mkdir('dir_d')
 
235
        else:
 
236
            t.mkdir_multi(['dir_c', 'dir_d'])
 
237
 
 
238
        if self.readonly:
 
239
            self.assertRaises(TransportNotPossible,
 
240
                    t.mkdir_multi, iter(['dir_e', 'dir_f']))
 
241
            os.mkdir('dir_e')
 
242
            os.mkdir('dir_f')
 
243
        else:
 
244
            t.mkdir_multi(iter(['dir_e', 'dir_f']))
 
245
        self.assertEqual(list(t.has_multi(
 
246
            ['dir_a', 'dir_b', 'dir_c', 'dir_q',
 
247
             'dir_d', 'dir_e', 'dir_f', 'dir_b'])),
 
248
            [True, True, True, False,
 
249
             True, True, True, True])
 
250
        for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_f']:
 
251
            self.assert_(os.path.isdir(d))
 
252
 
 
253
        if not self.readonly:
 
254
            self.assertRaises(NoSuchFile, t.mkdir, 'path/doesnt/exist')
 
255
            self.assertRaises(FileExists, t.mkdir, 'dir_a') # Creating a directory again should fail
 
256
 
 
257
        # Make sure the transport recognizes when a
 
258
        # directory is created by other means
 
259
        # Caching Transports will fail, because dir_e was already seen not
 
260
        # to exist. So instead, we will search for a new directory
 
261
        #os.mkdir('dir_e')
 
262
        #if not self.readonly:
 
263
        #    self.assertRaises(FileExists, t.mkdir, 'dir_e')
 
264
 
 
265
        os.mkdir('dir_g')
 
266
        if not self.readonly:
 
267
            self.assertRaises(FileExists, t.mkdir, 'dir_g')
 
268
 
 
269
        # Test get/put in sub-directories
 
270
        if self.readonly:
 
271
            open('dir_a/a', 'wb').write('contents of dir_a/a')
 
272
            open('dir_b/b', 'wb').write('contents of dir_b/b')
 
273
        else:
 
274
            self.assertEqual(
 
275
                t.put_multi([('dir_a/a', 'contents of dir_a/a'),
 
276
                             ('dir_b/b', 'contents of dir_b/b')])
 
277
                          , 2)
 
278
        for f in ('dir_a/a', 'dir_b/b'):
 
279
            self.assertEqual(t.get(f).read(), open(f).read())
 
280
 
 
281
    def test_copy_to(self):
 
282
        import tempfile
 
283
        from bzrlib.transport.local import LocalTransport
 
284
 
 
285
        t = self.get_transport()
 
286
 
 
287
        files = ['a', 'b', 'c', 'd']
 
288
        self.build_tree(files)
 
289
 
 
290
        dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
 
291
        dtmp_base = os.path.basename(dtmp)
 
292
        local_t = LocalTransport(dtmp)
 
293
 
 
294
        t.copy_to(files, local_t)
 
295
        for f in files:
 
296
            self.assertEquals(open(f).read(),
 
297
                    open(os.path.join(dtmp_base, f)).read())
 
298
 
 
299
        # Test that copying into a missing directory raises
 
300
        # NoSuchFile
 
301
        os.mkdir('e')
 
302
        open('e/f', 'wb').write('contents of e')
 
303
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
 
304
 
 
305
        os.mkdir(os.path.join(dtmp_base, 'e'))
 
306
        t.copy_to(['e/f'], local_t)
 
307
 
 
308
        del dtmp, dtmp_base, local_t
 
309
 
 
310
        dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
 
311
        dtmp_base = os.path.basename(dtmp)
 
312
        local_t = LocalTransport(dtmp)
 
313
 
 
314
        files = ['a', 'b', 'c', 'd']
 
315
        t.copy_to(iter(files), local_t)
 
316
        for f in files:
 
317
            self.assertEquals(open(f).read(),
 
318
                    open(os.path.join(dtmp_base, f)).read())
 
319
 
 
320
        del dtmp, dtmp_base, local_t
 
321
 
 
322
    def test_append(self):
 
323
        t = self.get_transport()
 
324
 
 
325
        if self.readonly:
 
326
            open('a', 'wb').write('diff\ncontents for\na\n')
 
327
            open('b', 'wb').write('contents\nfor b\n')
 
328
        else:
 
329
            t.put_multi([
 
330
                    ('a', 'diff\ncontents for\na\n'),
 
331
                    ('b', 'contents\nfor b\n')
 
332
                    ])
 
333
 
 
334
        if self.readonly:
 
335
            self.assertRaises(TransportNotPossible,
 
336
                    t.append, 'a', 'add\nsome\nmore\ncontents\n')
 
337
            _append('a', 'add\nsome\nmore\ncontents\n')
 
338
        else:
 
339
            t.append('a', 'add\nsome\nmore\ncontents\n')
 
340
 
 
341
        self.check_file_contents('a', 
 
342
            'diff\ncontents for\na\nadd\nsome\nmore\ncontents\n')
 
343
 
 
344
        if self.readonly:
 
345
            self.assertRaises(TransportNotPossible,
 
346
                    t.append_multi,
 
347
                        [('a', 'and\nthen\nsome\nmore\n'),
 
348
                         ('b', 'some\nmore\nfor\nb\n')])
 
349
            _append('a', 'and\nthen\nsome\nmore\n')
 
350
            _append('b', 'some\nmore\nfor\nb\n')
 
351
        else:
 
352
            t.append_multi([('a', 'and\nthen\nsome\nmore\n'),
 
353
                    ('b', 'some\nmore\nfor\nb\n')])
 
354
        self.check_file_contents('a', 
 
355
            'diff\ncontents for\na\n'
 
356
            'add\nsome\nmore\ncontents\n'
 
357
            'and\nthen\nsome\nmore\n')
 
358
        self.check_file_contents('b', 
 
359
                'contents\nfor b\n'
 
360
                'some\nmore\nfor\nb\n')
 
361
 
 
362
        if self.readonly:
 
363
            _append('a', 'a little bit more\n')
 
364
            _append('b', 'from an iterator\n')
 
365
        else:
 
366
            t.append_multi(iter([('a', 'a little bit more\n'),
 
367
                    ('b', 'from an iterator\n')]))
 
368
        self.check_file_contents('a', 
 
369
            'diff\ncontents for\na\n'
 
370
            'add\nsome\nmore\ncontents\n'
 
371
            'and\nthen\nsome\nmore\n'
 
372
            'a little bit more\n')
 
373
        self.check_file_contents('b', 
 
374
                'contents\nfor b\n'
 
375
                'some\nmore\nfor\nb\n'
 
376
                'from an iterator\n')
 
377
 
 
378
    def test_append_file(self):
 
379
        t = self.get_transport()
 
380
 
 
381
        contents = [
 
382
            ('f1', 'this is a string\nand some more stuff\n'),
 
383
            ('f2', 'here is some text\nand a bit more\n'),
 
384
            ('f3', 'some text for the\nthird file created\n'),
 
385
            ('f4', 'this is a string\nand some more stuff\n'),
 
386
            ('f5', 'here is some text\nand a bit more\n'),
 
387
            ('f6', 'some text for the\nthird file created\n')
 
388
        ]
 
389
        
 
390
        if self.readonly:
 
391
            for f, val in contents:
 
392
                open(f, 'wb').write(val)
 
393
        else:
 
394
            t.put_multi(contents)
 
395
 
 
396
        a1 = StringIO('appending to\none\n')
 
397
        if self.readonly:
 
398
            _append('f1', a1.read())
 
399
        else:
 
400
            t.append('f1', a1)
 
401
 
 
402
        del a1
 
403
 
 
404
        self.check_file_contents('f1', 
 
405
                'this is a string\nand some more stuff\n'
 
406
                'appending to\none\n')
 
407
 
 
408
        a2 = StringIO('adding more\ntext to two\n')
 
409
        a3 = StringIO('some garbage\nto put in three\n')
 
410
 
 
411
        if self.readonly:
 
412
            _append('f2', a2.read())
 
413
            _append('f3', a3.read())
 
414
        else:
 
415
            t.append_multi([('f2', a2), ('f3', a3)])
 
416
 
 
417
        del a2, a3
 
418
 
 
419
        self.check_file_contents('f2',
 
420
                'here is some text\nand a bit more\n'
 
421
                'adding more\ntext to two\n')
 
422
        self.check_file_contents('f3', 
 
423
                'some text for the\nthird file created\n'
 
424
                'some garbage\nto put in three\n')
 
425
 
 
426
        # Test that an actual file object can be used with put
 
427
        a4 = open('f1', 'rb')
 
428
        if self.readonly:
 
429
            _append('f4', a4.read())
 
430
        else:
 
431
            t.append('f4', a4)
 
432
 
 
433
        del a4
 
434
 
 
435
        self.check_file_contents('f4', 
 
436
                'this is a string\nand some more stuff\n'
 
437
                'this is a string\nand some more stuff\n'
 
438
                'appending to\none\n')
 
439
 
 
440
        a5 = open('f2', 'rb')
 
441
        a6 = open('f3', 'rb')
 
442
        if self.readonly:
 
443
            _append('f5', a5.read())
 
444
            _append('f6', a6.read())
 
445
        else:
 
446
            t.append_multi([('f5', a5), ('f6', a6)])
 
447
 
 
448
        del a5, a6
 
449
 
 
450
        self.check_file_contents('f5',
 
451
                'here is some text\nand a bit more\n'
 
452
                'here is some text\nand a bit more\n'
 
453
                'adding more\ntext to two\n')
 
454
        self.check_file_contents('f6',
 
455
                'some text for the\nthird file created\n'
 
456
                'some text for the\nthird file created\n'
 
457
                'some garbage\nto put in three\n')
 
458
 
 
459
    def test_delete(self):
 
460
        # TODO: Test Transport.delete
 
461
        pass
 
462
 
 
463
    def test_move(self):
 
464
        # TODO: Test Transport.move
 
465
        pass
 
466
 
 
467
    def test_connection_error(self):
 
468
        """ConnectionError is raised when connection is impossible"""
 
469
        if not hasattr(self, "get_bogus_transport"):
 
470
            return
 
471
        t = self.get_bogus_transport()
 
472
        self.assertRaises(ConnectionError, t.get, '.bzr/branch')
 
473
 
 
474
        
 
475
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
 
476
    def get_transport(self):
 
477
        from bzrlib.transport.local import LocalTransport
 
478
        return LocalTransport('.')
 
479
 
 
480
 
 
481
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
 
482
 
 
483
    readonly = True
 
484
 
 
485
    def get_transport(self):
 
486
        from bzrlib.transport.http import HttpTransport
 
487
        url = self.get_remote_url('.')
 
488
        return HttpTransport(url)
 
489
 
 
490
    def get_bogus_transport(self):
 
491
        from bzrlib.transport.http import HttpTransport
 
492
        return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
210
493
 
211
494
 
212
495
class TestMemoryTransport(TestCase):
213
496
 
214
497
    def test_get_transport(self):
215
 
        MemoryTransport()
 
498
        memory.MemoryTransport()
216
499
 
217
500
    def test_clone(self):
218
 
        transport = MemoryTransport()
219
 
        self.assertTrue(isinstance(transport, MemoryTransport))
220
 
        self.assertEqual("memory:///", transport.clone("/").base)
 
501
        transport = memory.MemoryTransport()
 
502
        self.failUnless(transport.clone() is transport)
221
503
 
222
504
    def test_abspath(self):
223
 
        transport = MemoryTransport()
224
 
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
225
 
 
226
 
    def test_abspath_of_root(self):
227
 
        transport = MemoryTransport()
228
 
        self.assertEqual("memory:///", transport.base)
229
 
        self.assertEqual("memory:///", transport.abspath('/'))
230
 
 
231
 
    def test_abspath_of_relpath_starting_at_root(self):
232
 
        transport = MemoryTransport()
233
 
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
 
505
        transport = memory.MemoryTransport()
 
506
        self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
 
507
 
 
508
    def test_relpath(self):
 
509
        transport = memory.MemoryTransport()
234
510
 
235
511
    def test_append_and_get(self):
236
 
        transport = MemoryTransport()
237
 
        transport.append_bytes('path', 'content')
 
512
        transport = memory.MemoryTransport()
 
513
        transport.append('path', StringIO('content'))
238
514
        self.assertEqual(transport.get('path').read(), 'content')
239
 
        transport.append_file('path', StringIO('content'))
 
515
        transport.append('path', StringIO('content'))
240
516
        self.assertEqual(transport.get('path').read(), 'contentcontent')
241
517
 
242
518
    def test_put_and_get(self):
243
 
        transport = MemoryTransport()
244
 
        transport.put_file('path', StringIO('content'))
 
519
        transport = memory.MemoryTransport()
 
520
        transport.put('path', StringIO('content'))
245
521
        self.assertEqual(transport.get('path').read(), 'content')
246
 
        transport.put_bytes('path', 'content')
 
522
        transport.put('path', StringIO('content'))
247
523
        self.assertEqual(transport.get('path').read(), 'content')
248
524
 
249
525
    def test_append_without_dir_fails(self):
250
 
        transport = MemoryTransport()
 
526
        transport = memory.MemoryTransport()
251
527
        self.assertRaises(NoSuchFile,
252
 
                          transport.append_bytes, 'dir/path', 'content')
 
528
                          transport.append, 'dir/path', StringIO('content'))
253
529
 
254
530
    def test_put_without_dir_fails(self):
255
 
        transport = MemoryTransport()
 
531
        transport = memory.MemoryTransport()
256
532
        self.assertRaises(NoSuchFile,
257
 
                          transport.put_file, 'dir/path', StringIO('content'))
 
533
                          transport.put, 'dir/path', StringIO('content'))
258
534
 
259
535
    def test_get_missing(self):
260
 
        transport = MemoryTransport()
 
536
        transport = memory.MemoryTransport()
261
537
        self.assertRaises(NoSuchFile, transport.get, 'foo')
262
538
 
263
539
    def test_has_missing(self):
264
 
        transport = MemoryTransport()
 
540
        transport = memory.MemoryTransport()
265
541
        self.assertEquals(False, transport.has('foo'))
266
542
 
267
543
    def test_has_present(self):
268
 
        transport = MemoryTransport()
269
 
        transport.append_bytes('foo', 'content')
 
544
        transport = memory.MemoryTransport()
 
545
        transport.append('foo', StringIO('content'))
270
546
        self.assertEquals(True, transport.has('foo'))
271
547
 
272
 
    def test_list_dir(self):
273
 
        transport = MemoryTransport()
274
 
        transport.put_bytes('foo', 'content')
275
 
        transport.mkdir('dir')
276
 
        transport.put_bytes('dir/subfoo', 'content')
277
 
        transport.put_bytes('dirlike', 'content')
278
 
 
279
 
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
280
 
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
281
 
 
282
548
    def test_mkdir(self):
283
 
        transport = MemoryTransport()
 
549
        transport = memory.MemoryTransport()
284
550
        transport.mkdir('dir')
285
 
        transport.append_bytes('dir/path', 'content')
 
551
        transport.append('dir/path', StringIO('content'))
286
552
        self.assertEqual(transport.get('dir/path').read(), 'content')
287
553
 
288
554
    def test_mkdir_missing_parent(self):
289
 
        transport = MemoryTransport()
 
555
        transport = memory.MemoryTransport()
290
556
        self.assertRaises(NoSuchFile,
291
557
                          transport.mkdir, 'dir/dir')
292
558
 
293
559
    def test_mkdir_twice(self):
294
 
        transport = MemoryTransport()
 
560
        transport = memory.MemoryTransport()
295
561
        transport.mkdir('dir')
296
562
        self.assertRaises(FileExists, transport.mkdir, 'dir')
297
563
 
298
564
    def test_parameters(self):
299
 
        transport = MemoryTransport()
 
565
        transport = memory.MemoryTransport()
300
566
        self.assertEqual(True, transport.listable())
301
567
        self.assertEqual(False, transport.should_cache())
302
 
        self.assertEqual(False, transport.is_readonly())
303
568
 
304
569
    def test_iter_files_recursive(self):
305
 
        transport = MemoryTransport()
 
570
        transport = memory.MemoryTransport()
306
571
        transport.mkdir('dir')
307
 
        transport.put_bytes('dir/foo', 'content')
308
 
        transport.put_bytes('dir/bar', 'content')
309
 
        transport.put_bytes('bar', 'content')
 
572
        transport.put('dir/foo', StringIO('content'))
 
573
        transport.put('dir/bar', StringIO('content'))
 
574
        transport.put('bar', StringIO('content'))
310
575
        paths = set(transport.iter_files_recursive())
311
576
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
312
577
 
313
578
    def test_stat(self):
314
 
        transport = MemoryTransport()
315
 
        transport.put_bytes('foo', 'content')
316
 
        transport.put_bytes('bar', 'phowar')
 
579
        transport = memory.MemoryTransport()
 
580
        transport.put('foo', StringIO('content'))
 
581
        transport.put('bar', StringIO('phowar'))
317
582
        self.assertEqual(7, transport.stat('foo').st_size)
318
583
        self.assertEqual(6, transport.stat('bar').st_size)
319
584
 
320
 
 
321
 
class ChrootDecoratorTransportTest(TestCase):
322
 
    """Chroot decoration specific tests."""
323
 
 
324
 
    def test_abspath(self):
325
 
        # The abspath is always relative to the chroot_url.
326
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
327
 
        server.setUp()
328
 
        transport = get_transport(server.get_url())
329
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
330
 
 
331
 
        subdir_transport = transport.clone('subdir')
332
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
333
 
        server.tearDown()
334
 
 
335
 
    def test_clone(self):
336
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
337
 
        server.setUp()
338
 
        transport = get_transport(server.get_url())
339
 
        # relpath from root and root path are the same
340
 
        relpath_cloned = transport.clone('foo')
341
 
        abspath_cloned = transport.clone('/foo')
342
 
        self.assertEqual(server, relpath_cloned.server)
343
 
        self.assertEqual(server, abspath_cloned.server)
344
 
        server.tearDown()
345
 
    
346
 
    def test_chroot_url_preserves_chroot(self):
347
 
        """Calling get_transport on a chroot transport's base should produce a
348
 
        transport with exactly the same behaviour as the original chroot
349
 
        transport.
350
 
 
351
 
        This is so that it is not possible to escape a chroot by doing::
352
 
            url = chroot_transport.base
353
 
            parent_url = urlutils.join(url, '..')
354
 
            new_transport = get_transport(parent_url)
355
 
        """
356
 
        server = ChrootServer(get_transport('memory:///path/subpath'))
357
 
        server.setUp()
358
 
        transport = get_transport(server.get_url())
359
 
        new_transport = get_transport(transport.base)
360
 
        self.assertEqual(transport.server, new_transport.server)
361
 
        self.assertEqual(transport.base, new_transport.base)
362
 
        server.tearDown()
363
 
        
364
 
    def test_urljoin_preserves_chroot(self):
365
 
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
366
 
        URL that escapes the intended chroot.
367
 
 
368
 
        This is so that it is not possible to escape a chroot by doing::
369
 
            url = chroot_transport.base
370
 
            parent_url = urlutils.join(url, '..')
371
 
            new_transport = get_transport(parent_url)
372
 
        """
373
 
        server = ChrootServer(get_transport('memory:///path/'))
374
 
        server.setUp()
375
 
        transport = get_transport(server.get_url())
376
 
        self.assertRaises(
377
 
            InvalidURLJoin, urlutils.join, transport.base, '..')
378
 
        server.tearDown()
379
 
 
380
 
 
381
 
class ChrootServerTest(TestCase):
382
 
 
383
 
    def test_construct(self):
384
 
        backing_transport = MemoryTransport()
385
 
        server = ChrootServer(backing_transport)
386
 
        self.assertEqual(backing_transport, server.backing_transport)
387
 
 
388
 
    def test_setUp(self):
389
 
        backing_transport = MemoryTransport()
390
 
        server = ChrootServer(backing_transport)
391
 
        server.setUp()
392
 
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
393
 
 
394
 
    def test_tearDown(self):
395
 
        backing_transport = MemoryTransport()
396
 
        server = ChrootServer(backing_transport)
397
 
        server.setUp()
398
 
        server.tearDown()
399
 
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
400
 
 
401
 
    def test_get_url(self):
402
 
        backing_transport = MemoryTransport()
403
 
        server = ChrootServer(backing_transport)
404
 
        server.setUp()
405
 
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
406
 
        server.tearDown()
407
 
 
408
 
 
409
 
class ReadonlyDecoratorTransportTest(TestCase):
410
 
    """Readonly decoration specific tests."""
411
 
 
412
 
    def test_local_parameters(self):
413
 
        import bzrlib.transport.readonly as readonly
414
 
        # connect to . in readonly mode
415
 
        transport = readonly.ReadonlyTransportDecorator('readonly+.')
416
 
        self.assertEqual(True, transport.listable())
417
 
        self.assertEqual(False, transport.should_cache())
418
 
        self.assertEqual(True, transport.is_readonly())
419
 
 
420
 
    def test_http_parameters(self):
421
 
        from bzrlib.tests.HttpServer import HttpServer
422
 
        import bzrlib.transport.readonly as readonly
423
 
        # connect to . via http which is not listable
424
 
        server = HttpServer()
425
 
        server.setUp()
426
 
        try:
427
 
            transport = get_transport('readonly+' + server.get_url())
428
 
            self.failUnless(isinstance(transport,
429
 
                                       readonly.ReadonlyTransportDecorator))
430
 
            self.assertEqual(False, transport.listable())
431
 
            self.assertEqual(True, transport.should_cache())
432
 
            self.assertEqual(True, transport.is_readonly())
433
 
        finally:
434
 
            server.tearDown()
435
 
 
436
 
 
437
 
class FakeNFSDecoratorTests(TestCaseInTempDir):
438
 
    """NFS decorator specific tests."""
439
 
 
440
 
    def get_nfs_transport(self, url):
441
 
        import bzrlib.transport.fakenfs as fakenfs
442
 
        # connect to url with nfs decoration
443
 
        return fakenfs.FakeNFSTransportDecorator('fakenfs+' + url)
444
 
 
445
 
    def test_local_parameters(self):
446
 
        # the listable, should_cache and is_readonly parameters
447
 
        # are not changed by the fakenfs decorator
448
 
        transport = self.get_nfs_transport('.')
449
 
        self.assertEqual(True, transport.listable())
450
 
        self.assertEqual(False, transport.should_cache())
451
 
        self.assertEqual(False, transport.is_readonly())
452
 
 
453
 
    def test_http_parameters(self):
454
 
        # the listable, should_cache and is_readonly parameters
455
 
        # are not changed by the fakenfs decorator
456
 
        from bzrlib.tests.HttpServer import HttpServer
457
 
        # connect to . via http which is not listable
458
 
        server = HttpServer()
459
 
        server.setUp()
460
 
        try:
461
 
            transport = self.get_nfs_transport(server.get_url())
462
 
            self.assertIsInstance(
463
 
                transport, bzrlib.transport.fakenfs.FakeNFSTransportDecorator)
464
 
            self.assertEqual(False, transport.listable())
465
 
            self.assertEqual(True, transport.should_cache())
466
 
            self.assertEqual(True, transport.is_readonly())
467
 
        finally:
468
 
            server.tearDown()
469
 
 
470
 
    def test_fakenfs_server_default(self):
471
 
        # a FakeNFSServer() should bring up a local relpath server for itself
472
 
        import bzrlib.transport.fakenfs as fakenfs
473
 
        server = fakenfs.FakeNFSServer()
474
 
        server.setUp()
475
 
        try:
476
 
            # the url should be decorated appropriately
477
 
            self.assertStartsWith(server.get_url(), 'fakenfs+')
478
 
            # and we should be able to get a transport for it
479
 
            transport = get_transport(server.get_url())
480
 
            # which must be a FakeNFSTransportDecorator instance.
481
 
            self.assertIsInstance(
482
 
                transport, fakenfs.FakeNFSTransportDecorator)
483
 
        finally:
484
 
            server.tearDown()
485
 
 
486
 
    def test_fakenfs_rename_semantics(self):
487
 
        # a FakeNFS transport must mangle the way rename errors occur to
488
 
        # look like NFS problems.
489
 
        transport = self.get_nfs_transport('.')
490
 
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
491
 
                        transport=transport)
492
 
        self.assertRaises(errors.ResourceBusy,
493
 
                          transport.rename, 'from', 'to')
494
 
 
495
 
 
496
 
class FakeVFATDecoratorTests(TestCaseInTempDir):
497
 
    """Tests for simulation of VFAT restrictions"""
498
 
 
499
 
    def get_vfat_transport(self, url):
500
 
        """Return vfat-backed transport for test directory"""
501
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
502
 
        return FakeVFATTransportDecorator('vfat+' + url)
503
 
 
504
 
    def test_transport_creation(self):
505
 
        from bzrlib.transport.fakevfat import FakeVFATTransportDecorator
506
 
        transport = self.get_vfat_transport('.')
507
 
        self.assertIsInstance(transport, FakeVFATTransportDecorator)
508
 
 
509
 
    def test_transport_mkdir(self):
510
 
        transport = self.get_vfat_transport('.')
511
 
        transport.mkdir('HELLO')
512
 
        self.assertTrue(transport.has('hello'))
513
 
        self.assertTrue(transport.has('Hello'))
514
 
 
515
 
    def test_forbidden_chars(self):
516
 
        transport = self.get_vfat_transport('.')
517
 
        self.assertRaises(ValueError, transport.has, "<NU>")
518
 
 
519
 
 
520
 
class BadTransportHandler(Transport):
521
 
    def __init__(self, base_url):
522
 
        raise DependencyNotPresent('some_lib', 'testing missing dependency')
523
 
 
524
 
 
525
 
class BackupTransportHandler(Transport):
526
 
    """Test transport that works as a backup for the BadTransportHandler"""
527
 
    pass
528
 
 
529
 
 
530
 
class TestTransportImplementation(TestCaseInTempDir):
531
 
    """Implementation verification for transports.
532
 
    
533
 
    To verify a transport we need a server factory, which is a callable
534
 
    that accepts no parameters and returns an implementation of
535
 
    bzrlib.transport.Server.
536
 
    
537
 
    That Server is then used to construct transport instances and test
538
 
    the transport via loopback activity.
539
 
 
540
 
    Currently this assumes that the Transport object is connected to the 
541
 
    current working directory.  So that whatever is done 
542
 
    through the transport, should show up in the working 
543
 
    directory, and vice-versa. This is a bug, because its possible to have
544
 
    URL schemes which provide access to something that may not be 
545
 
    result in storage on the local disk, i.e. due to file system limits, or 
546
 
    due to it being a database or some other non-filesystem tool.
547
 
 
548
 
    This also tests to make sure that the functions work with both
549
 
    generators and lists (assuming iter(list) is effectively a generator)
550
 
    """
551
 
    
552
 
    def setUp(self):
553
 
        super(TestTransportImplementation, self).setUp()
554
 
        self._server = self.transport_server()
555
 
        self._server.setUp()
556
 
        self.addCleanup(self._server.tearDown)
557
 
 
558
 
    def get_transport(self, relpath=None):
559
 
        """Return a connected transport to the local directory.
560
 
 
561
 
        :param relpath: a path relative to the base url.
562
 
        """
563
 
        base_url = self._server.get_url()
564
 
        url = self._adjust_url(base_url, relpath)
565
 
        # try getting the transport via the regular interface:
566
 
        t = get_transport(url)
567
 
        # vila--20070607 if the following are commented out the test suite
568
 
        # still pass. Is this really still needed or was it a forgotten
569
 
        # temporary fix ?
570
 
        if not isinstance(t, self.transport_class):
571
 
            # we did not get the correct transport class type. Override the
572
 
            # regular connection behaviour by direct construction.
573
 
            t = self.transport_class(url)
574
 
        return t
575
 
 
576
 
 
577
 
class TestLocalTransports(TestCase):
578
 
 
579
 
    def test_get_transport_from_abspath(self):
580
 
        here = os.path.abspath('.')
581
 
        t = get_transport(here)
582
 
        self.assertIsInstance(t, LocalTransport)
583
 
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
584
 
 
585
 
    def test_get_transport_from_relpath(self):
586
 
        here = os.path.abspath('.')
587
 
        t = get_transport('.')
588
 
        self.assertIsInstance(t, LocalTransport)
589
 
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
590
 
 
591
 
    def test_get_transport_from_local_url(self):
592
 
        here = os.path.abspath('.')
593
 
        here_url = urlutils.local_path_to_url(here) + '/'
594
 
        t = get_transport(here_url)
595
 
        self.assertIsInstance(t, LocalTransport)
596
 
        self.assertEquals(t.base, here_url)
597
 
 
598
 
    def test_local_abspath(self):
599
 
        here = os.path.abspath('.')
600
 
        t = get_transport(here)
601
 
        self.assertEquals(t.local_abspath(''), here)
602
 
 
603
 
 
604
 
class TestWin32LocalTransport(TestCase):
605
 
 
606
 
    def test_unc_clone_to_root(self):
607
 
        # Win32 UNC path like \\HOST\path
608
 
        # clone to root should stop at least at \\HOST part
609
 
        # not on \\
610
 
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
611
 
        for i in xrange(4):
612
 
            t = t.clone('..')
613
 
        self.assertEquals(t.base, 'file://HOST/')
614
 
        # make sure we reach the root
615
 
        t = t.clone('..')
616
 
        self.assertEquals(t.base, 'file://HOST/')
617
 
 
618
 
 
619
 
class TestConnectedTransport(TestCase):
620
 
    """Tests for connected to remote server transports"""
621
 
 
622
 
    def test_parse_url(self):
623
 
        t = ConnectedTransport('sftp://simple.example.com/home/source')
624
 
        self.assertEquals(t._host, 'simple.example.com')
625
 
        self.assertEquals(t._port, None)
626
 
        self.assertEquals(t._path, '/home/source/')
627
 
        self.failUnless(t._user is None)
628
 
        self.failUnless(t._password is None)
629
 
 
630
 
        self.assertEquals(t.base, 'sftp://simple.example.com/home/source/')
631
 
 
632
 
    def test_parse_quoted_url(self):
633
 
        t = ConnectedTransport('http://ro%62ey:h%40t@ex%41mple.com:2222/path')
634
 
        self.assertEquals(t._host, 'exAmple.com')
635
 
        self.assertEquals(t._port, 2222)
636
 
        self.assertEquals(t._user, 'robey')
637
 
        self.assertEquals(t._password, 'h@t')
638
 
        self.assertEquals(t._path, '/path/')
639
 
 
640
 
        # Base should not keep track of the password
641
 
        self.assertEquals(t.base, 'http://robey@exAmple.com:2222/path/')
642
 
 
643
 
    def test_parse_invalid_url(self):
644
 
        self.assertRaises(errors.InvalidURL,
645
 
                          ConnectedTransport,
646
 
                          'sftp://lily.org:~janneke/public/bzr/gub')
647
 
 
648
 
    def test_relpath(self):
649
 
        t = ConnectedTransport('sftp://user@host.com/abs/path')
650
 
 
651
 
        self.assertEquals(t.relpath('sftp://user@host.com/abs/path/sub'), 'sub')
652
 
        self.assertRaises(errors.PathNotChild, t.relpath,
653
 
                          'http://user@host.com/abs/path/sub')
654
 
        self.assertRaises(errors.PathNotChild, t.relpath,
655
 
                          'sftp://user2@host.com/abs/path/sub')
656
 
        self.assertRaises(errors.PathNotChild, t.relpath,
657
 
                          'sftp://user@otherhost.com/abs/path/sub')
658
 
        self.assertRaises(errors.PathNotChild, t.relpath,
659
 
                          'sftp://user@host.com:33/abs/path/sub')
660
 
        # Make sure it works when we don't supply a username
661
 
        t = ConnectedTransport('sftp://host.com/abs/path')
662
 
        self.assertEquals(t.relpath('sftp://host.com/abs/path/sub'), 'sub')
663
 
 
664
 
        # Make sure it works when parts of the path will be url encoded
665
 
        t = ConnectedTransport('sftp://host.com/dev/%path')
666
 
        self.assertEquals(t.relpath('sftp://host.com/dev/%path/sub'), 'sub')
667
 
 
668
 
    def test_connection_sharing_propagate_credentials(self):
669
 
        t = ConnectedTransport('foo://user@host.com/abs/path')
670
 
        self.assertIs(None, t._get_connection())
671
 
        self.assertIs(None, t._password)
672
 
        c = t.clone('subdir')
673
 
        self.assertEquals(None, c._get_connection())
674
 
        self.assertIs(None, t._password)
675
 
 
676
 
        # Simulate the user entering a password
677
 
        password = 'secret'
678
 
        connection = object()
679
 
        t._set_connection(connection, password)
680
 
        self.assertIs(connection, t._get_connection())
681
 
        self.assertIs(password, t._get_credentials())
682
 
        self.assertIs(connection, c._get_connection())
683
 
        self.assertIs(password, c._get_credentials())
684
 
 
685
 
        # credentials can be updated
686
 
        new_password = 'even more secret'
687
 
        c._update_credentials(new_password)
688
 
        self.assertIs(connection, t._get_connection())
689
 
        self.assertIs(new_password, t._get_credentials())
690
 
        self.assertIs(connection, c._get_connection())
691
 
        self.assertIs(new_password, c._get_credentials())
692
 
 
693
 
 
694
 
class TestReusedTransports(TestCase):
695
 
    """Tests for transport reuse"""
696
 
 
697
 
    def test_reuse_same_transport(self):
698
 
        t1 = get_transport('http://foo/')
699
 
        t2 = get_transport('http://foo/', possible_transports=[t1])
700
 
        self.assertIs(t1, t2)
701
 
 
702
 
        # Also check that final '/' are handled correctly
703
 
        t3 = get_transport('http://foo/path/')
704
 
        t4 = get_transport('http://foo/path', possible_transports=[t3])
705
 
        self.assertIs(t3, t4)
706
 
 
707
 
        t5 = get_transport('http://foo/path')
708
 
        t6 = get_transport('http://foo/path/', possible_transports=[t5])
709
 
        self.assertIs(t5, t6)
710
 
 
711
 
    def test_don_t_reuse_different_transport(self):
712
 
        t1 = get_transport('http://foo/path')
713
 
        t2 = get_transport('http://bar/path', possible_transports=[t1])
714
 
        self.assertIsNot(t1, t2)
715
 
 
716
 
 
717
 
def get_test_permutations():
718
 
    """Return transport permutations to be used in testing.
719
 
 
720
 
    This module registers some transports, but they're only for testing
721
 
    registration.  We don't really want to run all the transport tests against
722
 
    them.
723
 
    """
724
 
    return []