← 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: John Arbash Meinel
  • Date: 2006-11-10 15:38:16 UTC
  • mto: This revision was merged to the branch mainline in revision 2129.
  • Revision ID: john@arbash-meinel.com-20061110153816-46acf76fc86a512b
use try/finally to clean up a nested progress bar during weave fetching

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport.py
1
 
# Copyright (C) 2004, 2005, 2006, 2007 Canonical Ltd
 
1
# Copyright (C) 2004, 2005, 2006 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 (
25
 
    errors,
26
 
    urlutils,
27
 
    )
28
 
from bzrlib.errors import (ConnectionError,
29
 
                           DependencyNotPresent,
30
 
                           FileExists,
31
 
                           InvalidURLJoin,
32
 
                           NoSuchFile,
33
 
                           PathNotChild,
 
24
from bzrlib import urlutils
 
25
from bzrlib.errors import (NoSuchFile, FileExists,
34
26
                           TransportNotPossible,
35
27
                           ConnectionError,
36
28
                           DependencyNotPresent,
37
 
                           ReadError,
38
29
                           UnsupportedProtocol,
39
30
                           )
40
31
from bzrlib.tests import TestCase, TestCaseInTempDir
41
32
from bzrlib.transport import (_CoalescedOffset,
42
33
                              _get_protocol_handlers,
43
 
                              _set_protocol_handlers,
44
34
                              _get_transport_modules,
45
35
                              get_transport,
46
 
                              LateReadError,
47
36
                              register_lazy_transport,
48
 
                              register_transport_proto,
49
 
                              _clear_protocol_handlers,
 
37
                              _set_protocol_handlers,
50
38
                              Transport,
51
39
                              )
52
 
from bzrlib.transport.chroot import ChrootServer
53
40
from bzrlib.transport.memory import MemoryTransport
54
 
from bzrlib.transport.local import (LocalTransport,
55
 
                                    EmulatedWin32LocalTransport)
 
41
from bzrlib.transport.local import LocalTransport
56
42
 
57
43
 
58
44
# TODO: Should possibly split transport-specific tests into their own files.
63
49
 
64
50
    def test__get_set_protocol_handlers(self):
65
51
        handlers = _get_protocol_handlers()
66
 
        self.assertNotEqual([], handlers.keys( ))
 
52
        self.assertNotEqual({}, handlers)
67
53
        try:
68
 
            _clear_protocol_handlers()
69
 
            self.assertEqual([], _get_protocol_handlers().keys())
 
54
            _set_protocol_handlers({})
 
55
            self.assertEqual({}, _get_protocol_handlers())
70
56
        finally:
71
57
            _set_protocol_handlers(handlers)
72
58
 
75
61
        class SampleHandler(object):
76
62
            """I exist, isnt that enough?"""
77
63
        try:
78
 
            _clear_protocol_handlers()
79
 
            register_transport_proto('foo')
 
64
            my_handlers = {}
 
65
            _set_protocol_handlers(my_handlers)
80
66
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
81
 
            register_transport_proto('bar')
82
67
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
83
 
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
 
68
            self.assertEqual([SampleHandler.__module__],
84
69
                             _get_transport_modules())
85
70
        finally:
86
71
            _set_protocol_handlers(handlers)
89
74
        """Transport with missing dependency causes no error"""
90
75
        saved_handlers = _get_protocol_handlers()
91
76
        try:
92
 
            register_transport_proto('foo')
93
77
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
94
78
                    'BadTransportHandler')
95
79
            try:
110
94
        """Transport with missing dependency causes no error"""
111
95
        saved_handlers = _get_protocol_handlers()
112
96
        try:
113
 
            _clear_protocol_handlers()
114
 
            register_transport_proto('foo')
 
97
            _set_protocol_handlers({})
115
98
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
116
99
                    'BackupTransportHandler')
117
100
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
121
104
        finally:
122
105
            _set_protocol_handlers(saved_handlers)
123
106
 
124
 
    def test_LateReadError(self):
125
 
        """The LateReadError helper should raise on read()."""
126
 
        a_file = LateReadError('a path')
127
 
        try:
128
 
            a_file.read()
129
 
        except ReadError, error:
130
 
            self.assertEqual('a path', error.path)
131
 
        self.assertRaises(ReadError, a_file.read, 40)
