← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/selftest/testtransport.py

  • Committer: abentley
  • Date: 2005-10-14 03:50:50 UTC
  • mto: (1185.25.1)
  • mto: This revision was merged to the branch mainline in revision 1460.
  • Revision ID: abentley@lappy-20051014035050-d779472ccb599a51
semi-broke merge

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/selftest/testtransport.py
17
17
 
18
18
import os
19
19
from cStringIO import StringIO
20
 
 
21
 
from bzrlib.errors import (NoSuchFile, FileExists, TransportNotPossible,
22
 
                           ConnectionError)
23
 
from bzrlib.tests import TestCase, TestCaseInTempDir
24
 
from bzrlib.tests.HTTPTestUtil import TestCaseWithWebserver
25
 
from bzrlib.transport import memory, urlescape
26
 
 
 
20
from bzrlib.selftest import TestCaseInTempDir
 
21
from bzrlib.selftest.HTTPTestUtil import TestCaseWithWebserver
 
22
from bzrlib.errors import NoSuchFile, FileExists, TransportNotPossible
27
23
 
28
24
def _append(fn, txt):
29
25
    """Append the given text (file-like object) to the supplied filename."""
33
29
    f.close()
34
30
    del f
35
31
 
36
 
class TestTransport(TestCase):
37
 
    """Test the non transport-concrete class functionality."""
38
 
 
39
 
    def test_urlescape(self):
40
 
        self.assertEqual('%25', urlescape('%'))
