From 286b4c5a442c8039f22a892c6cb634f4d20e6a77 Mon Sep 17 00:00:00 2001 From: gecong1973 Date: Sun, 23 Dec 2018 18:18:32 -0800 Subject: [PATCH] Add ftp write_backup unit test The patch add ftp write_backup unit test Change-Id: I736a37418bc54cb1d34b1fa667e748115015d9b0 --- freezer/tests/unit/storages/test_ftp.py | 152 +++++++++++------------- 1 file changed, 70 insertions(+), 82 deletions(-) diff --git a/freezer/tests/unit/storages/test_ftp.py b/freezer/tests/unit/storages/test_ftp.py index 7ad3fb4e..12a8cd72 100644 --- a/freezer/tests/unit/storages/test_ftp.py +++ b/freezer/tests/unit/storages/test_ftp.py @@ -13,6 +13,7 @@ # limitations under the License. import shutil +import sys import tempfile import unittest @@ -107,19 +108,23 @@ class FtpStorageTestCase(unittest.TestCase): self.ftp_test_file_name = None self.ftp_test_file_dir = None + def create_ftpstorage_obj(self): + obj = ftp.FtpStorage( + storage_path=self.ftp_opt.ftp_storage_path, + remote_pwd=self.ftp_opt.ftp_remote_pwd, + remote_username=self.ftp_opt.ftp_remote_username, + remote_ip=self.ftp_opt.ftp_remote_ip, + port=self.ftp_opt.ftp_port, + max_segment_size=self.ftp_opt.ftp_max_segment_size) + return obj + @patch('ftplib.FTP') def test_init_fail_raise_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value seffect = mock.Mock(side_effect=Exception('create ftp failed error')) mock_ftp.raiseError.side_effect = seffect with self.assertRaises(Exception) as cm: # noqa - ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + self.create_ftpstorage_obj() the_exception = cm.exception self.assertIn('create ftp failed error', str(the_exception)) @@ -127,13 +132,7 @@ class FtpStorageTestCase(unittest.TestCase): @patch('ftplib.FTP') def test_init_ok_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value - ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + self.create_ftpstorage_obj() self.assertTrue(mock_ftp.set_pasv.called) self.assertTrue(mock_ftp.connect.called) self.assertTrue(mock_ftp.login.called) @@ -149,26 +148,14 @@ class FtpStorageTestCase(unittest.TestCase): def test_create_tempdir_FtpStorage(self, mock_ftp_constructor, mock_tempfile_constructor): mock_tempfile = mock_tempfile_constructor.return_value - ftp_obj = ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + ftp_obj = self.create_ftpstorage_obj() tmp = ftp_obj._create_tempdir() self.assertEqual(tmp, mock_tempfile) @patch('ftplib.FTP') def test_listdir_ok_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value - ftpobj = ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + ftpobj = self.create_ftpstorage_obj() test_dir = '/home/test' filelist = ['test1.py', 'test2.py', 'readme.txt', 'ftp.sh'] mock_ftp.nlst.return_value = filelist @@ -180,13 +167,7 @@ class FtpStorageTestCase(unittest.TestCase): @patch('ftplib.FTP') def test_listdir_fail_raise_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value - ftpobj = ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + ftpobj = self.create_ftpstorage_obj() test_dir = '/home/test' seffect = mock.Mock(side_effect=Exception()) mock_ftp.raiseError.side_effect = seffect @@ -196,13 +177,7 @@ class FtpStorageTestCase(unittest.TestCase): @patch('ftplib.FTP') def test_putfile_ok_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value - ftpobj = ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + ftpobj = self.create_ftpstorage_obj() self.create_ftp_test_file() frompath = self.ftp_test_file_dir + "/" + self.ftp_test_file_name topath = '/home/to' @@ -214,13 +189,7 @@ class FtpStorageTestCase(unittest.TestCase): @patch('ftplib.FTP') def test_getfile_ok_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value - ftpobj = ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + ftpobj = self.create_ftpstorage_obj() self.create_ftp_test_file() topath = self.ftp_test_file_dir + "/" + self.ftp_test_file_name frompath = '/home/from' @@ -232,13 +201,7 @@ class FtpStorageTestCase(unittest.TestCase): @patch('ftplib.FTP') def test_create_dirs_ok_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value - ftpobj = ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + ftpobj = self.create_ftpstorage_obj() path = '/' ftpobj.create_dirs(path) mock_ftp.cwd.assert_called_with(path) @@ -249,18 +212,47 @@ class FtpStorageTestCase(unittest.TestCase): @patch('ftplib.FTP') def test_rmtree_ok_FtpStorage(self, mock_ftp_constructor): mock_ftp = mock_ftp_constructor.return_value - ftpobj = ftp.FtpStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size) + ftpobj = self.create_ftpstorage_obj() path = '/home/tecs' ftpobj.rmtree(path) self.assertTrue(mock_ftp.dir.called) self.assertTrue(mock_ftp.rmd.called) + @unittest.skipIf(sys.version_info.major == 3, + 'Not supported on python v 3.x') + @patch("freezer.storage.ftp.BaseFtpStorage.create_dirs") + @patch("freezer.storage.ftp.BaseFtpStorage.put_file") + @patch("__builtin__.open") + @patch('ftplib.FTP') + def test_write_backup_FtpStorage(self, mock_ftp_constructor, + mock_open, + mock_put_file, + mock_create_dirs): + ftpobj = self.create_ftpstorage_obj() + + rich_queue = mock.MagicMock() + rich_queue.get_messages = mock.MagicMock() + rich_queue.get_messages.return_value = ['1'] + + backup = mock.MagicMock() + backup.copy = mock.MagicMock() + backup.copy.return_value = backup + + path = 'fakepath' + backup.data_path = mock.MagicMock() + backup.data_path.return_value = path + + b_file = mock.MagicMock() + b_file.write = mock.MagicMock() + + mock_open = mock.MagicMock() + mock_open.return_value = b_file + + ftpobj.write_backup(rich_queue=rich_queue, + backup=backup) + self.assertTrue(mock_create_dirs.called) + self.assertTrue(mock_put_file.called) + class FtpsStorageTestCase(unittest.TestCase): @@ -275,25 +267,8 @@ class FtpsStorageTestCase(unittest.TestCase): self.ftp_opt.ftp_keyfile = '/just/key.pem' self.ftp_opt.ftp_certfile = '/just/cert.pem' - def test_init_fail_FtpsStorage(self): - with self.assertRaises(Exception) as cm: # noqa - ftp.FtpsStorage( - storage_path=self.ftp_opt.ftp_storage_path, - remote_pwd=self.ftp_opt.ftp_remote_pwd, - remote_username=self.ftp_opt.ftp_remote_username, - remote_ip=self.ftp_opt.ftp_remote_ip, - port=self.ftp_opt.ftp_port, - max_segment_size=self.ftp_opt.ftp_max_segment_size, - keyfile=self.ftp_opt.ftp_keyfile, - certfile=self.ftp_opt.ftp_certfile) - the_exception = cm.exception - self.assertIn('create ftp failed error', - str(the_exception)) - - @patch('ftplib.FTP_TLS') - def test_init_ok_FtpsStorage(self, mock_ftp_constructor): - mock_ftp = mock_ftp_constructor.return_value - ftp.FtpsStorage( + def create_ftpsStorage_obj(self): + obj = ftp.FtpsStorage( storage_path=self.ftp_opt.ftp_storage_path, remote_pwd=self.ftp_opt.ftp_remote_pwd, remote_username=self.ftp_opt.ftp_remote_username, @@ -302,6 +277,19 @@ class FtpsStorageTestCase(unittest.TestCase): max_segment_size=self.ftp_opt.ftp_max_segment_size, keyfile=self.ftp_opt.ftp_keyfile, certfile=self.ftp_opt.ftp_certfile) + return obj + + def test_init_fail_FtpsStorage(self): + with self.assertRaises(Exception) as cm: # noqa + self.create_ftpsStorage_obj() + the_exception = cm.exception + self.assertIn('create ftp failed error', + str(the_exception)) + + @patch('ftplib.FTP_TLS') + def test_init_ok_FtpsStorage(self, mock_ftp_constructor): + mock_ftp = mock_ftp_constructor.return_value + self.create_ftpsStorage_obj() self.assertTrue(mock_ftp.set_pasv.called) self.assertTrue(mock_ftp.connect.called) self.assertTrue(mock_ftp.login.called)