132
 
        a_file.close()
133
 
 
134
107
    def test__combine_paths(self):
135
108
        t = Transport('/')
136
109
        self.assertEqual('/home/sarah/project/foo',
142
115
        self.assertEqual('/etc',
143
116
                         t._combine_paths('/home/sarah', '/etc'))
144
117
 
145
 
    def test_local_abspath_non_local_transport(self):
146
 
        # the base implementation should throw
147
 
        t = MemoryTransport()
148
 
        e = self.assertRaises(errors.NotLocalUrl, t.local_abspath, 't')
149
 
        self.assertEqual('memory:///t is not a local path.', str(e))
150
 
 
151
118
 
152
119
class TestCoalesceOffsets(TestCase):
153
120
    
316
283
        self.assertEqual(7, transport.stat('foo').st_size)
317
284
        self.assertEqual(6, transport.stat('bar').st_size)
318
285
 
319
 
 
320
 
class ChrootDecoratorTransportTest(TestCase):
321
 
    """Chroot decoration specific tests."""
322
 
 
323
 
    def test_abspath(self):
324
 
        # The abspath is always relative to the chroot_url.
325
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
326
 
        server.setUp()
327
 
        transport = get_transport(server.get_url())
328
 
        self.assertEqual(server.get_url(), transport.abspath('/'))
329
 
 
330
 
        subdir_transport = transport.clone('subdir')
331
 
        self.assertEqual(server.get_url(), subdir_transport.abspath('/'))
332
 
        server.tearDown()
333
 
 
334
 
    def test_clone(self):
335
 
        server = ChrootServer(get_transport('memory:///foo/bar/'))
336
 
        server.setUp()
337
 
        transport = get_transport(server.get_url())
338
 
        # relpath from root and root path are the same
339
 
        relpath_cloned = transport.clone('foo')
340
 
        abspath_cloned = transport.clone('/foo')
341
 
        self.assertEqual(server, relpath_cloned.server)
342
 
        self.assertEqual(server, abspath_cloned.server)
343
 
        server.tearDown()
344
 
    
345
 
    def test_chroot_url_preserves_chroot(self):
346
 
        """Calling get_transport on a chroot transport's base should produce a
347
 
        transport with exactly the same behaviour as the original chroot
348
 
        transport.
349
 
 
350
 
        This is so that it is not possible to escape a chroot by doing::
351
 
            url = chroot_transport.base
352
 
            parent_url = urlutils.join(url, '..')
353
 
            new_transport = get_transport(parent_url)
354
 
        """
355
 
        server = ChrootServer(get_transport('memory:///path/subpath'))
356
 
        server.setUp()
357
 
        transport = get_transport(server.get_url())
358
 
        new_transport = get_transport(transport.base)
359
 
        self.assertEqual(transport.server, new_transport.server)
360
 
        self.assertEqual(transport.base, new_transport.base)
361
 
        server.tearDown()
362
286
        
363
 
    def test_urljoin_preserves_chroot(self):
364
 
        """Using urlutils.join(url, '..') on a chroot URL should not produce a
365
 
        URL that escapes the intended chroot.
366
 
 
367
 
        This is so that it is not possible to escape a chroot by doing::
368
 
            url = chroot_transport.base
369
 
            parent_url = urlutils.join(url, '..')
370
 
            new_transport = get_transport(parent_url)
371
 
        """
372
 
        server = ChrootServer(get_transport('memory:///path/'))
373
 
        server.setUp()
374
 
        transport = get_transport(server.get_url())
375
 
        self.assertRaises(
376
 
            InvalidURLJoin, urlutils.join, transport.base, '..')
377
 
        server.tearDown()
378
 
 
379
 
 
380
 
class ChrootServerTest(TestCase):
381
 
 
382
 
    def test_construct(self):
383
 
        backing_transport = MemoryTransport()
384
 
        server = ChrootServer(backing_transport)
385
 
        self.assertEqual(backing_transport, server.backing_transport)
386
 
 
387
 
    def test_setUp(self):
388
 
        backing_transport = MemoryTransport()
389
 
        server = ChrootServer(backing_transport)
390
 
        server.setUp()
391
 
        self.assertTrue(server.scheme in _get_protocol_handlers().keys())
392
 
 
393
 
    def test_tearDown(self):
394
 
        backing_transport = MemoryTransport()
395
 
        server = ChrootServer(backing_transport)
396
 
        server.setUp()
397
 
        server.tearDown()
398
 
        self.assertFalse(server.scheme in _get_protocol_handlers().keys())
399
 
 
400
 
    def test_get_url(self):
401
 
        backing_transport = MemoryTransport()
402
 
        server = ChrootServer(backing_transport)
403
 
        server.setUp()
404
 
        self.assertEqual('chroot-%d:///' % id(server), server.get_url())
405
 
        server.tearDown()
406
 
 
407
 
 
408
287
class ReadonlyDecoratorTransportTest(TestCase):
409
288
    """Readonly decoration specific tests."""
410
289
 
417
296
        self.assertEqual(True, transport.is_readonly())
418
297
 
419
298
    def test_http_parameters(self):
420
 
        from bzrlib.tests.HttpServer import HttpServer
421
299
        import bzrlib.transport.readonly as readonly
 
300
        from bzrlib.transport.http import HttpServer
422
301
        # connect to . via http which is not listable
423
302
        server = HttpServer()
424
303
        server.setUp()
452
331
    def test_http_parameters(self):
453
332
        # the listable, should_cache and is_readonly parameters
454
333
        # are not changed by the fakenfs decorator
455
 
        from bzrlib.tests.HttpServer import HttpServer
 
334
        from bzrlib.transport.http import HttpServer
456
335
        # connect to . via http which is not listable
457
336
        server = HttpServer()
458
337
        server.setUp()
488
367
        transport = self.get_nfs_transport('.')
489
368
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
490
369
                        transport=transport)
