source: flex_extract.git/Source/Pythontest/TestTools.py @ 76c37a9

ctbtodev
Last change on this file since 76c37a9 was 76c37a9, checked in by Anne Philipp <anne.philipp@…>, 4 years ago

updated unit tests to work with updated code
  • Property mode set to 100644
File size: 13.0 KB
Line 
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4# for the gateway tests, the env vars of ECUID and ECGID have to be set upfront
5
6import os
7import sys
8import errno
9#from exceptions import OSError
10import subprocess
11import pipes
12
13try:
14    import exceptions
15except ImportError:
16    import builtins
17
18import pytest
19from mock import patch, call
20#from mock import PropertyMock
21
22
23import _config
24from . import _config_test
25from Classes.ControlFile import ControlFile
26from Mods.tools import (none_or_str, none_or_int, get_cmdline_args,
27                        read_ecenv, clean_up, my_error, send_mail,
28                        normal_exit, product, silent_remove,
29                        init128, to_param_id, get_list_as_string, make_dir,
30                        put_file_to_ecserver, submit_job_to_ecserver)
31
32class TestTools(object):
33    """Test the tools module."""
34
35    def setup_method(self):
36        self.testdir = _config_test.PATH_TEST_DIR
37        self.testfilesdir = _config_test.PATH_TESTFILES_DIR
38        self.c = ControlFile(self.testdir+'/Controls/CONTROL.test')
39
40    def test_nonestr_none_or_int(self):
41        assert None == none_or_int('None')
42
43    def test_intstr_none_or_int(self):
44        assert 42 == none_or_int('42')
45
46    def test_nonestr_none_or_str(self):
47        assert None == none_or_str('None')
48
49    def test_anystr_none_or_str(self):
50        assert 'test' == none_or_str('test')
51
52    def test_fail_get_cmdline_arguments(self):
53        sys.argv = ['dummy.py', '--wrong=1']
54        with pytest.raises(SystemExit):
55            results = get_cmdline_args()
56
57    def test_default_get_cmdline_arguments(self):
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}
78
79        sys.argv = ['dummy.py']
80
81        results = get_cmdline_args()
82
83        assert cmd_dict_control == vars(results)
84
85    def test_input_get_cmdline_arguments(self):
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}
105
106        sys.argv = ['dummy.py',
107                    '--start_date=20180101',
108                    '--end_date=20180101',
109                    '--date_chunk=3',
110                    '--basetime=12',
111                    '--step=1',
112                    '--levelist=1/to/10',
113                    '--area=50/10/60/20',
114                    '--inputdir=../work',
115                    '--outputdir=None',
116                    '--ppid=1234',
117                    '--job_template=job.sh',
118                    '--queue=ecgate',
119                    '--controlfile=CONTROL.WORK',
120                    '--debug=1',
121                    '--public=None',
122                    '--request=0',
123                    '--rrint=0',
124                    '--job_chunk=None',
125                    '--oper=0']
126
127        results = get_cmdline_args()
128
129        assert cmd_dict_control == vars(results)
130
131    def test_success_init128(self):
132        table128 = init128(_config.PATH_GRIBTABLE)
133        expected_sample = {'078': 'TCLW', '130': 'T', '034': 'SST'}
134        # check a sample of parameters which must have been read in
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)])
139    def test_fail_open_init128(self, mock_openfile):
140        with pytest.raises(SystemExit):
141            table128 = init128(_config.PATH_GRIBTABLE)
142
143    @pytest.mark.parametrize(
144        'ipar_str, opar_listint',
145        [('SP/LSP/SSHF', [134, 142, 146]),
146         ('T', [130]),
147         ('', []),
148         (None, []),
149         ('testtest', []),
150         ('130/142', [130, 142]),
151         (130, [130]),
152         (50.56, [])])
153    def test_to_param_id(self, ipar_str, opar_listint):
154        table128 = init128(_config.PATH_GRIBTABLE)
155        ipars = to_param_id(ipar_str, table128)
156        assert sorted(ipars) == sorted(opar_listint)
157
158    @patch('traceback.format_stack', return_value='empty trace')
159    @patch('Mods.tools.send_mail', return_value=0)
160    def test_success_my_error(self, mock_mail, mock_trace, capfd):
161        with pytest.raises(SystemExit):
162            my_error('Failed!')
163            out, err = capfd.readouterr()
164            assert out == "Failed!\n\nempty_trace\n"
165
166    @patch('subprocess.Popen')
167    @patch('os.getenv', return_value='user')
168    @patch('os.path.expandvars', return_value='user')
169    def test_success_userenv_twouser_send_mail(self, mock_os, mock_env, mock_popen, capfd):
170        mock_popen.return_value = subprocess.Popen(["echo", "Hello Test!"],
171                                                   stdout=subprocess.PIPE)
172        send_mail(['${USER}', 'any_user'], 'ERROR', message='error mail')
173        out, err = capfd.readouterr()
174        assert out == 'Email sent to user\nEmail sent to user\n'
175
176    @patch('subprocess.Popen')
177    @patch('os.path.expandvars', return_value='any_user')
178    def test_success_send_mail(self, mock_os,  mock_popen, capfd):
179        mock_popen.return_value = subprocess.Popen(["echo", "Hello Test!"],
180                                                   stdout=subprocess.PIPE)
181        send_mail(['any-user'], 'ERROR', message='error mail')
182        out, err = capfd.readouterr()
183        assert out == 'Email sent to any_user\n'
184
185    @patch('subprocess.Popen', side_effect=[ValueError, OSError])
186    @patch('os.path.expandvars', return_value='any_user')
187    def test_fail_valueerror_send_mail(self, mock_osvar, mock_popen):
188        with pytest.raises(SystemExit): # ValueError
189            send_mail(['any-user'], 'ERROR', message='error mail')
190        with pytest.raises(SystemExit): # OSError
191            send_mail(['any-user'], 'ERROR', message='error mail')
192
193    def test_success_read_ecenv(self):
194        envs_ref = {'ECUID': 'testuser',
195                    'ECGID': 'testgroup',
196                    'GATEWAY': 'gateway.test.ac.at',
197                    'DESTINATION': 'user@destination'
198                   }
199        envs = read_ecenv(self.testfilesdir + '/ECMWF_ENV.test')
200
201        assert envs_ref == envs
202
203    @patch('builtins.open', side_effect=[OSError(errno.EPERM)])
204    def test_fail_read_ecenv(self, mock_open):
205        with pytest.raises(SystemExit):
206            read_ecenv('any_file')
207
208    @patch('glob.glob', return_value=[])
209    @patch('Mods.tools.silent_remove')
210    def test_empty_clean_up(self, mock_rm, mock_clean):
211        clean_up(self.c)
212        mock_rm.assert_not_called()
213
214    @patch('glob.glob', return_value=['any_file','EIfile'])
215    @patch('os.remove', return_value=0)
216    def test_success_clean_up(self, mock_rm, mock_glob):
217
218        self.c.prefix = 'EI'
219        self.c.ecapi = False
220        clean_up(self.c)
221        mock_rm.assert_has_calls([call('any_file')])
222        mock_rm.reset_mock()
223
224
225    def test_default_normal_exit(self, capfd):
226        normal_exit()
227        out, err = capfd.readouterr()
228        assert out == 'Done!\n'
229
230    def test_message_normal_exit(self, capfd):
231        normal_exit('Hi there!')
232        out, err = capfd.readouterr()
233        assert out == 'Hi there!\n'
234
235    def test_int_normal_exit(self, capfd):
236        normal_exit(42)
237        out, err = capfd.readouterr()
238        assert out == '42\n'
239
240    @pytest.mark.parametrize(
241        'input1, input2, output_list',
242        [('ABC','xy',[('A','x'),('A','y'),('B','x'),('B','y'),('C','x'),('C','y')]),
243         (range(1), range(1), [(0,0),(0,1),(1,0),(1,1)])])
244    def test_success_product(self, input1, input2, output_list):
245        index = 0
246        for prod in product(input1, input2):
247            assert isinstance(prod, tuple)
248            assert prod == output_list[index]
249            index += 1
250
251    @pytest.mark.parametrize(
252        'input1, input2, output_list',
253        [(1,1,(1,1))])
254    def test_fail_product(self, input1, input2, output_list):
255        index = 0
256        with pytest.raises(SystemExit):
257            for prod in product(input1, input2):
258                assert isinstance(prod, tuple)
259                assert prod == output_list[index]
260                index += 1
261
262    def test_success_silent_remove(self):
263        testfile = self.testfilesdir + 'test.txt'
264        open(testfile, 'w').close()
265        silent_remove(testfile)
266        assert os.path.isfile(testfile) == False
267
268    @patch('os.remove', side_effect=OSError(errno.ENOENT))
269    def test_fail_notexist_silent_remove(self, mock_rm):
270        with pytest.raises(OSError) as pytest_wrapped_e:
271            silent_remove('any_dir')
272            assert pytest_wrapped_e.e.errno == errno.ENOENT
273
274    @patch('os.remove', side_effect=OSError(errno.EEXIST))
275    def test_fail_any_silent_remove(self, mock_rm):
276        with pytest.raises(OSError):
277            silent_remove('any_dir')
278
279    @pytest.mark.parametrize(
280        'input_list, output_list',
281        [([],''),
282         ([1, 2, 3.5, '...', 'testlist'], '1, 2, 3.5, ..., testlist'),
283         ('2', '2')])
284    def test_success_get_list_as_string(self, input_list, output_list):
285        assert output_list == get_list_as_string(input_list)
286
287    @patch('os.makedirs', side_effect=[OSError(errno.EEXIST)])
288    def test_warning_exist_make_dir(self, mock_make):
289        with pytest.raises(OSError) as pytest_wrapped_e:
290            make_dir('existing_dir')
291            assert pytest_wrapped_e.e.errno == errno.EEXIST
292
293    @patch('os.makedirs', side_effect=OSError)
294    def test_fail_any_make_dir(self, mock_makedir):
295        with pytest.raises(OSError):
296            make_dir('any_dir')
297
298    def test_fail_empty_make_dir(self):
299        with pytest.raises(OSError):
300            make_dir('')
301
302    def test_success_make_dir(self):
303        testdir = '/testing_mkdir'
304        make_dir(self.testdir + testdir)
305        assert os.path.exists(self.testdir + testdir) == True
306        os.rmdir(self.testdir + testdir)
307
308
309    @patch('subprocess.check_output', side_effect=[subprocess.CalledProcessError(1,'test')])
310    def test_fail_put_file_to_ecserver(self, mock_co):
311        with pytest.raises(SystemExit):
312            put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt',
313                                 'ecgate', 'ex_ECUID', 'ex_ECGID')
314
315    @patch('subprocess.check_output', return_value=0)
316    def test_general_success_put_file_to_ecserver(self, mock_co):
317        result = put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt',
318                                      'ecgate', 'ex_ECUID', 'ex_ECGID')
319        assert result == None
320
321    @pytest.mark.msuser_pw
322    @pytest.mark.gateway
323    @pytest.mark.skip(reason="easier to ignore for now - implement in final version")
324    def test_full_success_put_file_to_ecserver(self):
325        ecuid = os.environ['ECUID']
326        ecgid = os.environ['ECGID']
327        put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt',
328                             'ecgate', ecuid, ecgid)
329        assert subprocess.call(['ssh', ecuid+'@ecaccess.ecmwf.int' ,
330                                'test -e ' +
331                                pipes.quote('/home/ms/'+ecgid+'/'+ecuid)]) == 0
332
333    @patch('subprocess.check_output', side_effect=[subprocess.CalledProcessError(1,'test'),
334                                                   OSError])
335    def test_fail_submit_job_to_ecserver(self, mock_co):
336        with pytest.raises(SystemExit):
337            job_id = submit_job_to_ecserver('ecgate', 'job.ksh')
338
339    @pytest.mark.msuser_pw
340    @pytest.mark.gateway
341    @pytest.mark.skip(reason="easier to ignore for now - implement in final version")
342    def test_success_submit_job_to_ecserver(self):
343        job_id = submit_job_to_ecserver('ecgate',
344                                        os.path.join(self.testfilesdir,
345                                                     'test_put_to_ecserver.txt'))
346        assert job_id.strip().isdigit() == True
347
348
Note: See TracBrowser for help on using the repository browser.
hosted by ZAMG