Add ftp write_backup unit test

The patch add ftp write_backup unit test

Change-Id: I736a37418bc54cb1d34b1fa667e748115015d9b0
This commit is contained in:
gecong1973 2018-12-23 18:18:32 -08:00
parent 5185860441
commit 286b4c5a44
1 changed files with 70 additions and 82 deletions

View File

@ -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)