/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/stub_sftp.py

  • Committer: Jelmer Vernooij
  • Date: 2020-03-22 01:35:14 UTC
  • mfrom: (7490.7.6 work)
  • mto: This revision was merged to the branch mainline in revision 7499.
  • Revision ID: jelmer@jelmer.uk-20200322013514-7vw1ntwho04rcuj3
merge lp:brz/3.1.

Show diffs side-by-side

added added

removed removed

Lines of Context:
breezy/tests/stub_sftp.py
22
22
import os
23
23
import paramiko
24
24
import socket
25
 
try:
26
 
    import socketserver
27
 
except ImportError:
28
 
    import SocketServer as socketserver
 
25
import socketserver
29
26
import sys
30
27
import time
31
28
 
135
132
    def list_folder(self, path):
136
133
        path = self._realpath(path)
137
134
        try:
138
 
            out = [ ]
 
135
            out = []
139
136
            # TODO: win32 incorrectly lists paths with non-ascii if path is not
140
137
            # unicode. However on unix the server should only deal with
141
138
            # bytestreams and posix.listdir does the right thing
216
213
            return paramiko.SFTPServer.convert_errno(e.errno)
217
214
        return paramiko.SFTP_OK
218
215
 
 
216
    def symlink(self, target_path, path):
 
217
        path = self._realpath(path)
 
218
        try:
 
219
            os.symlink(target_path, path)
 
220
        except OSError as e:
 
221
            return paramiko.SFTPServer.convert_errno(e.errno)
 
222
        return paramiko.SFTP_OK
 
223
 
 
224
    def readlink(self, path):
 
225
        path = self._realpath(path)
 
226
        try:
 
227
            target_path = os.readlink(path)
 
228
        except OSError as e:
 
229
            return paramiko.SFTPServer.convert_errno(e.errno)
 
230
        return target_path
 
231
 
219
232
    def mkdir(self, path, attr):
220
233
        path = self._realpath(path)
221
234
        try:
240
253
            return paramiko.SFTPServer.convert_errno(e.errno)
241
254
        return paramiko.SFTP_OK
242
255
 
243
 
    # removed: chattr, symlink, readlink
 
256
    # removed: chattr
244
257
    # (nothing in bzr's sftp transport uses those)
245
258
 
246
259
 
247
260
# ------------- server test implementation --------------
248
261
 
249
 
STUB_SERVER_KEY = """
 
262
STUB_SERVER_KEY = """\
250
263
-----BEGIN RSA PRIVATE KEY-----
251
264
MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz
252
265
oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/
380
393
        # Re-import these as locals, so that they're still accessible during
381
394
        # interpreter shutdown (when all module globals get set to None, leading
382
395
        # to confusing errors like "'NoneType' object has no attribute 'error'".
 
396
 
383
397
        class FakeChannel(object):
384
398
            def get_transport(self):
385
399
                return self
 
400
 
386
401
            def get_log_channel(self):
387
402
                return 'brz.paramiko'
 
403
 
388
404
            def get_name(self):
389
405
                return '1'
 
406
 
390
407
            def get_hexdump(self):
391
408
                return False
 
409
 
392
410
            def close(self):
393
411
                pass
394
412
 
397
415
            FakeChannel(), 'sftp', StubServer(tcs), StubSFTPServer,
398
416
            root=tcs._root, home=tcs._server_homedir)
399
417
        self.sftp_server = sftp_server
400
 
        sys_stderr = sys.stderr # Used in error reporting during shutdown
 
418
        sys_stderr = sys.stderr  # Used in error reporting during shutdown
401
419
        try:
402
420
            sftp_server.start_subsystem(
403
421
                'sftp', None, ssh.SocketAsChannelAdapter(self.request))
477
495
    def start_server(self, backing_server=None):
478
496
        # XXX: TODO: make sftpserver back onto backing_server rather than local
479
497
        # disk.
480
 
        if not (backing_server is None or
481
 
                isinstance(backing_server, test_server.LocalURLServer)):
 
498
        if not (backing_server is None
 
499
                or isinstance(backing_server, test_server.LocalURLServer)):
482
500
            raise AssertionError(
483
501
                'backing_server should not be %r, because this can only serve '
484
502
                'the local current working directory.' % (backing_server,))
489
507
            # Normalize the path or it will be wrongly escaped
490
508
            self._homedir = osutils.normpath(self._homedir)
491
509
        else:
492
 
            # But unix SFTP servers should just deal in bytestreams
493
 
            self._homedir = self._homedir.encode('utf-8')
 
510
            self._homedir = self._homedir
494
511
        if self._server_homedir is None:
495
512
            self._server_homedir = self._homedir
496
513
        self._root = '/'
569
586
        server = super(SFTPSiblingAbsoluteServer, self).create_server()
570
587
        server._server_homedir = '/dev/noone/runs/tests/here'
571
588
        return server
572