← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport.py

  • Committer: Robert Collins
  • Date: 2007-03-08 04:06:06 UTC
  • mfrom: (2323.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 2442.
  • Revision ID: robertc@robertcollins.net-20070308040606-84gsniv56huiyjt4
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2004, 2005, 2006 by Canonical Ltd
2
 
 
 
1
# Copyright (C) 2004, 2005, 2006 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
21
21
from cStringIO import StringIO
22
22
 
23
23
import bzrlib
 
24
from bzrlib import urlutils
24
25
from bzrlib.errors import (NoSuchFile, FileExists,
25
26
                           TransportNotPossible,
26
27
                           ConnectionError,
27
28
                           DependencyNotPresent,
 
29
                           UnsupportedProtocol,
 
30
                           PathNotChild,
28
31
                           )
29
32
from bzrlib.tests import TestCase, TestCaseInTempDir
30
 
from bzrlib.transport import (_get_protocol_handlers,
 
33
from bzrlib.transport import (_CoalescedOffset,
 
34
                              _get_protocol_handlers,
31
35
                              _get_transport_modules,
32
36
                              get_transport,
33
37
                              register_lazy_transport,
34
38
                              _set_protocol_handlers,
35
 
                              urlescape,
36
39
                              Transport,
37
40
                              )
38
41
from bzrlib.transport.memory import MemoryTransport
39
 
from bzrlib.transport.local import LocalTransport
 
42
from bzrlib.transport.local import (LocalTransport,
 
43
                                    EmulatedWin32LocalTransport)
 
44
 
 
45
 
 
46
# TODO: Should possibly split transport-specific tests into their own files.
40
47
 
41
48
 
42
49
class TestTransport(TestCase):
43
50
    """Test the non transport-concrete class functionality."""
44
51
 
45
 
    def test_urlescape(self):
46
 
        self.assertEqual('%25', urlescape('%'))
47
 
 
48
52
    def test__get_set_protocol_handlers(self):
49
53
        handlers = _get_protocol_handlers()
50
54
        self.assertNotEqual({}, handlers)
74
78
        try:
75
79
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
76
80
                    'BadTransportHandler')
77
 
            t = get_transport('foo://fooserver/foo')
78
 
            # because we failed to load the transport
79
 
            self.assertTrue(isinstance(t, LocalTransport))
 
81
            try:
 
82
                get_transport('foo://fooserver/foo')
 
83
            except UnsupportedProtocol, e:
 
84
                e_str = str(e)
 
85
                self.assertEquals('Unsupported protocol'
 
86
                                  ' for url "foo://fooserver/foo":'
 
87
                                  ' Unable to import library "some_lib":'
 
88
                                  ' testing missing dependency', str(e))
 
89
            else:
 
90
                self.fail('Did not raise UnsupportedProtocol')
80
91
        finally:
81
92
            # restore original values
82
93
            _set_protocol_handlers(saved_handlers)
94
105
            self.assertTrue(isinstance(t, BackupTransportHandler))
95
106
        finally:
96
107
            _set_protocol_handlers(saved_handlers)
 
108
 
 
109
    def test__combine_paths(self):
 
110
        t = Transport('/')
 
111
        self.assertEqual('/home/sarah/project/foo',
 
112
                         t._combine_paths('/home/sarah', 'project/foo'))
 
113
        self.assertEqual('/etc',
 
114
                         t._combine_paths('/home/sarah', '../../etc'))
 
115
        self.assertEqual('/etc',
 
116
                         t._combine_paths('/home/sarah', '../../../etc'))
 
117
        self.assertEqual('/etc',
 
118
                         t._combine_paths('/home/sarah', '/etc'))
 
119
 
 
120
 
 
121
class TestCoalesceOffsets(TestCase):
 
122
    
 
123
    def check(self, expected, offsets, limit=0, fudge=0):
 
124
        coalesce = Transport._coalesce_offsets
 
125
        exp = [_CoalescedOffset(*x) for x in expected]
 
126
        out = list(coalesce(offsets, limit=limit, fudge_factor=fudge))
 
127
        self.assertEqual(exp, out)
 
128
 
 
129
    def test_coalesce_empty(self):
 
130
        self.check([], [])
 
131
 
 
132
    def test_coalesce_simple(self):
 
133
        self.check([(0, 10, [(0, 10)])], [(0, 10)])
 
134
 
 
135
    def test_coalesce_unrelated(self):
 
136
        self.check([(0, 10, [(0, 10)]),
 
137
                    (20, 10, [(0, 10)]),
 
138
                   ], [(0, 10), (20, 10)])
97
139
            
 
140
    def test_coalesce_unsorted(self):
 
141
        self.check([(20, 10, [(0, 10)]),
 
142
                    (0, 10, [(0, 10)]),
 
143
                   ], [(20, 10), (0, 10)])
 
144
 
 
145
    def test_coalesce_nearby(self):
 
146
        self.check([(0, 20, [(0, 10), (10, 10)])],
 
147
                   [(0, 10), (10, 10)])
 
148
 
 
149
    def test_coalesce_overlapped(self):
 
150
        self.check([(0, 15, [(0, 10), (5, 10)])],
 
151
                   [(0, 10), (5, 10)])
 
152
 
 
153
    def test_coalesce_limit(self):
 
154
        self.check([(10, 50, [(0, 10), (10, 10), (20, 10),
 
155
                              (30, 10), (40, 10)]),
 
156
                    (60, 50, [(0, 10), (10, 10), (20, 10),
 
157
                              (30, 10), (40, 10)]),
 
158
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
 
159
                       (50, 10), (60, 10), (70, 10), (80, 10),
 
160
                       (90, 10), (100, 10)],
 
161
                    limit=5)
 
162
 
 
163
    def test_coalesce_no_limit(self):
 
164
        self.check([(10, 100, [(0, 10), (10, 10), (20, 10),
 
165
                               (30, 10), (40, 10), (50, 10),
 
166
                               (60, 10), (70, 10), (80, 10),
 
167
                               (90, 10)]),
 
168
                   ], [(10, 10), (20, 10), (30, 10), (40, 10),
 
169
                       (50, 10), (60, 10), (70, 10), (80, 10),
 
170
                       (90, 10), (100, 10)])
 
171
 
 
172
    def test_coalesce_fudge(self):
 
173
        self.check([(10, 30, [(0, 10), (20, 10)]),
 
174
                    (100, 10, [(0, 10),]),
 
175
                   ], [(10, 10), (30, 10), (100, 10)],
 
176
                   fudge=10
 
177
                  )
 
178
 
98
179
 
99
180
class TestMemoryTransport(TestCase):
100
181
 
104
185
    def test_clone(self):
105
186
        transport = MemoryTransport()
106
187
        self.assertTrue(isinstance(transport, MemoryTransport))
 
188
        self.assertEqual("memory:///", transport.clone("/").base)
107
189
 
108
190
    def test_abspath(self):
109
191
        transport = MemoryTransport()
110
 
        self.assertEqual("memory:/relpath", transport.abspath('relpath'))
111
 
 
112
 
    def test_relpath(self):
113
 
        transport = MemoryTransport()
 
192
        self.assertEqual("memory:///relpath", transport.abspath('relpath'))
 
193
 
 
194
    def test_abspath_of_root(self):
 
195
        transport = MemoryTransport()
 
196
        self.assertEqual("memory:///", transport.base)
 
197
        self.assertEqual("memory:///", transport.abspath('/'))
 
198
 
 
199
    def test_abspath_of_relpath_starting_at_root(self):
 
200
        transport = MemoryTransport()
 
201
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
114
202
 
115
203
    def test_append_and_get(self):
116
204
        transport = MemoryTransport()
117
 
        transport.append('path', StringIO('content'))
 
205
        transport.append_bytes('path', 'content')
118
206
        self.assertEqual(transport.get('path').read(), 'content')
119
 
        transport.append('path', StringIO('content'))
 
207
        transport.append_file('path', StringIO('content'))
120
208
        self.assertEqual(transport.get('path').read(), 'contentcontent')
121
209
 
122
210
    def test_put_and_get(self):
123
211
        transport = MemoryTransport()
124
 
        transport.put('path', StringIO('content'))
 
212
        transport.put_file('path', StringIO('content'))
125
213
        self.assertEqual(transport.get('path').read(), 'content')
126
 
        transport.put('path', StringIO('content'))
 
214
        transport.put_bytes('path', 'content')
127
215
        self.assertEqual(transport.get('path').read(), 'content')
128
216
 
129
217
    def test_append_without_dir_fails(self):
130
218
        transport = MemoryTransport()
131
219
        self.assertRaises(NoSuchFile,
132
 
                          transport.append, 'dir/path', StringIO('content'))
 
220
                          transport.append_bytes, 'dir/path', 'content')
