← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport.py

  • Committer: John Arbash Meinel
  • Date: 2005-07-18 17:13:34 UTC
  • mto: (1185.11.1)
  • mto: This revision was merged to the branch mainline in revision 1396.
  • Revision ID: john@arbash-meinel.com-20050718171333-52cdbb6c84ba1044
Created mkdir_multi, modified branch to use _multi forms when initializing a branch, creating a full test routine for transports

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport.py
5
5
"""
6
6
 
7
7
from bzrlib.trace import mutter
 
8
from bzrlib.errors import BzrError
8
9
 
9
10
_protocol_handlers = {
10
11
}
20
21
        mutter('registering transport: %s => %s' % (prefix, klass.__name__))
21
22
        _protocol_handlers[prefix] = klass
22
23
 
23
 
class AsyncError(Exception):
24
 
    pass
25
 
 
26
 
class TransportNotPossibleError(Exception):
 
24
class TransportError(BzrError):
 
25
    pass
 
26
 
 
27
class AsyncError(TransportError):
 
28
    pass
 
29
 
 
30
class TransportNotPossibleError(TransportError):
27
31
    """This is for transports where a specific function is explicitly not
28
32
    possible. Such as pushing files to an HTTP server.
29
33
    """
243
247
        """Create a directory at the given path."""
244
248
        raise NotImplementedError
245
249
 
 
250
    def mkdir_multi(self, relpaths, pb=None):
 
251
        """Create a group of directories"""
 
252
        return self._iterate_over(relpaths, self.mkdir, pb, 'mkdir', expand=False)
 
253
 
246
254
    def append(self, relpath, f, encode=False):
247
255
        """Append the text in the file-like or string object to 
248
256
        the supplied location.
378
386
    # which has a lookup of None
379
387
    return _protocol_handlers[None](base)
380
388
 
 
389
def transport_test(tester, t):
 
390
    """Test a transport object. Basically, it assumes that the
 
391
    Transport object is connected to the current working directory.
 
392
    So that whatever is done through the transport, should show
 
393
    up in the working directory, and vice-versa.
 
394
 
 
395
    This also tests to make sure that the functions work with both
 
396
    generators and lists (assuming iter(list) is effectively a generator)
 
397
    """
 
398
    import tempfile, os
 
399
    from local_transport import LocalTransport
 
400
 
 
401
    # Test has
 
402
    files = ['a', 'b', 'e', 'g']
 
403
    tester.build_tree(files)
 
404
    tester.assertEqual(t.has('a'), True)
 
405
    tester.assertEqual(t.has('c'), False)
 
406
    tester.assertEqual(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']),
 
407
            [True, True, False, False, True, False, True, False])
 
