Passed
Push — master ( c3033d...75a62f )
by
unknown
54s queued 12s
created

TestPartialConfigFile.test_env_variables_in_config()   A

Complexity

Conditions 1

Size

Total Lines 14
Code Lines 13

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 13
dl 0
loc 14
rs 9.75
c 0
b 0
f 0
cc 1
nop 5
1
import os
2
import unittest
3
from argparse import Namespace
4
from tempfile import NamedTemporaryFile
5
import tabpy
6
from tabpy.tabpy_server.app.util import validate_cert
7
from tabpy.tabpy_server.app.app import TabPyApp
8
9
from unittest.mock import patch
10
11
12
class TestConfigEnvironmentCalls(unittest.TestCase):
13
    @patch('tabpy.tabpy_server.app.app.TabPyApp._parse_cli_arguments',
14
           return_value=Namespace(config=None))
15
    @patch('tabpy.tabpy_server.app.app.TabPyState')
16
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
17
    @patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
18
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
19
    @patch('tabpy.tabpy_server.app.app.os')
20
    def test_no_config_file(self, mock_os,
21
                            mock_path_exists, mock_psws,
22
                            mock_management_util, mock_tabpy_state,
23
                            mock_parse_arguments):
24
        pkg_path = os.path.dirname(tabpy.__file__)
25
        obj_path = os.path.join(pkg_path, 'tmp', 'query_objects')
26
        state_path = os.path.join(pkg_path, 'tabpy_server')
27
        mock_os.environ = {
28
            'TABPY_PORT': '9004', 'TABPY_QUERY_OBJECT_PATH': obj_path,
29
            'TABPY_STATE_PATH': state_path}
30
31
        TabPyApp(None)
32
33
        self.assertEqual(len(mock_psws.mock_calls), 1)
34
        self.assertEqual(len(mock_tabpy_state.mock_calls), 1)
35
        self.assertEqual(len(mock_path_exists.mock_calls), 1)
36
        self.assertTrue(len(mock_management_util.mock_calls) > 0)
37
        mock_os.makedirs.assert_not_called()
38
39
    @patch('tabpy.tabpy_server.app.app.TabPyApp._parse_cli_arguments',
40
           return_value=Namespace(config=None))
41
    @patch('tabpy.tabpy_server.app.app.TabPyState')
42
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
43
    @patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
44
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=False)
45
    @patch('tabpy.tabpy_server.app.app.os')
46
    def test_no_state_ini_file_or_state_dir(self, mock_os,
47
                                            mock_path_exists, mock_psws,
48
                                            mock_management_util,
49
                                            mock_tabpy_state,
50
                                            mock_parse_arguments):
51
        TabPyApp(None)
52
        self.assertEqual(len(mock_os.makedirs.mock_calls), 1)
53
54
55
class TestPartialConfigFile(unittest.TestCase):
56
    def setUp(self):
57
        self.config_file = NamedTemporaryFile(delete=False)
58
59
    def tearDown(self):
60
        os.remove(self.config_file.name)
61
        self.config_file = None
62
63
    @patch('tabpy.tabpy_server.app.app.TabPyApp._parse_cli_arguments')
64
    @patch('tabpy.tabpy_server.app.app.TabPyState')
65
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
66
    @patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
67
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
68
    @patch('tabpy.tabpy_server.app.app.os')
69
    def test_config_file_present(self, mock_os, mock_path_exists,
70
                                 mock_psws, mock_management_util,
71
                                 mock_tabpy_state, mock_parse_arguments):
72
        self.assertTrue(self.config_file is not None)
73
        config_file = self.config_file
74
        config_file.write('[TabPy]\n'
75
                          'TABPY_QUERY_OBJECT_PATH = foo\n'
76
                          'TABPY_STATE_PATH = bar\n'.encode())
77
        config_file.close()
78
79
        mock_parse_arguments.return_value = Namespace(config=config_file.name)
80
        mock_os.path.realpath.return_value = 'bar'
81
        mock_os.environ = {'TABPY_PORT': '1234'}
82
83
        app = TabPyApp(config_file.name)
84
85
        self.assertEqual(app.settings['port'], '1234')
86
        self.assertEqual(app.settings['server_version'],
87
                         open('tabpy/VERSION').read().strip())
88
        self.assertEqual(app.settings['upload_dir'], 'foo')
89
        self.assertEqual(app.settings['state_file_path'], 'bar')
90
        self.assertEqual(app.settings['transfer_protocol'], 'http')