133
221
 
134
222
    def test_put_without_dir_fails(self):
135
223
        transport = MemoryTransport()
136
224
        self.assertRaises(NoSuchFile,
137
 
                          transport.put, 'dir/path', StringIO('content'))
 
225
                          transport.put_file, 'dir/path', StringIO('content'))
138
226
 
139
227
    def test_get_missing(self):
140
228
        transport = MemoryTransport()
146
234
 
147
235
    def test_has_present(self):
148
236
        transport = MemoryTransport()
149
 
        transport.append('foo', StringIO('content'))
 
237
        transport.append_bytes('foo', 'content')
150
238
        self.assertEquals(True, transport.has('foo'))
151
239
 
 
240
    def test_list_dir(self):
 
241
        transport = MemoryTransport()
 
242
        transport.put_bytes('foo', 'content')
 
243
        transport.mkdir('dir')
 
244
        transport.put_bytes('dir/subfoo', 'content')
 
245
        transport.put_bytes('dirlike', 'content')
 
246
 
 
247
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
248
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
249
 
152
250
    def test_mkdir(self):
153
251
        transport = MemoryTransport()
154
252
        transport.mkdir('dir')
155
 
        transport.append('dir/path', StringIO('content'))
 
253
        transport.append_bytes('dir/path', 'content')
