Bug 1133685 - Added tests for TransferMixin. r=pmoore
--- a/mozharness/base/transfer.py
+++ b/mozharness/base/transfer.py
@@ -32,17 +32,22 @@ class TransferMixin(object):
rsync_options=None,
error_level=ERROR,
create_remote_directory=True,
):
"""
Create a remote directory and upload the contents of
a local directory to it via rsync+ssh.
- Return None on success, not None on failure.
+ Returns:
+ None: on success
+ -1: if local_path is not a directory
+ -2: if the remote_directory cannot be created
+ (it only makes sense if create_remote_directory is True)
+ -3: rsync fails to copy to the remote directory
"""
dirs = self.query_abs_dirs()
self.info("Uploading the contents of %s to %s:%s" % (local_path, remote_host, remote_path))
rsync = self.query_exe("rsync")
ssh = self.query_exe("ssh")
if rsync_options is None:
rsync_options = ['-azv']
if not os.path.isdir(local_path):
@@ -73,20 +78,22 @@ class TransferMixin(object):
return -3
def rsync_download_directory(self, ssh_key, ssh_user, remote_host,
remote_path, local_path,
rsync_options=None,
error_level=ERROR,
):
"""
- Create a remote directory and upload the contents of
- a local directory to it via rsync+ssh.
+ rsync+ssh the content of a remote directory to local_path
- Return None on success, not None on failure.
+ Returns:
+ None: on success
+ -1: if local_path is not a directory
+ -3: rsync fails to download from the remote directory
"""
self.info("Downloading the contents of %s:%s to %s" % (remote_host, remote_path, local_path))
rsync = self.query_exe("rsync")
ssh = self.query_exe("ssh")
if rsync_options is None:
rsync_options = ['-azv']
if not os.path.isdir(local_path):
self.log("%s isn't a directory!" % local_path,
new file mode 100644
--- /dev/null
+++ b/test/test_base_transfer.py
@@ -0,0 +1,127 @@
+import unittest
+import mock
+
+from mozharness.base.transfer import TransferMixin
+
+
+class GoodMockMixin(object):
+ def query_abs_dirs(self):
+ return {'abs_work_dir': ''}
+
+ def query_exe(self, exe):
+ return exe
+
+ def info(self, msg):
+ pass
+
+ def log(self, msg, level):
+ pass
+
+ def run_command(*args, **kwargs):
+ return 0
+
+
+class BadMockMixin(GoodMockMixin):
+ def run_command(*args, **kwargs):
+ return 1
+
+
+class GoodTransferMixin(TransferMixin, GoodMockMixin):
+ pass
+
+
+class BadTransferMixin(TransferMixin, BadMockMixin):
+ pass
+
+
+class TestTranferMixin(unittest.TestCase):
+ @mock.patch('mozharness.base.transfer.os')
+ def test_rsync_upload_dir_not_a_dir(self, os_mock):
+ # simulates upload dir but dir is a file
+ os_mock.path.isdir.return_value = False
+ tm = GoodTransferMixin()
+ self.assertEqual(tm.rsync_upload_directory(
+ local_path='',
+ ssh_key='my ssh key',
+ ssh_user='my ssh user',
+ remote_host='remote host',
+ remote_path='remote path',), -1)
+
+ @mock.patch('mozharness.base.transfer.os')
+ def test_rsync_upload_fails_create_remote_dir(self, os_mock):
+ # we cannot create the remote directory
+ os_mock.path.isdir.return_value = True
+ tm = BadTransferMixin()
+ self.assertEqual(tm.rsync_upload_directory(
+ local_path='',
+ ssh_key='my ssh key',
+ ssh_user='my ssh user',
+ remote_host='remote host',
+ remote_path='remote path',
+ create_remote_directory=True), -2)
+
+ @mock.patch('mozharness.base.transfer.os')
+ def test_rsync_upload_fails_do_not_create_remote_dir(self, os_mock):
+ # upload fails, remote directory is not created
+ os_mock.path.isdir.return_value = True
+ tm = BadTransferMixin()
+ self.assertEqual(tm.rsync_upload_directory(
+ local_path='',
+ ssh_key='my ssh key',
+ ssh_user='my ssh user',
+ remote_host='remote host',
+ remote_path='remote path',
+ create_remote_directory=False), -3)
+
+ @mock.patch('mozharness.base.transfer.os')
+ def test_rsync_upload(self, os_mock):
+ # simulates an upload with no errors
+ os_mock.path.isdir.return_value = True
+ tm = GoodTransferMixin()
+ self.assertEqual(tm.rsync_upload_directory(
+ local_path='',
+ ssh_key='my ssh key',
+ ssh_user='my ssh user',
+ remote_host='remote host',
+ remote_path='remote path',
+ create_remote_directory=False), None)
+
+ @mock.patch('mozharness.base.transfer.os')
+ def test_rsync_download_in_not_a_dir(self, os_mock):
+ # local path is not a directory
+ os_mock.path.isdir.return_value = False
+ tm = GoodTransferMixin()
+ self.assertEqual(tm.rsync_download_directory(
+ local_path='',
+ ssh_key='my ssh key',
+ ssh_user='my ssh user',
+ remote_host='remote host',
+ remote_path='remote path',), -1)
+
+ @mock.patch('mozharness.base.transfer.os')
+ def test_rsync_download(self, os_mock):
+ # successful rsync
+ os_mock.path.isdir.return_value = True
+ tm = GoodTransferMixin()
+ self.assertEqual(tm.rsync_download_directory(
+ local_path='',
+ ssh_key='my ssh key',
+ ssh_user='my ssh user',
+ remote_host='remote host',
+ remote_path='remote path',), None)
+
+ @mock.patch('mozharness.base.transfer.os')
+ def test_rsync_download_fail(self, os_mock):
+ # ops download has failed
+ os_mock.path.isdir.return_value = True
+ tm = BadTransferMixin()
+ self.assertEqual(tm.rsync_download_directory(
+ local_path='',
+ ssh_key='my ssh key',
+ ssh_user='my ssh user',
+ remote_host='remote host',
+ remote_path='remote path',), -3)
+
+
+if __name__ == '__main__':
+ unittest.main()