python - setup SFTP server on temp folder?

Question

Is it possible to setup SFTP (SSH File Transfer Protocol) on temp folder? I want this for testing purposes (first setup SFTP and after tests tear it down).

Or maybe there is some mock approach where I could mimic as if SFTP exists on that temp folder?

Something like:

import tempfile
import unittest


class TestSFTP(unittest.TestCase):

    @classmethod
    def setupClass(cls):
        folder = tempfile.TemporaryDirectory()
        # use temp folder and setup temp sftp to be used in tests

    # define some test methods and use sftp.

    @classmethod
    def tearDownClass(cls):
        # destroy temp sftp.

P.S. Normally to create SFTP, you need to use sudo, restart some services etc, so such approach would be unfeasible for testing purposes.

Update:

So I was able to set up test class, that it would run sftp server, but I've got issues when I need to stop sftp server properly. Here is the code I've got so far..:

import sftpserver
import paramiko
import os
import sh
import threading

from odoo.tests import common
from odoo.modules.module import get_module_path


def _start_thread(target, args=None, kwargs=None):
    """Run target object in thread, in a background."""
    if not args:
        args = []
    if not kwargs:
        kwargs = {}
    thread = threading.Thread(target=target, args=args, kwargs=kwargs)
    thread.daemon = True
    thread.start()
    return thread


class TestExchangeCommon(common.SavepointCase):
    """Common class for exchange module tests."""

    tests_path = os.path.join(get_module_path('exchange'), 'tests')
    rsa_key_path = os.path.join(tests_path, 'extra/test_rsa.key')

    @classmethod
    def _start_sftp(cls):
        sftpserver.start_server('localhost', 3373, cls.rsa_key_path, 'INFO')

    @classmethod
    def setUpClass(cls):
        """Set up data to be used by all test classes."""
        import logging
        _logger = logging.getLogger()
        super(TestExchangeCommon, cls).setUpClass()
        cls.thread = _start_thread(target=cls._start_sftp)
        pkey = paramiko.RSAKey.from_private_key_file(cls.rsa_key_path)
        cls.transport = paramiko.Transport(('localhost', 3373))
        cls.transport.connect(username='admin', password='admin', pkey=pkey)
        cls.sftp = paramiko.SFTPClient.from_transport(cls.transport)
        _logger.warn(cls.sftp.listdir('.'))

    @classmethod
    def tearDownClass(cls):
        """Kill sftp server to stop running it in a background."""
        cls.sftp.close()
        cls.transport.close()
        sh.fuser('-k', '3373/tcp')  # this kills main application...

In order for sftp server to run, I had to put in a thread, so it would not stall my main application. But when I need to stop sftp server after tests are done, if I kill on port 3373 (sftp server is run), it kills main application too, which is actually run on port 8069. Is there a way to close sftpserver instance properly via python?

Answer 1

Here is my implementation for sftpserver setup that can be used in unittests.

With threading, it was problematic to stop sftpserver, but with multiprocessing it does seem to work properly.

Also note that I added sleep just after process is started, because at least in my case, sometimes sftp server would not start in time and I would get an error that connection was refused.

(Paths implementation for pkey and sftpserver location might be different in different application though):

import sftpserver
import paramiko
import os
import sh
import multiprocessing
import time

from odoo.tests import common
from odoo.modules.module import get_module_path


def _start_process(target, args=None, kwargs=None):
    """Run target object in new process, in a background."""
    if not args:
        args = []
    if not kwargs:
        kwargs = {}
    process = multiprocessing.Process(target=target, args=args, kwargs=kwargs)
    process.daemon = True
    process.start()
    return process


class TestExchangeCommon(common.SavepointCase):
    """Common class for exchange module tests."""

    tests_path = os.path.join(get_module_path('exchange'), 'tests')
    rsa_key_path = os.path.join(tests_path, 'extra/test_rsa.key')

    @classmethod
    def _start_sftp(cls):
        sftpserver.start_server('localhost', 3373, cls.rsa_key_path, 'INFO')

    @classmethod
    def setUpClass(cls):
        """Set up data to be used by all test classes."""
        import logging
        _logger = logging.getLogger()
        super(TestExchangeCommon, cls).setUpClass()
        cls.process = _start_process(target=cls._start_sftp)
        time.sleep(0.01)  # give time for server to start up.
        pkey = paramiko.RSAKey.from_private_key_file(cls.rsa_key_path)
        cls.transport = paramiko.Transport(('localhost', 3373))
        cls.transport.connect(username='admin', password='admin', pkey=pkey)
        cls.sftp = paramiko.SFTPClient.from_transport(cls.transport)
        _logger.warn(cls.sftp.listdir('.'))

    @classmethod
    def tearDownClass(cls):
        """Kill sftp server to stop running it in a background."""
        cls.sftp.close()
        cls.transport.close()
        sh.fuser('-k', '3373/tcp')