156
254
        self.assertEqual(transport.get('dir/path').read(), 'content')
157
255
 
158
256
    def test_mkdir_missing_parent(self):
174
272
    def test_iter_files_recursive(self):
175
273
        transport = MemoryTransport()
176
274
        transport.mkdir('dir')
177
 
        transport.put('dir/foo', StringIO('content'))
178
 
        transport.put('dir/bar', StringIO('content'))
179
 
        transport.put('bar', StringIO('content'))
 
275
        transport.put_bytes('dir/foo', 'content')
 
276
        transport.put_bytes('dir/bar', 'content')
 
277
        transport.put_bytes('bar', 'content')
180
278
        paths = set(transport.iter_files_recursive())
181
279
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
182
280
 
183
281
    def test_stat(self):
184
282
        transport = MemoryTransport()
185
 
        transport.put('foo', StringIO('content'))
186
 
        transport.put('bar', StringIO('phowar'))
 
283
        transport.put_bytes('foo', 'content')
 
284
        transport.put_bytes('bar', 'phowar')
187
285
        self.assertEqual(7, transport.stat('foo').st_size)
188
286
        self.assertEqual(6, transport.stat('bar').st_size)
189
287
 
190
 
        
 
288
 
 
289
class ChrootDecoratorTransportTest(TestCase):
 
290
    """Chroot decoration specific tests."""
 
291
 
 
292
    def test_construct(self):
 
293
        from bzrlib.transport import chroot
 
294
        transport = chroot.ChrootTransportDecorator('chroot+memory:///pathA/')
 
295
        self.assertEqual('memory:///pathA/', transport.chroot_url)
 
296
 
 
297
        transport = chroot.ChrootTransportDecorator(
 
298
            'chroot+memory:///path/B', chroot='memory:///path/')
 
299
        self.assertEqual('memory:///path/', transport.chroot_url)
 
300
 
 
301
    def test_append_file(self):
 
302
        transport = get_transport('chroot+memory:///foo/bar')
 
303
        self.assertRaises(PathNotChild, transport.append_file, '/foo', None)
 
304
 
 
305
    def test_append_bytes(self):
 
306
        transport = get_transport('chroot+memory:///foo/bar')
 
