← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_transport_implementations.py

Merge in bzrdir work to enable checkout improvements.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_transport_implementations.py
26
26
import sys
27
27
 
28
28
from bzrlib.errors import (NoSuchFile, FileExists,
 
29
                           LockError,
29
30
                           TransportNotPossible, ConnectionError)
30
31
from bzrlib.tests import TestCaseInTempDir, TestSkipped
31
32
from bzrlib.transport import memory, urlescape
35
36
def _append(fn, txt):
36
37
    """Append the given text (file-like object) to the supplied filename."""
37
38
    f = open(fn, 'ab')
38
 
    f.write(txt.read())
39
 
    f.flush()
40
 
    f.close()
41
 
    del f
 
39
    try:
 
40
        f.write(txt.read())
 
41
    finally:
 
42
        f.close()
42
43
 
43
44
 
44
45
class TestTransportImplementation(TestCaseInTempDir):
74
75
        
75
76
    def check_transport_contents(self, content, transport, relpath):
76
77
        """Check that transport.get(relpath).read() == content."""
77
 
        self.assertEqual(content, transport.get(relpath).read())
 
78
        self.assertEqualDiff(content, transport.get(relpath).read())
78
79
 
79
80
    def get_transport(self):
80
81
        """Return a connected transport to the local directory."""
86
87
        return t
87
88
 
88
89
    def assertListRaises(self, excClass, func, *args, **kwargs):
89
 
        """Many transport functions can return generators this makes sure
 
90
        """Fail unless excClass is raised when the iterator from func is used.
 
91
        
 
92
        Many transport functions can return generators this makes sure
90
93
        to wrap them in a list() call to make sure the whole generator
91
94
        is run, and that the proper exception is raised.
