source: flex_extract.git/python/pythontest/TestTools.py @ efa05d7

dev
Last change on this file since efa05d7 was efa05d7, checked in by Anne Philipp <anne.philipp@…>, 10 months ago

added more testcases

  • Property mode set to 100644
File size: 6.5 KB
Line 
1#!/usr/bin/env python
2# -*- coding: utf-8 -*-
3
4import os
5import sys
6import subprocess
7import pipes
8import pytest
9from unittest.mock import patch
10
11sys.path.append('../')
12import _config
13from tools import (get_cmdline_arguments, read_ecenv, clean_up, my_error,
14                   normal_exit, product, silent_remove, init128, to_param_id,
15                   get_list_as_string, make_dir, put_file_to_ecserver,
16                   submit_job_to_ecserver)
17
18
19class TestTools():
20    '''
21    '''
22
23    def setUp(self):
24        pass
25
26    def test_get_cmdline_arguments(self):
27        '''
28        '''
29        cmd_dict_control = {'start_date':'20180101',
30                            'end_date':'20180101',
31                            'date_chunk':'3',
32                            'basetime':'12',
33                            'step':'1',
34                            'levelist':'1/to/10',
35                            'area':'50/10/60/20',
36                            'inputdir':'../work',
37                            'outputdir':'../work',
38                            'flexpart_root_scripts':'../',
39                            'ppid':'1234',
40                            'job_template':'job.sh',
41                            'queue':'ecgate',
42                            'controlfile':'CONTROL.WORK',
43                            'debug':'1'}
44
45        sys.argv = ['dummy.py',
46                    '--start_date=20180101',
47                    '--end_date=20180101',
48                    '--date_chunk=3',
49                    '--basetime=12',
50                    '--step=1',
51                    '--levelist=1/to/10',
52                    '--area=50/10/60/20',
53                    '--inputdir=../work',
54                    '--outputdir=../work',
55                    '--flexpart_root_scripts=../',
56                    '--ppid=1234',
57                    '--job_template=job.sh',
58                    '--queue=ecgate',
59                    '--controlfile=CONTROL.WORK',
60                    '--debug=1']
61
62        results = get_cmdline_arguments()
63
64        assert cmd_dict_control == vars(results)
65
66    def test_init128(self):
67        '''
68        '''
69        table128 = init128(_config.PATH_GRIBTABLE)
70        expected = {'078': 'TCLW', '130': 'T', '034': 'SST'}
71        # check a sample of parameters which must have been read in
72        result = all((k in table128 and table128[k]==v) for k,v in expected.iteritems())
73        assert result == True
74
75    def test_to_param_id(self):
76        '''
77        '''
78        table128 = init128(_config.PATH_GRIBTABLE)
79        pars = to_param_id("T/SP/LSP/SSHF", table128)
80        for par in pars:
81            assert par in [130, 134, 142, 146]
82
83    def test_my_error(self):
84        '''
85        '''
86        with pytest.raises(SystemExit) as pytest_wrapped_e:
87            my_error(['${USER}', 'anne.philipp@univie.ac.at'], 'Failed!')
88        assert pytest_wrapped_e.type == SystemExit
89        assert pytest_wrapped_e.value.code == 1
90
91    def test_read_ecenv(self):
92        '''
93        '''
94        envs_ref = {'ECUID': 'km4a',
95                    'ECGID': 'at',
96                    'GATEWAY': 'srvx8.img.univie.ac.at',
97                    'DESTINATION': 'annep@genericSftp'
98                   }
99        envs = read_ecenv(os.getcwd() + '/TestData/ECMWF_ENV')
100
101        assert envs_ref == envs
102
103    def test_clean_up(self):
104        assert True
105
106    def test_normal_exit(self):
107        assert True
108
109    def test_product(self):
110        assert True
111
112    def test_success_silent_remove(self, capfd):
113        testfile = 'testfile.test'
114        open(testfile, 'w').close()
115        silent_remove(testfile)
116        out, err = capfd.readouterr()
117        assert os.path.isfile(testfile) == False
118        assert out == ''
119
120    def test_failnotexist_silent_remove(self, capfd):
121        testfile = 'testfile.test'
122        silent_remove(testfile)
123        out, err = capfd.readouterr()
124        assert os.path.isfile(testfile) == False
125        assert out == ''
126
127    @patch(silent_remove)
128    def test_failany_silent_remove(self, mock_silent_remove):
129        testfile = 'testfileany.test'
130        mock_silent_remove.side_effect = OSError
131        with pytest.raises(OSError) as pytest_wrapped_e:
132            silent_remove(testfile)
133        #out, err = capfd.readouterr()
134        #assert os.path.isfile(testfile) == False
135        #assert out == ''
136
137    def test_get_list_as_string(self):
138        assert True
139
140    def test_warningexist_make_dir(self, capfd):
141        testdir = 'TestData'
142        make_dir(testdir)
143        out, err = capfd.readouterr()
144        assert out.strip() == 'WARNING: Directory {0} already exists!'.format(testdir)
145
146    def test_failany_make_dir(self):
147        testdir = '/test' # force a permission denied error
148        with pytest.raises(OSError) as pytest_wrapped_e:
149            make_dir(testdir)
150        assert pytest_wrapped_e.type == OSError
151
152    def test_success_make_dir(self):
153        testdir = 'testing_mkdir'
154        make_dir(testdir)
155        assert os.path.exists(testdir) == True
156        os.rmdir(testdir)
157
158    def test_fail_put_file_to_ecserver(self):
159        ecuid=os.environ['ECUID']
160        ecgid=os.environ['ECGID']
161        with pytest.raises(SystemExit) as pytest_wrapped_e:
162            put_file_to_ecserver('TestData/', 'testfil.txt',
163                                 'ecgate', ecuid, ecgid)
164        assert pytest_wrapped_e.type == SystemExit
165        assert pytest_wrapped_e.value.code == '... ECACCESS-FILE-PUT FAILED!'
166
167    def test_success_put_file_to_ecserver(self):
168        ecuid=os.environ['ECUID']
169        ecgid=os.environ['ECGID']
170        result = put_file_to_ecserver('TestData/', 'testfile.txt',
171                                      'ecgate', ecuid, ecgid)
172        assert result == ''
173
174    @pytest.mark.msuser_pw
175    def test_fullsuccess_put_file_to_ecserver(self):
176        ecuid=os.environ['ECUID']
177        ecgid=os.environ['ECGID']
178        put_file_to_ecserver('TestData/', 'testfile.txt', 'ecgate', ecuid, ecgid)
179        assert subprocess.call(['ssh', ecuid+'@ecaccess.ecmwf.int' ,
180                                'test -e ' +
181                                pipes.quote('/home/ms/'+ecgid+'/'+ecuid)]) == 0
182
183    def test_fail_submit_job_to_ecserver(self):
184        with pytest.raises(SystemExit) as pytest_wrapped_e:
185            submit_job_to_ecserver('ecgate', 'job.ksh')
186        assert pytest_wrapped_e.type == SystemExit
187        assert pytest_wrapped_e.value.code == '... ECACCESS-JOB-SUBMIT FAILED!'
188
189    def test_success_submit_job_to_ecserver(self):
190        result = submit_job_to_ecserver('ecgate', 'TestData/testfile.txt')
191        assert result.strip().isdigit() == True
192
193
194if __name__ == "__main__":
195    unittest.main()
Note: See TracBrowser for help on using the repository browser.
hosted by ZAMG