← 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: Canonical.com Patch Queue Manager
  • Date: 2007-06-12 02:17:42 UTC
  • mfrom: (2521.1.1 56322)
  • Revision ID: pqm@pqm.ubuntu.com-20070612021742-uetsy3g747iq3xkk
merge init --create-prefix

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
 
1
# Copyright (C) 2004, 2005, 2006, 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
21
21
from cStringIO import StringIO
22
22
 
23
23
import bzrlib
24
 
from bzrlib.errors import (NoSuchFile, FileExists,
 
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,
25
34
                           TransportNotPossible,
26
 
                           ConnectionError,
27
 
                           DependencyNotPresent,
28
35
                           UnsupportedProtocol,
29
36
                           )
30
37
from bzrlib.tests import TestCase, TestCaseInTempDir
31
38
from bzrlib.transport import (_CoalescedOffset,
32
39
                              _get_protocol_handlers,
 
40
                              _set_protocol_handlers,
33
41
                              _get_transport_modules,
34
42
                              get_transport,
35
43
                              register_lazy_transport,
36
 
                              _set_protocol_handlers,
 
44
                              register_transport_proto,
 
45
                              _clear_protocol_handlers,
37
46
                              Transport,
38
47
                              )
 
48
from bzrlib.transport.chroot import ChrootServer
39
49
from bzrlib.transport.memory import MemoryTransport
40
 
from bzrlib.transport.local import LocalTransport
 
50
from bzrlib.transport.local import (LocalTransport,
 
51
                                    EmulatedWin32LocalTransport)
 
52
 
 
53
 
 
54
# TODO: Should possibly split transport-specific tests into their own files.
41
55
 
42
56
 
43
57
class TestTransport(TestCase):
45
59
 
46
60
    def test__get_set_protocol_handlers(self):
47
61
        handlers = _get_protocol_handlers()
48
 
        self.assertNotEqual({}, handlers)
 
62
        self.assertNotEqual([], handlers.keys( ))
49
63
        try:
50
 
            _set_protocol_handlers({})
51
 
            self.assertEqual({}, _get_protocol_handlers())
 
64
            _clear_protocol_handlers()
 
65
            self.assertEqual([], _get_protocol_handlers().keys())
52
66
        finally:
53
67
            _set_protocol_handlers(handlers)
54
68
 
57
71
        class SampleHandler(object):
58
72
            """I exist, isnt that enough?"""
59
73
        try:
60
 
            my_handlers = {}
61
 
            _set_protocol_handlers(my_handlers)
 
74
            _clear_protocol_handlers()
 
75
            register_transport_proto('foo')
62
76
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
77
            register_transport_proto('bar')
63
78
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
64
 
            self.assertEqual([SampleHandler.__module__],
 
79
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
65
80
                             _get_transport_modules())
66
81
        finally:
67
82
            _set_protocol_handlers(handlers)
70
85
        """Transport with missing dependency causes no error"""
71
86
        saved_handlers = _get_protocol_handlers()
72
87
        try:
 
88
            register_transport_proto('foo')
73
89
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
74
90
                    'BadTransportHandler')
75
91
            try:
90
106
        """Transport with missing dependency causes no error"""
91
107
        saved_handlers = _get_protocol_handlers()
92
108
        try:
93
 
            _set_protocol_handlers({})
 
109
            _clear_protocol_handlers()
 
110
            register_transport_proto('foo')
94
111
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
95
112
                    'BackupTransportHandler')
96
113
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
100
117
        finally:
101
118
            _set_protocol_handlers(saved_handlers)
102
119
 
 
120
    def test__combine_paths(self):
 
121
        t = Transport('/')
 
122
        self.assertEqual('/home/sarah/project/foo',
 
123
                         t._combine_paths('/home/sarah', 'project/foo'))
 
124
        self.assertEqual('/etc',
 
125
                         t._combine_paths('/home/sarah', '../../etc'))
 
126
        self.assertEqual('/etc',
 
127
                         t._combine_paths('/home/sarah', '../../../etc'))
 
128
        self.assertEqual('/etc',
 
129
                         t._combine_paths('/home/sarah', '/etc'))
 
130
 
 
131
    def test_local_abspath_non_local_transport(self):
 
132
        # the base implementation should throw
 
133
        t = MemoryTransport()
 
134
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
 
135
        self.assertEqual('memory:///t is not a local path.', str(e))
 
136
 
103
137
 
104
138
class TestCoalesceOffsets(TestCase):
105
139
    
179
213
        self.assertEqual("memory:///", transport.base)
180
214
        self.assertEqual("memory:///", transport.abspath('/'))
