diff options
author | Jeff Forcier <jeff@bitprophet.org> | 2018-06-14 15:55:56 -0700 |
---|---|---|
committer | Jeff Forcier <jeff@bitprophet.org> | 2018-06-14 15:55:56 -0700 |
commit | 1ed67ad760d9e4d3e3b528600e0f70786607127d (patch) | |
tree | 35b492f9753feb3882f57e09fa00a95413992606 /tests/stub_sftp.py | |
parent | 54384b3b8abaef9e5f80a582e97941b117d4a24a (diff) | |
parent | 6efe46d6ab0ad2daba436dc0aaed29f57b13bd2c (diff) |
Merge branch 'master' into 1212-int
Diffstat (limited to 'tests/stub_sftp.py')
-rw-r--r-- | tests/stub_sftp.py | 58 |
1 files changed, 34 insertions, 24 deletions
diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py index 19545865..ffae635d 100644 --- a/tests/stub_sftp.py +++ b/tests/stub_sftp.py @@ -24,13 +24,21 @@ import os import sys from paramiko import ( - ServerInterface, SFTPServerInterface, SFTPServer, SFTPAttributes, - SFTPHandle, SFTP_OK, SFTP_FAILURE, AUTH_SUCCESSFUL, OPEN_SUCCEEDED, + ServerInterface, + SFTPServerInterface, + SFTPServer, + SFTPAttributes, + SFTPHandle, + SFTP_OK, + SFTP_FAILURE, + AUTH_SUCCESSFUL, + OPEN_SUCCEEDED, ) from paramiko.common import o666 -class StubServer (ServerInterface): +class StubServer(ServerInterface): + def check_auth_password(self, username, password): # all are allowed return AUTH_SUCCESSFUL @@ -39,7 +47,8 @@ class StubServer (ServerInterface): return OPEN_SUCCEEDED -class StubSFTPHandle (SFTPHandle): +class StubSFTPHandle(SFTPHandle): + def stat(self): try: return SFTPAttributes.from_stat(os.fstat(self.readfile.fileno())) @@ -56,11 +65,11 @@ class StubSFTPHandle (SFTPHandle): return SFTPServer.convert_errno(e.errno) -class StubSFTPServer (SFTPServerInterface): +class StubSFTPServer(SFTPServerInterface): # assume current folder is a fine root # (the tests always create and eventually delete a subfolder, so there shouldn't be any mess) ROOT = os.getcwd() - + def _realpath(self, path): return self.ROOT + self.canonicalize(path) @@ -70,7 +79,9 @@ class StubSFTPServer (SFTPServerInterface): out = [] flist = os.listdir(path) for fname in flist: - attr = SFTPAttributes.from_stat(os.stat(os.path.join(path, fname))) + attr = SFTPAttributes.from_stat( + os.stat(os.path.join(path, fname)) + ) attr.filename = fname out.append(attr) return out @@ -94,9 +105,9 @@ class StubSFTPServer (SFTPServerInterface): def open(self, path, flags, attr): path = self._realpath(path) try: - binary_flag = getattr(os, 'O_BINARY', 0) + binary_flag = getattr(os, "O_BINARY", 0) flags |= binary_flag - mode = getattr(attr, 'st_mode', None) + mode = getattr(attr, "st_mode", None) if mode is not None: fd = os.open(path, flags, mode) else: @@ -110,17 +121,17 @@ class StubSFTPServer (SFTPServerInterface): SFTPServer.set_file_attr(path, attr) if flags & os.O_WRONLY: if flags & os.O_APPEND: - fstr = 'ab' + fstr = "ab" else: - fstr = 'wb' + fstr = "wb" elif flags & os.O_RDWR: if flags & os.O_APPEND: - fstr = 'a+b' + fstr = "a+b" else: - fstr = 'r+b' + fstr = "r+b" else: # O_RDONLY (== 0) - fstr = 'rb' + fstr = "rb" try: f = os.fdopen(fd, fstr) except OSError as e: @@ -159,7 +170,6 @@ class StubSFTPServer (SFTPServerInterface): return SFTPServer.convert_errno(e.errno) return SFTP_OK - def mkdir(self, path, attr): path = self._realpath(path) try: @@ -188,18 +198,18 @@ class StubSFTPServer (SFTPServerInterface): def symlink(self, target_path, path): path = self._realpath(path) - if (len(target_path) > 0) and (target_path[0] == '/'): + if (len(target_path) > 0) and (target_path[0] == "/"): # absolute symlink target_path = os.path.join(self.ROOT, target_path[1:]) - if target_path[:2] == '//': + if target_path[:2] == "//": # bug in os.path.join target_path = target_path[1:] else: # compute relative to path abspath = os.path.join(os.path.dirname(path), target_path) - if abspath[:len(self.ROOT)] != self.ROOT: + if abspath[: len(self.ROOT)] != self.ROOT: # this symlink isn't going to work anyway -- just break it immediately - target_path = '<error>' + target_path = "<error>" try: os.symlink(target_path, path) except OSError as e: @@ -214,10 +224,10 @@ class StubSFTPServer (SFTPServerInterface): return SFTPServer.convert_errno(e.errno) # if it's absolute, remove the root if os.path.isabs(symlink): - if symlink[:len(self.ROOT)] == self.ROOT: - symlink = symlink[len(self.ROOT):] - if (len(symlink) == 0) or (symlink[0] != '/'): - symlink = '/' + symlink + if symlink[: len(self.ROOT)] == self.ROOT: + symlink = symlink[len(self.ROOT) :] + if (len(symlink) == 0) or (symlink[0] != "/"): + symlink = "/" + symlink else: - symlink = '<error>' + symlink = "<error>" return symlink |