Changeset 96e1533 in flex_extract.git for source/pythontest

Timestamp:

Nov 10, 2018, 9:37:36 PM (5 years ago)

Author:

Anne Philipp <anne.philipp@…>

Branches:

master, ctbto, dev

Children:

1abf820

Parents:

8500ba1

Message:

redefined test data dir and completed unittests for tools module

Location:

source/pythontest

Files:

• TestTools.py (modified) (4 diffs)
• _config_test.py (added)
• conftest.py (added)
• pytest.ini (modified) (1 diff)

source/pythontest/TestTools.py

@@ -1 @@
-#!/usr/bin/env python
+#!/opt/anaconda/bin/python
 # -*- coding: utf-8 -*-
+
+# for the gateway tests, the env vars of ECUID and ECGID have to be set upfront
 
 import os
 import sys
+import errno
+from exceptions import OSError
 import subprocess
 import pipes
 import pytest
-
-sys.path.append('../python')
+from mock import patch, call
+#from mock import PropertyMock
+
+
 import _config
-from mods.tools import (get_cmdline_arguments, read_ecenv, clean_up, my_error,
-                        normal_exit, product, silent_remove, init128,
-                        to_param_id, get_list_as_string, make_dir,
+import _config_test
+from classes.ControlFile import ControlFile
+from mods.tools import (none_or_str, none_or_int, get_cmdline_arguments,
+                        read_ecenv, clean_up, my_error, send_mail,
+                        normal_exit, product, silent_remove,
+                        init128, to_param_id, get_list_as_string, make_dir,
                         put_file_to_ecserver, submit_job_to_ecserver)
 
-
-class TestTools():
+class TestTools(object):
     '''
     '''
 
-    def test_get_cmdline_arguments(self):
+    def setup_method(self):
+        self.testdir = _config_test.PATH_TEST_DIR
+        self.testfilesdir = _config_test.PATH_TESTFILES_DIR
+        self.c = ControlFile(self.testdir+'/Controls/CONTROL.test')
+
+    def test_nonestr_none_or_int(self):
+        assert None == none_or_int('None')
+
+    def test_intstr_none_or_int(self):
+        assert 42 == none_or_int('42')
+
+    def test_nonestr_none_or_str(self):
+        assert None == none_or_str('None')
+
+    def test_anystr_none_or_str(self):
+        assert 'test' == none_or_str('test')
+
+    def test_default_get_cmdline_arguments(self):
+        cmd_dict_control = {'start_date':None,
+                            'end_date':None,
+                            'date_chunk':None,
+                            'basetime':None,
+                            'step':None,
+                            'levelist':None,
+                            'area':None,
+                            'inputdir':None,
+                            'outputdir':None,
+                            'flexpart_root_scripts':None,
+                            'ppid':None,
+                            'job_template':'job.temp',
+                            'queue':None,
+                            'controlfile':'CONTROL.temp',
+                            'debug':None,
+                            'public':None,
+                            'request':None}
+
+        sys.argv = ['dummy.py']
+
+        results = get_cmdline_arguments()
+
+        assert cmd_dict_control == vars(results)
+
+
+    def test_input_get_cmdline_arguments(self):
         cmd_dict_control = {'start_date':'20180101',
                             'end_date':'20180101',
-                            'date_chunk':'3',
-                            'basetime':'12',
+                            'date_chunk':3,
+                            'basetime':12,
                             'step':'1',
                             'levelist':'1/to/10',
                             'area':'50/10/60/20',
                             'inputdir':'../work',
-                            'outputdir':'../work',
+                            'outputdir':None,
                             'flexpart_root_scripts':'../',
-                            'ppid':'1234',
+                            'ppid':1234,
                             'job_template':'job.sh',
                             'queue':'ecgate',
                             'controlfile':'CONTROL.WORK',
-                            'debug':'1',
-                            'request':'0'}
+                            'debug':1,
+                            'public':None,
+                            'request':0}
 
         sys.argv = ['dummy.py',
@@ -47 +99 @@
                     '--area=50/10/60/20',
                     '--inputdir=../work',
-                    '--outputdir=../work',
+                    '--outputdir=None',
                     '--flexpart_root_scripts=../',
                     '--ppid=1234',
@@ -54 +106 @@
                     '--controlfile=CONTROL.WORK',
                     '--debug=1',
+                    '--public=None',
                     '--request=0']
 