91
        self.assertTrue('certificate_file' not in app.settings)
92
        self.assertTrue('key_file' not in app.settings)
93
        self.assertEqual(app.settings['log_request_context'], False)
94
        self.assertEqual(app.settings['evaluate_timeout'], 30)
95
96
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
97
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
98
    @patch('tabpy.tabpy_server.app.app.TabPyState')
99
    def test_custom_evaluate_timeout_valid(self, mock_state,
100
                                           mock_get_state_from_file,
101
                                           mock_path_exists):
102
        self.assertTrue(self.config_file is not None)
103
        config_file = self.config_file
104
        config_file.write('[TabPy]\n'
105
                          'TABPY_EVALUATE_TIMEOUT = 1996'.encode())
106
        config_file.close()
107
108
        app = TabPyApp(self.config_file.name)
109
        self.assertEqual(app.settings['evaluate_timeout'], 1996.0)
110
111
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
112
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
113
    @patch('tabpy.tabpy_server.app.app.TabPyState')
114
    def test_custom_evaluate_timeout_invalid(self, mock_state,
115
                                             mock_get_state_from_file,
116
                                             mock_path_exists):
117
        self.assertTrue(self.config_file is not None)
118
        config_file = self.config_file
119
        config_file.write('[TabPy]\n'
120
                          'TABPY_EVALUATE_TIMEOUT = "im not a float"'.encode())
121
        config_file.close()
122
123
        app = TabPyApp(self.config_file.name)
124
        self.assertEqual(app.settings['evaluate_timeout'], 30.0)
125
126
    @patch('tabpy.tabpy_server.app.app.os')
127
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
128
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
129
    @patch('tabpy.tabpy_server.app.app.TabPyState')
130
    def test_env_variables_in_config(self, mock_state, mock_get_state,
131
                                     mock_path_exists, mock_os):
132
        mock_os.environ = {'foo': 'baz'}
133
        config_file = self.config_file
134
        config_file.write('[TabPy]\n'
135
                          'TABPY_PORT = %(foo)sbar'.encode())
136
        config_file.close()
137
138
        app = TabPyApp(self.config_file.name)
139
        self.assertEqual(app.settings['port'], 'bazbar')
140
141
142
class TestTransferProtocolValidation(unittest.TestCase):
143
    def assertTabPyAppRaisesRuntimeError(self, expected_message):
144
        with self.assertRaises(RuntimeError) as err:
145
            TabPyApp(self.fp.name)
146
        self.assertEqual(err.exception.args[0], expected_message)
147
148
    @staticmethod
149
    def mock_isfile(target_file, existing_files):
150
        if target_file in existing_files:
151
            return True
152
        return False
153
154
    @staticmethod
155
    def raise_attribute_error():
156
        raise AttributeError()
157
158
    def __init__(self, *args, **kwargs):
159
        super(TestTransferProtocolValidation, self).__init__(*args, **kwargs)
160
        self.fp = None
161
162
    def setUp(self):
163
        self.fp = NamedTemporaryFile(mode='w+t', delete=False)
164
165
    def tearDown(self):
166
        os.remove(self.fp.name)
167
        self.fp = None
168
169
    def test_http(self):
170
        self.fp.write("[TabPy]\n"
171
                      "TABPY_TRANSFER_PROTOCOL = http")
172
        self.fp.close()
173
174
        app = TabPyApp(self.fp.name)
175
        self.assertEqual(app.settings['transfer_protocol'], 'http')
176
177
    def test_https_without_cert_and_key(self):
178
        self.fp.write("[TabPy]\n"
179
                      "TABPY_TRANSFER_PROTOCOL = https")
180
        self.fp.close()
181
182
        self.assertTabPyAppRaisesRuntimeError('Error using HTTPS: The paramete'
183
                                              'r(s) TABPY_CERTIFICATE_FILE and'
184
                                              ' TABPY_KEY_FILE must be set.')
185
186
    def test_https_without_cert(self):
187
        self.fp.write(
188
            "[TabPy]\n"
189
            "TABPY_TRANSFER_PROTOCOL = https\n"
190
            "TABPY_KEY_FILE = foo")
191
        self.fp.close()
192
193
        self.assertTabPyAppRaisesRuntimeError(
194
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE must '
195
            'be set.')
196
197
    def test_https_without_key(self):
198
        self.fp.write("[TabPy]\n"
199
                      "TABPY_TRANSFER_PROTOCOL = https\n"
200
                      "TABPY_CERTIFICATE_FILE = foo")
