source: flex_extract.git/python/pythontest/TestTools.py @ 97e09f4

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

completed testing and adaptations of ecacces job-submit and file-put functions

  • Property mode set to 100644
File size: 5.0 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('../')
11import _config
12from tools import (init128, to_param_id, my_error, read_ecenv,
13                   get_cmdline_arguments, submit_job_to_ecserver,
14                   put_file_to_ecserver)
15
16
17class TestTools():
18    '''
19    '''
20
21    def setUp(self):
22        pass
23
24    def test_get_cmdline_arguments(self):
25        '''
26        '''
27        cmd_dict_control = {'start_date':'20180101',
28                            'end_date':'20180101',
29                            'date_chunk':'3',
30                            'basetime':'12',
31                            'step':'1',
32                            'levelist':'1/to/10',
33                            'area':'50/10/60/20',
34                            'inputdir':'../work',
35                            'outputdir':'../work',
36                            'flexpart_root_scripts':'../',
37                            'ppid':'1234',
38                            'job_template':'job.sh',
39                            'queue':'ecgate',
40                            'controlfile':'CONTROL.WORK',
41                            'debug':'1'}
42
43        sys.argv = ['dummy.py',
44                    '--start_date=20180101',
45                    '--end_date=20180101',
46                    '--date_chunk=3',
47                    '--basetime=12',
48                    '--step=1',
49                    '--levelist=1/to/10',
50                    '--area=50/10/60/20',
51                    '--inputdir=../work',
52                    '--outputdir=../work',
53                    '--flexpart_root_scripts=../',
54                    '--ppid=1234',
55                    '--job_template=job.sh',
56                    '--queue=ecgate',
57                    '--controlfile=CONTROL.WORK',
58                    '--debug=1']
59
60        results = get_cmdline_arguments()
61
62        assert cmd_dict_control == vars(results)
63
64    def test_init128(self):
65        '''
66        '''
67        table128 = init128(_config.PATH_GRIBTABLE)
68        expected = {'078': 'TCLW', '130': 'T', '034': 'SST'}
69        # check a sample of parameters which must have been read in
70        result = all((k in table128 and table128[k]==v) for k,v in expected.iteritems())
71        assert result == True
72
73    def test_to_param_id(self):
74        '''
75        '''
76        table128 = init128(_config.PATH_GRIBTABLE)
77        pars = to_param_id("T/SP/LSP/SSHF", table128)
78        for par in pars:
79            assert par in [130, 134, 142, 146]
80
81    def test_my_error(self):
82        '''
83        '''
84        with pytest.raises(SystemExit) as pytest_wrapped_e:
85            my_error(['${USER}', 'anne.philipp@univie.ac.at'], 'Failed!')
86        assert pytest_wrapped_e.type == SystemExit
87        assert pytest_wrapped_e.value.code == 1
88
89    def test_read_ecenv(self):
90        '''
91        '''
92        envs_ref = {'ECUID': 'km4a',
93                    'ECGID': 'at',
94                    'GATEWAY': 'srvx8.img.univie.ac.at',
95                    'DESTINATION': 'annep@genericSftp'
96                   }
97        envs = read_ecenv(os.getcwd() + '/TestData/ECMWF_ENV')
98
99        assert envs_ref == envs
100
101    def test_clean_up(self):
102        assert True
103
104    def test_normal_exit(self):
105        assert True
106
107    def test_product(self):
108        assert True
109
110    def test_silent_remove(self):
111        assert True
112
113    def test_get_list_as_string(self):
114        assert True
115
116    def test_make_dir(self):
117        assert True
118
119    def test_fail_put_file_to_ecserver(self):
120        ecuid=os.environ['ECUID']
121        ecgid=os.environ['ECGID']
122        with pytest.raises(SystemExit) as pytest_wrapped_e:
123            put_file_to_ecserver('TestData/', 'testfil.txt',
124                                 'ecgate', ecuid, ecgid)
125        assert pytest_wrapped_e.type == SystemExit
126        assert pytest_wrapped_e.value.code == '... ECACCESS-FILE-PUT FAILED!'
127
128    def test_success_put_file_to_ecserver(self):
129        ecuid=os.environ['ECUID']
130        ecgid=os.environ['ECGID']
131        result = put_file_to_ecserver('TestData/', 'testfile.txt',
132                                      'ecgate', ecuid, ecgid)
133        assert result == ''
134
135    @pytest.mark.msuser_pw
136    def test_fullsuccess_put_file_to_ecserver(self):
137        ecuid=os.environ['ECUID']
138        ecgid=os.environ['ECGID']
139        put_file_to_ecserver('TestData/', 'testfile.txt', 'ecgate', ecuid, ecgid)
140        assert subprocess.call(['ssh', ecuid+'@ecaccess.ecmwf.int' ,
141                                'test -e ' +
142                                pipes.quote('/home/ms/'+ecgid+'/'+ecuid)]) == 0
143
144    def test_fail_submit_job_to_ecserver(self):
145        with pytest.raises(SystemExit) as pytest_wrapped_e:
146            submit_job_to_ecserver('ecgate', 'job.ksh')
147        assert pytest_wrapped_e.type == SystemExit
148        assert pytest_wrapped_e.value.code == '... ECACCESS-JOB-SUBMIT FAILED!'
149
150    def test_success_submit_job_to_ecserver(self):
151        result = submit_job_to_ecserver('ecgate', 'TestData/testfile.txt')
152        assert result.strip().isdigit() == True
153
154
155if __name__ == "__main__":
156    unittest.main()
Note: See TracBrowser for help on using the repository browser.
hosted by ZAMG