307
        self.assertRaises(PathNotChild, transport.append_bytes, '/foo', 'bytes')
 
308
 
 
309
    def test_clone(self):
 
310
        transport = get_transport('chroot+memory:///foo/bar')
 
311
        self.assertRaises(PathNotChild, transport.clone, '/foo')
 
312
 
 
313
    def test_delete(self):
 
314
        transport = get_transport('chroot+memory:///foo/bar')
 
315
        self.assertRaises(PathNotChild, transport.delete, '/foo')
 
316
 
 
317
    def test_delete_tree(self):
 
318
        transport = get_transport('chroot+memory:///foo/bar')
 
319
        self.assertRaises(PathNotChild, transport.delete_tree, '/foo')
 
320
 
 
321
    def test_get(self):
 
322
        transport = get_transport('chroot+memory:///foo/bar')
 
323
        self.assertRaises(PathNotChild, transport.get, '/foo')
 
324
 
 
325
    def test_get_bytes(self):
 
326
        transport = get_transport('chroot+memory:///foo/bar')
 
327
        self.assertRaises(PathNotChild, transport.get_bytes, '/foo')
 
328
 
 
329
    def test_has(self):
 
330
        transport = get_transport('chroot+memory:///foo/bar')
 
331
        self.assertRaises(PathNotChild, transport.has, '/foo')
 
332
 
 
333
    def test_list_dir(self):
 
334
        transport = get_transport('chroot+memory:///foo/bar')
 
335
        self.assertRaises(PathNotChild, transport.list_dir, '/foo')
 
336
 
 
337
    def test_lock_read(self):
 
338
        transport = get_transport('chroot+memory:///foo/bar')
 
339
        self.assertRaises(PathNotChild, transport.lock_read, '/foo')
 
340
 
 
341
    def test_lock_write(self):
 
342
        transport = get_transport('chroot+memory:///foo/bar')
 
343
        self.assertRaises(PathNotChild, transport.lock_write, '/foo')
 
344
 
 
345
    def test_mkdir(self):
 
346
        transport = get_transport('chroot+memory:///foo/bar')
 
347
        self.assertRaises(PathNotChild, transport.mkdir, '/foo')
 
348
 
 
349
    def test_put_bytes(self):
 
350
        transport = get_transport('chroot+memory:///foo/bar')
 
351
        self.assertRaises(PathNotChild, transport.put_bytes, '/foo', 'bytes')
 
352
 
 
353
    def test_put_file(self):
 
354
        transport = get_transport('chroot+memory:///foo/bar')
 
355
        self.assertRaises(PathNotChild, transport.put_file, '/foo', None)
 
356
 
 
357
    def test_rename(self):
 
358
        transport = get_transport('chroot+memory:///foo/bar')
 
359
        self.assertRaises(PathNotChild, transport.rename, '/aaa', 'bbb')
 
360
        self.assertRaises(PathNotChild, transport.rename, 'ccc', '/d')
 
361
 
 
362
    def test_rmdir(self):
 
363
        transport = get_transport('chroot+memory:///foo/bar')
 
364
        self.assertRaises(PathNotChild, transport.rmdir, '/foo')
 
365
 
 
366
    def test_stat(self):
 
367
        transport = get_transport('chroot+memory:///foo/bar')
 
368
        self.assertRaises(PathNotChild, transport.stat, '/foo')
 
369
 
 
370
 
191
371
class ReadonlyDecoratorTransportTest(TestCase):
192
372
    """Readonly decoration specific tests."""
193
373
 
200
380
        self.assertEqual(True, transport.is_readonly())
201
381
 
202
382
    def test_http_parameters(self):
 
383
        from bzrlib.tests.HttpServer import HttpServer
203
384
        import bzrlib.transport.readonly as readonly
204
 
        from bzrlib.transport.http import HttpServer
205
385
        # connect to . via http which is not listable
206
386
        server = HttpServer()
207
387
        server.setUp()
235
415
    def test_http_parameters(self):
236
416
        # the listable, should_cache and is_readonly parameters
237
417
        # are not changed by the fakenfs decorator
