← 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: Aaron Bentley
  • Date: 2007-06-12 18:10:06 UTC
  • mfrom: (2525 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2527.
  • Revision ID: abentley@panoramicfeedback.com-20070612181006-kinkxenc2dmgb4lp
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 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 import urlutils
25
 
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,
26
34
                           TransportNotPossible,
27
 
                           ConnectionError,
28
 
                           DependencyNotPresent,
29
35
                           UnsupportedProtocol,
30
 
                           PathNotChild,
31
36
                           )
32
37
from bzrlib.tests import TestCase, TestCaseInTempDir
33
38
from bzrlib.transport import (_CoalescedOffset,
34
39
                              _get_protocol_handlers,
 
40
                              _set_protocol_handlers,
35
41
                              _get_transport_modules,
36
42
                              get_transport,
37
43
                              register_lazy_transport,
38
 
                              _set_protocol_handlers,
 
44
                              register_transport_proto,
 
45
                              _clear_protocol_handlers,
39
46
                              Transport,
40
47
                              )
 
48
from bzrlib.transport.chroot import ChrootServer
41
49
from bzrlib.transport.memory import MemoryTransport
42
 
from bzrlib.transport.local import LocalTransport
 
50
from bzrlib.transport.local import (LocalTransport,
 
51
                                    EmulatedWin32LocalTransport)
43
52
 
44
53
 
45
54
# TODO: Should possibly split transport-specific tests into their own files.
50
59
 
51
60
    def test__get_set_protocol_handlers(self):
52
61
        handlers = _get_protocol_handlers()
53
 
        self.assertNotEqual({}, handlers)
 
62
        self.assertNotEqual([], handlers.keys( ))
54
63
        try:
55
 
            _set_protocol_handlers({})
56
 
            self.assertEqual({}, _get_protocol_handlers())
 
64
            _clear_protocol_handlers()
 
65
            self.assertEqual([], _get_protocol_handlers().keys())
57
66
        finally:
58
67
            _set_protocol_handlers(handlers)
59
68
 
62
71
        class SampleHandler(object):
63
72
            """I exist, isnt that enough?"""
64
73
        try:
65
 
            my_handlers = {}
66
 
            _set_protocol_handlers(my_handlers)
 
74
            _clear_protocol_handlers()
 
75
            register_transport_proto('foo')
67
76
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
77
            register_transport_proto('bar')
68
78
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
69
 
            self.assertEqual([SampleHandler.__module__],
 
79
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
70
80
                             _get_transport_modules())
71
81
        finally:
72
82
            _set_protocol_handlers(handlers)
75
85
        """Transport with missing dependency causes no error"""
76
86
        saved_handlers = _get_protocol_handlers()
77
87
        try:
 
88
            register_transport_proto('foo')
78
89
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
79
90
                    'BadTransportHandler')
80
91
            try:
95
106
        """Transport with missing dependency causes no error"""
96
107
        saved_handlers = _get_protocol_handlers()
97
108
        try:
98
 
            _set_protocol_handlers({})
 
109
            _clear_protocol_handlers()
 
110
            register_transport_proto('foo')
99
111
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
100
112
                    'BackupTransportHandler')
101
113
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
116
128
        self.assertEqual('/etc',
117
129
                         t._combine_paths('/home/sarah', '/etc'))
118
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
 
119
137
 
120
138
class TestCoalesceOffsets(TestCase):
121
139
    
288
306
class ChrootDecoratorTransportTest(TestCase):
289
307
    """Chroot decoration specific tests."""
290
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()
 
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
 
291
368
    def test_construct(self):
292
 
        from bzrlib.transport import chroot
293
 
        transport = chroot.ChrootTransportDecorator('chroot+memory:///pathA/')
294
 
        self.assertEqual('memory:///pathA/', transport.chroot_url)
295
 
 
296
 
        transport = chroot.ChrootTransportDecorator(
297
 
            'chroot+memory:///path/B', chroot='memory:///path/')
298
 
        self.assertEqual('memory:///path/', transport.chroot_url)
299
 
 
300
 
    def test_append_file(self):
301
 
        transport = get_transport('chroot+memory:///foo/bar')
302
 
        self.assertRaises(PathNotChild, transport.append_file, '/foo', None)
303
 
 
304
 
    def test_append_bytes(self):
305
 
        transport = get_transport('chroot+memory:///foo/bar')
