summaryrefslogtreecommitdiffhomepage
path: root/tests/stub_sftp.py
diff options
context:
space:
mode:
authorJeff Forcier <jeff@bitprophet.org>2018-06-14 15:55:56 -0700
committerJeff Forcier <jeff@bitprophet.org>2018-06-14 15:55:56 -0700
commit1ed67ad760d9e4d3e3b528600e0f70786607127d (patch)
tree35b492f9753feb3882f57e09fa00a95413992606 /tests/stub_sftp.py
parent54384b3b8abaef9e5f80a582e97941b117d4a24a (diff)
parent6efe46d6ab0ad2daba436dc0aaed29f57b13bd2c (diff)
Merge branch 'master' into 1212-int
Diffstat (limited to 'tests/stub_sftp.py')
-rw-r--r--tests/stub_sftp.py58
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