python - настроить SFTP-сервер на временную папку? - PullRequest
Новая
       18

python - настроить SFTP-сервер на временную папку?

0 голосов
/ 29 января 2019

Можно ли настроить SFTP (SSH File Transfer Protocol) для временной папки?Я хочу это для целей тестирования (сначала настройте SFTP, а после тестирования разорвите его).

Или, может быть, есть какой-то ложный подход, в котором я мог бы имитировать, как если бы SFTP существовал в этой временной папке?

Что-то вроде: 

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.

PS Обычно для создания SFTP необходимо использовать sudo, перезапустить некоторые службы и т. Д., Поэтому такой подход был бы невозможен для целей тестирования.

Обновление:

Итак, я смог настроить тестовый класс, чтобы он запускал sftp-сервер, но у меня есть проблемы, когда мне нужно правильно остановить sftp-сервер.Вот код, который я получил до сих пор ..: 

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...

Для того, чтобы сервер sftp работал, мне пришлось добавить поток, чтобы он не остановил мое основное приложение.Но когда мне нужно остановить сервер sftp после того, как тесты завершены, если я уничтожу порт 3373 (сервер sftp запущен), он также убивает основное приложение, которое фактически запускается на порту 8069. Есть ли способ правильно закрыть экземпляр sftpserver черезпитон

1 Ответ

0 голосов
/ 29 января 2019

Вот моя реализация для настройки sftpserver, которую можно использовать в тестах модулей.

При работе с потоками было трудно остановить sftpserver, но при многопроцессорной обработке он, похоже, работает правильно.

Также обратите внимание, что я добавил sleep сразу после запуска процесса, потому что, по крайней мере, в моем случае иногда сервер sftp не запускается вовремя, и я получаю сообщение об ошибке, что в соединении отказано.

(Реализация путей для расположения pkey и sftpserver может отличаться в разных приложениях): 

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')
...