from ftplib import error_perm from unittest.mock import patch from vector_utils.connections import ftp from vector_utils.connections.connection_info import ConnectionInfo import os import pytest @pytest.fixture def valid_conn_obj(ftp_server): return ConnectionInfo({ 'authenticate_type': 'password', 'connection_type': 'ftp', 'domain_name': 'localhost', 'port': ftp_server.server_port, 'user_name': ftp_server.username, 'password': ftp_server.password }) @pytest.fixture def invalid_conn_obj(ftp_server): return ConnectionInfo({ 'authenticate_type': 'password', 'connection_type': 'ftp', 'domain_name': 'localhost', 'port': ftp_server.server_port, 'user_name': 'wrong', 'password': 'wrong' }) @pytest.fixture def local_test_files(tmpdir): file_names = ['file_1.txt', 'file_2.txt', 'file_3.txt'] for file_name in file_names: f = tmpdir.join(file_name) f.write('content') return ['{}/{}'.format(tmpdir, x) for x in file_names] @pytest.fixture def uploaded_file_path(tmpdir, ftp_server): """This fixture puts a file to ftp_server for a test.""" filename = 'test_file.txt' f = tmpdir.join(filename) f.write('content') file_local_path = os.path.join(tmpdir, filename) remote_filename = 'download_test_file.txt' ftp_server.put_files({ 'src': file_local_path, 'dest': remote_filename}) os.remove(file_local_path) yield remote_filename ftp_server.reset_tmp_dirs() def mock_ls(): def retrlines(self, cmd, callback): sample_ls = [ 'drwxrwxrwx 1 user group ' '0 Jul 26 2018 digital_orchard', 'drwxrwxrwx 1 user group ' '187 Mar 26 2018 DMS_214', 'drwxrwxrwx 1 user group ' '15 Mar 24 2014 DMS_287', 'drwxrwxrwx 1 user group ' '0 Jan 9 2017 DMS_453', 'drwxrwxrwx 1 user group ' '0 Dec 9 2016 DMS_484'] for line in sample_ls: callback(line) return retrlines def mock_ls2(): def retrlines(self, cmd, callback): sample_ls = [ '08-26-19 01:43PM 194491040796', '08-26-19 01:37PM 4044693130831', '08-26-19 01:39PM 608917282825', '08-26-19 01:40PM 886447922064', '08-26-19 02:30PM 279 manifest.txt' ] for line in sample_ls: callback(line) return retrlines def test_create_filesize_dict(): """Test create_filesize_dict function from ftp module.""" test_input = \ 'drwxrwxrwx 1 user group ' \ '15 Jul 26 2018 digital_orchard' expected = {'digital_orchard': 15} assert ftp.create_filesize_dict(test_input) == expected def test_ftp_connection_connect(valid_conn_obj): """Test FTP connection class.""" ftp.FtpConnection(valid_conn_obj) def test_ftp_connection_login(valid_conn_obj): """Test FTP connection class.""" ftp.FtpConnection(valid_conn_obj) def test_ftp_connection_login_invalid(invalid_conn_obj): """Test FTP connection class.""" with pytest.raises(error_perm) as excinfo: ftp.FtpConnection(invalid_conn_obj) assert 'Authentication failed' in str(excinfo.value) def test_ftp_connection_check_connection(valid_conn_obj): """Test FTP connection class check_connection.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) ftp_conn.check_connection('/') assert ftp_conn.connection.pwd() == '/' def test_ftp_connection_check_connection_no_connection(valid_conn_obj): """Test FTP connection class check_connection on no connection.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) ftp_conn.close_connection() with pytest.raises(AttributeError) as excinfo: ftp_conn.check_connection('/') assert 'no attribute \'sendall\'' in str(excinfo.value) def test_ftp_connection_mkdir(valid_conn_obj): """Test FTP class recursive directory creation.""" test_dir_name = '/test/this/is/working/' ftp_conn = ftp.FtpConnection(valid_conn_obj) with pytest.raises(error_perm) as excinfo: ftp_conn.connection.cwd(test_dir_name) assert 'No such file or directory' in str(excinfo.value) assert ftp_conn.mkdir(test_dir_name) ftp_conn.connection.cwd(test_dir_name) assert ftp_conn.connection.pwd() == test_dir_name.rstrip('/') def test_ftp_connection_mkdir_root(valid_conn_obj): """Test FTP class mkdir handles root path properly.""" test_dir_name = '/' ftp_conn = ftp.FtpConnection(valid_conn_obj) assert not ftp_conn.mkdir(test_dir_name) assert ftp_conn.connection.pwd() == '/' def test_ftp_connection_mkdir_empty(valid_conn_obj): """Test FTP class mkdir handles empty path gracefully.""" test_dir_name = '' ftp_conn = ftp.FtpConnection(valid_conn_obj) assert not ftp_conn.mkdir(test_dir_name) @patch.object(ftp.FTP, 'cwd') @patch.object(ftp.FTP, 'retrlines', new_callable=mock_ls) @patch('vector_utils.connections.ftp.create_filesize_dict') def test_ftp_connection_scandir( mock_create_filesize_dict, mock_cwd, mock_retrlines, valid_conn_obj): """Test FTP class scandir.""" mock_create_filesize_dict.side_effect = [ {'digital_orchard': 0}, {'DMS_214': 187}, {'DMS_287': 15}, {'DMS_453': 0}, {'DMS_484': 0} ] expected_dict = { 'digital_orchard': 0, 'DMS_214': 187, 'DMS_287': 15, 'DMS_453': 0, 'DMS_484': 0 } ftp_conn = ftp.FtpConnection(valid_conn_obj) filesize_dict = ftp_conn.scan_dir('/') assert expected_dict == filesize_dict @patch.object(ftp.FTP, 'cwd') @patch.object(ftp.FTP, 'retrlines', new_callable=mock_ls2) def test_ftp_connection_scandir_windows( mock_cwd, mock_retrlines, valid_conn_obj): """Test FTP class scandir.""" expected_dict = { '194491040796': 0, '4044693130831': 0, '608917282825': 0, '886447922064': 0, 'manifest.txt': 279 } ftp_conn = ftp.FtpConnection(valid_conn_obj) filesize_dict = ftp_conn.scan_dir('/') assert expected_dict == filesize_dict def test_upload_file(valid_conn_obj, local_test_files): """Test FTP class upload file.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) for file in local_test_files: filename = os.path.basename(file) ftp_conn._upload_file(file, '/{}'.format(filename)) remote_dir_list = ftp_conn.scan_dir('/') assert filename in remote_dir_list.keys() assert remote_dir_list[filename] == os.stat(file).st_size def test_download_file( valid_conn_obj, uploaded_file_path, tmpdir): """Test FTP class download file.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) download_path = os.path.join(tmpdir, 'download_test.txt') ftp_conn._download_file(uploaded_file_path, download_path) remote_dir_list = ftp_conn.scan_dir('/') assert os.path.isfile(download_path) assert remote_dir_list[uploaded_file_path] == \ os.stat(download_path).st_size def test_ftp_connection_transfer_files_upload( valid_conn_obj, local_test_files): """Test FTP connection class uploads files.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) transfer_files_list = [ {'local': x, 'remote': '{}/{}'.format('', os.path.basename(x))} for x in local_test_files] ftp_conn.transfer_files(transfer_files_list) dir_list = ftp_conn.scan_dir('/') for filepath in local_test_files: assert os.path.basename(filepath) in dir_list.keys() def test_ftp_connection_transfer_files_upload_with_reconnect( valid_conn_obj, local_test_files): """Test FTP connection class uploads files with reconnect.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) # We close this connection intentionally to test reconnect during transfer ftp_conn.close_connection() transfer_files_list = [ {'local': x, 'remote': '{}/{}'.format('', os.path.basename(x))} for x in local_test_files] ftp_conn.transfer_files(transfer_files_list) dir_list = ftp_conn.scan_dir('/') for filepath in local_test_files: assert os.path.basename(filepath) in dir_list.keys() def test_ftp_connection_transfer_files_download( valid_conn_obj, uploaded_file_path, tmpdir): """Test FTP connection class downloads files.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) local_file_location = '{}/{}'.format(tmpdir, uploaded_file_path) transfer_files_list = [{ 'local': local_file_location, 'remote': '/{}'.format(uploaded_file_path) }] ftp_conn.transfer_files(transfer_files_list, transfer_mode='download') assert tmpdir.listdir() == [local_file_location] @patch('vector_utils.connections.ftp.sleep') def test_ftp_connection_transfer_files_download_err( mock_sleep, valid_conn_obj, uploaded_file_path, tmpdir): """Test FTP connection class downloads files error.""" ftp_conn = ftp.FtpConnection(valid_conn_obj) local_file_location = '{}/{}'.format(tmpdir, 'somefile.txt') transfer_files_list = [{ 'local': local_file_location, 'remote': '/{}-1'.format(uploaded_file_path) }] with pytest.raises(Exception) as excinfo: ftp_conn.transfer_files(transfer_files_list, transfer_mode='download') assert str(excinfo.value) == ('The file /download_test_file.txt-1 you are ' 'attempting to download doesn\'t exist.')