← 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-07-04 08:08:13 UTC
  • mfrom: (2572 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2587.
  • Revision ID: robertc@robertcollins.net-20070704080813-wzebx0r88fvwj5rq
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
 
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
35
                           ConnectionError,
27
36
                           DependencyNotPresent,
31
40
from bzrlib.tests import TestCase, TestCaseInTempDir
32
41
from bzrlib.transport import (_CoalescedOffset,
33
42
                              _get_protocol_handlers,
 
43
                              _set_protocol_handlers,
34
44
                              _get_transport_modules,
35
45
                              get_transport,
36
46
                              LateReadError,
37
47
                              register_lazy_transport,
38
 
                              _set_protocol_handlers,
 
48
                              register_transport_proto,
 
49
                              _clear_protocol_handlers,
39
50
                              Transport,
40
51
                              )
 
52
from bzrlib.transport.chroot import ChrootServer
41
53
from bzrlib.transport.memory import MemoryTransport
42
 
from bzrlib.transport.local import LocalTransport
 
54
from bzrlib.transport.local import (LocalTransport,
 
55
                                    EmulatedWin32LocalTransport)
 
56
 
 
57
 
 
58
# TODO: Should possibly split transport-specific tests into their own files.
43
59
 
44
60
 
45
61
class TestTransport(TestCase):
47
63
 
48
64
    def test__get_set_protocol_handlers(self):
49
65
        handlers = _get_protocol_handlers()
50
 
        self.assertNotEqual({}, handlers)
 
66
        self.assertNotEqual([], handlers.keys( ))
51
67
        try:
52
 
            _set_protocol_handlers({})
53
 
            self.assertEqual({}, _get_protocol_handlers())
 
68
            _clear_protocol_handlers()
 
69
            self.assertEqual([], _get_protocol_handlers().keys())
54
70
        finally:
55
71
            _set_protocol_handlers(handlers)
56
72
 
59
75
        class SampleHandler(object):
60
76
            """I exist, isnt that enough?"""
61
77
        try:
62
 
            my_handlers = {}
63
 
            _set_protocol_handlers(my_handlers)
 
78
            _clear_protocol_handlers()
 
79
            register_transport_proto('foo')
64
80
            register_lazy_transport('foo', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
 
81
            register_transport_proto('bar')
65
82
            register_lazy_transport('bar', 'bzrlib.tests.test_transport', 'TestTransport.SampleHandler')
66
 
            self.assertEqual([SampleHandler.__module__],
 
83
            self.assertEqual([SampleHandler.__module__, 'bzrlib.transport.chroot'],
67
84
                             _get_transport_modules())
68
85
        finally:
69
86
            _set_protocol_handlers(handlers)
72
89
        """Transport with missing dependency causes no error"""
73
90
        saved_handlers = _get_protocol_handlers()
74
91
        try:
 
92
            register_transport_proto('foo')
75
93
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
76
94
                    'BadTransportHandler')
77
95
            try:
92
110
        """Transport with missing dependency causes no error"""
93
111
        saved_handlers = _get_protocol_handlers()
94
112
        try:
95
 
            _set_protocol_handlers({})
 
113
            _clear_protocol_handlers()
 
114
            register_transport_proto('foo')
96
115
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
97
116
                    'BackupTransportHandler')
98
117
            register_lazy_transport('foo', 'bzrlib.tests.test_transport',
112
131
        self.assertRaises(ReadError, a_file.read, 40)
113
132
        a_file.close()
114
133
 
 
134
    def test__combine_paths(self):
 
135
        t = Transport('/')
 
136
        self.assertEqual('/home/sarah/project/foo',
 
137
                         t._combine_paths('/home/sarah', 'project/foo'))
 
138
        self.assertEqual('/etc',
 
139
                         t._combine_paths('/home/sarah', '../../etc'))
 
140
        self.assertEqual('/etc',
 
141
                         t._combine_paths('/home/sarah', '../../../etc'))
 
142
        self.assertEqual('/etc',
 
143
                         t._combine_paths('/home/sarah', '/etc'))
 
144
 
 
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
 
115
151
 
116
152
class TestCoalesceOffsets(TestCase):
117
153
    
191
227
        self.assertEqual("memory:///", transport.base)
192
228
        self.assertEqual("memory:///", transport.abspath('/'))
193
229
 
194
 
    def test_relpath(self):
 
230
    def test_abspath_of_relpath_starting_at_root(self):
195
231
        transport = MemoryTransport()
 
232
        self.assertEqual("memory:///foo", transport.abspath('/foo'))
196
233
 
197
234
    def test_append_and_get(self):
198
235
        transport = MemoryTransport()
231
268
        transport.append_bytes('foo', 'content')
232
269
        self.assertEquals(True, transport.has('foo'))
233
270
 
 
271
    def test_list_dir(self):
 
272
        transport = MemoryTransport()
 
273
        transport.put_bytes('foo', 'content')
 
274
        transport.mkdir('dir')
 
275
        transport.put_bytes('dir/subfoo', 'content')
 
276
        transport.put_bytes('dirlike', 'content')
 
277
 
 
278
        self.assertEquals(['dir', 'dirlike', 'foo'], sorted(transport.list_dir('.')))
 
279
        self.assertEquals(['subfoo'], sorted(transport.list_dir('dir')))
 
280
 
234
281
    def test_mkdir(self):
235
282
        transport = MemoryTransport()
236
283
        transport.mkdir('dir')
269
316
        self.assertEqual(7, transport.stat('foo').st_size)
270
317
        self.assertEqual(6, transport.stat('bar').st_size)
271
318
 
 
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()
272
362
        
 
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
 
273
408
class ReadonlyDecoratorTransportTest(TestCase):
274
409
    """Readonly decoration specific tests."""
275
410
 
282
417
        self.assertEqual(True, transport.is_readonly())
283
418
 
284
419
    def test_http_parameters(self):
 
420
        from bzrlib.tests.HttpServer import HttpServer
285
421
        import bzrlib.transport.readonly as readonly
286
 
        from bzrlib.transport.http import HttpServer
287
422
        # connect to . via http which is not listable
288
423
        server = HttpServer()
289
424
        server.setUp()
317
452
    def test_http_parameters(self):
318
453
        # the listable, should_cache and is_readonly parameters
319
454
        # are not changed by the fakenfs decorator
320
 
        from bzrlib.transport.http import HttpServer
 
455
        from bzrlib.tests.HttpServer import HttpServer
321
456
        # connect to . via http which is not listable
322
457
        server = HttpServer()
323
458
        server.setUp()
337
472
        server = fakenfs.FakeNFSServer()
338
473
        server.setUp()
339
474
        try:
340
 
            # the server should be a relpath localhost server
341
 
            self.assertEqual(server.get_url(), 'fakenfs+.')
 
475
            # the url should be decorated appropriately
 
476
            self.assertStartsWith(server.get_url(), 'fakenfs+')
342
477
            # and we should be able to get a transport for it
343
478
            transport = get_transport(server.get_url())
344
479
            # which must be a FakeNFSTransportDecorator instance.
353
488
        transport = self.get_nfs_transport('.')
354
489
        self.build_tree(['from/', 'from/foo', 'to/', 'to/bar'],
355
490
                        transport=transport)
356
 
        self.assertRaises(bzrlib.errors.ResourceBusy,
 
491
        self.assertRaises(errors.ResourceBusy,
357
492
                          transport.rename, 'from', 'to')
358
493
 
359
494
 
417
552
        super(TestTransportImplementation, self).setUp()
418
553
        self._server = self.transport_server()
419
554
        self._server.setUp()
 
555
        self.addCleanup(self._server.tearDown)
420
556
 
421
 
    def tearDown(self):
422
 
        super(TestTransportImplementation, self).tearDown()
423
 
        self._server.tearDown()
424
 
        
425
557
    def get_transport(self):
426
558
        """Return a connected transport to the local directory."""
427
559
        base_url = self._server.get_url()
428
560
        # try getting the transport via the regular interface:
429
561
        t = get_transport(base_url)
430
 
        if not isinstance(t, self.transport_class): 
 
562
        if not isinstance(t, self.transport_class):
431
563
            # we did not get the correct transport class type. Override the
432
564
            # regular connection behaviour by direct construction.
433
565
            t = self.transport_class(base_url)
434
566
        return t
 
567
 
 
568
 
 
569
class TestLocalTransports(TestCase):
 
570
 
 
571
    def test_get_transport_from_abspath(self):
 
572
        here = os.path.abspath('.')
 
573
        t = get_transport(here)
 
574
        self.assertIsInstance(t, LocalTransport)
 
575
        self.assertEquals(t.base, urlutils.local_path_to_url(here) + '/')
 
576
 
 
577
    def test_get_transport_from_relpath(self):
 
578
        here = os.path.abspath('.')
 
579
        t = get_transport('.')
 
580
        self.assertIsInstance(t, LocalTransport)
 
581
        self.assertEquals(t.base, urlutils.local_path_to_url('.') + '/')
 
582
 
 
583
    def test_get_transport_from_local_url(self):
 
584
        here = os.path.abspath('.')
 
585
        here_url = urlutils.local_path_to_url(here) + '/'
 
586
        t = get_transport(here_url)
 
587
        self.assertIsInstance(t, LocalTransport)
 
588
        self.assertEquals(t.base, here_url)
 
589
 
 
590
    def test_local_abspath(self):
 
591
        here = os.path.abspath('.')
 
592
        t = get_transport(here)
 
593
        self.assertEquals(t.local_abspath(''), here)
 
594
 
 
595
 
 
596
class TestWin32LocalTransport(TestCase):
 
597
 
 
598
    def test_unc_clone_to_root(self):
 
599
        # Win32 UNC path like \\HOST\path
 
600
        # clone to root should stop at least at \\HOST part
 
601
        # not on \\
 
602
        t = EmulatedWin32LocalTransport('file://HOST/path/to/some/dir/')
 
603
        for i in xrange(4):
 
604
            t = t.clone('..')
 
605
        self.assertEquals(t.base, 'file://HOST/')
 
606
        # make sure we reach the root
 
607
        t = t.clone('..')
 
608
        self.assertEquals(t.base, 'file://HOST/')
 
609
 
 
610
 
 
611
def get_test_permutations():
 
612
    """Return transport permutations to be used in testing.
 
613
 
 
614
    This module registers some transports, but they're only for testing
 
615
    registration.  We don't really want to run all the transport tests against
 
616
    them.
 
617
    """
 
618
    return []