408
    tester.assertEqual(t.has_multi(iter(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h'])),
 
409
            [True, True, False, False, True, False, True, False])
 
410
 
 
411
    # Test get
 
412
    tester.assertEqual(t.get('a').read(), open('a').read())
 
413
    content_f = t.get_multi(files)
 
414
    for path,f in zip(files, content_f):
 
415
        tester.assertEqual(open(path).read(), f.read())
 
416
 
 
417
    content_f = t.get_multi(iter(files))
 
418
    for path,f in zip(files, content_f):
 
419
        tester.assertEqual(open(path).read(), f.read())
 
420
 
 
421
    tester.assertRaises(TransportError, t.get, 'c')
 
422
    try:
 
423
        files = list(t.get_multi(['a', 'b', 'c']))
 
424
    except TransportError:
 
425
        pass
 
426
    else:
 
427
        tester.fail('Failed to raise TransportError for missing file in get_multi')
 
428
    try:
 
429
        files = list(t.get_multi(iter(['a', 'b', 'c', 'e'])))
 
430
    except TransportError:
 
431
        pass
 
432
    else:
 
433
        tester.fail('Failed to raise TransportError for missing file in get_multi')
 
434
 
 
435
    # Test put
 
436
    t.put('c', 'some text for c\n')
 
437
    tester.assert_(os.path.exists('c'))
 
438
    tester.assertEqual(open('c').read(), 'some text for c\n')
 
439
    # Make sure 'has' is updated
 
440
    tester.assertEqual(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']),
 
441
            [True, True, True, False, True, False, True, False])
 
442
    # Put also replaces contents
 
443
    tester.assertEqual(t.put_multi([('a', 'new\ncontents for\na\n'),
 
444
                                  ('d', 'contents\nfor d\n')]),
 
445
                     2)
 
446
    tester.assertEqual(t.has_multi(['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']),
 
447
            [True, True, True, True, True, False, True, False])
 
448
    tester.assertEqual(open('a').read(), 'new\ncontents for\na\n')
 
449
    tester.assertEqual(open('d').read(), 'contents\nfor d\n')
 
450
 
 
451
    tester.assertEqual(t.put_multi(iter([('a', 'diff\ncontents for\na\n'),
 
452
                                  ('d', 'another contents\nfor d\n')])),
 
453
                     2)
 
454
    tester.assertEqual(open('a').read(), 'diff\ncontents for\na\n')
 
455
    tester.assertEqual(open('d').read(), 'another contents\nfor d\n')
 
456
 
 
457
    tester.assertRaises(TransportError, t.put, 'path/doesnt/exist/c')
 
458
 
 
459
    # Test mkdir
 
460
    os.mkdir('dir_a')
 
461
    tester.assertEqual(t.has('dir_a'), True)
 
462
    tester.assertEqual(t.has('dir_b'), False)
 
463
 
 
464
    t.mkdir('dir_b')
 
465
    tester.assertEqual(t.has('dir_b'), True)
 
466
    tester.assert_(os.path.isdir('dir_b'))
 
467
 
 
468
    t.mkdir_multi(['dir_c', 'dir_d'])
 
469
    tester.assertEqual(t.has_multi(['dir_a', 'dir_b', 'dir_c', 'dir_d', 'dir_e', 'dir_b']),
 
470
            [True, True, True, True, False, True])
 
471
    for d in ['dir_a', 'dir_b', 'dir_c', 'dir_d']:
 
472
        tester.assert_(os.path.isdir(d))
 
473
 
 
474
    tester.assertRaises(TransportError, t.mkdir, 'path/doesnt/exist')
 
475
    tester.assertRaises(TransportError, t.mkdir, 'dir_a') # Creating a directory again should fail
 
476
 
 
477
    # This one may fail for some transports.
 
478
    # Specifically, I know RsyncTransport doesn't check for the directory
 
479
    # existing, before it creates it. The reason is that it seems to
 
480
    # expensive, it does check to see if the local directory already exists,
 
481
    # and will throw an exception for that
 
482
    # FIXME: Make this work everywhere
 
483
    #os.mkdir('dir_e')
 
484
    #tester.assertRaises(TransportError, t.mkdir, 'dir_e')
 
485
 
 
486
    # Test get/put in sub-directories
 
487
    tester.assertEqual(t.put_multi([('dir_a/a', 'contents of dir_a/a'),
 
488
                                    ('dir_b/b', 'contents of dir_b/b')]
 
489
                      , 2))
 
490
    for f in ('dir_a/a', 'dir_b/b'):
 
491
        tester.assertEqual(t.get(f).read(), open(f).read())
 
492
 
 
493
    # Test copy_to
 
494
    dtmp = tempfile.mkdtemp(dir='.', prefix='test-transport-')
 
495
    dtmp_base = os.path.basename(dtmp)
 
496
    local_t = LocalTransport(dtmp)
 
497
 
 
498
    files = ['a', 'b', 'c', 'd']
 
499
    t.copy_to(files, local_t)
 
500
    for f in files:
 
501
        tester.assertEquals(open(f).read(), open(os.path.join(dtmp_base, f)).read())
 
502
 
381
503
# Local transport should always be initialized
382
504
import local_transport