/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 breezy/tests/test_sftp_transport.py

  • Committer: Jelmer Vernooij
  • Date: 2018-11-16 23:19:12 UTC
  • mfrom: (7180 work)
  • mto: This revision was merged to the branch mainline in revision 7294.
  • Revision ID: jelmer@jelmer.uk-20181116231912-e043vpq22bdkxa6q
Merge trunk.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/test_sftp_transport.py
38
38
    TestSkipped,
39
39
    )
40
40
from breezy.tests.http_server import HttpServer
41
 
import breezy.transport.http
42
41
 
43
42
if features.paramiko.available():
44
43
    from breezy.transport import sftp as _mod_sftp
127
126
        self.assertIsSameRealPath(root_parent + '/sibling',
128
127
                                  t._remote_path('../sibling'))
129
128
        # /  should be illegal ?
130
 
        ### FIXME decide and then test for all transports. RBC20051208
 
129
        # FIXME decide and then test for all transports. RBC20051208
131
130
 
132
131
 
133
132
class SFTPTransportTestRelativeRoot(TestCaseWithSFTPServer):
203
202
 
204
203
        self.assertEqual(b2.last_revision(), b'a1')
205
204
 
206
 
        with open('a/foo', 'wt') as f: f.write('something new in foo\n')
 
205
        with open('a/foo', 'wt') as f:
 
206
            f.write('something new in foo\n')
207
207
        t.commit('new', rev_id=b'a2')
208
208
        b2.pull(b)
209
209
 
219
219
      'loopback': Doesn't use ssh, just uses a local socket. Most tests are
220
220
                  done this way to save the handshaking time, so it is not
221
221
                  tested again here
222
 
      'none':     This uses paramiko's built-in ssh client and server, and layers
223
 
                  sftp on top of it.
 
222
      'none':     This uses paramiko's built-in ssh client and server, and
 
223
                  layers sftp on top of it.
224
224
      None:       If 'ssh' exists on the machine, then it will be spawned as a
225
225
                  child process.
226
226
    """
283
283
 
284
284
    def set_vendor(self, vendor, subprocess_stderr=None):
285
285
        from breezy.transport import ssh
286
 
        self.overrideAttr(ssh._ssh_vendor_manager, '_cached_ssh_vendor', vendor)
 
286
        self.overrideAttr(ssh._ssh_vendor_manager,
 
287
                          '_cached_ssh_vendor', vendor)
287
288
        if subprocess_stderr is not None:
288
289
            self.overrideAttr(ssh.SubprocessVendor, "_stderr_target",
289
 
                subprocess_stderr)
 
290
                              subprocess_stderr)
290
291
 
291
292
    def test_bad_connection_paramiko(self):
292
293
        """Test that a real connection attempt raises the right error"""
318
319
        start_time = time.time()
319
320
        self.get_server().add_latency = 0.5
320
321
        transport = self.get_transport()
321
 
        transport.has('not me') # Force connection by issuing a request
 
322
        transport.has('not me')  # Force connection by issuing a request
322
323
        with_latency_knob_time = time.time() - start_time
323
324
        self.assertTrue(with_latency_knob_time > 0.4)
324
325
 
328
329
        raise TestSkipped('Timing-sensitive test')
329
330
        start_time = time.time()
330
331
        transport = self.get_transport()
331
 
        transport.has('not me') # Force connection by issuing a request
 
332
        transport.has('not me')  # Force connection by issuing a request
332
333
        regular_time = time.time() - start_time
333
334
        self.assertTrue(regular_time < 0.5)
334
335
 
393
394
 
394
395
    def test_bandwidth(self):
395
396
        sending = FakeSocket()
396
 
        receiving = stub_sftp.SocketDelay(sending, 0, bandwidth=8.0/(1024*1024),
397
 
                                          really_sleep=False)
 
397
        receiving = stub_sftp.SocketDelay(
 
398
            sending, 0, bandwidth=8.0 / (1024 * 1024), really_sleep=False)
398
399
        # check that simulated time is charged only per round-trip:
399
400
        t1 = stub_sftp.SocketDelay.simulated_time
400
401
        receiving.send("connect")
413
414
 
414
415
    def readv(self, requests):
415
416
        for start, length in requests:
416
 
            yield self._data[start:start+length]
 
417
            yield self._data[start:start + length]
417
418
 
418
419
    def close(self):
419
420
        pass
428
429
    def checkGetRequests(self, expected_requests, offsets):
429
430
        self.requireFeature(features.paramiko)
430
431
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
431
 
            _null_report_activity)
 
432
                                            _null_report_activity)
432
433
        self.assertEqual(expected_requests, helper._get_requests())
433
434
 
434
435
    def test__get_requests(self):
448
449
    def checkRequestAndYield(self, expected, data, offsets):
449
450
        self.requireFeature(features.paramiko)
450
451
        helper = _mod_sftp._SFTPReadvHelper(offsets, 'artificial_test',
451
 
            _null_report_activity)
 
452
                                            _null_report_activity)
452
453
        data_f = ReadvFile(data)
453
454
        result = list(helper.request_and_yield_offsets(data_f))
454
455
        self.assertEqual(expected, result)
463
464
        # Out of order requests. The requests should get combined, but then be
464
465
        # yielded out-of-order. We also need one that is at the end of a
465
466
        # previous range. See bug #293746
466
 
        self.checkRequestAndYield([(0, b'a'), (10, b'k'), (4, b'efg'), (1, b'bcd')],
467
 
                                  data, [(0, 1), (10, 1), (4, 3), (1, 3)])
 
467
        self.checkRequestAndYield(
 
468
            [(0, b'a'), (10, b'k'), (4, b'efg'), (1, b'bcd')],
 
469
            data, [(0, 1), (10, 1), (4, 3), (1, 3)])
468
470
 
469
471
 
470
472
class TestUsesAuthConfig(TestCaseWithSFTPServer):