#!/opt/anaconda/bin/python # -*- coding: utf-8 -*- # for the gateway tests, the env vars of ECUID and ECGID have to be set upfront import os import sys import errno from exceptions import OSError import subprocess import pipes import pytest from mock import patch, call #from mock import PropertyMock import _config import _config_test from classes.ControlFile import ControlFile from mods.tools import (none_or_str, none_or_int, get_cmdline_arguments, read_ecenv, clean_up, my_error, send_mail, normal_exit, product, silent_remove, init128, to_param_id, get_list_as_string, make_dir, put_file_to_ecserver, submit_job_to_ecserver) class TestTools(object): ''' ''' def setup_method(self): self.testdir = _config_test.PATH_TEST_DIR self.testfilesdir = _config_test.PATH_TESTFILES_DIR self.c = ControlFile(self.testdir+'/Controls/CONTROL.test') def test_nonestr_none_or_int(self): assert None == none_or_int('None') def test_intstr_none_or_int(self): assert 42 == none_or_int('42') def test_nonestr_none_or_str(self): assert None == none_or_str('None') def test_anystr_none_or_str(self): assert 'test' == none_or_str('test') def test_default_get_cmdline_arguments(self): cmd_dict_control = {'start_date':None, 'end_date':None, 'date_chunk':None, 'basetime':None, 'step':None, 'levelist':None, 'area':None, 'inputdir':None, 'outputdir':None, 'flexpart_root_scripts':None, 'ppid':None, 'job_template':'job.temp', 'queue':None, 'controlfile':'CONTROL.temp', 'debug':None, 'public':None, 'request':None} sys.argv = ['dummy.py'] results = get_cmdline_arguments() assert cmd_dict_control == vars(results) def test_input_get_cmdline_arguments(self): cmd_dict_control = {'start_date':'20180101', 'end_date':'20180101', 'date_chunk':3, 'basetime':12, 'step':'1', 'levelist':'1/to/10', 'area':'50/10/60/20', 'inputdir':'../work', 'outputdir':None, 'flexpart_root_scripts':'../', 'ppid':1234, 'job_template':'job.sh', 'queue':'ecgate', 'controlfile':'CONTROL.WORK', 'debug':1, 'public':None, 'request':0} sys.argv = ['dummy.py', '--start_date=20180101', '--end_date=20180101', '--date_chunk=3', '--basetime=12', '--step=1', '--levelist=1/to/10', '--area=50/10/60/20', '--inputdir=../work', '--outputdir=None', '--flexpart_root_scripts=../', '--ppid=1234', '--job_template=job.sh', '--queue=ecgate', '--controlfile=CONTROL.WORK', '--debug=1', '--public=None', '--request=0'] results = get_cmdline_arguments() assert cmd_dict_control == vars(results) def test_success_init128(self): table128 = init128(_config.PATH_GRIBTABLE) expected_sample = {'078': 'TCLW', '130': 'T', '034': 'SST'} # check a sample of parameters which must have been read in assert all((k in table128 and table128[k]==v) for k,v in expected_sample.iteritems()) @patch('__builtin__.open', side_effect=[OSError(errno.EEXIST)]) def test_fail_open_init128(self, mock_openfile): with pytest.raises(SystemExit): table128 = init128(_config.PATH_GRIBTABLE) @pytest.mark.parametrize( 'ipar_str, opar_listint', [('SP/LSP/SSHF', [134, 142, 146]), ('T', [130]), ('', []), (None, []), ('testtest', []), ('130/142', [130, 142]), (130, [130]), (50.56, [])]) def test_to_param_id(self, ipar_str, opar_listint): table128 = init128(_config.PATH_GRIBTABLE) ipars = to_param_id(ipar_str, table128) assert sorted(ipars) == sorted(opar_listint) @patch('traceback.format_stack', return_value='empty trace') @patch('mods.tools.send_mail', return_value=0) def test_success_my_error(self, mock_mail, mock_trace, capfd): with pytest.raises(SystemExit): my_error(['any_user'], 'Failed!') out, err = capfd.readouterr() assert out == "Failed!\n\nempty_trace\n" @patch('subprocess.Popen') @patch('os.getenv', return_value='user') @patch('os.path.expandvars', return_value='user') def test_success_userenv_twouser_send_mail(self, mock_os, mock_env, mock_popen, capfd): mock_popen.return_value = subprocess.Popen(["echo", "Hello Test!"], stdout=subprocess.PIPE) send_mail(['${USER}', 'any_user'], 'ERROR', message='error mail') out, err = capfd.readouterr() assert out == b'Email sent to user\nEmail sent to user\n' @patch('subprocess.Popen') @patch('os.path.expandvars', return_value='any_user') def test_success_send_mail(self, mock_os, mock_popen, capfd): mock_popen.return_value = subprocess.Popen(["echo", "Hello Test!"], stdout=subprocess.PIPE) send_mail(['any-user'], 'ERROR', message='error mail') out, err = capfd.readouterr() assert out == b'Email sent to any_user\n' @patch('subprocess.Popen', side_effect=[ValueError, OSError]) @patch('os.path.expandvars', return_value='any_user') def test_fail_valueerror_send_mail(self, mock_osvar, mock_popen): with pytest.raises(SystemExit): # ValueError send_mail(['any-user'], 'ERROR', message='error mail') with pytest.raises(SystemExit): # OSError send_mail(['any-user'], 'ERROR', message='error mail') def test_success_read_ecenv(self): envs_ref = {'ECUID': 'testuser', 'ECGID': 'testgroup', 'GATEWAY': 'gateway.test.ac.at', 'DESTINATION': 'user@destination' } envs = read_ecenv(self.testfilesdir + '/ECMWF_ENV.test') assert envs_ref == envs @patch('__builtin__.open', side_effect=[OSError(errno.EPERM)]) def test_fail_read_ecenv(self, mock_open): with pytest.raises(SystemExit): read_ecenv('any_file') @patch('glob.glob', return_value=[]) @patch('mods.tools.silent_remove') def test_empty_clean_up(self, mock_rm, mock_clean): clean_up(self.c) mock_rm.assert_not_called() @patch('glob.glob', return_value=['any_file','EIfile']) @patch('os.remove', return_value=0) def test_success_clean_up(self, mock_rm, mock_glob): # ectrans=0; ecstorage=0; ecapi=None; prefix not in filename clean_up(self.c) mock_rm.assert_has_calls([call('any_file'), call('EIfile')]) mock_rm.reset_mock() # ectrans=0; ecstorage=0; ecapi=False; prefix in filename self.c.prefix = 'EI' self.c.ecapi = False clean_up(self.c) mock_rm.assert_has_calls([call('any_file')]) mock_rm.reset_mock() # ectrans=0; ecstorage=0; ecapi=True; prefix in filename self.c.prefix = 'EI' self.c.ecapi = True clean_up(self.c) mock_rm.assert_has_calls([call('any_file')]) mock_rm.reset_mock() # ectrans=1; ecstorage=0; ecapi=True; prefix in filename self.c.prefix = 'EI' self.c.ecapi = True self.c.ectrans = 1 clean_up(self.c) mock_rm.assert_has_calls([call('any_file')]) mock_rm.reset_mock() # ectrans=1; ecstorage=0; ecapi=False; prefix in filename self.c.prefix = 'EI' self.c.ecapi = False self.c.ectrans = 1 clean_up(self.c) mock_rm.assert_has_calls([call('any_file'), call('EIfile')]) mock_rm.reset_mock() # ectrans=1; ecstorage=1; ecapi=False; prefix in filename self.c.prefix = 'EI' self.c.ecapi = False self.c.ectrans = 1 self.c.ecstorage = 1 clean_up(self.c) mock_rm.assert_has_calls([call('any_file'), call('EIfile')]) mock_rm.reset_mock() def test_default_normal_exit(self, capfd): normal_exit() out, err = capfd.readouterr() assert out == 'Done!\n' def test_message_normal_exit(self, capfd): normal_exit('Hi there!') out, err = capfd.readouterr() assert out == 'Hi there!\n' def test_int_normal_exit(self, capfd): normal_exit(42) out, err = capfd.readouterr() assert out == '42\n' @pytest.mark.parametrize( 'input1, input2, output_list', [('ABC','xy',[('A','x'),('A','y'),('B','x'),('B','y'),('C','x'),('C','y')]), (range(1), range(1), [(0,0),(0,1),(1,0),(1,1)])]) def test_success_product(self, input1, input2, output_list): index = 0 for prod in product(input1, input2): assert isinstance(prod, tuple) assert prod == output_list[index] index += 1 @pytest.mark.parametrize( 'input1, input2, output_list', [(1,1,(1,1))]) def test_fail_product(self, input1, input2, output_list): index = 0 with pytest.raises(SystemExit): for prod in product(input1, input2): assert isinstance(prod, tuple) assert prod == output_list[index] index += 1 def test_success_silent_remove(self): testfile = self.testfilesdir + 'test.txt' open(testfile, 'w').close() silent_remove(testfile) assert os.path.isfile(testfile) == False @patch('os.remove', side_effect=OSError(errno.ENOENT)) def test_fail_notexist_silent_remove(self, mock_rm): with pytest.raises(OSError) as pytest_wrapped_e: silent_remove('any_dir') assert pytest_wrapped_e.e.errno == errno.ENOENT @patch('os.remove', side_effect=OSError(errno.EEXIST)) def test_fail_any_silent_remove(self, mock_rm): with pytest.raises(OSError): silent_remove('any_dir') @pytest.mark.parametrize( 'input_list, output_list', [([],''), ([1, 2, 3.5, '...', 'testlist'], '1, 2, 3.5, ..., testlist'), ('2', '2')]) def test_success_get_list_as_string(self, input_list, output_list): assert output_list == get_list_as_string(input_list) @patch('os.makedirs', side_effect=[OSError(errno.EEXIST)]) def test_warning_exist_make_dir(self, mock_make): with pytest.raises(OSError) as pytest_wrapped_e: make_dir('existing_dir') assert pytest_wrapped_e.e.errno == errno.EEXIST @patch('os.makedirs', side_effect=OSError) def test_fail_any_make_dir(self, mock_makedir): with pytest.raises(OSError): make_dir('any_dir') def test_fail_empty_make_dir(self): with pytest.raises(OSError): make_dir('') def test_success_make_dir(self): testdir = '/testing_mkdir' make_dir(self.testdir + testdir) assert os.path.exists(self.testdir + testdir) == True os.rmdir(self.testdir + testdir) @patch('subprocess.check_output', side_effect=[subprocess.CalledProcessError(1,'test')]) def test_fail_put_file_to_ecserver(self, mock_co): with pytest.raises(SystemExit): put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt', 'ecgate', 'ex_ECUID', 'ex_ECGID') @patch('subprocess.check_output', return_value=0) def test_general_success_put_file_to_ecserver(self, mock_co): result = put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt', 'ecgate', 'ex_ECUID', 'ex_ECGID') assert result == None @pytest.mark.msuser_pw @pytest.mark.gateway @pytest.mark.skip(reason="easier to ignore for now - implement in final version") def test_full_success_put_file_to_ecserver(self): ecuid = os.environ['ECUID'] ecgid = os.environ['ECGID'] put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt', 'ecgate', ecuid, ecgid) assert subprocess.call(['ssh', ecuid+'@ecaccess.ecmwf.int' , 'test -e ' + pipes.quote('/home/ms/'+ecgid+'/'+ecuid)]) == 0 @patch('subprocess.check_output', side_effect=[subprocess.CalledProcessError(1,'test'), OSError]) def test_fail_submit_job_to_ecserver(self, mock_co): with pytest.raises(SystemExit): job_id = submit_job_to_ecserver('ecgate', 'job.ksh') @pytest.mark.msuser_pw @pytest.mark.gateway @pytest.mark.skip(reason="easier to ignore for now - implement in final version") def test_success_submit_job_to_ecserver(self): job_id = submit_job_to_ecserver('ecgate', os.path.join(self.testfilesdir, 'test_put_to_ecserver.txt')) assert job_id.strip().isdigit() == True