@@ -60 +113 @@
         assert cmd_dict_control == vars(results)
 
-    def test_init128(self):
+    def test_success_init128(self):
         table128 = init128(_config.PATH_GRIBTABLE)
-        expected = {'078': 'TCLW', '130': 'T', '034': 'SST'}
+        expected_sample = {'078': 'TCLW', '130': 'T', '034': 'SST'}
         # check a sample of parameters which must have been read in
-        result = all((k in table128 and table128[k]==v) for k,v in expected.iteritems())
-        assert result == True
-
-    def test_to_param_id(self):
+        assert all((k in table128 and table128[k]==v)
+                   for k,v in expected_sample.iteritems())
+
+    @patch('__builtin__.open', side_effect=[OSError(errno.EEXIST)])
+    def test_fail_open_init128(self, mock_openfile):
+        with pytest.raises(SystemExit):
+            table128 = init128(_config.PATH_GRIBTABLE)
+
+    @pytest.mark.parametrize(
+        'ipar_str, opar_listint',
+        [('SP/LSP/SSHF', [134, 142, 146]),
+         ('T', [130]),
+         ('', []),
+         (None, []),
+         ('testtest', []),
+         ('130/142', [130, 142]),
+         (130, [130]),
+         (50.56, [])])
+    def test_to_param_id(self, ipar_str, opar_listint):
         table128 = init128(_config.PATH_GRIBTABLE)