306
 
        self.assertRaises(PathNotChild, transport.append_bytes, '/foo', 'bytes')
307
 
 
308
 
    def test_clone(self):
309
 
        transport = get_transport('chroot+memory:///foo/bar')
310
 
        self.assertRaises(PathNotChild, transport.clone, '/foo')
311
 
 
312
 
    def test_delete(self):
313
 
        transport = get_transport('chroot+memory:///foo/bar')
314
 
        self.assertRaises(PathNotChild, transport.delete, '/foo')
315
 
 
316
 
    def test_delete_tree(self):
317
 
        transport = get_transport('chroot+memory:///foo/bar')
318
 
        self.assertRaises(PathNotChild, transport.delete_tree, '/foo')
319
 
 
320
 
    def test_get(self):
321
 
        transport = get_transport('chroot+memory:///foo/bar')
322
 
        self.assertRaises(PathNotChild, transport.get, '/foo')
323
 
 
324
 
    def test_get_bytes(self):
325
 
        transport = get_transport('chroot+memory:///foo/bar')
326
 
        self.assertRaises(PathNotChild, transport.get_bytes, '/foo')
327
 
 
328
 
    def test_has(self):
329
 
        transport = get_transport('chroot+memory:///foo/bar')
330
 
        self.assertRaises(PathNotChild, transport.has, '/foo')
331
 
 
332
 
    def test_list_dir(self):
333
 
        transport = get_transport('chroot+memory:///foo/bar')
334
 
        self.assertRaises(PathNotChild, transport.list_dir, '/foo')
335
 
 
336
 
    def test_lock_read(self):
337
 
        transport = get_transport('chroot+memory:///foo/bar')
338
 
        self.assertRaises(PathNotChild, transport.lock_read, '/foo')
339
 
 
340
 
    def test_lock_write(self):
341
 
        transport = get_transport('chroot+memory:///foo/bar')
342
 
        self.assertRaises(PathNotChild, transport.lock_write, '/foo')
343
 
 
344
 
    def test_mkdir(self):
345
 
        transport = get_transport('chroot+memory:///foo/bar')
346
 
        self.assertRaises(PathNotChild, transport.mkdir, '/foo')
347
 
 
348
 
    def test_put_bytes(self):
349
 
        transport = get_transport('chroot+memory:///foo/bar')
350
 
        self.assertRaises(PathNotChild, transport.put_bytes, '/foo', 'bytes')
351
 
 
352
 
    def test_put_file(self):
353
 
        transport = get_transport('chroot+memory:///foo/bar')
354
 
        self.assertRaises(PathNotChild, transport.put_file, '/foo', None)
355
 
 
356
 
    def test_rename(self):
357
 
        transport = get_transport('chroot+memory:///foo/bar')
358
 
        self.assertRaises(PathNotChild, transport.rename, '/aaa', 'bbb')
359
 
        self.assertRaises(PathNotChild, transport.rename, 'ccc', '/d')
360
 
 
361
 
    def test_rmdir(self):
362
 
        transport = get_transport('chroot+memory:///foo/bar')
363
 
        self.assertRaises(PathNotChild, transport.rmdir, '/foo')
364
 
 
365
 
    def test_stat(self):
366
 
        transport = get_transport('chroot+memory:///foo/bar')
367
 
        self.assertRaises(PathNotChild, transport.stat, '/foo')
 
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()
368
392
 
369
393
 
370
394
class ReadonlyDecoratorTransportTest(TestCase):
450
474
        transport = self.get_nfs_transport('.')
451
475
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
452
476
                        transport=transport)
453
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
477
        self.assertRaises(errors.ResourceBusy,
454
478
                          transport.rename, 'from', 'to')
455
479
 
456
480
 
514
538
        super(TestTransportImplementation, self).setUp()
515
539
        self._server = self.transport_server()
516
540
        self._server.setUp()
 
541
        self.addCleanup(self._server.tearDown)
517
542
 
518
 
    def tearDown(self):
519
 
        super(TestTransportImplementation, self).tearDown()
520
 
        self._server.tearDown()
521
 
        
522
543
    def get_transport(self):
523
544
        """Return a connected transport to the local directory."""
524
545
        base_url = self._server.get_url()
551
572
        t = get_transport(here_url)
552
573
        self.assertIsInstance(t, LocalTransport)
553
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 []