Improved SFTP storage (#177)

* Added parameters to SFTP __init__ and fix File.close()

* Added tests for SFTP

* Moved SFTP documentation from comments to docs

docs/backends/sftp.rst

@@ -1,5 +1,68 @@
 SFTP
 ====
 
-Take a look at the top of the backend's file for the documentation.
+Settings
+--------
 
+``SFTP_STORAGE_HOST``
+
+The hostname where you want the files to be saved.
+
+``SFTP_STORAGE_ROOT``
+
+The root directory on the remote host into which files should be placed.
+Should work the same way that ``STATIC_ROOT`` works for local files. Must
+include a trailing slash.
+
+``SFTP_STORAGE_PARAMS`` (Optional)
+
+A dictionary containing connection parameters to be passed as keyword
+arguments to ``paramiko.SSHClient().connect()`` (do not include hostname here).
+See `paramiko SSHClient.connect() documentation`_ for details
+
+.. _`paramiko SSHClient.connect() documentation`: http://docs.paramiko.org/en/latest/api/client.html#paramiko.client.SSHClient.connect
+
+``SFTP_STORAGE_INTERACTIVE`` (Optional)
+
+A boolean indicating whether to prompt for a password if the connection cannot
+be made using keys, and there is not already a password in
+``SFTP_STORAGE_PARAMS``. You can set this to ``True`` to enable interactive
+login when running ``manage.py collectstatic``, for example.
+
+.. warning::
+
+  DO NOT set SFTP_STORAGE_INTERACTIVE to True if you are using this storage
+  for files being uploaded to your site by users, because you'll have no way
+  to enter the password when they submit the form..
+
+``SFTP_STORAGE_FILE_MODE`` (Optional)
+
+A bitmask for setting permissions on newly-created files. See
+`Python os.chmod documentation`_ for acceptable values.
+
+
+``SFTP_STORAGE_DIR_MODE`` (Optional)
+
+A bitmask for setting permissions on newly-created directories. See
+`Python os.chmod documentation`_ for acceptable values.
+
+.. note::
+
+  Hint: if you start the mode number with a 0 you can express it in octal
+  just like you would when doing "chmod 775 myfile" from bash.
+
+.. _`Python os.chmod documentation`: http://docs.python.org/library/os.html#os.chmod
+
+``SFTP_STORAGE_UID`` (Optional)
+
+UID of the account that should be set as owner of the files on the remote
+host. You may have to be root to set this.
+
+``SFTP_STORAGE_GID`` (Optional)
+
+GID of the group that should be set on the files on the remote host. You have
+to be a member of the group to set this.
+
+``SFTP_KNOWN_HOST_FILE`` (Optional)
+
+Absolute path of know host file, if it isn't set ``"~/.ssh/known_hosts"`` will be used.

requirements-tests.txt

@@ -3,3 +3,4 @@ pytest-cov==2.2.1
 boto>=2.32.0
 dropbox>=3.24
 mock
+paramiko

storages/backends/sftpstorage.py

@@ -4,49 +4,6 @@
 # License: MIT
 #
 # Modeled on the FTP storage by Rafal Jonca <[email protected]>
-#
-# Settings:
-#
-# SFTP_STORAGE_HOST - The hostname where you want the files to be saved.
-#
-# SFTP_STORAGE_ROOT - The root directory on the remote host into which files
-# should be placed.  Should work the same way that STATIC_ROOT works for local
-# files.  Must include a trailing slash.
-#
-# SFTP_STORAGE_PARAMS (Optional) - A dictionary containing connection
-# parameters to be passed as keyword arguments to
-# paramiko.SSHClient().connect() (do not include hostname here).  See
-# http://docs.paramiko.org/en/latest/api/client.html#paramiko.client.SSHClient.connect
-# for details
-#
-# SFTP_STORAGE_INTERACTIVE (Optional) - A boolean indicating whether to prompt
-# for a password if the connection cannot be made using keys, and there is not
-# already a password in SFTP_STORAGE_PARAMS.  You can set this to True to
-# enable interactive login when running 'manage.py collectstatic', for example.
-#
-#   DO NOT set SFTP_STORAGE_INTERACTIVE to True if you are using this storage
-#   for files being uploaded to your site by users, because you'll have no way
-#   to enter the password when they submit the form..
-#
-# SFTP_STORAGE_FILE_MODE (Optional) - A bitmask for setting permissions on
-# newly-created files.  See http://docs.python.org/library/os.html#os.chmod for
-# acceptable values.
-#
-# SFTP_STORAGE_DIR_MODE (Optional) - A bitmask for setting permissions on
-# newly-created directories.  See
-# http://docs.python.org/library/os.html#os.chmod for acceptable values.
-#
-#   Hint: if you start the mode number with a 0 you can express it in octal
-#   just like you would when doing "chmod 775 myfile" from bash.
-#
-# SFTP_STORAGE_UID (Optional) - uid of the account that should be set as owner
-# of the files on the remote host.  You have to be root to set this.
-#
-# SFTP_STORAGE_GID (Optional) - gid of the group that should be set on the
-# files on the remote host.  You have to be a member of the group to set this.
-# SFTP_KNOWN_HOST_FILE (Optional) - absolute path of know host file, if it isn't
-# set "~/.ssh/known_hosts" will be used
-
 
 import getpass
 import os
@@ -59,29 +16,32 @@
 from django.core.files.base import File
 
 from storages.compat import urlparse, BytesIO, Storage
+from storages.utils import setting
 
 
 class SFTPStorage(Storage):
 
-    def __init__(self):
-        self._host = settings.SFTP_STORAGE_HOST
+    def __init__(self, host, params=None, interactive=None, file_mode=None,
+                 dir_mode=None, uid=None, gid=None, known_host_file=None,
+                 root_path=None, base_url=None):
+        self._host = host or settings('SFTP_STORAGE_HOST')
 
-        # if present, settings.SFTP_STORAGE_PARAMS should be a dict with params
-        # matching the keyword arguments to paramiko.SSHClient().connect().  So
-        # you can put username/password there.  Or you can omit all that if
-        # you're using keys.
-        self._params = getattr(settings, 'SFTP_STORAGE_PARAMS', {})
-        self._interactive = getattr(settings, 'SFTP_STORAGE_INTERACTIVE',
-                                    False)
-        self._file_mode = getattr(settings, 'SFTP_STORAGE_FILE_MODE', None)
-        self._dir_mode = getattr(settings, 'SFTP_STORAGE_DIR_MODE', None)
+        self._params = params or setting('SFTP_STORAGE_PARAMS', {})
+        self._interactive = setting('SFTP_STORAGE_INTERACTIVE', False) \
+            if interactive is None else interactive
+        self._file_mode = setting('SFTP_STORAGE_FILE_MODE') \
+            if file_mode is None else file_mode
+        self._dir_mode = setting('SFTP_STORAGE_DIR_MODE') if \
+            dir_mode is None else dir_mode
 
-        self._uid = getattr(settings, 'SFTP_STORAGE_UID', None)
-        self._gid = getattr(settings, 'SFTP_STORAGE_GID', None)
-        self._known_host_file = getattr(settings, 'SFTP_KNOWN_HOST_FILE', None)
+        self._uid = setting('SFTP_STORAGE_UID') if uid is None else uid
+        self._gid = setting('SFTP_STORAGE_GID') if gid is None else gid
+        self._known_host_file = setting('SFTP_KNOWN_HOST_FILE') \
+            if known_host_file is None else known_host_file
 
-        self._root_path = settings.SFTP_STORAGE_ROOT
-        self._base_url = settings.MEDIA_URL
+        self._root_path = setting('SFTP_STORAGE_ROOT', '') \
+            if root_path is None else root_path
+        self._base_url = setting('MEDIA_URL') if base_url is None else base_url
 
         # for now it's all posix paths.  Maybe someday we'll support figuring
         # out if the remote host is windows.
@@ -263,5 +223,5 @@ def write(self, content):
 
     def close(self):
         if self._is_dirty:
-            self._storage._save(self._name, self.file.getvalue())
+            self._storage._save(self._name, self)
         self.file.close()

tests/test_sftp.py

@@ -0,0 +1,146 @@
+import stat
+from datetime import datetime
+try:
+    from unittest.mock import patch, MagicMock
+except ImportError:  # Python 3.2 and below
+    from mock import patch, MagicMock
+from django.test import TestCase
+from django.core.files.base import File
+from django.utils.six import BytesIO
+from storages.backends import sftpstorage
+
+
+class SFTPStorageTest(TestCase):
+    def setUp(self):
+        self.storage = sftpstorage.SFTPStorage('foo')
+
+    def test_init(self):
+        pass
+
+    @patch('paramiko.SSHClient')
+    def test_connect(self, mock_ssh):
+        self.storage._connect()
+        self.assertEqual('foo', mock_ssh.return_value.connect.call_args[0][0])
+
+    def test_open(self):
+        file_ = self.storage._open('foo')
+        self.assertIsInstance(file_, sftpstorage.SFTPStorageFile)
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp')
+    def test_read(self, mock_sftp):
+        file_ = self.storage._read('foo')
+        self.assertTrue(mock_sftp.open.called)
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp')
+    def test_chown(self, mock_sftp):
+        self.storage._chown('foo', 1, 1)
+        self.assertEqual(mock_sftp.chown.call_args[0], ('foo', 1, 1))
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp')
+    def test_mkdir(self, mock_sftp):
+        self.storage._mkdir('foo')
+        self.assertEqual(mock_sftp.mkdir.call_args[0], ('foo',))
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'stat.side_effect': (IOError(), True)
+    })
+    def test_mkdir_parent(self, mock_sftp):
+        self.storage._mkdir('bar/foo')
+        self.assertEqual(mock_sftp.mkdir.call_args_list[0][0], ('bar',))
+        self.assertEqual(mock_sftp.mkdir.call_args_list[1][0], ('bar/foo',))
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp')
+    def test_save(self, mock_sftp):
+        self.storage._save('foo', File(BytesIO(b'foo'), 'foo'))
+        self.assertTrue(mock_sftp.open.return_value.write.called)
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'stat.side_effect': (IOError(), True)
+    })
+    def test_save_in_subdir(self, mock_sftp):
+        self.storage._save('bar/foo', File(BytesIO(b'foo'), 'foo'))
+        self.assertEqual(mock_sftp.mkdir.call_args_list[0][0], ('bar',))
+        self.assertTrue(mock_sftp.open.return_value.write.called)
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp')
+    def test_delete(self, mock_sftp):
+        self.storage.delete('foo')
+        self.assertEqual(mock_sftp.remove.call_args_list[0][0], ('foo',))
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp')
+    def test_exists(self, mock_sftp):
+        self.assertTrue(self.storage.exists('foo'))
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'stat.side_effect': IOError()
+    })
+    def test_not_exists(self, mock_sftp):
+        self.assertFalse(self.storage.exists('foo'))
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'listdir_attr.return_value':
+            [MagicMock(filename='foo', st_mode=stat.S_IFDIR),
+             MagicMock(filename='bar', st_mode=None)]})
+    def test_listdir(self, mock_sftp):
+        dirs, files = self.storage.listdir('/')
+        self.assertTrue(dirs)
+        self.assertTrue(files)
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'stat.return_value.st_size': 42,
+    })
+    def test_size(self, mock_sftp):
+        self.assertEqual(self.storage.size('foo'), 42)
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'stat.return_value.st_atime': 1469674684.000000,
+    })
+    def test_accessed_time(self, mock_sftp):
+        self.assertEqual(self.storage.accessed_time('foo'),
+                         datetime(2016, 7, 27, 21, 58, 4))
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'stat.return_value.st_mtime': 1469674684.000000,
+    })
+    def test_modified_time(self, mock_sftp):
+        self.assertEqual(self.storage.modified_time('foo'),
+                         datetime(2016, 7, 27, 21, 58, 4))
+
+    def test_url(self):
+        self.assertEqual(self.storage.url('foo'), '/media/foo')
+        # Test custom
+        self.storage._base_url = 'http://bar.pt/'
+        self.assertEqual(self.storage.url('foo'), 'http://bar.pt/foo')
+        # Test error
+        with self.assertRaises(ValueError):
+            self.storage._base_url = None
+            self.storage.url('foo')
+
+
+class SFTPStorageFileTest(TestCase):
+    def setUp(self):
+        self.storage = sftpstorage.SFTPStorage('foo')
+        self.file = sftpstorage.SFTPStorageFile('bar', self.storage, 'wb')
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'stat.return_value.st_size': 42,
+    })
+    def test_size(self, mock_sftp):
+        self.assertEqual(self.file.size, 42)
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp', **{
+        'open.return_value.read.return_value': b'foo',
+    })
+    def test_read(self, mock_sftp):
+        self.assertEqual(self.file.read(), b'foo')
+        self.assertTrue(mock_sftp.open.called)
+
+    def test_write(self):
+        self.file.write(b'foo')
+        self.assertEqual(self.file.file.read(), b'foo')
+
+    @patch('storages.backends.sftpstorage.SFTPStorage.sftp')
+    def test_close(self, mock_sftp):
+        self.file.write(b'foo')
+        self.file.close()
+        self.assertTrue(mock_sftp.open.return_value.write.called)

tox.ini

@@ -18,3 +18,4 @@ deps =
     boto>=2.32.0
     pytest-cov==2.2.1
     dropbox>=3.24
+    paramiko