92
95
        """
178
181
        if t.is_readonly():
179
182
            return
180
183
        t.put('mode644', StringIO('test text\n'), mode=0644)
181
 
        self.assertMode(t, 'mode644', 0644)
 
184
        self.assertTransportMode(t, 'mode644', 0644)
182
185
        t.put('mode666', StringIO('test text\n'), mode=0666)
183
 
        self.assertMode(t, 'mode666', 0666)
 
186
        self.assertTransportMode(t, 'mode666', 0666)
184
187
        t.put('mode600', StringIO('test text\n'), mode=0600)
185
 
        self.assertMode(t, 'mode600', 0600)
 
188
        self.assertTransportMode(t, 'mode600', 0600)
186
189
        # Yes, you can put a file such that it becomes readonly
187
190
        t.put('mode400', StringIO('test text\n'), mode=0400)
188
 
        self.assertMode(t, 'mode400', 0400)
 
191
        self.assertTransportMode(t, 'mode400', 0400)
189
192
        t.put_multi([('mmode644', StringIO('text\n'))], mode=0644)
190
 
        self.assertMode(t, 'mmode644', 0644)
 
193
        self.assertTransportMode(t, 'mmode644', 0644)
191
194
        
192
195
    def test_mkdir(self):
193
196
        t = self.get_transport()
243
246
            return
244
247
        # Test mkdir with a mode
245
248
        t.mkdir('dmode755', mode=0755)
246
 
        self.assertMode(t, 'dmode755', 0755)
 
249
        self.assertTransportMode(t, 'dmode755', 0755)
247
250
        t.mkdir('dmode555', mode=0555)
248
 
        self.assertMode(t, 'dmode555', 0555)
 
251
        self.assertTransportMode(t, 'dmode555', 0555)
249
252
        t.mkdir('dmode777', mode=0777)
250
 
        self.assertMode(t, 'dmode777', 0777)
 
253
        self.assertTransportMode(t, 'dmode777', 0777)
251
254
        t.mkdir('dmode700', mode=0700)
252
 
        self.assertMode(t, 'dmode700', 0700)
 
255
        self.assertTransportMode(t, 'dmode700', 0700)
253
256
        # TODO: jam 20051215 test mkdir_multi with a mode
254
257
        t.mkdir_multi(['mdmode755'], mode=0755)
255
 
        self.assertMode(t, 'mdmode755', 0755)
 
258
        self.assertTransportMode(t, 'mdmode755', 0755)
256
259
 
257
260
    def test_copy_to(self):
 
261
        # FIXME: test:   same server to same server (partly done)
 
262
        # same protocol two servers
 
263
        # and    different protocols (done for now except for MemoryTransport.
 
264
        # - RBC 20060122
258
265
        from bzrlib.transport.memory import MemoryTransport
 
266
 
 
267
        def simple_copy_files(transport_from, transport_to):
 
268
            files = ['a', 'b', 'c', 'd']
 
269
            self.build_tree(files, transport=transport_from)
 
270
            transport_from.copy_to(files, transport_to)
 
271
            for f in files:
 
272
                self.check_transport_contents(transport_to.get(f).read(),
 
273
                                              transport_from, f)
 
274
 
259
275
        t = self.get_transport()
260
 
 
261
 
        files = ['a', 'b', 'c', 'd']
262
 
        self.build_tree(files, transport=t)
263
 
 
264
276
        temp_transport = MemoryTransport('memory:/')
 
277
        simple_copy_files(t, temp_transport)
 
278
        if not t.is_readonly():
 
279
            t.mkdir('copy_to_simple')
 
280
            t2 = t.clone('copy_to_simple')
 
281
            simple_copy_files(t, t2)
265
282
 
266
 
        t.copy_to(files, temp_transport)
267
 
        for f in files:
268
 
            self.check_transport_contents(temp_transport.get(f).read(),
269
 
                                          t, f)
270
283
 
271
284
        # Test that copying into a missing directory raises
272
285
        # NoSuchFile
273
286
        if t.is_readonly():
274
 
            os.mkdir('e')
275
 
            open('e/f', 'wb').write('contents of e')
 
287
            self.build_tree(['e/', 'e/f'])
276
288
        else:
277
289
            t.mkdir('e')
278
290
            t.put('e/f', StringIO('contents of e'))
294
306
            temp_transport = MemoryTransport("memory:/")
295
307
            t.copy_to(files, temp_transport, mode=mode)
296
308
            for f in files:
297
 
                self.assertMode(temp_transport, f, mode)
 
309
                self.assertTransportMode(temp_transport, f, mode)
298
310
 
299
311
    def test_append(self):
300
312
        t = self.get_transport()
483
495
        self.check_transport_contents(t.get('f2').read(), t, 'c')
484
496
        self.check_transport_contents(t.get('f3').read(), t, 'd')
485
497
 
486
 
 
487
498
    def test_delete(self):
488
499
        # TODO: Test Transport.delete
489
500
        t = self.get_transport()
490
501
 
491
502
        # Not much to do with a readonly transport
492
503
        if t.is_readonly():
 
504
            self.assertRaises(TransportNotPossible, t.delete, 'missing')
493
505
            return
494
506
 
495
507
        t.put('a', StringIO('a little bit of text\n'))
527
539
        # plain "listdir".
528
540
        # self.assertEqual([], os.listdir('.'))
529
541
 
 
542
    def test_rmdir(self):
 
543
        t = self.get_transport()
 
544
        # Not much to do with a readonly transport
 
545
        if t.is_readonly():
 
546
            self.assertRaises(TransportNotPossible, t.rmdir, 'missing')
 
547
            return
 
548
        t.mkdir('adir')
 
549
        t.mkdir('adir/bdir')
 
550
        t.rmdir('adir/bdir')
 
551
        self.assertRaises(NoSuchFile, t.stat, 'adir/bdir')
 
552
        t.rmdir('adir')
 
553
        self.assertRaises(NoSuchFile, t.stat, 'adir')
 
554
 
 
555
    def test_delete_tree(self):
 
556
        t = self.get_transport()
 
557
 
 
558
        # Not much to do with a readonly transport
 
559
        if t.is_readonly():
 
560
            self.assertRaises(TransportNotPossible, t.delete_tree, 'missing')
 
561
            return
 
562
 
 
563
        # and does it like listing ?
 
564
        t.mkdir('adir')
 
565
        try:
 
566
            t.delete_tree('adir')
 
567
        except TransportNotPossible:
 
568
            # ok, this transport does not support delete_tree
 
569
            return
 
570
        
 
571
        # did it delete that trivial case?
 
572
        self.assertRaises(NoSuchFile, t.stat, 'adir')
 
573
 
 
574
        self.build_tree(['adir/',
 
575
                         'adir/file', 
 
576
                         'adir/subdir/', 
 
577
                         'adir/subdir/file', 
 
578
                         'adir/subdir2/',
 
579
                         'adir/subdir2/file',
 
580
                         ], transport=t)
 
581
 
 
582
        t.delete_tree('adir')
 
583
        # adir should be gone now.
 
584
        self.assertRaises(NoSuchFile, t.stat, 'adir')
 
585
 
530
586
    def test_move(self):
531
587
        t = self.get_transport()
532
588
 
629
685
 
630
686
        self.assertListRaises(NoSuchFile, t.stat_multi, ['a', 'c', 'd'])
631
687
        self.assertListRaises(NoSuchFile, t.stat_multi, iter(['a', 'c', 'd']))
 
688
        self.build_tree(['subdir/', 'subdir/file'], transport=t)
 
689
        subdir = t.clone('subdir')
 
690
        subdir.stat('./file')
 
691
        subdir.stat('.')
632
692
 
633
693
    def test_list_dir(self):
634
694
        # TODO: Test list_dir, just try once, and if it throws, stop testing
645
705
 
646
706
        # SftpServer creates control files in the working directory
647
707
        # so lets move down a directory to avoid those.
648
 
        t.mkdir('wd')
 
708
        if not t.is_readonly():
 
709
            t.mkdir('wd')
 
710
        else:
 
711
            os.mkdir('wd')
649
712
        t = t.clone('wd')
650
713
 
651
714
        self.assertEqual([], sorted_list(u'.'))
652
 
        self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e'], transport=t)
 
715
        # c2 is precisely one letter longer than c here to test that
 
716
        # suffixing is not confused.
 
717
        if not t.is_readonly():
 
718
            self.build_tree(['a', 'b', 'c/', 'c/d', 'c/e', 'c2/'], transport=t)
 
719
        else:
 
720
            self.build_tree(['wd/a', 'wd/b', 'wd/c/', 'wd/c/d', 'wd/c/e', 'wd/c2/'])
653
721
 
654
 
        self.assertEqual([u'a', u'b', u'c'], sorted_list(u'.'))
 
722
        self.assertEqual([u'a', u'b', u'c', u'c2'], sorted_list(u'.'))
655
723
        self.assertEqual([u'd', u'e'], sorted_list(u'c'))
656
724
 
657
 
        t.delete('c/d')
658
 
        t.delete('b')
659
 
        self.assertEqual([u'a', u'c'], sorted_list('.'))
 
725
        if not t.is_readonly():
 
726
            t.delete('c/d')
 
727
            t.delete('b')
 
728
        else:
 
729
            os.unlink('wd/c/d')
 
730
            os.unlink('wd/b')
 
731
            
 
732
        self.assertEqual([u'a', u'c', u'c2'], sorted_list('.'))
660
733
        self.assertEqual([u'e'], sorted_list(u'c'))
661
734
 
662
735
        self.assertListRaises(NoSuchFile, t.list_dir, 'q')
729
802
        transport = transport.clone('isolated')
730
803
        paths = set(transport.iter_files_recursive())
731
804
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
 
805
 
 
806
    def test_connect_twice_is_same_content(self):
 
807
        # check that our server (whatever it is) is accessable reliably
 
808
        # via get_transport and multiple connections share content.
 
809
        transport = self.get_transport()
 
810
        if transport.is_readonly():
 
811
            return
 
812
        transport.put('foo', StringIO('bar'))
 
813
        transport2 = self.get_transport()
 
814
        self.check_transport_contents('bar', transport2, 'foo')
 
815
        # its base should be usable.
 
816
        transport2 = bzrlib.transport.get_transport(transport.base)
 
817
        self.check_transport_contents('bar', transport2, 'foo')
 
818
 
 
819
        # now opening at a relative url should give use a sane result:
 
820
        transport.mkdir('newdir')
 
821
        transport2 = bzrlib.transport.get_transport(transport.base + "newdir")
 
822
        transport2 = transport2.clone('..')
 
823
        self.check_transport_contents('bar', transport2, 'foo')
 
824
 
 
825
    def test_lock_write(self):
 
826
        transport = self.get_transport()
 
827
        if transport.is_readonly():
 
828
            self.assertRaises(TransportNotPossible, transport.lock_write, 'foo')
 
829
            return
 
830
        transport.put('lock', StringIO())
 
831
        lock = transport.lock_write('lock')
 
832
        # TODO make this consistent on all platforms:
 
833
        # self.assertRaises(LockError, transport.lock_write, 'lock')
 
834
        lock.unlock()
 
835
 
 
836
    def test_lock_read(self):
 
837
        transport = self.get_transport()
 
838
        if transport.is_readonly():
 
839
            file('lock', 'w').close()
 
840
        else:
 
841
            transport.put('lock', StringIO())
 
842
        lock = transport.lock_read('lock')
 
843
        # TODO make this consistent on all platforms:
 
844
        # self.assertRaises(LockError, transport.lock_read, 'lock')
 
845
        lock.unlock()