Changeset 76c37a9 in flex_extract.git for Source/Pythontest/TestTools.py
- Timestamp:
- Dec 19, 2019, 8:11:12 PM (4 years ago)
- Branches:
- master, ctbto, dev
- Children:
- d720895
- Parents:
- d2b7217
- File:
-
- 1 edited
Legend:
- Unmodified
- Added
- Removed
-
Source/Pythontest/TestTools.py
rba99230 r76c37a9 7 7 import sys 8 8 import errno 9 from exceptions import OSError9 #from exceptions import OSError 10 10 import subprocess 11 11 import pipes 12 13 try: 14 import exceptions 15 except ImportError: 16 import builtins 17 12 18 import pytest 13 19 from mock import patch, call … … 16 22 17 23 import _config 18 import _config_test19 from classes.ControlFile import ControlFile20 from mods.tools import (none_or_str, none_or_int, get_cmdline_arguments,24 from . import _config_test 25 from Classes.ControlFile import ControlFile 26 from Mods.tools import (none_or_str, none_or_int, get_cmdline_args, 21 27 read_ecenv, clean_up, my_error, send_mail, 22 28 normal_exit, product, silent_remove, … … 25 31 26 32 class TestTools(object): 27 ''' 28 ''' 33 """Test the tools module.""" 29 34 30 35 def setup_method(self): … … 48 53 sys.argv = ['dummy.py', '--wrong=1'] 49 54 with pytest.raises(SystemExit): 50 results = get_cmdline_arg uments()55 results = get_cmdline_args() 51 56 52 57 def test_default_get_cmdline_arguments(self): 53 cmd_dict_control = {'start_date':None, 54 'end_date':None, 55 'date_chunk':None, 56 'basetime':None, 57 'step':None, 58 'levelist':None, 59 'area':None, 60 'inputdir':None, 61 'outputdir':None, 62 'flexpart_root_scripts':None, 63 'ppid':None, 64 'job_template':'job.temp', 65 'queue':None, 66 'controlfile':'CONTROL.temp', 67 'debug':None, 68 'public':None, 69 'request':None} 58 cmd_dict_control = {'start_date': None, 59 'end_date': None, 60 'date_chunk': None, 61 'basetime': None, 62 'step': None, 63 'levelist': None, 64 'area': None, 65 'inputdir': None, 66 'outputdir': None, 67 'job_template': None, 68 'job_chunk': None, 69 'ppid': None, 70 'job_template': 'job.temp', 71 'queue': None, 72 'controlfile': 'CONTROL_EA5', 73 'debug': None, 74 'public': None, 75 'request': None, 76 'oper': None, 77 'rrint': None} 70 78 71 79 sys.argv = ['dummy.py'] 72 80 73 results = get_cmdline_arg uments()81 results = get_cmdline_args() 74 82 75 83 assert cmd_dict_control == vars(results) 76 84 77 85 def test_input_get_cmdline_arguments(self): 78 cmd_dict_control = {'start_date':'20180101', 79 'end_date':'20180101', 80 'date_chunk':3, 81 'basetime':12, 82 'step':'1', 83 'levelist':'1/to/10', 84 'area':'50/10/60/20', 85 'inputdir':'../work', 86 'outputdir':None, 87 'flexpart_root_scripts':'../', 88 'ppid':1234, 89 'job_template':'job.sh', 90 'queue':'ecgate', 91 'controlfile':'CONTROL.WORK', 92 'debug':1, 93 'public':None, 94 'request':0} 86 cmd_dict_control = {'start_date': '20180101', 87 'end_date': '20180101', 88 'date_chunk': 3, 89 'basetime': 12, 90 'step': '1', 91 'levelist': '1/to/10', 92 'area': '50/10/60/20', 93 'inputdir': '../work', 94 'outputdir': None, 95 'ppid': '1234', 96 'job_template': 'job.sh', 97 'queue': 'ecgate', 98 'controlfile': 'CONTROL.WORK', 99 'debug': 1, 100 'public': None, 101 'request': 0, 102 'rrint': 0, 103 'job_chunk': None, 104 'oper': 0} 95 105 96 106 sys.argv = ['dummy.py', … … 104 114 '--inputdir=../work', 105 115 '--outputdir=None', 106 '--flexpart_root_scripts=../',107 116 '--ppid=1234', 108 117 '--job_template=job.sh', … … 111 120 '--debug=1', 112 121 '--public=None', 113 '--request=0'] 114 115 results = get_cmdline_arguments() 122 '--request=0', 123 '--rrint=0', 124 '--job_chunk=None', 125 '--oper=0'] 126 127 results = get_cmdline_args() 116 128 117 129 assert cmd_dict_control == vars(results) … … 121 133 expected_sample = {'078': 'TCLW', '130': 'T', '034': 'SST'} 122 134 # check a sample of parameters which must have been read in 123 assert all((k in table128 and table128[k] ==v)124 for k, v in expected_sample.iteritems())125 126 @patch(' __builtin__.open', side_effect=[OSError(errno.EEXIST)])135 assert all((k in table128 and table128[k] == v) 136 for k, v in expected_sample.items()) 137 138 @patch('builtins.open', side_effect=[OSError(errno.EEXIST)]) 127 139 def test_fail_open_init128(self, mock_openfile): 128 140 with pytest.raises(SystemExit): … … 145 157 146 158 @patch('traceback.format_stack', return_value='empty trace') 147 @patch(' mods.tools.send_mail', return_value=0)159 @patch('Mods.tools.send_mail', return_value=0) 148 160 def test_success_my_error(self, mock_mail, mock_trace, capfd): 149 161 with pytest.raises(SystemExit): 150 my_error( ['any_user'],'Failed!')162 my_error('Failed!') 151 163 out, err = capfd.readouterr() 152 164 assert out == "Failed!\n\nempty_trace\n" … … 160 172 send_mail(['${USER}', 'any_user'], 'ERROR', message='error mail') 161 173 out, err = capfd.readouterr() 162 assert out == b'Email sent to user\nEmail sent to user\n'174 assert out == 'Email sent to user\nEmail sent to user\n' 163 175 164 176 @patch('subprocess.Popen') … … 169 181 send_mail(['any-user'], 'ERROR', message='error mail') 170 182 out, err = capfd.readouterr() 171 assert out == b'Email sent to any_user\n'183 assert out == 'Email sent to any_user\n' 172 184 173 185 @patch('subprocess.Popen', side_effect=[ValueError, OSError]) … … 189 201 assert envs_ref == envs 190 202 191 @patch(' __builtin__.open', side_effect=[OSError(errno.EPERM)])203 @patch('builtins.open', side_effect=[OSError(errno.EPERM)]) 192 204 def test_fail_read_ecenv(self, mock_open): 193 205 with pytest.raises(SystemExit): … … 195 207 196 208 @patch('glob.glob', return_value=[]) 197 @patch(' mods.tools.silent_remove')209 @patch('Mods.tools.silent_remove') 198 210 def test_empty_clean_up(self, mock_rm, mock_clean): 199 211 clean_up(self.c) … … 203 215 @patch('os.remove', return_value=0) 204 216 def test_success_clean_up(self, mock_rm, mock_glob): 205 # ectrans=0; ecstorage=0; ecapi=None; prefix not in filename 206 clean_up(self.c) 207 mock_rm.assert_has_calls([call('any_file'), call('EIfile')]) 208 mock_rm.reset_mock() 209 210 # ectrans=0; ecstorage=0; ecapi=False; prefix in filename 217 211 218 self.c.prefix = 'EI' 212 219 self.c.ecapi = False … … 215 222 mock_rm.reset_mock() 216 223 217 # ectrans=0; ecstorage=0; ecapi=True; prefix in filename218 self.c.prefix = 'EI'219 self.c.ecapi = True220 clean_up(self.c)221 mock_rm.assert_has_calls([call('any_file')])222 mock_rm.reset_mock()223 224 # ectrans=1; ecstorage=0; ecapi=True; prefix in filename225 self.c.prefix = 'EI'226 self.c.ecapi = True227 self.c.ectrans = 1228 clean_up(self.c)229 mock_rm.assert_has_calls([call('any_file')])230 mock_rm.reset_mock()231 232 # ectrans=1; ecstorage=0; ecapi=False; prefix in filename233 self.c.prefix = 'EI'234 self.c.ecapi = False235 self.c.ectrans = 1236 clean_up(self.c)237 mock_rm.assert_has_calls([call('any_file'), call('EIfile')])238 mock_rm.reset_mock()239 240 # ectrans=1; ecstorage=1; ecapi=False; prefix in filename241 self.c.prefix = 'EI'242 self.c.ecapi = False243 self.c.ectrans = 1244 self.c.ecstorage = 1245 clean_up(self.c)246 mock_rm.assert_has_calls([call('any_file'), call('EIfile')])247 mock_rm.reset_mock()248 224 249 225 def test_default_normal_exit(self, capfd):
Note: See TracChangeset
for help on using the changeset viewer.