491
 
        self.assertRaises(errors.ResourceBusy,
 
370
        self.assertRaises(bzrlib.errors.ResourceBusy,
492
371
                          transport.rename, 'from', 'to')
493
372
 
494
373
 
552
431
        super(TestTransportImplementation, self).setUp()
553
432
        self._server = self.transport_server()
554
433
        self._server.setUp()
555
 
        self.addCleanup(self._server.tearDown)
556
 
 
557
 
    def get_transport(self, relpath=None):
558
 
        """Return a connected transport to the local directory.
559
 
 
560
 
        :param relpath: a path relative to the base url.
561
 
        """
 
434
 
 
435
    def tearDown(self):
 
436
        super(TestTransportImplementation, self).tearDown()
 
437
        self._server.tearDown()
 
438
        
 
439
    def get_transport(self):
 
440
        """Return a connected transport to the local directory."""
562
441
        base_url = self._server.get_url()
563
 
        url = self._adjust_url(base_url, relpath)
564
442
        # try getting the transport via the regular interface:
565
 
        t = get_transport(url)
 
443
        t = get_transport(base_url)
566
444
        if not isinstance(t, self.transport_class):
567
445
            # we did not get the correct transport class type. Override the
568
446
            # regular connection behaviour by direct construction.
569
 
            t = self.transport_class(url)
 
447
            t = self.transport_class(base_url)
570
448
        return t
571
449
 
572
450
 
590
468
        t = get_transport(here_url)
591
469
        self.assertIsInstance(t, LocalTransport)
592
470
        self.assertEquals(t.base, here_url)
593
 
 
594
 
    def test_local_abspath(self):
595
 
        here = os.path.abspath('.')
596
 
        t = get_transport(here)
597
 
        self.assertEquals(t.local_abspath(''), here)
598
 
 
599
 
 
600
 
class TestWin32LocalTransport(TestCase):
601
 
 
602
 
    def test_unc_clone_to_root(self):
603
 
        # Win32 UNC path like \\HOST\path
604
 
        # clone to root should stop at least at \\HOST part
605
 
        # not on \\
606
 
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
607
 
        for i in xrange(4):
608
 
            t = t.clone('..')
609
 
        self.assertEquals(t.base, 'file://HOST/')
610
 
        # make sure we reach the root
611
 
        t = t.clone('..')
612
 
        self.assertEquals(t.base, 'file://HOST/')
613
 
 
614
 
 
615
 
def get_test_permutations():
616
 
    """Return transport permutations to be used in testing.
617
 
 
618
 
    This module registers some transports, but they're only for testing
619
 
    registration.  We don't really want to run all the transport tests against
620
 
    them.
621
 
    """
622
 
    return []