server_tests.test_config.TestPartialConfigFile.test_custom_evaluate_timeout_invalid() - Code Metrics - Inspection of "Migrate to Tornado 6 and other improvements (#329)" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( d71f34...c856ab )

by Oleksandr

created 2019-07-31 20:42 UTC

TestPartialConfigFile.test_custom_evaluate_timeout_invalid() A

↳ Parent: server_tests.test_config

Complexity

Conditions

Size

Total Lines	14
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	13
dl	0
loc	14
rs	9.75
c	0
b	0
f	0
cc	1
nop	4

import os
import unittest
from argparse import Namespace
from tempfile import NamedTemporaryFile

from tabpy_server.app.util import validate_cert
from tabpy_server.app.app import TabPyApp

from unittest.mock import patch, call


def assert_raises_runtime_error(message, fn, args={}):
    try:
        fn(*args)
        assert False
    except RuntimeError as err:
        assert err.args[0] == message


class TestConfigEnvironmentCalls(unittest.TestCase):
    @patch('tabpy_server.app.app.TabPyApp._parse_cli_arguments',
           return_value=Namespace(config=None))
    @patch('tabpy_server.app.app.TabPyState')
    @patch('tabpy_server.app.app._get_state_from_file')
    @patch('tabpy_server.app.app.PythonServiceHandler')
    @patch('tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy_server.app.app.os.path.isfile', return_value=False)
    @patch('tabpy_server.app.app.os')
    def test_no_config_file(self, mock_os, mock_file_exists,
                            mock_path_exists, mock_psws,
                            mock_management_util, mock_tabpy_state,
                            mock_parse_arguments):
        TabPyApp(None)

        getenv_calls = [call('TABPY_PORT', 9004),
                        call('TABPY_QUERY_OBJECT_PATH', '/tmp/query_objects'),
                        call('TABPY_STATE_PATH',
                             './tabpy-server/tabpy_server')]
        mock_os.getenv.assert_has_calls(getenv_calls, any_order=True)
        self.assertEqual(len(mock_file_exists.mock_calls), 2)
        self.assertEqual(len(mock_psws.mock_calls), 1)
        self.assertEqual(len(mock_tabpy_state.mock_calls), 1)
        self.assertEqual(len(mock_path_exists.mock_calls), 1)
        self.assertTrue(len(mock_management_util.mock_calls) > 0)
        mock_os.makedirs.assert_not_called()

    @patch('tabpy_server.app.app.TabPyApp._parse_cli_arguments',
           return_value=Namespace(config=None))
    @patch('tabpy_server.app.app.TabPyState')
    @patch('tabpy_server.app.app._get_state_from_file')
    @patch('tabpy_server.app.app.PythonServiceHandler')
    @patch('tabpy_server.app.app.os.path.exists', return_value=False)
    @patch('tabpy_server.app.app.os.path.isfile', return_value=False)
    @patch('tabpy_server.app.app.os')
    def test_no_state_ini_file_or_state_dir(self, mock_os, mock_file_exists,
                                            mock_path_exists, mock_psws,
                                            mock_management_util,
                                            mock_tabpy_state,
                                            mock_parse_arguments):
        TabPyApp(None)
        self.assertEqual(len(mock_os.makedirs.mock_calls), 1)


class TestPartialConfigFile(unittest.TestCase):
    def setUp(self):
        self.config_file = NamedTemporaryFile(delete=False)

    def tearDown(self):
        os.remove(self.config_file.name)
        self.config_file = None

    @patch('tabpy_server.app.app.TabPyApp._parse_cli_arguments')
    @patch('tabpy_server.app.app.TabPyState')
    @patch('tabpy_server.app.app._get_state_from_file')
    @patch('tabpy_server.app.app.PythonServiceHandler')
    @patch('tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy_server.app.app.os')
    def test_config_file_present(self, mock_os, mock_path_exists,
                                 mock_psws, mock_management_util,
                                 mock_tabpy_state, mock_parse_arguments):
        self.assertTrue(self.config_file is not None)
        config_file = self.config_file
        config_file.write('[TabPy]\n'
                          'TABPY_QUERY_OBJECT_PATH = foo\n'
                          'TABPY_STATE_PATH = bar\n'.encode())
        config_file.close()

        mock_parse_arguments.return_value = Namespace(config=config_file.name)

        mock_os.getenv.side_effect = [1234]
        mock_os.path.realpath.return_value = 'bar'

        app = TabPyApp(config_file.name)
        getenv_calls = [call('TABPY_PORT', 9004)]

        mock_os.getenv.assert_has_calls(getenv_calls, any_order=True)
        self.assertEqual(app.settings['port'], 1234)
        self.assertEqual(app.settings['server_version'],
                         open('VERSION').read().strip())
        self.assertEqual(app.settings['upload_dir'], 'foo')
        self.assertEqual(app.settings['state_file_path'], 'bar')
        self.assertEqual(app.settings['transfer_protocol'], 'http')
        self.assertTrue('certificate_file' not in app.settings)
        self.assertTrue('key_file' not in app.settings)
        self.assertEqual(app.settings['log_request_context'], False)
        self.assertEqual(app.settings['evaluate_timeout'], 30)

    @patch('tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy_server.app.app._get_state_from_file')
    @patch('tabpy_server.app.app.TabPyState')
    def test_custom_evaluate_timeout_valid(self, mock_state,
                                           mock_get_state_from_file,
                                           mock_path_exists):
        self.assertTrue(self.config_file is not None)
        config_file = self.config_file
        config_file.write('[TabPy]\n'
                          'TABPY_EVALUATE_TIMEOUT = 1996'.encode())
        config_file.close()

        app = TabPyApp(self.config_file.name)
        self.assertEqual(app.settings['evaluate_timeout'], 1996.0)

    @patch('tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy_server.app.app._get_state_from_file')
    @patch('tabpy_server.app.app.TabPyState')
    def test_custom_evaluate_timeout_invalid(self, mock_state,
                                             mock_get_state_from_file,
                                             mock_path_exists):
        self.assertTrue(self.config_file is not None)
        config_file = self.config_file
        config_file.write('[TabPy]\n'
                          'TABPY_EVALUATE_TIMEOUT = "im not a float"'.encode())
        config_file.close()

        app = TabPyApp(self.config_file.name)
        self.assertEqual(app.settings['evaluate_timeout'], 30.0)