238
 
        from bzrlib.transport.http import HttpServer
 
418
        from bzrlib.tests.HttpServer import HttpServer
239
419
        # connect to . via http which is not listable
240
420
        server = HttpServer()
241
421
        server.setUp()
255
435
        server = fakenfs.FakeNFSServer()
256
436
        server.setUp()
257
437
        try:
258
 
            # the server should be a relpath localhost server
259
 
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
438
            # the url should be decorated appropriately
 
439
            self.assertStartsWith(server.get_url(), 'fakenfs+')
260
440
            # and we should be able to get a transport for it
261
441
            transport = get_transport(server.get_url())
262
442
            # which must be a FakeNFSTransportDecorator instance.
307
487
class BackupTransportHandler(Transport):
308
488
    """Test transport that works as a backup for the BadTransportHandler"""
309
489
    pass
 
490
 
 
491
 
 
492
class TestTransportImplementation(TestCaseInTempDir):
 
493
    """Implementation verification for transports.
 
494
    
 
495
    To verify a transport we need a server factory, which is a callable
 
496
    that accepts no parameters and returns an implementation of
 
497
    bzrlib.transport.Server.
 
498
    
 
499
    That Server is then used to construct transport instances and test
 
500
    the transport via loopback activity.
 
501
 
 
502
    Currently this assumes that the Transport object is connected to the 
 
503
    current working directory.  So that whatever is done 
 
504
    through the transport, should show up in the working 
 
505
    directory, and vice-versa. This is a bug, because its possible to have
 
506
    URL schemes which provide access to something that may not be 
 
507
    result in storage on the local disk, i.e. due to file system limits, or 
 
508
    due to it being a database or some other non-filesystem tool.
 
509
 
 
510
    This also tests to make sure that the functions work with both
 
511
    generators and lists (assuming iter(list) is effectively a generator)
 
512
    """
 
513
    
 
514
    def setUp(self):
 
515
        super(TestTransportImplementation, self).setUp()
 
516
        self._server = self.transport_server()
 
517
        self._server.setUp()
 
518
 
 
519
    def tearDown(self):
 
520
        super(TestTransportImplementation, self).tearDown()
 
521
        self._server.tearDown()
 
522
        
 
523
    def get_transport(self):
 
524
        """Return a connected transport to the local directory."""
 
525
        base_url = self._server.get_url()
 
526
        # try getting the transport via the regular interface:
 
527
        t = get_transport(base_url)
 
528
        if not isinstance(t, self.transport_class):
 
529
            # we did not get the correct transport class type. Override the
 
530
            # regular connection behaviour by direct construction.
 
531
            t = self.transport_class(base_url)
 
532
        return t
 
533
 
 
534
 
 
535
class TestLocalTransports(TestCase):
 
536
 
 
537
    def test_get_transport_from_abspath(self):
 
538
        here = os.path.abspath('.')
 
539
        t = get_transport(here)
 
540
        self.assertIsInstance(t, LocalTransport)
 
541
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
 
542
 
 
543
    def test_get_transport_from_relpath(self):
 
544
        here = os.path.abspath('.')
 
545
        t = get_transport('.')
 
546
        self.assertIsInstance(t, LocalTransport)
 
547
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
 
548
 
 
549
    def test_get_transport_from_local_url(self):
 
550
        here = os.path.abspath('.')
 
551
        here_url = urlutils.local_path_to_url(here) + '/'
 
552
        t = get_transport(here_url)
 
553
        self.assertIsInstance(t, LocalTransport)
 
554
        self.assertEquals(t.base, here_url)
 
555
 
 
556
 
 
557
class TestWin32LocalTransport(TestCase):
 
558
 
 
559
    def test_unc_clone_to_root(self):
 
560
        # Win32 UNC path like \\HOST\path
 
561
        # clone to root should stop at least at \\HOST part
 
562
        # not on \\
 
563
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
564
        for i in xrange(4):
 
565
            t = t.clone('..')
 
566
        self.assertEquals(t.base, 'file://HOST/')
 
567
        # make sure we reach the root
 
568
        t = t.clone('..')
 
569
        self.assertEquals(t.base, 'file://HOST/')