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

ctbtodev
Last change on this file since 3232589 was 222aa11, checked in by Anne Philipp <anne.philipp@…>, 6 years ago

added pathes in config file and according testcases

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