From 84a00d8e8a8353a662aa4e5b6662cf0ef7caa35f Mon Sep 17 00:00:00 2001 From: fazledyn-or Date: Fri, 6 Oct 2023 12:25:06 +0600 Subject: [PATCH 1/4] Feature: Added support & test for FTP_TLS --- storages/backends/ftp.py | 13 ++- tests/test_ftp.py | 208 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 217 insertions(+), 4 deletions(-) diff --git a/storages/backends/ftp.py b/storages/backends/ftp.py index f467e70b..8c559459 100644 --- a/storages/backends/ftp.py +++ b/storages/backends/ftp.py @@ -6,11 +6,13 @@ # Usage: # # Add below to settings.py: -# FTP_STORAGE_LOCATION = '[a]ftp://:@:/[path]' +# FTP_STORAGE_LOCATION = '[a]ftp[s]://:@:/[path]' # # In models.py you can write: # from FTPStorage import FTPStorage # fs = FTPStorage() +# For a TLS configuration, you must use it below: +# fs = FTPStorage(secure=True) # class FTPTest(models.Model): # file = models.FileField(upload_to='a/b/c/', storage=fs) @@ -37,7 +39,7 @@ class FTPStorageException(Exception): class FTPStorage(Storage): """FTP Storage class for Django pluggable storage system.""" - def __init__(self, location=None, base_url=None, encoding=None): + def __init__(self, location=None, base_url=None, encoding=None, secure=False): location = location or setting("FTP_STORAGE_LOCATION") if location is None: raise ImproperlyConfigured( @@ -51,13 +53,14 @@ def __init__(self, location=None, base_url=None, encoding=None): self._config = self._decode_location(location) self._base_url = base_url self._connection = None + self._secure = secure def _decode_location(self, location): """Return splitted configuration data from location.""" splitted_url = urlparse(location) config = {} - if splitted_url.scheme not in ("ftp", "aftp"): + if splitted_url.scheme not in ("ftp", "aftp", "ftps"): raise ImproperlyConfigured("FTPStorage works only with FTP protocol!") if splitted_url.hostname == "": raise ImproperlyConfigured("You must at least provide hostname!") @@ -84,11 +87,13 @@ def _start_connection(self): # Real reconnect if self._connection is None: - ftp = ftplib.FTP() + ftp = ftplib.FTP_TLS() if self._secure else ftplib.FTP() ftp.encoding = self.encoding try: ftp.connect(self._config["host"], self._config["port"]) ftp.login(self._config["user"], self._config["passwd"]) + if self._secure: + ftp.prot_p() if self._config["active"]: ftp.set_pasv(False) if self._config["path"] != "": diff --git a/tests/test_ftp.py b/tests/test_ftp.py index e4fed1c7..209b0670 100644 --- a/tests/test_ftp.py +++ b/tests/test_ftp.py @@ -14,6 +14,9 @@ URL = "ftp://{user}:{passwd}@{host}:{port}/".format( user=USER, passwd=PASSWORD, host=HOST, port=PORT ) +URL_TLS = "ftps://{user}:{passwd}@{host}:{port}/".format( + user=USER, passwd=PASSWORD, host=HOST, port=PORT +) LIST_FIXTURE = """drwxr-xr-x 2 ftp nogroup 4096 Jul 27 09:46 dir -rw-r--r-- 1 ftp nogroup 1024 Jul 27 09:45 fi @@ -239,3 +242,208 @@ def test_close(self, mock_ftp, mock_storage): file_.is_dirty = True file_.read() file_.close() + + +class FTPTLSTest(TestCase): + def setUp(self): + self.storage = ftp.FTPStorage(location=URL_TLS, secure=True) + + def test_init_no_location(self): + with self.assertRaises(ImproperlyConfigured): + ftp.FTPStorage(secure=True) + + @patch("storages.backends.ftp.setting", return_value=URL_TLS) + def test_init_location_from_setting(self, mock_setting): + storage = ftp.FTPStorage(secure=True) + self.assertTrue(mock_setting.called) + self.assertEqual(storage.location, URL_TLS) + + def test_decode_location(self): + config = self.storage._decode_location(URL_TLS) + wanted_config = { + "passwd": "b@r", + "host": "localhost", + "user": "foo", + "active": False, + "path": "/", + "port": 2121, + } + self.assertEqual(config, wanted_config) + + def test_decode_location_error(self): + with self.assertRaises(ImproperlyConfigured): + self.storage._decode_location("foo") + with self.assertRaises(ImproperlyConfigured): + self.storage._decode_location("http://foo.pt") + # TODO: Cannot not provide a port + # with self.assertRaises(ImproperlyConfigured): + # self.storage._decode_location('ftp://') + + @patch("ftplib.FTP_TLS") + def test_start_connection(self, mock_ftp): + self.storage._start_connection() + self.assertIsNotNone(self.storage._connection) + # Start active + storage = ftp.FTPStorage(location=URL_TLS, secure=True) + storage._start_connection() + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.side_effect": IOError()}) + def test_start_connection_timeout(self, mock_ftp): + self.storage._start_connection() + self.assertIsNotNone(self.storage._connection) + + @patch("ftplib.FTP_TLS", **{"return_value.connect.side_effect": IOError()}) + def test_start_connection_error(self, mock_ftp): + with self.assertRaises(ftp.FTPStorageException): + self.storage._start_connection() + + @patch("ftplib.FTP_TLS", **{"return_value.quit.return_value": None}) + def test_disconnect(self, mock_ftp_quit): + self.storage._start_connection() + self.storage.disconnect() + self.assertIsNone(self.storage._connection) + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) + def test_mkremdirs(self, mock_ftp): + self.storage._start_connection() + self.storage._mkremdirs("foo/bar") + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) + def test_mkremdirs_n_subdirectories(self, mock_ftp): + self.storage._start_connection() + self.storage._mkremdirs("foo/bar/null") + + @patch( + "ftplib.FTP_TLS", + **{ + "return_value.pwd.return_value": "foo", + "return_value.storbinary.return_value": None, + }, + ) + def test_put_file(self, mock_ftp): + self.storage._start_connection() + self.storage._put_file("foo", File(io.BytesIO(b"foo"), "foo")) + + @patch( + "ftplib.FTP_TLS", + **{ + "return_value.pwd.return_value": "foo", + "return_value.storbinary.side_effect": IOError(), + }, + ) + def test_put_file_error(self, mock_ftp): + self.storage._start_connection() + with self.assertRaises(ftp.FTPStorageException): + self.storage._put_file("foo", File(io.BytesIO(b"foo"), "foo")) + + def test_open(self): + remote_file = self.storage._open("foo") + self.assertIsInstance(remote_file, ftp.FTPStorageFile) + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) + def test_read(self, mock_ftp): + self.storage._start_connection() + self.storage._read("foo") + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.side_effect": IOError()}) + def test_read2(self, mock_ftp): + self.storage._start_connection() + with self.assertRaises(ftp.FTPStorageException): + self.storage._read("foo") + + @patch( + "ftplib.FTP_TLS", + **{ + "return_value.pwd.return_value": "foo", + "return_value.storbinary.return_value": None, + }, + ) + def test_save(self, mock_ftp): + self.storage._save("foo", File(io.BytesIO(b"foo"), "foo")) + + @patch("ftplib.FTP_TLS", **{"return_value.retrlines": list_retrlines}) + def test_listdir(self, mock_retrlines): + dirs, files = self.storage.listdir("/") + self.assertEqual(len(dirs), 1) + self.assertEqual(dirs, ["dir"]) + self.assertEqual(len(files), 2) + self.assertEqual(sorted(files), sorted(["fi", "fi2"])) + + @patch("ftplib.FTP_TLS", **{"return_value.retrlines.side_effect": IOError()}) + def test_listdir_error(self, mock_ftp): + with self.assertRaises(ftp.FTPStorageException): + self.storage.listdir("/") + + @patch("ftplib.FTP_TLS", **{"return_value.nlst.return_value": ["foo", "foo2"]}) + def test_exists(self, mock_ftp): + self.assertTrue(self.storage.exists("foo")) + self.assertFalse(self.storage.exists("bar")) + + @patch("ftplib.FTP_TLS", **{"return_value.nlst.side_effect": IOError()}) + def test_exists_error(self, mock_ftp): + with self.assertRaises(ftp.FTPStorageException): + self.storage.exists("foo") + + @patch( + "ftplib.FTP_TLS", + **{ + "return_value.delete.return_value": None, + "return_value.nlst.return_value": ["foo", "foo2"], + }, + ) + def test_delete(self, mock_ftp): + self.storage.delete("foo") + self.assertTrue(mock_ftp.return_value.delete.called) + + @patch("ftplib.FTP_TLS", **{"return_value.retrlines": list_retrlines}) + def test_size(self, mock_ftp): + self.assertEqual(1024, self.storage.size("fi")) + self.assertEqual(2048, self.storage.size("fi2")) + self.assertEqual(0, self.storage.size("bar")) + + @patch("ftplib.FTP_TLS", **{"return_value.retrlines.side_effect": IOError()}) + def test_size_error(self, mock_ftp): + self.assertEqual(0, self.storage.size("foo")) + + def test_url(self): + with self.assertRaises(ValueError): + self.storage._base_url = None + self.storage.url("foo") + self.storage = ftp.FTPStorage(location=URL_TLS, base_url="http://foo.bar/", secure=True) + self.assertEqual("http://foo.bar/foo", self.storage.url("foo")) + + +class FTPTLSStorageFileTest(TestCase): + def setUp(self): + self.storage = ftp.FTPStorage(location=URL_TLS, secure=True) + + @patch("ftplib.FTP_TLS", **{"return_value.retrlines": list_retrlines}) + def test_size(self, mock_ftp): + file_ = ftp.FTPStorageFile("fi", self.storage, "wb") + self.assertEqual(file_.size, 1024) + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) + @patch("storages.backends.ftp.FTPStorage._read", return_value=io.BytesIO(b"foo")) + def test_readlines(self, mock_ftp, mock_storage): + file_ = ftp.FTPStorageFile("fi", self.storage, "wb") + self.assertEqual([b"foo"], file_.readlines()) + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) + @patch("storages.backends.ftp.FTPStorage._read", return_value=io.BytesIO(b"foo")) + def test_read(self, mock_ftp, mock_storage): + file_ = ftp.FTPStorageFile("fi", self.storage, "wb") + self.assertEqual(b"foo", file_.read()) + + def test_write(self): + file_ = ftp.FTPStorageFile("fi", self.storage, "wb") + file_.write(b"foo") + file_.seek(0) + self.assertEqual(file_.file.read(), b"foo") + + @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) + @patch("storages.backends.ftp.FTPStorage._read", return_value=io.BytesIO(b"foo")) + def test_close(self, mock_ftp, mock_storage): + file_ = ftp.FTPStorageFile("fi", self.storage, "wb") + file_.is_dirty = True + file_.read() + file_.close() From 099196a250e943b9ec572b0d52373d4259314672 Mon Sep 17 00:00:00 2001 From: fazledyn-or Date: Mon, 9 Oct 2023 13:33:02 +0600 Subject: [PATCH 2/4] Refactor: Updated the code & test for FTP_TLS --- storages/backends/ftp.py | 16 ++-- tests/test_ftp.py | 171 ++------------------------------------- 2 files changed, 17 insertions(+), 170 deletions(-) diff --git a/storages/backends/ftp.py b/storages/backends/ftp.py index 8c559459..05f639f1 100644 --- a/storages/backends/ftp.py +++ b/storages/backends/ftp.py @@ -11,8 +11,7 @@ # In models.py you can write: # from FTPStorage import FTPStorage # fs = FTPStorage() -# For a TLS configuration, you must use it below: -# fs = FTPStorage(secure=True) +# For a TLS configuration, you must use 'ftps' protocol # class FTPTest(models.Model): # file = models.FileField(upload_to='a/b/c/', storage=fs) @@ -39,7 +38,7 @@ class FTPStorageException(Exception): class FTPStorage(Storage): """FTP Storage class for Django pluggable storage system.""" - def __init__(self, location=None, base_url=None, encoding=None, secure=False): + def __init__(self, location=None, base_url=None, encoding=None): location = location or setting("FTP_STORAGE_LOCATION") if location is None: raise ImproperlyConfigured( @@ -53,7 +52,6 @@ def __init__(self, location=None, base_url=None, encoding=None, secure=False): self._config = self._decode_location(location) self._base_url = base_url self._connection = None - self._secure = secure def _decode_location(self, location): """Return splitted configuration data from location.""" @@ -69,6 +67,12 @@ def _decode_location(self, location): config["active"] = True else: config["active"] = False + + if splitted_url.scheme == "ftps": + config["secure"] = True + else: + config["secure"] = False + config["path"] = splitted_url.path config["host"] = splitted_url.hostname config["user"] = splitted_url.username @@ -87,12 +91,12 @@ def _start_connection(self): # Real reconnect if self._connection is None: - ftp = ftplib.FTP_TLS() if self._secure else ftplib.FTP() + ftp = ftplib.FTP_TLS() if self._config["secure"] else ftplib.FTP() ftp.encoding = self.encoding try: ftp.connect(self._config["host"], self._config["port"]) ftp.login(self._config["user"], self._config["passwd"]) - if self._secure: + if self._config["secure"]: ftp.prot_p() if self._config["active"]: ftp.set_pasv(False) diff --git a/tests/test_ftp.py b/tests/test_ftp.py index 209b0670..5baf977e 100644 --- a/tests/test_ftp.py +++ b/tests/test_ftp.py @@ -51,6 +51,7 @@ def test_decode_location(self): "active": False, "path": "/", "port": 2121, + "secure": False } self.assertEqual(config, wanted_config) # Test active FTP @@ -62,6 +63,7 @@ def test_decode_location(self): "active": True, "path": "/", "port": 2121, + "secure": False } self.assertEqual(config, wanted_config) @@ -246,15 +248,15 @@ def test_close(self, mock_ftp, mock_storage): class FTPTLSTest(TestCase): def setUp(self): - self.storage = ftp.FTPStorage(location=URL_TLS, secure=True) + self.storage = ftp.FTPStorage(location=URL_TLS) def test_init_no_location(self): with self.assertRaises(ImproperlyConfigured): - ftp.FTPStorage(secure=True) + ftp.FTPStorage() @patch("storages.backends.ftp.setting", return_value=URL_TLS) def test_init_location_from_setting(self, mock_setting): - storage = ftp.FTPStorage(secure=True) + storage = ftp.FTPStorage() self.assertTrue(mock_setting.called) self.assertEqual(storage.location, URL_TLS) @@ -267,6 +269,7 @@ def test_decode_location(self): "active": False, "path": "/", "port": 2121, + "secure": True } self.assertEqual(config, wanted_config) @@ -284,166 +287,6 @@ def test_start_connection(self, mock_ftp): self.storage._start_connection() self.assertIsNotNone(self.storage._connection) # Start active - storage = ftp.FTPStorage(location=URL_TLS, secure=True) + storage = ftp.FTPStorage(location=URL_TLS) storage._start_connection() - @patch("ftplib.FTP_TLS", **{"return_value.pwd.side_effect": IOError()}) - def test_start_connection_timeout(self, mock_ftp): - self.storage._start_connection() - self.assertIsNotNone(self.storage._connection) - - @patch("ftplib.FTP_TLS", **{"return_value.connect.side_effect": IOError()}) - def test_start_connection_error(self, mock_ftp): - with self.assertRaises(ftp.FTPStorageException): - self.storage._start_connection() - - @patch("ftplib.FTP_TLS", **{"return_value.quit.return_value": None}) - def test_disconnect(self, mock_ftp_quit): - self.storage._start_connection() - self.storage.disconnect() - self.assertIsNone(self.storage._connection) - - @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) - def test_mkremdirs(self, mock_ftp): - self.storage._start_connection() - self.storage._mkremdirs("foo/bar") - - @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) - def test_mkremdirs_n_subdirectories(self, mock_ftp): - self.storage._start_connection() - self.storage._mkremdirs("foo/bar/null") - - @patch( - "ftplib.FTP_TLS", - **{ - "return_value.pwd.return_value": "foo", - "return_value.storbinary.return_value": None, - }, - ) - def test_put_file(self, mock_ftp): - self.storage._start_connection() - self.storage._put_file("foo", File(io.BytesIO(b"foo"), "foo")) - - @patch( - "ftplib.FTP_TLS", - **{ - "return_value.pwd.return_value": "foo", - "return_value.storbinary.side_effect": IOError(), - }, - ) - def test_put_file_error(self, mock_ftp): - self.storage._start_connection() - with self.assertRaises(ftp.FTPStorageException): - self.storage._put_file("foo", File(io.BytesIO(b"foo"), "foo")) - - def test_open(self): - remote_file = self.storage._open("foo") - self.assertIsInstance(remote_file, ftp.FTPStorageFile) - - @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) - def test_read(self, mock_ftp): - self.storage._start_connection() - self.storage._read("foo") - - @patch("ftplib.FTP_TLS", **{"return_value.pwd.side_effect": IOError()}) - def test_read2(self, mock_ftp): - self.storage._start_connection() - with self.assertRaises(ftp.FTPStorageException): - self.storage._read("foo") - - @patch( - "ftplib.FTP_TLS", - **{ - "return_value.pwd.return_value": "foo", - "return_value.storbinary.return_value": None, - }, - ) - def test_save(self, mock_ftp): - self.storage._save("foo", File(io.BytesIO(b"foo"), "foo")) - - @patch("ftplib.FTP_TLS", **{"return_value.retrlines": list_retrlines}) - def test_listdir(self, mock_retrlines): - dirs, files = self.storage.listdir("/") - self.assertEqual(len(dirs), 1) - self.assertEqual(dirs, ["dir"]) - self.assertEqual(len(files), 2) - self.assertEqual(sorted(files), sorted(["fi", "fi2"])) - - @patch("ftplib.FTP_TLS", **{"return_value.retrlines.side_effect": IOError()}) - def test_listdir_error(self, mock_ftp): - with self.assertRaises(ftp.FTPStorageException): - self.storage.listdir("/") - - @patch("ftplib.FTP_TLS", **{"return_value.nlst.return_value": ["foo", "foo2"]}) - def test_exists(self, mock_ftp): - self.assertTrue(self.storage.exists("foo")) - self.assertFalse(self.storage.exists("bar")) - - @patch("ftplib.FTP_TLS", **{"return_value.nlst.side_effect": IOError()}) - def test_exists_error(self, mock_ftp): - with self.assertRaises(ftp.FTPStorageException): - self.storage.exists("foo") - - @patch( - "ftplib.FTP_TLS", - **{ - "return_value.delete.return_value": None, - "return_value.nlst.return_value": ["foo", "foo2"], - }, - ) - def test_delete(self, mock_ftp): - self.storage.delete("foo") - self.assertTrue(mock_ftp.return_value.delete.called) - - @patch("ftplib.FTP_TLS", **{"return_value.retrlines": list_retrlines}) - def test_size(self, mock_ftp): - self.assertEqual(1024, self.storage.size("fi")) - self.assertEqual(2048, self.storage.size("fi2")) - self.assertEqual(0, self.storage.size("bar")) - - @patch("ftplib.FTP_TLS", **{"return_value.retrlines.side_effect": IOError()}) - def test_size_error(self, mock_ftp): - self.assertEqual(0, self.storage.size("foo")) - - def test_url(self): - with self.assertRaises(ValueError): - self.storage._base_url = None - self.storage.url("foo") - self.storage = ftp.FTPStorage(location=URL_TLS, base_url="http://foo.bar/", secure=True) - self.assertEqual("http://foo.bar/foo", self.storage.url("foo")) - - -class FTPTLSStorageFileTest(TestCase): - def setUp(self): - self.storage = ftp.FTPStorage(location=URL_TLS, secure=True) - - @patch("ftplib.FTP_TLS", **{"return_value.retrlines": list_retrlines}) - def test_size(self, mock_ftp): - file_ = ftp.FTPStorageFile("fi", self.storage, "wb") - self.assertEqual(file_.size, 1024) - - @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) - @patch("storages.backends.ftp.FTPStorage._read", return_value=io.BytesIO(b"foo")) - def test_readlines(self, mock_ftp, mock_storage): - file_ = ftp.FTPStorageFile("fi", self.storage, "wb") - self.assertEqual([b"foo"], file_.readlines()) - - @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) - @patch("storages.backends.ftp.FTPStorage._read", return_value=io.BytesIO(b"foo")) - def test_read(self, mock_ftp, mock_storage): - file_ = ftp.FTPStorageFile("fi", self.storage, "wb") - self.assertEqual(b"foo", file_.read()) - - def test_write(self): - file_ = ftp.FTPStorageFile("fi", self.storage, "wb") - file_.write(b"foo") - file_.seek(0) - self.assertEqual(file_.file.read(), b"foo") - - @patch("ftplib.FTP_TLS", **{"return_value.pwd.return_value": "foo"}) - @patch("storages.backends.ftp.FTPStorage._read", return_value=io.BytesIO(b"foo")) - def test_close(self, mock_ftp, mock_storage): - file_ = ftp.FTPStorageFile("fi", self.storage, "wb") - file_.is_dirty = True - file_.read() - file_.close() From 38b338ab52a8042327af2520d92122477f5cc194 Mon Sep 17 00:00:00 2001 From: fazledyn-or Date: Tue, 17 Oct 2023 10:22:30 +0600 Subject: [PATCH 3/4] Removed unnecessary tests for `tests/test_ftp.py` for FTP_TLS --- tests/test_ftp.py | 29 +++-------------------------- 1 file changed, 3 insertions(+), 26 deletions(-) diff --git a/tests/test_ftp.py b/tests/test_ftp.py index 5baf977e..3d25f182 100644 --- a/tests/test_ftp.py +++ b/tests/test_ftp.py @@ -250,18 +250,7 @@ class FTPTLSTest(TestCase): def setUp(self): self.storage = ftp.FTPStorage(location=URL_TLS) - def test_init_no_location(self): - with self.assertRaises(ImproperlyConfigured): - ftp.FTPStorage() - - @patch("storages.backends.ftp.setting", return_value=URL_TLS) - def test_init_location_from_setting(self, mock_setting): - storage = ftp.FTPStorage() - self.assertTrue(mock_setting.called) - self.assertEqual(storage.location, URL_TLS) - def test_decode_location(self): - config = self.storage._decode_location(URL_TLS) wanted_config = { "passwd": "b@r", "host": "localhost", @@ -271,22 +260,10 @@ def test_decode_location(self): "port": 2121, "secure": True } - self.assertEqual(config, wanted_config) - - def test_decode_location_error(self): - with self.assertRaises(ImproperlyConfigured): - self.storage._decode_location("foo") - with self.assertRaises(ImproperlyConfigured): - self.storage._decode_location("http://foo.pt") - # TODO: Cannot not provide a port - # with self.assertRaises(ImproperlyConfigured): - # self.storage._decode_location('ftp://') + self.assertEqual(self.storage._config, wanted_config) @patch("ftplib.FTP_TLS") - def test_start_connection(self, mock_ftp): + def test_start_connection_calls_prot_p(self, mock_ftp): self.storage._start_connection() - self.assertIsNotNone(self.storage._connection) - # Start active - storage = ftp.FTPStorage(location=URL_TLS) - storage._start_connection() + self.storage._connection.prot_p.assert_called_once() From 34a8d19a33108813e81548a01f98325ea71e378e Mon Sep 17 00:00:00 2001 From: fazledyn-or Date: Wed, 18 Oct 2023 10:20:35 +0600 Subject: [PATCH 4/4] Formatted `test_ftp.py` using black --- tests/test_ftp.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/tests/test_ftp.py b/tests/test_ftp.py index 3d25f182..d472f96e 100644 --- a/tests/test_ftp.py +++ b/tests/test_ftp.py @@ -51,7 +51,7 @@ def test_decode_location(self): "active": False, "path": "/", "port": 2121, - "secure": False + "secure": False, } self.assertEqual(config, wanted_config) # Test active FTP @@ -63,7 +63,7 @@ def test_decode_location(self): "active": True, "path": "/", "port": 2121, - "secure": False + "secure": False, } self.assertEqual(config, wanted_config) @@ -258,7 +258,7 @@ def test_decode_location(self): "active": False, "path": "/", "port": 2121, - "secure": True + "secure": True, } self.assertEqual(self.storage._config, wanted_config) @@ -266,4 +266,3 @@ def test_decode_location(self): def test_start_connection_calls_prot_p(self, mock_ftp): self.storage._start_connection() self.storage._connection.prot_p.assert_called_once() -