server_tests.test_config.TestTransferProtocolValidation.test_https_without_cert() - Code Metrics - Inspection of "v 0.8.9 (#353)" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 305552...00a4d3 )

by Oleksandr

created 2019-10-22 00:09 UTC

TestTransferProtocolValidation.test_https_without_cert() A

↳ Parent: server_tests.test_config

Complexity

Conditions

Size

Total Lines	9
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	9
rs	10
c	0
b	0
f	0
cc	1
nop	1

import os
import unittest
from argparse import Namespace
from tempfile import NamedTemporaryFile
import tabpy
from tabpy.tabpy_server.app.util import validate_cert
from tabpy.tabpy_server.app.app import TabPyApp

from unittest.mock import patch


class TestConfigEnvironmentCalls(unittest.TestCase):
    def test_config_file_does_not_exist(self):
        app = TabPyApp('/folder_does_not_exit/file_does_not_exist.conf')

        self.assertEqual(app.settings['port'], 9004)
        self.assertEqual(app.settings['server_version'],
                         open('tabpy/VERSION').read().strip())
        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.tabpy_server.app.app.TabPyApp._parse_cli_arguments',
           return_value=Namespace(config=None))
    @patch('tabpy.tabpy_server.app.app.TabPyState')
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
    @patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy.tabpy_server.app.app.os')
    def test_no_config_file(self, mock_os,
                            mock_path_exists, mock_psws,
                            mock_management_util, mock_tabpy_state,
                            mock_parse_arguments):
        pkg_path = os.path.dirname(tabpy.__file__)
        obj_path = os.path.join(pkg_path, 'tmp', 'query_objects')
        state_path = os.path.join(pkg_path, 'tabpy_server')
        mock_os.environ = {
            'TABPY_PORT': '9004', 'TABPY_QUERY_OBJECT_PATH': obj_path,
            'TABPY_STATE_PATH': state_path}

        TabPyApp(None)

        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.tabpy_server.app.app.TabPyApp._parse_cli_arguments',
           return_value=Namespace(config=None))
    @patch('tabpy.tabpy_server.app.app.TabPyState')
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
    @patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=False)
    @patch('tabpy.tabpy_server.app.app.os')
    def test_no_state_ini_file_or_state_dir(self, mock_os,
                                            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.tabpy_server.app.app.TabPyApp._parse_cli_arguments')
    @patch('tabpy.tabpy_server.app.app.TabPyState')
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
    @patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy.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.path.realpath.return_value = 'bar'
        mock_os.environ = {'TABPY_PORT': '1234'}

        app = TabPyApp(config_file.name)

        self.assertEqual(app.settings['port'], '1234')
        self.assertEqual(app.settings['server_version'],
                         open('tabpy/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.tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
    @patch('tabpy.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.tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
    @patch('tabpy.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)

    @patch('tabpy.tabpy_server.app.app.os')
    @patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
    @patch('tabpy.tabpy_server.app.app._get_state_from_file')
    @patch('tabpy.tabpy_server.app.app.TabPyState')
    def test_env_variables_in_config(self, mock_state, mock_get_state,
                                     mock_path_exists, mock_os):
        mock_os.environ = {'foo': 'baz'}
        config_file = self.config_file
        config_file.write('[TabPy]\n'
                          'TABPY_PORT = %(foo)sbar'.encode())
        config_file.close()

        app = TabPyApp(self.config_file.name)
        self.assertEqual(app.settings['port'], 'bazbar')


class TestTransferProtocolValidation(unittest.TestCase):
    def assertTabPyAppRaisesRuntimeError(self, expected_message):
        with self.assertRaises(RuntimeError) as err:
            TabPyApp(self.fp.name)
        self.assertEqual(err.exception.args[0], expected_message)

    @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_invalid_protocol(self):
        self.fp.write("[TabPy]\n"
                      "TABPY_TRANSFER_PROTOCOL = gopher")
        self.fp.close()

        self.assertTabPyAppRaisesRuntimeError(
            'Unsupported transfer protocol: gopher')

    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()

        self.assertTabPyAppRaisesRuntimeError('Error using HTTPS: The paramete'
                                              'r(s) TABPY_CERTIFICATE_FILE and'
                                              ' TABPY_KEY_FILE must be set.')

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

        self.assertTabPyAppRaisesRuntimeError(
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE must '
            'be set.')

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

        self.assertTabPyAppRaisesRuntimeError(
            'Error using HTTPS: The parameter(s) TABPY_KEY_FILE must be set.')

    @patch('tabpy.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})

        self.assertTabPyAppRaisesRuntimeError(
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE and '
            'TABPY_KEY_FILE must point to an existing file.')

    @patch('tabpy.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'})

        self.assertTabPyAppRaisesRuntimeError(
            'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE '
            'must point to an existing file.')

    @patch('tabpy.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'})

        self.assertTabPyAppRaisesRuntimeError(
            'Error using HTTPS: The parameter(s) TABPY_KEY_FILE '
            'must point to an existing file.')

    @patch('tabpy.tabpy_server.app.app.os.path.isfile', return_value=True)
    @patch('tabpy.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 assertValidateCertRaisesRuntimeError(self, expected_message, path):
        with self.assertRaises(RuntimeError) as err:
            validate_cert(path)
        self.assertEqual(err.exception.args[0], expected_message)

    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.')
        self.assertValidateCertRaisesRuntimeError(message, 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.')
        self.assertValidateCertRaisesRuntimeError(message, 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			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			def test_config_file_does_not_exist(self):
14			app = TabPyApp('/folder_does_not_exit/file_does_not_exist.conf')
15
16			self.assertEqual(app.settings['port'], 9004)
17			self.assertEqual(app.settings['server_version'],
18			open('tabpy/VERSION').read().strip())
19			self.assertEqual(app.settings['transfer_protocol'], 'http')
20			self.assertTrue('certificate_file' not in app.settings)
21			self.assertTrue('key_file' not in app.settings)
22			self.assertEqual(app.settings['log_request_context'], False)
23			self.assertEqual(app.settings['evaluate_timeout'], 30)
24
25			@patch('tabpy.tabpy_server.app.app.TabPyApp._parse_cli_arguments',
26			return_value=Namespace(config=None))
27			@patch('tabpy.tabpy_server.app.app.TabPyState')
28			@patch('tabpy.tabpy_server.app.app._get_state_from_file')
29			@patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
30			@patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
31			@patch('tabpy.tabpy_server.app.app.os')
32			def test_no_config_file(self, mock_os,
33			mock_path_exists, mock_psws,
34			mock_management_util, mock_tabpy_state,
35			mock_parse_arguments):
36			pkg_path = os.path.dirname(tabpy.__file__)
37			obj_path = os.path.join(pkg_path, 'tmp', 'query_objects')
38			state_path = os.path.join(pkg_path, 'tabpy_server')
39			mock_os.environ = {
40			'TABPY_PORT': '9004', 'TABPY_QUERY_OBJECT_PATH': obj_path,
41			'TABPY_STATE_PATH': state_path}
42
43			TabPyApp(None)
44
45			self.assertEqual(len(mock_psws.mock_calls), 1)
46			self.assertEqual(len(mock_tabpy_state.mock_calls), 1)
47			self.assertEqual(len(mock_path_exists.mock_calls), 1)
48			self.assertTrue(len(mock_management_util.mock_calls) > 0)
49			mock_os.makedirs.assert_not_called()
50
51			@patch('tabpy.tabpy_server.app.app.TabPyApp._parse_cli_arguments',
52			return_value=Namespace(config=None))
53			@patch('tabpy.tabpy_server.app.app.TabPyState')
54			@patch('tabpy.tabpy_server.app.app._get_state_from_file')
55			@patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
56			@patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=False)
57			@patch('tabpy.tabpy_server.app.app.os')
58			def test_no_state_ini_file_or_state_dir(self, mock_os,
59			mock_path_exists, mock_psws,
60			mock_management_util,
61			mock_tabpy_state,
62			mock_parse_arguments):
63			TabPyApp(None)
64			self.assertEqual(len(mock_os.makedirs.mock_calls), 1)
65
66
67			class TestPartialConfigFile(unittest.TestCase):
68			def setUp(self):
69			self.config_file = NamedTemporaryFile(delete=False)
70
71			def tearDown(self):
72			os.remove(self.config_file.name)
73			self.config_file = None
74
75			@patch('tabpy.tabpy_server.app.app.TabPyApp._parse_cli_arguments')
76			@patch('tabpy.tabpy_server.app.app.TabPyState')
77			@patch('tabpy.tabpy_server.app.app._get_state_from_file')
78			@patch('tabpy.tabpy_server.app.app.PythonServiceHandler')
79			@patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
80			@patch('tabpy.tabpy_server.app.app.os')
81			def test_config_file_present(self, mock_os, mock_path_exists,
82			mock_psws, mock_management_util,
83			mock_tabpy_state, mock_parse_arguments):
84			self.assertTrue(self.config_file is not None)
85			config_file = self.config_file
86			config_file.write('[TabPy]\n'
87			'TABPY_QUERY_OBJECT_PATH = foo\n'
88			'TABPY_STATE_PATH = bar\n'.encode())
89			config_file.close()
90
91			mock_parse_arguments.return_value = Namespace(config=config_file.name)
92			mock_os.path.realpath.return_value = 'bar'
93			mock_os.environ = {'TABPY_PORT': '1234'}
94
95			app = TabPyApp(config_file.name)
96
97			self.assertEqual(app.settings['port'], '1234')
98			self.assertEqual(app.settings['server_version'],
99			open('tabpy/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.tabpy_server.app.app.os.path.exists', return_value=True)
109			@patch('tabpy.tabpy_server.app.app._get_state_from_file')
110			@patch('tabpy.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.tabpy_server.app.app.os.path.exists', return_value=True)
124			@patch('tabpy.tabpy_server.app.app._get_state_from_file')
125			@patch('tabpy.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			@patch('tabpy.tabpy_server.app.app.os')
139			@patch('tabpy.tabpy_server.app.app.os.path.exists', return_value=True)
140			@patch('tabpy.tabpy_server.app.app._get_state_from_file')
141			@patch('tabpy.tabpy_server.app.app.TabPyState')
142			def test_env_variables_in_config(self, mock_state, mock_get_state,
143			mock_path_exists, mock_os):
144			mock_os.environ = {'foo': 'baz'}
145			config_file = self.config_file
146			config_file.write('[TabPy]\n'
147			'TABPY_PORT = %(foo)sbar'.encode())
148			config_file.close()
149
150			app = TabPyApp(self.config_file.name)
151			self.assertEqual(app.settings['port'], 'bazbar')
152
153
154			class TestTransferProtocolValidation(unittest.TestCase):
155			def assertTabPyAppRaisesRuntimeError(self, expected_message):
156			with self.assertRaises(RuntimeError) as err:
157			TabPyApp(self.fp.name)
158			self.assertEqual(err.exception.args[0], expected_message)
159
160			@staticmethod
161			def mock_isfile(target_file, existing_files):
162			if target_file in existing_files:
163			return True
164			return False
165
166			@staticmethod
167			def raise_attribute_error():
168			raise AttributeError()
169
170			def __init__(self, args, *kwargs):
171			super(TestTransferProtocolValidation, self).__init__(args, *kwargs)
172			self.fp = None
173
174			def setUp(self):
175			self.fp = NamedTemporaryFile(mode='w+t', delete=False)
176
177			def tearDown(self):
178			os.remove(self.fp.name)
179			self.fp = None
180
181			def test_invalid_protocol(self):
182			self.fp.write("[TabPy]\n"
183			"TABPY_TRANSFER_PROTOCOL = gopher")
184			self.fp.close()
185
186			self.assertTabPyAppRaisesRuntimeError(
187			'Unsupported transfer protocol: gopher')
188
189			def test_http(self):
190			self.fp.write("[TabPy]\n"
191			"TABPY_TRANSFER_PROTOCOL = http")
192			self.fp.close()
193
194			app = TabPyApp(self.fp.name)
195			self.assertEqual(app.settings['transfer_protocol'], 'http')
196
197			def test_https_without_cert_and_key(self):
198			self.fp.write("[TabPy]\n"
199			"TABPY_TRANSFER_PROTOCOL = https")
200			self.fp.close()
201
202			self.assertTabPyAppRaisesRuntimeError('Error using HTTPS: The paramete'
203			'r(s) TABPY_CERTIFICATE_FILE and'
204			' TABPY_KEY_FILE must be set.')
205
206			def test_https_without_cert(self):
207			self.fp.write(
208			"[TabPy]\n"
209			"TABPY_TRANSFER_PROTOCOL = https\n"
210			"TABPY_KEY_FILE = foo")
211			self.fp.close()
212
213			self.assertTabPyAppRaisesRuntimeError(
214			'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE must '
215			'be set.')
216
217			def test_https_without_key(self):
218			self.fp.write("[TabPy]\n"
219			"TABPY_TRANSFER_PROTOCOL = https\n"
220			"TABPY_CERTIFICATE_FILE = foo")
221			self.fp.close()
222
223			self.assertTabPyAppRaisesRuntimeError(
224			'Error using HTTPS: The parameter(s) TABPY_KEY_FILE must be set.')
225
226			@patch('tabpy.tabpy_server.app.app.os.path')
227			def test_https_cert_and_key_file_not_found(self, mock_path):
228			self.fp.write("[TabPy]\n"
229			"TABPY_TRANSFER_PROTOCOL = https\n"
230			"TABPY_CERTIFICATE_FILE = foo\n"
231			"TABPY_KEY_FILE = bar")
232			self.fp.close()
233
234			mock_path.isfile.side_effect = lambda x: self.mock_isfile(
235			x, {self.fp.name})
236
237			self.assertTabPyAppRaisesRuntimeError(
238			'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE and '
239			'TABPY_KEY_FILE must point to an existing file.')
240
241			@patch('tabpy.tabpy_server.app.app.os.path')
242			def test_https_cert_file_not_found(self, mock_path):
243			self.fp.write("[TabPy]\n"
244			"TABPY_TRANSFER_PROTOCOL = https\n"
245			"TABPY_CERTIFICATE_FILE = foo\n"
246			"TABPY_KEY_FILE = bar")
247			self.fp.close()
248
249			mock_path.isfile.side_effect = lambda x: self.mock_isfile(
250			x, {self.fp.name, 'bar'})
251
252			self.assertTabPyAppRaisesRuntimeError(
253			'Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE '
254			'must point to an existing file.')
255
256			@patch('tabpy.tabpy_server.app.app.os.path')
257			def test_https_key_file_not_found(self, mock_path):
258			self.fp.write("[TabPy]\n"
259			"TABPY_TRANSFER_PROTOCOL = https\n"
260			"TABPY_CERTIFICATE_FILE = foo\n"
261			"TABPY_KEY_FILE = bar")
262			self.fp.close()
263
264			mock_path.isfile.side_effect = lambda x: self.mock_isfile(
265			x, {self.fp.name, 'foo'})
266
267			self.assertTabPyAppRaisesRuntimeError(
268			'Error using HTTPS: The parameter(s) TABPY_KEY_FILE '
269			'must point to an existing file.')
270
271			@patch('tabpy.tabpy_server.app.app.os.path.isfile', return_value=True)
272			@patch('tabpy.tabpy_server.app.util.validate_cert')
273			def test_https_success(self, mock_isfile, mock_validate_cert):
274			self.fp.write("[TabPy]\n"
275			"TABPY_TRANSFER_PROTOCOL = HtTpS\n"
276			"TABPY_CERTIFICATE_FILE = foo\n"
277			"TABPY_KEY_FILE = bar")
278			self.fp.close()
279
280			app = TabPyApp(self.fp.name)
281
282			self.assertEqual(app.settings['transfer_protocol'], 'https')
283			self.assertEqual(app.settings['certificate_file'], 'foo')
284			self.assertEqual(app.settings['key_file'], 'bar')
285
286
287			class TestCertificateValidation(unittest.TestCase):
288			def assertValidateCertRaisesRuntimeError(self, expected_message, path):
289			with self.assertRaises(RuntimeError) as err:
290			validate_cert(path)
291			self.assertEqual(err.exception.args[0], expected_message)
292
293			def __init__(self, args, *kwargs):
294			super(TestCertificateValidation, self).__init__(args, *kwargs)
295			self.resources_path = os.path.join(
296			os.path.dirname(__file__), 'resources')
297
298			def test_expired_cert(self):
299			path = os.path.join(self.resources_path, 'expired.crt')
300			message = ('Error using HTTPS: The certificate provided expired '
301			'on 2018-08-18 19:47:18.')
302			self.assertValidateCertRaisesRuntimeError(message, path)
303
304			def test_future_cert(self):
305			path = os.path.join(self.resources_path, 'future.crt')
306			message = ('Error using HTTPS: The certificate provided is not valid '
307			'until 3001-01-01 00:00:00.')
308			self.assertValidateCertRaisesRuntimeError(message, path)
309
310			def test_valid_cert(self):
311			path = os.path.join(self.resources_path, 'valid.crt')
312			validate_cert(path)
313

tableau / TabPy

Push — master ( 305552...00a4d3 )

TestTransferProtocolValidation.test_https_without_cert() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like