41
 
 
42
 
 
43
32
class TestTransportMixIn(object):
44
33
    """Subclass this, and it will provide a series of tests for a Transport.
45
34
    It assumes that the Transport object is connected to the 
59
48
    def test_has(self):
60
49
        t = self.get_transport()
61
50
 
62
 
        files = ['a', 'b', 'e', 'g', '%']
 
51
        files = ['a', 'b', 'e', 'g']
63
52
        self.build_tree(files)
64
53
        self.assertEqual(t.has('a'), True)
65
54
        self.assertEqual(t.has('c'), False)
66
 
        self.assertEqual(t.has(urlescape('%')), True)
67
55
        self.assertEqual(list(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
68
56
                [True, True, False, False, True, False, True, False])
69
 
        self.assertEqual(t.has_any(['a', 'b', 'c']), True)
70
 
        self.assertEqual(t.has_any(['c', 'd', 'f', urlescape('%%')]), False)
71
57
        self.assertEqual(list(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']))),
72
58
                [True, True, False, False, True, False, True, False])
73
 
        self.assertEqual(t.has_any(['c', 'c', 'c']), False)
74
 
        self.assertEqual(t.has_any(['b', 'b', 'b']), True)
75
59
 
76
60
    def test_get(self):
77
61
        t = self.get_transport()
296
280
            self.assertEquals(open(f).read(),
297
281
                    open(os.path.join(dtmp_base, f)).read())
298
282
 
299
 
        # Test that copying into a missing directory raises
300
 
        # NoSuchFile
301
 
        os.mkdir('e')
302
 
        open('e/f', 'wb').write('contents of e')
303
 
        self.assertRaises(NoSuchFile, t.copy_to, ['e/f'], local_t)
304
 
 
305
 
        os.mkdir(os.path.join(dtmp_base, 'e'))
306
 
        t.copy_to(['e/f'], local_t)
307
 
 
308
283
        del dtmp, dtmp_base, local_t
309
284
 
310
285
        dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
456
431
                'some text for the\nthird file created\n'
457
432
                'some garbage\nto put in three\n')
458
433
 
 
434
    def test_get_partial(self):
 
435
        t = self.get_transport()
 
436
 
 
437
        contents = [
 
438
            ('f1', 
 
439
                'here is some text\nand a bit more\n'
 
440
                'adding more\ntext to two\n'),
 
441
            ('f2',
 
442
                'this is a string\nand some more stuff\n'
 
443
                'appending to\none\n'),
 
444
            ('f3',
 
445
                'some text for the\nthird file created\n'
 
446
                'some garbage\nto put in three\n')
 
447
        ]
 
448
        if self.readonly:
 
449
            for f, val in contents:
 
450
                open(f, 'wb').write(val)
 
451
        else:
 
452
            t.put_multi(contents)
 
453
 
 
454
        self.assertRaises(NoSuchFile,
 
455
                t.get_partial, 'a-missing-file', 20)
 
456
        self.assertRaises(NoSuchFile,
 
457
                t.get_partial, 'another-missing-file', 20, 30)
 
458
        f = t.get_partial('f1', 33)
 
459
        self.assertEqual(f.read(), 
 
460
                'adding more\ntext to two\n')
 
461
        f = t.get_partial('f1', 33, 10)
 
462
        self.assertEqual(f.read(10), 
 
463
                'adding mor')
 
464
 
 
465
        del f
 
466
 
 
467
        offsets = [('f2', 37), ('f3', 20, 10), ('f1', 10, 20)]
 
468
        values = ['appending to\none\n',
 
469
                  'ird file c',
 
470
                  'me text\nand a bit mo'
 
471
                 ]
 
472
        contents_f = t.get_partial_multi(offsets)
 
473
        count = 0
 
474
        for f, val in zip(contents_f, values):
 
475
            count += 1
 
476
            self.assertEqual(val, f.read(len(val)))
 
477
        # Make sure we saw all values, and no extra
 
478
        self.assertEqual(len(values), count)
 
479
        self.assertEqual(list(contents_f), [])
 
480
 
 
481
        # Do the same thing with an iterator
 
482
        offsets = iter([('f2', 34), ('f3', 18, 10), ('f1', 15, 15)])
 
483
        values = ['ff\nappending to\none\n',
 
484
                  'third file',
 
485
                  'xt\nand a bit mo'
 
486
                 ]
 
487
        contents_f = t.get_partial_multi(offsets)
 
488
        count = 0
 
489
        for f, val in zip(contents_f, values):
 
490
            count += 1
 
491
            self.assertEqual(val, f.read(len(val)))
 
492
        self.assertEqual(len(values), count)
 
493
        self.assertEqual(list(contents_f), [])
 
494
 
 
495
 
459
496
    def test_delete(self):
460
497
        # TODO: Test Transport.delete
461
498
        pass
464
501
        # TODO: Test Transport.move
465
502
        pass
466
503
 
467
 
    def test_connection_error(self):
468
 
        """ConnectionError is raised when connection is impossible"""
469
 
        if not hasattr(self, "get_bogus_transport"):
470
 
            return
471
 
        t = self.get_bogus_transport()
472
 
        try:
473
 
            t.get('.bzr/branch')
474
 
        except (ConnectionError, NoSuchFile), e:
475
 
            pass
476
 
        except (Exception), e:
477
 
            self.failIf(True, 'Wrong exception thrown: %s' % e)
478
 
        else:
479
 
            self.failIf(True, 'Did not get the expected exception.')
480
 
 
481
 
        
482
504
class LocalTransportTest(TestCaseInTempDir, TestTransportMixIn):
483
505
    def get_transport(self):
484
506
        from bzrlib.transport.local import LocalTransport
485
507
        return LocalTransport('.')
486
508
 
487
 
 
488
509
class HttpTransportTest(TestCaseWithWebserver, TestTransportMixIn):
489
 
 
490
510
    readonly = True
491
 
 
492
511
    def get_transport(self):
493
512
        from bzrlib.transport.http import HttpTransport
494
513
        url = self.get_remote_url('.')
495
514
        return HttpTransport(url)
496
515
 
497
 
    def get_bogus_transport(self):
498
 
        from bzrlib.transport.http import HttpTransport
499
 
        return HttpTransport('http://jasldkjsalkdjalksjdkljasd')
500
 
 
501
 
 
502
 
class TestMemoryTransport(TestCase):
503
 
 
504
 
    def test_get_transport(self):
505
 
        memory.MemoryTransport()
506
 
 
507
 
    def test_clone(self):
508
 
        transport = memory.MemoryTransport()
509
 
        self.failUnless(transport.clone() is transport)
510
 
 
511
 
    def test_abspath(self):
512
 
        transport = memory.MemoryTransport()
513
 
        self.assertEqual("in-memory:relpath", transport.abspath('relpath'))
514
 
 
515
 
    def test_relpath(self):
516
 
        transport = memory.MemoryTransport()
517
 
 
518
 
    def test_append_and_get(self):
519
 
        transport = memory.MemoryTransport()
520
 
        transport.append('path', StringIO('content'))
521
 
        self.assertEqual(transport.get('path').read(), 'content')
522
 
        transport.append('path', StringIO('content'))
523
 
        self.assertEqual(transport.get('path').read(), 'contentcontent')
524
 
 
525
 
    def test_put_and_get(self):
526
 
        transport = memory.MemoryTransport()
527
 
        transport.put('path', StringIO('content'))
528
 
        self.assertEqual(transport.get('path').read(), 'content')
529
 
        transport.put('path', StringIO('content'))
530
 
        self.assertEqual(transport.get('path').read(), 'content')
531
 
 
532
 
    def test_append_without_dir_fails(self):
533
 
        transport = memory.MemoryTransport()
534
 
        self.assertRaises(NoSuchFile,
535
 
                          transport.append, 'dir/path', StringIO('content'))
536
 
 
537
 
    def test_put_without_dir_fails(self):
538
 
        transport = memory.MemoryTransport()
539
 
        self.assertRaises(NoSuchFile,
540
 
                          transport.put, 'dir/path', StringIO('content'))
541
 
 
542
 
    def test_get_missing(self):
543
 
        transport = memory.MemoryTransport()
544
 
        self.assertRaises(NoSuchFile, transport.get, 'foo')
545
 
 
546
 
    def test_has_missing(self):
547
 
        transport = memory.MemoryTransport()
548
 
        self.assertEquals(False, transport.has('foo'))
549
 
 
550
 
    def test_has_present(self):
551
 
        transport = memory.MemoryTransport()
552
 
        transport.append('foo', StringIO('content'))
553
 
        self.assertEquals(True, transport.has('foo'))
554
 
 
555
 
    def test_mkdir(self):
556
 
        transport = memory.MemoryTransport()
557
 
        transport.mkdir('dir')
558
 
        transport.append('dir/path', StringIO('content'))
559
 
        self.assertEqual(transport.get('dir/path').read(), 'content')
560
 
 
561
 
    def test_mkdir_missing_parent(self):
562
 
        transport = memory.MemoryTransport()
563
 
        self.assertRaises(NoSuchFile,
564
 
                          transport.mkdir, 'dir/dir')
565
 
 
566
 
    def test_mkdir_twice(self):
567
 
        transport = memory.MemoryTransport()
568
 
        transport.mkdir('dir')
569
 
        self.assertRaises(FileExists, transport.mkdir, 'dir')
570
 
 
571
 
    def test_parameters(self):
572
 
        transport = memory.MemoryTransport()
573
 
        self.assertEqual(True, transport.listable())
574
 
        self.assertEqual(False, transport.should_cache())
575
 
 
576
 
    def test_iter_files_recursive(self):
577
 
        transport = memory.MemoryTransport()
578
 
        transport.mkdir('dir')
579
 
        transport.put('dir/foo', StringIO('content'))
580
 
        transport.put('dir/bar', StringIO('content'))
581
 
        transport.put('bar', StringIO('content'))
582
 
        paths = set(transport.iter_files_recursive())
583
 
        self.assertEqual(set(['dir/foo', 'dir/bar', 'bar']), paths)
584
 
 
585
 
    def test_stat(self):
586
 
        transport = memory.MemoryTransport()
587
 
        transport.put('foo', StringIO('content'))
588
 
        transport.put('bar', StringIO('phowar'))
589
 
        self.assertEqual(7, transport.stat('foo').st_size)
590
 
        self.assertEqual(6, transport.stat('bar').st_size)
591