181
215
 
182
 
    def test_relpath(self):
 
216
    def test_abspath_of_relpath_starting_at_root(self):
183
217
        transport = MemoryTransport()
 
218
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
184
219
 
185
220
    def test_append_and_get(self):
186
221
        transport = MemoryTransport()
219
254
        transport.append_bytes('foo', 'content')
220
255
        self.assertEquals(True, transport.has('foo'))
221
256
 
 
257
    def test_list_dir(self):
 
258
        transport = MemoryTransport()
 
259
        transport.put_bytes('foo', 'content')
 
260
        transport.mkdir('dir')
 
261
        transport.put_bytes('dir/subfoo', 'content')
 
262
        transport.put_bytes('dirlike', 'content')
 
263
 
 
264
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
265
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
266
 
222
267
    def test_mkdir(self):
223
268
        transport = MemoryTransport()
224
269
        transport.mkdir('dir')
257
302
        self.assertEqual(7, transport.stat('foo').st_size)
258
303
        self.assertEqual(6, transport.stat('bar').st_size)
259
304
 
 
305
 
 
306
class ChrootDecoratorTransportTest(TestCase):
 
307
    """Chroot decoration specific tests."""
 
308
 
 
309
    def test_abspath(self):
 
310
        # The abspath is always relative to the chroot_url.
 
311
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
312
        server.setUp()
 
313
        transport = get_transport(server.get_url())
 
314
        self.assertEqual(server.get_url(), transport.abspath('/'))
 
315
 
 
316
        subdir_transport = transport.clone('subdir')
 
317
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
 
318
        server.tearDown()
 
319
 
 
320
    def test_clone(self):
 
321
        server = ChrootServer(get_transport('memory:///foo/bar/'))
 
322
        server.setUp()
 
323
        transport = get_transport(server.get_url())
 
324
        # relpath from root and root path are the same
 
325
        relpath_cloned = transport.clone('foo')
 
326
        abspath_cloned = transport.clone('/foo')
 
327
        self.assertEqual(server, relpath_cloned.server)
 
328
        self.assertEqual(server, abspath_cloned.server)
 
329
        server.tearDown()
 
330
    
 
331
    def test_chroot_url_preserves_chroot(self):
 
332
        """Calling get_transport on a chroot transport's base should produce a
 
333
        transport with exactly the same behaviour as the original chroot
 
334
        transport.
 
335
 
 
336
        This is so that it is not possible to escape a chroot by doing::
 
337
            url = chroot_transport.base
 
338
            parent_url = urlutils.join(url, '..')
 
339
            new_transport = get_transport(parent_url)
 
340
        """
 
341
        server = ChrootServer(get_transport('memory:///path/subpath'))
 
342
        server.setUp()
 
343
        transport = get_transport(server.get_url())
 
344
        new_transport = get_transport(transport.base)
 
345
        self.assertEqual(transport.server, new_transport.server)
 
346
        self.assertEqual(transport.base, new_transport.base)
 
347
        server.tearDown()
260
348
        
 
349
    def test_urljoin_preserves_chroot(self):
 
350
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
 
351
        URL that escapes the intended chroot.
 
352
 
 
353
        This is so that it is not possible to escape a chroot by doing::
 
354
            url = chroot_transport.base
 
355
            parent_url = urlutils.join(url, '..')
 
356
            new_transport = get_transport(parent_url)
 