201
        self.fp.close()
202
203
        self.assertTabPyAppRaisesRuntimeError(
204
            'Error using HTTPS: The parameter(s) TABPY_KEY_FILE must be set.')
205
206
    @patch('tabpy.tabpy_server.app.app.os.path')
207
    def test_https_cert_and_key_file_not_found(self, mock_path):
208
        self.fp.write("[TabPy]\n"
209
                      "TABPY_TRANSFER_PROTOCOL = https\n"
210
                      "TABPY_CERTIFICATE_FILE = foo\n"
211
                      "TABPY_KEY_FILE = bar")
212
        self.fp.close()
213
214
        mock_path.isfile.side_effect = lambda x: self.mock_isfile(
215
            x, {self.fp.name})
216
217
        self.assertTabPyAppRaisesRuntimeError(
218
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE and '
219
            'TABPY_KEY_FILE must point to an existing file.')
220
221
    @patch('tabpy.tabpy_server.app.app.os.path')
222
    def test_https_cert_file_not_found(self, mock_path):
223
        self.fp.write("[TabPy]\n"
224
                      "TABPY_TRANSFER_PROTOCOL = https\n"
225
                      "TABPY_CERTIFICATE_FILE = foo\n"
226
                      "TABPY_KEY_FILE = bar")
227
        self.fp.close()
228
229
        mock_path.isfile.side_effect = lambda x: self.mock_isfile(
230
            x, {self.fp.name, 'bar'})
231
232
        self.assertTabPyAppRaisesRuntimeError(
233
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE '
234
            'must point to an existing file.')
235
236
    @patch('tabpy.tabpy_server.app.app.os.path')
237
    def test_https_key_file_not_found(self, mock_path):
238
        self.fp.write("[TabPy]\n"
239
                      "TABPY_TRANSFER_PROTOCOL = https\n"
240
                      "TABPY_CERTIFICATE_FILE = foo\n"
241
                      "TABPY_KEY_FILE = bar")
242
        self.fp.close()
243
244
        mock_path.isfile.side_effect = lambda x: self.mock_isfile(
245
            x, {self.fp.name, 'foo'})
246
247
        self.assertTabPyAppRaisesRuntimeError(
248
            'Error using HTTPS: The parameter(s) TABPY_KEY_FILE '
249
            'must point to an existing file.')
250
251
    @patch('tabpy.tabpy_server.app.app.os.path.isfile', return_value=True)
252
    @patch('tabpy.tabpy_server.app.util.validate_cert')
253
    def test_https_success(self, mock_isfile, mock_validate_cert):
254
        self.fp.write("[TabPy]\n"
255
                      "TABPY_TRANSFER_PROTOCOL = HtTpS\n"
256
                      "TABPY_CERTIFICATE_FILE = foo\n"
257
                      "TABPY_KEY_FILE = bar")
258
        self.fp.close()
259
260
        app = TabPyApp(self.fp.name)
261
262
        self.assertEqual(app.settings['transfer_protocol'], 'https')
263
        self.assertEqual(app.settings['certificate_file'], 'foo')
264
        self.assertEqual(app.settings['key_file'], 'bar')
265
266
267
class TestCertificateValidation(unittest.TestCase):
268
    def assertValidateCertRaisesRuntimeError(self, expected_message, path):
269
        with self.assertRaises(RuntimeError) as err:
270
            validate_cert(path)
271
        self.assertEqual(err.exception.args[0], expected_message)
272
273
    def __init__(self, *args, **kwargs):
274
        super(TestCertificateValidation, self).__init__(*args, **kwargs)
275
        self.resources_path = os.path.join(
276
            os.path.dirname(__file__), 'resources')
277
278
    def test_expired_cert(self):
279
        path = os.path.join(self.resources_path, 'expired.crt')
280
        message = ('Error using HTTPS: The certificate provided expired '
281
                   'on 2018-08-18 19:47:18.')
282
        self.assertValidateCertRaisesRuntimeError(message, path)
283
284
    def test_future_cert(self):
285
        path = os.path.join(self.resources_path, 'future.crt')
286
        message = ('Error using HTTPS: The certificate provided is not valid '
287
                   'until 3001-01-01 00:00:00.')
288
        self.assertValidateCertRaisesRuntimeError(message, path)
289
290
    def test_valid_cert(self):
291
        path = os.path.join(self.resources_path, 'valid.crt')
292
        validate_cert(path)
293