ulope / pytest-sftpserver

Local SFTP server fixture plugin for pytest

pytest-sftpserver

pytest-sftpserver is a plugin for pytest that provides a local SFTP-Server fixture.

The SFTP-Server provided by this fixture serves content not from files but directly from Python objects.

Quickstart

Assume you want to test a function that downloads a file from an SFTP-Server:

from contextlib import closing
import paramiko
def get_sftp_file(host, port, username, password, path):
    with closing(paramiko.Transport((host, port))) as transport:
        transport.connect(username=username, password=password)
        with closing(paramiko.SFTPClient.from_transport(transport)) as sftpclient:
            with sftpclient.open(path, "r") as sftp_file:
                return sftp_file.read()

This plugin allows to test such functions without having to spin up an external SFTP-Server by providing a pytest fixture called sftpserver. You use it simply by adding a parameter named sftpserver to your test function:

def test_sftp_fetch(sftpserver):
    with sftpserver.serve_content({'a_dir': {'somefile.txt': "File content"}}):
        assert get_sftp_file(sftpserver.host, sftpserver.port, "user",
                             "pw", "/a_dir/somefile.txt") == "File content"

As can be seen from this example sftpserver serves content directly from python objects instead of files.

Installation

pip install pytest-sftpserver

Supported Python versions

This package supports the following Python versions:

• 2.7, 3.5 - 3.7

TODO

• Add more documentation
• Add more usage examples
• Add TODOs :)

Version History

1.3.0 - 2019-09-16

• Updated supported Python versions to 2.7, 3.5 - 3.7.

  Droped (official) support for 3.4.

