Newer
Older
Javier Vegas-Regidor
committed
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
# coding=utf-8
from unittest import TestCase
import mock
from earthdiagnostics.utils import TempFile, Utils
class TestTempFile(TestCase):
def setUp(self):
TempFile.scratch_folder = '/tmp'
TempFile.prefix = 'prefix'
def test_get(self):
self.assertEquals(TempFile.get('tempfile', clean=False), '/tmp/tempfile')
self.assertEquals(TempFile.get('tempfile2', clean=True), '/tmp/tempfile2')
self.assertNotIn('/tmp/tempfile', TempFile.files)
self.assertIn('/tmp/tempfile2', TempFile.files)
TempFile.autoclean = True
self.assertEquals(TempFile.get('tempfile3'), '/tmp/tempfile3')
self.assertIn('/tmp/tempfile3', TempFile.files)
TempFile.autoclean = False
self.assertEquals(TempFile.get('tempfile4'), '/tmp/tempfile4')
self.assertNotIn('/tmp/tempfile4', TempFile.files)
with mock.patch('tempfile.mkstemp') as mkstemp_mock:
with mock.patch('os.close') as close_mock:
mkstemp_mock.return_value = (34, 'path_to_tempfile')
TempFile.get()
TempFile.get(suffix='suffix')
mkstemp_mock.assert_has_calls((mock.call(dir='/tmp', prefix='prefix', suffix='.nc'),
mock.call(dir='/tmp', prefix='prefix', suffix='suffix')))
close_mock.assert_has_calls((mock.call(34), mock.call(34)))
def test_clean(self):
with mock.patch('os.path.exists') as exists_mock:
with mock.patch('tempfile.mkstemp'):
with mock.patch('os.close'):
with mock.patch('os.remove'):
TempFile.clean()
TempFile.clean()
exists_mock.side_effect = [True, False]
TempFile.autoclean = True
TempFile.get('tempfile')
TempFile.get('tempfile2')
TempFile.clean()
self.assertEquals(len(TempFile.files), 0)
class TestUtils(TestCase):
def test_rename_variable(self):
with mock.patch('earthdiagnostics.utils.Utils.rename_variables') as rename_mock:
Utils.rename_variable('file', 'old', 'new')
Utils.rename_variable('file', 'old', 'new', False, True)
rename_mock.assert_has_calls((mock.call('file', {'old': 'new'}, True, False),
mock.call('file', {'old': 'new'}, False, True)))
def test_rename_variables(self):
mock_handler = mock.Mock()
mock_handler.variables = dict()
mock_handler.dimensions = dict()
mock_handler.variables['old'] = mock.Mock()
mock_handler.variables['old_var'] = mock.Mock()
mock_handler.dimensions['old'] = mock.Mock()
with mock.patch('earthdiagnostics.utils.Utils.openCdf') as opencdf_mock:
opencdf_mock.return_value = mock_handler
Utils.rename_variables('file', {'old': 'old_var'})
Utils.rename_variables('file', {'old': 'new'}, False, True)
Utils.rename_variables('file', {'new': 'new'}, False)
Utils.rename_variables('file', {'old_var': 'new'}, False, True)
with self.assertRaises(Exception):
Utils.rename_variables('file', {'new': 'new'})
with self.assertRaises(Exception):
Utils.rename_variables('file', {'old_var': 'new'}, rename_dimension=True)
def test_convert2netcdf4(self):
mock_handler = mock.Mock()
with mock.patch('earthdiagnostics.utils.Utils.openCdf') as opencdf_mock:
with mock.patch('earthdiagnostics.utils.Utils.execute_shell_command') as execute_mock:
with mock.patch('earthdiagnostics.utils.TempFile.get') as tempfile_mock:
with mock.patch('shutil.move'):
tempfile_mock.return_value = 'tempfile'
opencdf_mock.return_value = mock_handler
mock_handler.file_format = 'NETCDF4'
Utils.convert2netcdf4('file', False)
mock_handler.file_format = 'OTHER'
Utils.convert2netcdf4('file2', False)
execute_mock.assert_called_with(['nccopy', '-4', '-d4', '-s', 'file2', 'tempfile'])
mock_handler.file_format = 'NETCDF4'
Utils.convert2netcdf4('file3', True)
execute_mock.assert_called_with(['nccopy', '-4', '-d4', '-s', 'file3', 'tempfile'])
self.assertEqual(execute_mock.call_count, 2)