/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_sftp_transport.py

1st cut merge of bzr.dev r3907

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_sftp_transport.py
29
29
 
30
30
from bzrlib import (
31
31
    bzrdir,
 
32
    config,
32
33
    errors,
 
34
    tests,
 
35
    transport as _mod_transport,
33
36
    )
34
37
from bzrlib.osutils import (
35
38
    pathjoin,
46
49
import bzrlib.transport.http
47
50
 
48
51
if paramiko_loaded:
 
52
    from bzrlib.transport import sftp as _mod_sftp
49
53
    from bzrlib.transport.sftp import (
50
54
        SFTPAbsoluteServer,
51
55
        SFTPHomeDirServer,
76
80
        set_test_transport_to_sftp(self)
77
81
 
78
82
 
79
 
class SFTPLockTests (TestCaseWithSFTPServer):
 
83
class SFTPLockTests(TestCaseWithSFTPServer):
80
84
 
81
85
    def test_sftp_locks(self):
82
86
        from bzrlib.errors import LockError
459
463
        self.assertAlmostEqual(t2 - t1, 100 + 7)
460
464
 
461
465
 
 
466
class ReadvFile(object):
 
467
    """An object that acts like Paramiko's SFTPFile.readv()"""
 
468
 
 
469
    def __init__(self, data):
 
470
        self._data = data
 
471
 
 
472
    def readv(self, requests):
 
473
        for start, length in requests:
 
474
            yield self._data[start:start+length]
 
475
 
 
476
 
 
477
class Test_SFTPReadvHelper(tests.TestCase):
 
478
 
 
479
    def checkGetRequests(self, expected_requests, offsets):
 
480
        if not paramiko_loaded:
 
481
            raise TestSkipped('you must have paramiko to run this test')
 
482
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test')
 
483
        self.assertEqual(expected_requests, helper._get_requests())
 
484
 
 
485
    def test__get_requests(self):
 
486
        # Small single requests become a single readv request
 
487
        self.checkGetRequests([(0, 100)],
 
488
                              [(0, 20), (30, 50), (20, 10), (80, 20)])
 
489
        # Non-contiguous ranges are given as multiple requests
 
490
        self.checkGetRequests([(0, 20), (30, 50)],
 
491
                              [(10, 10), (30, 20), (0, 10), (50, 30)])
 
492
        # Ranges larger than _max_request_size (32kB) are broken up into
 
493
        # multiple requests, even if it actually spans multiple logical
 
494
        # requests
 
495
        self.checkGetRequests([(0, 32768), (32768, 32768), (65536, 464)],
 
496
                              [(0, 40000), (40000, 100), (40100, 1900),
 
497
                               (42000, 24000)])
 
498
 
 
499
    def checkRequestAndYield(self, expected, data, offsets):
 
500
        if not paramiko_loaded:
 
501
            raise TestSkipped('you must have paramiko to run this test')
 
502
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test')
 
503
        data_f = ReadvFile(data)
 
504
        result = list(helper.request_and_yield_offsets(data_f))
 
505
        self.assertEqual(expected, result)
 
506
 
 
507
    def test_request_and_yield_offsets(self):
 
508
        data = 'abcdefghijklmnopqrstuvwxyz'
 
509
        self.checkRequestAndYield([(0, 'a'), (5, 'f'), (10, 'klm')], data,
 
510
                                  [(0, 1), (5, 1), (10, 3)])
 
511
        # Should combine requests, and split them again
 
512
        self.checkRequestAndYield([(0, 'a'), (1, 'b'), (10, 'klm')], data,
 
513
                                  [(0, 1), (1, 1), (10, 3)])
 
514
        # Out of order requests. The requests should get combined, but then be
 
515
        # yielded out-of-order. We also need one that is at the end of a
 
516
        # previous range. See bug #293746
 
517
        self.checkRequestAndYield([(0, 'a'), (10, 'k'), (4, 'efg'), (1, 'bcd')],
 
518
                                  data, [(0, 1), (10, 1), (4, 3), (1, 3)])
 
519
 
 
520
 
 
521
class TestUsesAuthConfig(TestCaseWithSFTPServer):
 
522
    """Test that AuthenticationConfig can supply default usernames."""
 
523
 
 
524
    def get_transport_for_connection(self, set_config):
 
525
        port = self.get_server()._listener.port
 
526
        if set_config:
 
527
            conf = config.AuthenticationConfig()
 
528
            conf._get_config().update(
 
529
                {'sftptest': {'scheme': 'ssh', 'port': port, 'user': 'bar'}})
 
530
            conf._save()
 
531
        t = get_transport('sftp://localhost:%d' % port)
 
532
        # force a connection to be performed.
 
533
        t.has('foo')
 
534
        return t
 
535
 
 
536
    def test_sftp_uses_config(self):
 
537
        t = self.get_transport_for_connection(set_config=True)
 
538
        self.assertEqual('bar', t._get_credentials()[0])
 
539
 
 
540
    def test_sftp_is_none_if_no_config(self):
 
541
        t = self.get_transport_for_connection(set_config=False)
 
542
        self.assertIs(None, t._get_credentials()[0])