class TestTransferProtocolValidation(unittest.TestCase):
    @staticmethod
    def mock_isfile(target_file, existing_files):
        if target_file in existing_files:
            return True
        return False

    @staticmethod
    def raise_attribute_error():
        raise AttributeError()

    def __init__(self, *args, **kwargs):
        super(TestTransferProtocolValidation, self).__init__(*args, **kwargs)
        self.fp = None

    def setUp(self):
        self.fp = NamedTemporaryFile(mode='w+t', delete=False)

    def tearDown(self):
        os.remove(self.fp.name)
        self.fp = None

    def test_http(self):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = http")
        self.fp.close()

        app = TabPyApp(self.fp.name)
        self.assertEqual(app.settings['transfer_protocol'], 'http')

    def test_https_without_cert_and_key(self):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = https")
        self.fp.close()

        assert_raises_runtime_error(
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE '
            'and TABPY_KEY_FILE must be set.',
            TabPyApp, {self.fp.name})

    def test_https_without_cert(self):
        self.fp.write(
            "[TabPy]\n"
            "TABPY_TRANSFER_PROTOCOL = https\n"
            "TABPY_KEY_FILE = foo")
        self.fp.close()

        assert_raises_runtime_error('Error using HTTPS: The parameter(s) '
                                    'TABPY_CERTIFICATE_FILE must be set.',
                                    TabPyApp, {self.fp.name})

    def test_https_without_key(self):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = https\n"
                      "TABPY_CERTIFICATE_FILE = foo")
        self.fp.close()

        assert_raises_runtime_error('Error using HTTPS: The parameter(s) '
                                    'TABPY_KEY_FILE must be set.',
                                    TabPyApp, {self.fp.name})

    @patch('tabpy_server.app.app.os.path')
    def test_https_cert_and_key_file_not_found(self, mock_path):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = https\n"
                      "TABPY_CERTIFICATE_FILE = foo\n"
                      "TABPY_KEY_FILE = bar")
        self.fp.close()

        mock_path.isfile.side_effect = lambda x: self.mock_isfile(
            x, {self.fp.name})

        assert_raises_runtime_error(
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE and '
            'TABPY_KEY_FILE must point to an existing file.',
            TabPyApp, {self.fp.name})

    @patch('tabpy_server.app.app.os.path')
    def test_https_cert_file_not_found(self, mock_path):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = https\n"
                      "TABPY_CERTIFICATE_FILE = foo\n"
                      "TABPY_KEY_FILE = bar")
        self.fp.close()

        mock_path.isfile.side_effect = lambda x: self.mock_isfile(
            x, {self.fp.name, 'bar'})

        assert_raises_runtime_error(
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE '
            'must point to an existing file.',
            TabPyApp, {self.fp.name})

    @patch('tabpy_server.app.app.os.path')
    def test_https_key_file_not_found(self, mock_path):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = https\n"
                      "TABPY_CERTIFICATE_FILE = foo\n"
                      "TABPY_KEY_FILE = bar")
        self.fp.close()

        mock_path.isfile.side_effect = lambda x: self.mock_isfile(
            x, {self.fp.name, 'foo'})

        assert_raises_runtime_error(
            'Error using HTTPS: The parameter(s) TABPY_KEY_FILE '
            'must point to an existing file.',
            TabPyApp, {self.fp.name})

    @patch('tabpy_server.app.app.os.path.isfile', return_value=True)
    @patch('tabpy_server.app.util.validate_cert')
    def test_https_success(self, mock_isfile, mock_validate_cert):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = HtTpS\n"
                      "TABPY_CERTIFICATE_FILE = foo\n"
                      "TABPY_KEY_FILE = bar")
        self.fp.close()

        app = TabPyApp(self.fp.name)

        self.assertEqual(app.settings['transfer_protocol'], 'https')
        self.assertEqual(app.settings['certificate_file'], 'foo')
        self.assertEqual(app.settings['key_file'], 'bar')


class TestCertificateValidation(unittest.TestCase):

    def __init__(self, *args, **kwargs):
        super(TestCertificateValidation, self).__init__(*args, **kwargs)
        self.resources_path = os.path.join(
            os.path.dirname(__file__), 'resources')

    def test_expired_cert(self):
        path = os.path.join(self.resources_path, 'expired.crt')
        message = ('Error using HTTPS: The certificate provided expired '
                   'on 2018-08-18 19:47:18.')
        assert_raises_runtime_error(message, validate_cert, {path})

    def test_future_cert(self):
        path = os.path.join(self.resources_path, 'future.crt')
        message = ('Error using HTTPS: The certificate provided is not valid '
                   'until 3001-01-01 00:00:00.')
        assert_raises_runtime_error(message, validate_cert, {path})

    def test_valid_cert(self):
        path = os.path.join(self.resources_path, 'valid.crt')
        validate_cert(path)


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

tableau / TabPy

Push — master ( d71f34...c856ab )

TestPartialConfigFile.test_custom_evaluate_timeout_invalid() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like