• Check / format code with black, isort and flake8.
• Fix return type of .read(). (#15, thanks @WeatherGod)
• Support the offset parameter on write operations. (#11, #16, thanks @DrNecromant)

1.2.0 - 2018-03-28

• Updated supported Python versions to 2.7, 3.4 - 3.6. Droped (official) support for 2.6 and 3.2, 3.3.
• Now always uses posixpath internally to avoid problems when running on Windows (#7, #8, thanks @dundeemt)
• Fixed broken readme badges (#14, thanks @movermeyer)

1.1.2 - 2015-06-01

• Fixed a bug in stat size calculation (#4)
• Fixed mkdir() overwriting existing content (#5)

Thanks to @zerok for both bug reports and accompanying tests.

1.1.1 - 2015-04-04

• Fixed broken chmod() behaviour for non-existing 'files' (Thanks @dundeemt)

1.1.0 - 2014-10-15

• Fixed broken stat() behaviour for non-existing 'files'
• Slightly increased test coverage

1.0.2 - 2014-07-27

• Fixed broken test on Python 2.6

1.0.1 - 2014-07-27

• Added Python 3.2 support
• Cleaned up tox configuration

1.0.0 - 2014-07-18

• Initial release

License

Licensed unter the MIT License. See file LICENSE.

Inspiration

The implementation and idea for this plugin is in part based upon:

• pytest-localserver
• sftpserver
• The Twisted Conch in 60 Seconds series (although I ended up not using twisted, this was very helpful understanding SFTP internals)

pytest-sftpserver's Issues

Package the LICENSE file on source distributions

The LICENSE file isn't included in the source archive on PyPI for some reason. Most licenses require the file to be included in all distributions of the software. This is currently breaking the conda-forge build, which enforces packages to have a license file (conda-forge/pytest-sftpserver-feedstock#4). We can work around that for v1.3.0 but it would be nice if future releases packaged the file.

I tried to reproduce this locally but when I run python setup.py sdist the file is included in there for some reason (even though there is no MANIFEST.in file). So I'm not sure why it's not on the PyPI archive. What is the process used to build and deploy to PyPI?

Cannot put if the server storage starts empty.

pytest-sftpserver/pytest_sftpserver/sftp/content_provider.py

Lines 75 to 77 in 8f90e80

	def _find_object_for_path(self, path):
	if not self.content_object:
	return None

makes it impossible to create things starting from scratch using SFTPServer(content_object={}) because e.g. mkdir('test') invokes self._get_path_components('test') which returns ('', 'test') and then self._find_object_for_path('') returns None instead of the global storage object.

(the default content_object should be {} and not None too)

Fixed ContentProvider is

class FixedContentProvider(ContentProvider):  # fixes default content_object
    def __init__(self, content_object=None):
        self.content_object = content_object or {}

    def is_dir(self, path):  # fixes wrong mode attr on binary uploads
        return not isinstance(self.get(path), (bytes, str, int))

    def _find_object_for_path(self, path):  # fixes adding to empty storage
        if path == '':
            return self.content_object
        else:
            return super()._find_object_for_path(path)

Binary file not handled correctly by is_dir

https://github.com/ulope/pytest-sftpserver/blob/master/pytest_sftpserver/sftp/content_provider.py#L66

def is_dir(self, path):
    return not isinstance(self.get(path), string_types + integer_types)

It should include has well bytes type.

Tests hanging forever

Thanks for this great utility, I'm trying to get it working now with a very simple example:

def test_connect_to_sftp(sftpserver):
    with sftpserver.serve_content({'a_dir': {'somefile.txt': "Some content"}}):
        trans, client = sftp_utils.connect_to_sftp(
            host=sftpserver.host,
            port=sftpserver.port,
            username='user',
            password='pw',
            dsp_name='name',
        )

Where the connect is very simple and just does this

def connect_to_sftp(host, port, username, password, dsp_name):
    logger.info('Connecting to %s: %s:%s', dsp_name, host, port)
    transport = Transport((host, port))
    transport.connect(username=username, password=password)
    logger.info('Connected to %s', dsp_name)

    return transport, SFTPClient.from_transport(transport)

It works in theory but the test runner never quits it just hangs there forever.
Any idea why it's happening?
Is there something I can do to kill the sftpserver thread?
(I tried shutdown already but nothing..)

Feature Request: ability to set default current working directory on connection

When connecting to a standard sftpserver, you are left in the users home directory. i.e. username=test, when you login in you normally are left at /home/test (unless modified by sshd configurations)

However, using the pytest-sftpserver plugin, after the initial connection your are left at /

Do you think it would be possible (within reason) to specify the default directory location if you have a CONTENT structure like:

{'home': {
    'test': {
        'file1.txt': 'file1 contents'
             }
         } 
}

and when the connection is set up for username='test', a request for pwd would result in '/home/test' ?
It wouldn't even have to be that smart, maybe just a default directory arg that if set, results in the above irregardless of login details. If you set a 'bad' default directory it would throw an OSError, if required. Otherwise the error could just be thrown if they try to access a bad file path, like it does currently.

I've implemented a number of tests using the plugin (for the pysftp project), however now that I'm finished I don't like the file system differences between the two test types and I am working on making them go away, but will require that I add a call in the setup for the plugin type tests that sets the directory to the /home/test -- this would be required on all plugin style tests, thus making them a bit different than the tests I have against a local sftpserver.

When using pytest-sftpserver in django test, threads never conclude

Atfinity is using your SFTP Server in our Django tests (without pytest) nicely like so:

mock_sftp_server = SFTPServer()

class TestSFTP(APITestCase):
    fixtures = ['simple.json']
    
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        mock_sftp_server.block_on_close = False
        mock_sftp_server.start()
        print('Started Mock SFTP Server ...')  # noqa
    
    @classmethod
    def tearDownClass(cls) -> None:
        super().tearDownClass()
        mock_sftp_server.server_close()
        mock_sftp_server.shutdown()
        mock_sftp_server.join(0)
        print('Shut down Mock SFTP Server ...')  # noqa

...

However, threads never close. For this, we had to set the threads created by ThreadingMixin to be demon threads and not block like so:

class TestSFTPServer(SFTPServer):
    daemon_threads = True
    block_on_close = False

As this seems to have no negative effect on the functionality, we suggest to do this for SFTPServer directly to make this package useful for django tests. Happy to provide more info how to use this within django tests.

feature request: buggy sftpserver

Just came across this package yesterday, and it has really helped me improve my test suites for some of my projects that involves shipping files via sftp. Right now, I can write test cases for nominal behavior, but I haven't been able to really exercise my error handling code. This is code that has developed organically as I have encountered real-world situations, but is extremely difficult to test, such as flaky connections, and intermittently missing ssh banners.

Do you imagine it might be possible to construct a fixture that might be able to mimic some of that behavior in a predictable manner?

How to mock wrong credentials?

class AllowAllAuthHandler(ServerInterface):
    def check_auth_none(self, username):
        return AUTH_SUCCESSFUL

    def check_auth_password(self, username, password):
        return AUTH_SUCCESSFUL

    def check_auth_publickey(self, username, key):
        return AUTH_SUCCESSFUL

    def check_channel_request(self, kind, chanid):
        return OPEN_SUCCEEDED

As you can see it allows any username password combinations, so it's difficult to mock wrong credentials.

Downloaded files get modified time in the future

I do something like

with server.serve_content({'a_dir': {'somefile.txt': "File content"}}):
    ...

If I then download "somefile.txt" with my pysftp client that I'm trying to test, the file gets a timestamp one hour ahead of now.

I believe this is because of this line

mtime = calendar.timegm(datetime.now().timetuple())

which appears in two places in the code. It takes the current local time (=CET in my case) and transforms it to a timestamp as though it were GMT, resulting in the offset. I recommend that it should call "datetime.utcnow()" instead.

Posix pathing should be used at all times

If running tests on windows, a path that should be /home/test/file.txt is no longer accessible. The pytest-sftpserver is putting it at \home\test\file.txt and the test results as "IOError: [Errno 2] No such file "

Python switches os.path to posixpath or ntpath based on the running environment. In interface.py and util.py path handling is being pulled from os.path (so when running tests in a windows envrionment os.path is using ntpath instead.) and should really always pull from posixpath.

If you import posixpath and replaced all occurrences of os.path with posixpath in those two modules, then sftpserver would also present a standard posix interface.

Put method fails on Python 3.x

It seems like the stats call returns a wrong filesize on Python 3 compared to Python 2. Here is a test-case to reproduce it: https://gist.github.com/zerok/8880cf0db30c4419ac02

Mkdir destroys pre-existing directories

If you execute mkdir on a pre-existing directory path, that directory is replaced with a new empty one. Paramiko itself would raise an IOError on real servers. A test-case for this can be found here: https://gist.github.com/zerok/4e2d313f440436750588

File Like Objects

I can only see examples for persistent files on the host. Would it be possible to enable the use of file like objects instead.

It would make test setup and modification much easier.

Problem using put with sftpclient

I'm having some problems using put on sftpclient, I always get an exception and it does not work for me.

In the most simple example I have pytest and pytest-sftpserver installed in a virtual environment (I have tried with both python 2.7 and python 3.8).

Using this test:

from paramiko import Transport
from paramiko.channel import Channel
from paramiko.sftp_client import SFTPClient

def test_sftp(sftpserver):
    transport = Transport((sftpserver.host, sftpserver.port))
    transport.connect(username="a", password="b")
    sftpclient = SFTPClient.from_transport(transport)
    
    try:
        assert sftpclient.listdir("/") == []
        sftpclient.put("temp.txt", "/a/test.txt")
    finally:
        sftpclient.close()
        transport.close()

I get this error:

test.py:12:

---

local/lib/python2.7/site-packages/paramiko/sftp_client.py:759: in put
return self.putfo(fl, remotepath, file_size, callback, confirm)
local/lib/python2.7/site-packages/paramiko/sftp_client.py:720: in putfo
s = self.stat(remotepath)
local/lib/python2.7/site-packages/paramiko/sftp_client.py:493: in stat
t, msg = self._request(CMD_STAT, path)
local/lib/python2.7/site-packages/paramiko/sftp_client.py:813: in _request
return self._read_response(num)
local/lib/python2.7/site-packages/paramiko/sftp_client.py:865: in _read_response
self._convert_status(msg)

---

self = <paramiko.sftp_client.SFTPClient object at 0x7f360daa0fd0>, msg = paramiko.Message('\x00\x00\x00\x07\x00\x00\x00\x02\x00\x00\x00\x0cNo such file\x00\x00\x00\x00')

def _convert_status(self, msg):
    """
    Raises EOFError or IOError on error status; otherwise does nothing.
    """
    code = msg.get_int()
    text = msg.get_text()
    if code == SFTP_OK:
        return
    elif code == SFTP_EOF:
        raise EOFError(text)
    elif code == SFTP_NO_SUCH_FILE:
        # clever idea from john a. meinel: map the error codes to errno
        raise IOError(errno.ENOENT, text)

E IOError: [Errno 2] No such file

I don't understand what is wrong. Any help would be greatly appreciated.

IOError not raised when chmod a non-existent file

def test_chmod_not_exist(sftpserver):
    '''verify error if trying to chmod something that isn't there'''
    with sftpserver.serve_content(CONTENT):
        with pysftp.Connection(**conn(sftpserver)) as psftp:
            with pytest.raises(IOError):
                psftp.chmod('i-do-not-exist.txt', 666)

sftpserver = <SFTPServer(Thread-1, started daemon 139983241848576)>

    def test_chmod_not_exist(sftpserver):
        '''verify error if trying to chmod something that isn't there'''
        with sftpserver.serve_content(CONTENT):
            with pysftp.Connection(**conn(sftpserver)) as psftp:
                with pytest.raises(IOError):
>                   psftp.chmod('i-do-not-exist.txt', 666)
E                   Failed: DID NOT RAISE

Content provider can't handle big byte data

I've been using this test sftpserver with Paramiko.

If a file that is being "transferred" is bigger than the MAX_REQUEST_SIZE in Paramiko then the test sftpserver will not append the leftover data from subsequent packets.

I've narrowed down the line to this one https://github.com/ulope/pytest-sftpserver/blob/master/pytest_sftpserver/sftp/interface.py#L44 which just needs to account for the binary_data type in six.

Server cannot handle directory named 'get'

Scenario

Given a directory with name 'get', the SFTP server does not process this directory. Instead, a OSError (text = 'Failure') is thrown.

This error also occurs for folders with other names matching methods of dict, e.g. items. (https://docs.python.org/3.4/library/stdtypes.html#dict)

Root cause

The content provider invokes the builtin getattr on the content_object dict. 'get' however is a method of dict, which throws TypeError if invoked without arguments.

This typeerror is subsequently caught by the server and wrapped into a failure.

Reproduction steps

def test_sftp_fetch(sftpserver):
    with sftpserver.serve_content({'get': {'somefile.txt': "File content"}}):
        assert get_sftp_file(sftpserver.host, sftpserver.port, "user",
                             "pw", "/get/somefile.txt") == "File content"

Stacktrace

=================================== FAILURES ===================================
_______________________________ test_sftp_fetch ________________________________

sftpserver = <SFTPServer(Thread-1, started daemon 139954133071616)>

    def test_sftp_fetch(sftpserver):
        with sftpserver.serve_content({'get': {'somefile.txt': "File content"}}):
>           assert get_sftp_file(sftpserver.host, sftpserver.port, "user",
                                 "pw", "/get/somefile.txt") == "File content"

test.py:12: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
test.py:7: in get_sftp_file
    with sftpclient.open(path, "r") as sftp_file:
python3.4/dist-packages/paramiko/sftp_client.py:327: in open
    t, msg = self._request(CMD_OPEN, filename, imode, attrblock)
python3.4/dist-packages/paramiko/sftp_client.py:729: in _request
    return self._read_response(num)
python3.4/dist-packages/paramiko/sftp_client.py:776: in _read_response
    self._convert_status(msg)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <paramiko.sftp_client.SFTPClient object at 0x7f499c635668>
msg = paramiko.Message(b'\x00\x00\x00\x01\x00\x00\x00\x04\x00\x00\x00\x07Failure\x00\x00\x00\x00')

    def _convert_status(self, msg):
        """
            Raises EOFError or IOError on error status; otherwise does nothing.
            """
        code = msg.get_int()
        text = msg.get_text()
        if code == SFTP_OK:
            return
        elif code == SFTP_EOF:
            raise EOFError(text)
        elif code == SFTP_NO_SUCH_FILE:
            # clever idea from john a. meinel: map the error codes to errno
            raise IOError(errno.ENOENT, text)
        elif code == SFTP_PERMISSION_DENIED:
            raise IOError(errno.EACCES, text)
        else:
>           raise IOError(text)
E           OSError: Failure

content_provider.py:

 82        obj = self.content_object
 83        for part in path.split(separator):
 84            if part:
 85                try:
 86                    new_obj = getattr(obj, part)
 87                except (AttributeError, TypeError):
 88                    try:
 89                        new_obj = obj[part]
 90                    except (KeyError, TypeError, IndexError):
 91                        if part.isdigit():
 92                            try:
 93                                new_obj = obj[int(part)]
 94                            except (KeyError, TypeError, IndexError):
 95                                return None
 96                        else:
 97                            return None
 98                obj = new_obj
 99                if callable(obj):
100                    obj = obj()  # <built-in method get of dict object at 0x7f99f4f0ec48>
101        return obj

sftp_server.py:

110            try:
111                self._process(t, request_number, msg)
112            except Exception as e:  # {TypeError} get expected at least 1 arguments, got 0
113                self._log(DEBUG, 'Exception in server processing: ' + str(e))
114                self._log(DEBUG, util.tb_strings())
115                # send some kind of failure message, at least
116                try:
117                    self._send_status(request_number, SFTP_FAILURE)
118                except:
119                    pass