Add ftp write_backup unit test
The patch add ftp write_backup unit test Change-Id: I736a37418bc54cb1d34b1fa667e748115015d9b0
This commit is contained in:
parent
5185860441
commit
286b4c5a44
|
@ -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)
|
||||
|
|
Loading…
Reference in New Issue