source: flex_extract.git/python/pythontest/TestTools.py @ 70a0bec

devfeature/makefilesorigin/task/language-editingtask/language-editing
Last change on this file since 70a0bec was 70a0bec, checked in by Anne Philipp <anne.philipp@…>, 2 years ago

added test for put_file_to_ecserver and corrected a potential bug; test is not perfect since it needs user password. TODO: CHANGE

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