Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions pytest_sftpserver/sftp/interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,12 @@
from paramiko.sftp_attr import SFTPAttributes
from paramiko.sftp_handle import SFTPHandle
from paramiko.sftp_si import SFTPServerInterface
from six import string_types, text_type
from six import binary_type, string_types, text_type

from pytest_sftpserver.sftp.util import abspath

ACCEPTABLE_TYPES = string_types + (binary_type,)


class VirtualSFTPHandle(SFTPHandle):
def __init__(self, path, content_provider, flags=0):
Expand All @@ -41,7 +43,7 @@ def write(self, offset, data):
if content is None:
return SFTP_OK if self.content_provider.put(self.path, data) else SFTP_NO_SUCH_FILE

if not isinstance(content, string_types):
if not isinstance(content, ACCEPTABLE_TYPES):
# Can't offset write into a 'directory' or integer
return SFTP_FAILURE

Expand Down
13 changes: 11 additions & 2 deletions tests/test_sftp.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,9 @@ def sftpclient(sftpserver):

@pytest.yield_fixture
def content(sftpserver):
with sftpserver.serve_content(deepcopy(CONTENT_OBJ)):
yield
actual_content = deepcopy(CONTENT_OBJ)
with sftpserver.serve_content(actual_content):
yield actual_content


@pytest.mark.xfail(sys.version_info < (2, 7), reason="Intermittently broken on 2.6")
Expand Down Expand Up @@ -105,6 +106,14 @@ def test_sftpserver_put_file(content, sftpclient, tmpdir):
assert set(sftpclient.listdir("/a")) == set(["test.txt", "b", "c", "f"])


def test_sftpserver_put_bigger_file(content, sftpclient, tmpdir):
tmpfile = tmpdir.join("test.txt")
file_size = 40000
tmpfile.write("x" * file_size)
sftpclient.put(str(tmpfile), "/a/test.txt")
assert len(content["a"]["test.txt"]) == file_size


def test_sftpserver_round_trip(content, sftpclient, tmpdir):
tmpfile = tmpdir.join("test.txt")
thetext = u"Just some plain, normal text"
Expand Down