-        pars = to_param_id("T/SP/LSP/SSHF", table128)
-        for par in pars:
-            assert par in [130, 134, 142, 146]
-
-    def test_my_error(self):
-        with pytest.raises(SystemExit) as pytest_wrapped_e:
-            my_error(['${USER}', 'anne.philipp@univie.ac.at'], 'Failed!')
-        assert pytest_wrapped_e.type == SystemExit
-        assert pytest_wrapped_e.value.code == 1
-
-    def test_read_ecenv(self):
-        envs_ref = {'ECUID': 'km4a',
-                    'ECGID': 'at',
-                    'GATEWAY': 'srvx8.img.univie.ac.at',
-                    'DESTINATION': 'annep@genericSftp'
+        ipars = to_param_id(ipar_str, table128)
+        assert sorted(ipars) == sorted(opar_listint)
+
+    @patch('traceback.format_stack', return_value='empty trace')
+    @patch('mods.tools.send_mail', return_value=0)
+    def test_success_my_error(self, mock_mail, mock_trace, capfd):
+        with pytest.raises(SystemExit):
+            my_error(['any_user'], 'Failed!')
+            out, err = capfd.readouterr()
+            assert out == "Failed!\n\nempty_trace\n"
+
+    @patch('subprocess.Popen')
+    @patch('os.getenv', return_value='user')
+    @patch('os.path.expandvars', return_value='user')
+    def test_success_userenv_twouser_send_mail(self, mock_os, mock_env, mock_popen, capfd):
+        mock_popen.return_value = subprocess.Popen(["echo", "Hello Test!"],
+                                                   stdout=subprocess.PIPE)
+        send_mail(['${USER}', 'any_user'], 'ERROR', message='error mail')
+        out, err = capfd.readouterr()
+        assert out == b'Email sent to user\nEmail sent to user\n'
+
+    @patch('subprocess.Popen')
+    @patch('os.path.expandvars', return_value='any_user')
+    def test_success_send_mail(self, mock_os,  mock_popen, capfd):
+        mock_popen.return_value = subprocess.Popen(["echo", "Hello Test!"],
+                                                   stdout=subprocess.PIPE)
+        send_mail(['any-user'], 'ERROR', message='error mail')
+        out, err = capfd.readouterr()
+        assert out == b'Email sent to any_user\n'
+
+    @patch('subprocess.Popen', side_effect=[ValueError, OSError])
+    @patch('os.path.expandvars', return_value='any_user')
+    def test_fail_valueerror_send_mail(self, mock_osvar, mock_popen):
+        with pytest.raises(SystemExit): # ValueError
+            send_mail(['any-user'], 'ERROR', message='error mail')
+        with pytest.raises(SystemExit): # OSError
+            send_mail(['any-user'], 'ERROR', message='error mail')
+
+    def test_success_read_ecenv(self):
+        envs_ref = {'ECUID': 'testuser',
+                    'ECGID': 'testgroup',
+                    'GATEWAY': 'gateway.test.ac.at',
+                    'DESTINATION': 'user@destination'
                    }
-        envs = read_ecenv(os.getcwd() + '/TestData/ECMWF_ENV')
+        envs = read_ecenv(self.testfilesdir + '/ECMWF_ENV.test')
 
         assert envs_ref == envs
 
-    def test_clean_up(self):
-
-        assert True
-
-    def test_normal_exit(self):
-        assert True
-
-    def test_product(self):
-        assert True
-
-    def test_success_silent_remove(self, capfd):
-        testfile = 'testfile.test'
+    @patch('__builtin__.open', side_effect=[OSError(errno.EPERM)])
+    def test_fail_read_ecenv(self, mock_open):
+        with pytest.raises(SystemExit):
+            read_ecenv('any_file')
+
+    @patch('glob.glob', return_value=[])
+    @patch('mods.tools.silent_remove')
+    def test_empty_clean_up(self, mock_rm, mock_clean):
+        clean_up(self.c)
+        mock_rm.assert_not_called()
+
+    @patch('glob.glob', return_value=['any_file','EIfile'])
+    @patch('os.remove', return_value=0)
+    def test_success_clean_up(self, mock_rm, mock_glob):
+        # ectrans=0; ecstorage=0; ecapi=None; prefix not in filename
+        clean_up(self.c)
+        mock_rm.assert_has_calls([call('any_file'), call('EIfile')])
+        mock_rm.reset_mock()
+
+        # ectrans=0; ecstorage=0; ecapi=False; prefix in filename
+        self.c.prefix = 'EI'
+        self.c.ecapi = False
+        clean_up(self.c)
+        mock_rm.assert_has_calls([call('any_file')])
+        mock_rm.reset_mock()
+
+        # ectrans=0; ecstorage=0; ecapi=True; prefix in filename
+        self.c.prefix = 'EI'
+        self.c.ecapi = True
+        clean_up(self.c)
+        mock_rm.assert_has_calls([call('any_file')])
+        mock_rm.reset_mock()
+
+        # ectrans=1; ecstorage=0; ecapi=True; prefix in filename
+        self.c.prefix = 'EI'
+        self.c.ecapi = True
+        self.c.ectrans = 1
+        clean_up(self.c)
+        mock_rm.assert_has_calls([call('any_file')])
+        mock_rm.reset_mock()
+
+        # ectrans=1; ecstorage=0; ecapi=False; prefix in filename
+        self.c.prefix = 'EI'
+        self.c.ecapi = False
+        self.c.ectrans = 1
+        clean_up(self.c)
+        mock_rm.assert_has_calls([call('any_file'), call('EIfile')])
+        mock_rm.reset_mock()
+
+        # ectrans=1; ecstorage=1; ecapi=False; prefix in filename
+        self.c.prefix = 'EI'
+        self.c.ecapi = False
+        self.c.ectrans = 1
+        self.c.ecstorage = 1
+        clean_up(self.c)
+        mock_rm.assert_has_calls([call('any_file'), call('EIfile')])
+        mock_rm.reset_mock()
+
+    def test_default_normal_exit(self, capfd):
+        normal_exit()
+        out, err = capfd.readouterr()
+        assert out == 'Done!\n'
+
+    def test_message_normal_exit(self, capfd):
+        normal_exit('Hi there!')
+        out, err = capfd.readouterr()
+        assert out == 'Hi there!\n'
+
+    def test_int_normal_exit(self, capfd):
+        normal_exit(42)
+        out, err = capfd.readouterr()
+        assert out == '42\n'
+
+    @pytest.mark.parametrize(
+        'input1, input2, output_list',
+        [('ABC','xy',[('A','x'),('A','y'),('B','x'),('B','y'),('C','x'),('C','y')]),
+         (range(1), range(1), [(0,0),(0,1),(1,0),(1,1)])])
+    def test_success_product(self, input1, input2, output_list):
+        index = 0
+        for prod in product(input1, input2):
+            assert isinstance(prod, tuple)
+            assert prod == output_list[index]
+            index += 1
+
+    @pytest.mark.parametrize(
+        'input1, input2, output_list',
+        [(1,1,(1,1))])
+    def test_fail_product(self, input1, input2, output_list):
+        index = 0
+        with pytest.raises(SystemExit):
+            for prod in product(input1, input2):
+                assert isinstance(prod, tuple)
+                assert prod == output_list[index]
+                index += 1
+
+    def test_success_silent_remove(self):
+        testfile = self.testfilesdir + 'test.txt'
         open(testfile, 'w').close()
         silent_remove(testfile)
-        out, err = capfd.readouterr()
         assert os.path.isfile(testfile) == False
-        assert out == ''
-
-    def test_failnotexist_silent_remove(self, capfd):
-        testfile = 'testfile.test'
-        silent_remove(testfile)
-        out, err = capfd.readouterr()
-        assert os.path.isfile(testfile) == False
-        assert out == ''
-
-    @pytest.mark.skip(reason="no way of currently testing this")
-    def test_failany_silent_remove(self):
-        testfile = 'testfileany.test'
+
+    @patch('os.remove', side_effect=OSError(errno.ENOENT))
+    def test_fail_notexist_silent_remove(self, mock_rm):
         with pytest.raises(OSError) as pytest_wrapped_e:
-            silent_remove(testfile)
-        #out, err = capfd.readouterr()
-        #assert os.path.isfile(testfile) == False
-        #assert out == ''
-
-    def test_success_get_list_as_string(self):
-        list_object =  [1, 2, 3, '...', 'testlist']
-        list_as_string = '1, 2, 3, ..., testlist'
-        assert list_as_string == get_list_as_string(list_object)
-
-    @pytest.mark.skip(reason="no way of currently testing this")
-    def test_fail_get_list_as_string(self):
-        list_object =  [1, 2, 3, '...', 'testlist']
-        list_as_string = '1, 2, 3, ..., testlist'
-        with pytest.raises(Exception) as pytest_wrapped_e:
-            result = get_list_as_string(list_object)
-        assert result == list_as_string
-
-    def test_warningexist_make_dir(self, capfd):
-        testdir = 'TestData'
-        make_dir(testdir)
-        out, err = capfd.readouterr()
-        assert out.strip() == 'WARNING: Directory {0} already exists!'.format(testdir)
-
-    def test_failany_make_dir(self):
-        testdir = '/test' # force a permission denied error
+            silent_remove('any_dir')
+            assert pytest_wrapped_e.e.errno == errno.ENOENT
+
+    @patch('os.remove', side_effect=OSError(errno.EEXIST))
+    def test_fail_any_silent_remove(self, mock_rm):
+        with pytest.raises(OSError):
+            silent_remove('any_dir')
+
+    @pytest.mark.parametrize(
+        'input_list, output_list',
+        [([],''),
+         ([1, 2, 3.5, '...', 'testlist'], '1, 2, 3.5, ..., testlist'),
+         ('2', '2')])
+    def test_success_get_list_as_string(self, input_list, output_list):
+        assert output_list == get_list_as_string(input_list)
+
+    @patch('os.makedirs', side_effect=[OSError(errno.EEXIST)])
+    def test_warning_exist_make_dir(self, mock_make):
         with pytest.raises(OSError) as pytest_wrapped_e:
-            make_dir(testdir)
-        assert pytest_wrapped_e.type == OSError
+            make_dir('existing_dir')
+            assert pytest_wrapped_e.e.errno == errno.EEXIST
+
+    @patch('os.makedirs', side_effect=OSError)
+    def test_fail_any_make_dir(self, mock_makedir):
+        with pytest.raises(OSError):
+            make_dir('any_dir')
+
+    def test_fail_empty_make_dir(self):
+        with pytest.raises(OSError):
+            make_dir('')
 
     def test_success_make_dir(self):
-        testdir = 'testing_mkdir'
-        make_dir(testdir)
-        assert os.path.exists(testdir) == True
-        os.rmdir(testdir)
-
-    def test_fail_put_file_to_ecserver(self):
-        ecuid=os.environ['ECUID']
-        ecgid=os.environ['ECGID']
-        with pytest.raises(SystemExit) as pytest_wrapped_e:
-            put_file_to_ecserver('TestData/', 'testfil.txt',
-                                 'ecgate', ecuid, ecgid)
-        assert pytest_wrapped_e.type == SystemExit
-        assert pytest_wrapped_e.value.code == '... ECACCESS-FILE-PUT FAILED!'
-
-    def test_success_put_file_to_ecserver(self):
-        ecuid=os.environ['ECUID']
-        ecgid=os.environ['ECGID']
-        result = put_file_to_ecserver('TestData/', 'testfile.txt',
-                                      'ecgate', ecuid, ecgid)
-        assert result == ''
+        testdir = '/testing_mkdir'
+        make_dir(self.testdir + testdir)
+        assert os.path.exists(self.testdir + testdir) == True
+        os.rmdir(self.testdir + testdir)
+
+
+    @patch('subprocess.check_output', side_effect=[subprocess.CalledProcessError(1,'test')])
+    def test_fail_put_file_to_ecserver(self, mock_co):
+        with pytest.raises(SystemExit):
+            put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt',
+                                 'ecgate', 'ex_ECUID', 'ex_ECGID')
+
+    @patch('subprocess.check_output', return_value=0)
+    def test_general_success_put_file_to_ecserver(self, mock_co):
+        result = put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt',
+                                      'ecgate', 'ex_ECUID', 'ex_ECGID')
+        assert result == None
 
     @pytest.mark.msuser_pw