357
        """
 
358
        server = ChrootServer(get_transport('memory:///path/'))
 
359
        server.setUp()
 
360
        transport = get_transport(server.get_url())
 
361
        self.assertRaises(
 
362
            InvalidURLJoin, urlutils.join, transport.base, '..')
 
363
        server.tearDown()
 
364
 
 
365
 
 
366
class ChrootServerTest(TestCase):
 
367
 
 
368
    def test_construct(self):
 
369
        backing_transport = MemoryTransport()
 
370
        server = ChrootServer(backing_transport)
 
371
        self.assertEqual(backing_transport, server.backing_transport)
 
372
 
 
373
    def test_setUp(self):
 
374
        backing_transport = MemoryTransport()
 
375
        server = ChrootServer(backing_transport)
 
376
        server.setUp()
 
377
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
 
378
 
 
379
    def test_tearDown(self):
 
380
        backing_transport = MemoryTransport()
 
381
        server = ChrootServer(backing_transport)
 
382
        server.setUp()
 
383
        server.tearDown()
 
384
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
 
385
 
 
386
    def test_get_url(self):
 
387
        backing_transport = MemoryTransport()
 
388
        server = ChrootServer(backing_transport)
 
389
        server.setUp()
 
390
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
 
391
        server.tearDown()
 
392
 
 
393
 
261
394
class ReadonlyDecoratorTransportTest(TestCase):
262
395
    """Readonly decoration specific tests."""
263
396
 
270
403
        self.assertEqual(True, transport.is_readonly())
271
404
 
272
405
    def test_http_parameters(self):
 
406
        from bzrlib.tests.HttpServer import HttpServer
273
407
        import bzrlib.transport.readonly as readonly
274
 
        from bzrlib.transport.http import HttpServer
275
408
        # connect to . via http which is not listable
276
409
        server = HttpServer()
277
410
        server.setUp()
305
438
    def test_http_parameters(self):
306
439
        # the listable, should_cache and is_readonly parameters
307
440
        # are not changed by the fakenfs decorator
308
 
        from bzrlib.transport.http import HttpServer
 
441
        from bzrlib.tests.HttpServer import HttpServer
309
442
        # connect to . via http which is not listable
310
443
        server = HttpServer()
311
444
        server.setUp()
325
458
        server = fakenfs.FakeNFSServer()
326
459
        server.setUp()
327
460
        try:
328
 
            # the server should be a relpath localhost server
329
 
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
461
            # the url should be decorated appropriately
 
462
            self.assertStartsWith(server.get_url(), 'fakenfs+')
330
463
            # and we should be able to get a transport for it
331
464
            transport = get_transport(server.get_url())
332
465
            # which must be a FakeNFSTransportDecorator instance.
341
474
        transport = self.get_nfs_transport('.')
342
475
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
343
476
                        transport=transport)
344
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
477
        self.assertRaises(errors.ResourceBusy,
345
478
                          transport.rename, 'from', 'to')
346
479
 
347
480
 
405
538
        super(TestTransportImplementation, self).setUp()
406
539
        self._server = self.transport_server()
407
540
        self._server.setUp()
 
541
        self.addCleanup(self._server.tearDown)
408
542
 
409
 
    def tearDown(self):
410
 
        super(TestTransportImplementation, self).tearDown()
411
 
        self._server.tearDown()
412
 
        
413
543
    def get_transport(self):
414
544
        """Return a connected transport to the local directory."""
415
545
        base_url = self._server.get_url()
416
546
        # try getting the transport via the regular interface:
417
547
        t = get_transport(base_url)
418
 
        if not isinstance(t, self.transport_class): 
 
548
        if not isinstance(t, self.transport_class):
419
549
            # we did not get the correct transport class type. Override the
420
550
            # regular connection behaviour by direct construction.
421
551
            t = self.transport_class(base_url)
422
552
        return t
 
553
 
 
554
 
 
555
class TestLocalTransports(TestCase):
 
556
 
 
557
    def test_get_transport_from_abspath(self):
 
558
        here = os.path.abspath('.')
 
559
        t = get_transport(here)
 
560
        self.assertIsInstance(t, LocalTransport)
 
561
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
 
562
 
 
563
    def test_get_transport_from_relpath(self):
 
564
        here = os.path.abspath('.')
 
565
        t = get_transport('.')
 
566
        self.assertIsInstance(t, LocalTransport)
 
567
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
 
568
 
 
569
    def test_get_transport_from_local_url(self):
 
570
        here = os.path.abspath('.')
 
571
        here_url = urlutils.local_path_to_url(here) + '/'
 
572
        t = get_transport(here_url)
 
573
        self.assertIsInstance(t, LocalTransport)
 
574
        self.assertEquals(t.base, here_url)
 
575
 
 
576
    def test_local_abspath(self):
 
577
        here = os.path.abspath('.')
 
578
        t = get_transport(here)
 
579
        self.assertEquals(t.local_abspath(''), here)
 
580
 
 
581
 
 
582
class TestWin32LocalTransport(TestCase):
 
583
 
 
584
    def test_unc_clone_to_root(self):
 
585
        # Win32 UNC path like \\HOST\path
 
586
        # clone to root should stop at least at \\HOST part
 
587
        # not on \\
 
588
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
589
        for i in xrange(4):
 
590
            t = t.clone('..')
 
591
        self.assertEquals(t.base, 'file://HOST/')
 
592
        # make sure we reach the root
 
593
        t = t.clone('..')
 
594
        self.assertEquals(t.base, 'file://HOST/')
 
595
 
 
596
 
 
597
def get_test_permutations():
 
598
    """Return transport permutations to be used in testing.
 
599
 
 
600
    This module registers some transports, but they're only for testing
 
601
    registration.  We don't really want to run all the transport tests against
 
602
    them.
 
603
    """
 
604
    return []