SFTPのモックサーバをたてユニットテストをする

Python

SFTP サーバを介してデータをやりとりするシステムのユニットテストを書くかもしれないので、どのように組むのか試してみました。

ひとまず、 Django で試します。

必須ソフトウェア

以下のソフトを利用します。

  • Django (1.5 利用)
  • paramiko (1.10.0 利用)
  • sftpserver (0.2 利用)

やること

SFTP サーバからファイルをとってくる関数を作成し、そのユニットテストを書きます。

ディレクトリ構成

  • ディレクトリ直下で virtualenv 作ってます。
  • settings にてこのディレクトリの . の位置を PRJ_ROOT としています。
  • test_rsa.key と test_rsa.key.pub は秘密鍵と公開鍵です。
  • test.txt はテストで利用します。
  • var は　（今回は利用しませんが） SQLite3 データ保存用に利用します。
$ tree
.
├── app
│   ├── __init__.py
│   ├── manage.py
│   ├── settings.py
│   ├── sftp
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── tasks.py
│   │   ├── tests.py
│   │   └── views.py
│   ├── urls.py
│   └── wsgi.py
├── misc
│   ├── test_rsa.key
│   ├── test_rsa.key.pub
│   └── test.txt
├── README.rst
├── setup.cfg
├── setup.py
└── var

コード

次のような tasks.py を作成します。

  • ざっくりとユニットテストが動けばいいので雑で済みません。
  • transport.connect() の password は秘密鍵のパスワードです。
# -*- coding: utf-8 -*-
import os

import paramiko
from django.conf import settings


def get_remote_file(filename, local_dir):
    """ 
    SFTP サーバから filename ファイルを取得し、
    dst_dir に配置する。
    """
    pkey = paramiko.RSAKey.from_private_key_file(
        '{0}'.format(settings.PRIV_KEY))
    transport = paramiko.Transport(('localhost', 3373))                                                                       
    transport.connect(password='', pkey=pkey)
    sftp = paramiko.SFTPClient.from_transport(transport)
    sftp.get(
        '{0}'.format(os.path.join(os.curdir, os.sep, filename)),
        '{0}'.format(os.path.join(local_dir, filename)))
    sftp.close()

テストコード

次のような tests.py を作成します。

# -*- coding: utf-8 -*-                                                                                                       
import os
import hashlib
from time import sleep
from subprocess import Popen
from tempfile import mkdtemp
from shutil import (
    copyfile,
    rmtree,
)

from django.conf import settings
from django.test import TestCase

from app.sftp.tasks import get_remote_file


class SftpTest(TestCase):

    def setUp(self):
        """テストのための準備を整える

        1. テスト用フォルダを作成する。
        2. テストファイルを配置する。
        3. サーバのルートに移動する。
        4. 作成済みの秘密鍵を利用して、
           SFTP サーバを起動する。
        5. ローカルの作業ディレクトリの作成
        """
        self.sftproot = mkdtemp()

        self.testfile = 'test.txt'
        copyfile(
            os.path.join(settings.PRJ_ROOT, 'misc', self.testfile),
            os.path.join(self.sftproot, self.testfile))

        self.curdir = os.getcwd()
        os.chdir(self.sftproot)

        self.proc = Popen([
            '{0}'.format(os.path.join(settings.PRJ_ROOT, 'bin', 'sftpserver')),
            '-k',
            '{0}'.format(os.path.join(settings.PRJ_ROOT, 'misc', 'test_rsa.key')),
            '-l',
            'WARNING',
        ])
        # 起動に多少時間がかかるので、1秒待つ。
        sleep(1)

        self.localdir = mkdtemp()

    def tearDown(self):
        """テストの後始末をする

        1. テスト時の作業ディレクトリに移動する。
        2. SFTP サーバを停止する。
        3. テスト用フォルダを削除する。
        4. 作業用フォルダを削除する。
        """
        os.chdir(self.curdir)
        self.proc.terminate()
        rmtree(self.sftproot)
        rmtree(self.localdir)

    def test_basic_addition(self):
        """取得したファイルのハッシュが期待したものであること"""
        get_remote_file('test.txt', self.localdir)
        with open(os.path.join(self.localdir, 'test.txt')) as fp:
            md5alg = hashlib.md5()
            md5alg.update(fp.read())
            md5 = md5alg.hexdigest()
        # ファイルの中身は hoge のみ
        self.assertEqual(md5, 'c59548c3c576228486a1f0037eb16a1b')

まとめ

上記でテストが出きるようになると思います。ただ、 sftpserver の最終更新日が 2011/12/12 なのはちょっと古いのかもしれません、そこは注意願います。

参考

割と検索ったー