+    @pytest.mark.gateway
     @pytest.mark.skip(reason="easier to ignore for now - implement in final version")
-    def test_fullsuccess_put_file_to_ecserver(self):
-        ecuid=os.environ['ECUID']
-        ecgid=os.environ['ECGID']
-        put_file_to_ecserver('TestData/', 'testfile.txt', 'ecgate', ecuid, ecgid)
+    def test_full_success_put_file_to_ecserver(self):
+        ecuid = os.environ['ECUID']
+        ecgid = os.environ['ECGID']
+        put_file_to_ecserver(self.testfilesdir, 'test_put_to_ecserver.txt',
+                             'ecgate', ecuid, ecgid)
         assert subprocess.call(['ssh', ecuid+'@ecaccess.ecmwf.int' ,
                                 'test -e ' +
                                 pipes.quote('/home/ms/'+ecgid+'/'+ecuid)]) == 0
 
-    def test_fail_submit_job_to_ecserver(self):
-        with pytest.raises(SystemExit) as pytest_wrapped_e:
-            submit_job_to_ecserver('ecgate', 'job.ksh')
-        assert pytest_wrapped_e.type == SystemExit
-        assert pytest_wrapped_e.value.code == '... ECACCESS-JOB-SUBMIT FAILED!'
-
+    @patch('subprocess.check_output', side_effect=[subprocess.CalledProcessError(1,'test'),
+                                                   OSError])
+    def test_fail_submit_job_to_ecserver(self, mock_co):
+        with pytest.raises(SystemExit):
+            job_id = submit_job_to_ecserver('ecgate', 'job.ksh')
+
+    @pytest.mark.msuser_pw
+    @pytest.mark.gateway
+    @pytest.mark.skip(reason="easier to ignore for now - implement in final version")
     def test_success_submit_job_to_ecserver(self):
-        result = submit_job_to_ecserver('ecgate', 'TestData/testfile.txt')
-        assert result.strip().isdigit() == True
-
+        job_id = submit_job_to_ecserver('ecgate',
+                                        os.path.join(self.testfilesdir,
+                                                     'test_put_to_ecserver.txt'))
+        assert job_id.strip().isdigit() == True
+
+

source/pythontest/pytest.ini

@@ -3 +3 @@
 markers =
     msuser_pw: Test that can be executed only as a member-state user. Password required.
+    gateway: Test that can be executed only in the gateway mode.