source: flex_extract.git/source/pythontest/TestTools.py @ 4971f63

ctbtodev
Last change on this file since 4971f63 was 25b14be, checked in by Anne Philipp <anne.philipp@…>, 6 years ago

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