tests.unit.server_tests.test_config.TestPartialConfigFile.test_gzip_setting_off_valid() - Code Metrics - Inspection of "Support Gzip requests" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#543)

unknown

created 2022-03-03 00:45 UTC

TestPartialConfigFile.test_gzip_setting_off_valid() A

↳ Parent: tests.unit.server_tests.test_config

Complexity

Conditions

Size

Total Lines	13
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	11
dl	0
loc	13
rs	9.85
c	0
b	0
f	0
cc	1
nop	4

import os
import unittest
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.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,
    ):
        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.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,
    ):
        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.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,
    ):
        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_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()

        with self.assertRaises(ValueError):
            TabPyApp(self.config_file.name)

    @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")

    @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_gzip_setting_on_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_GZIP_ENABLE = true".encode())
        config_file.close()

        app = TabPyApp(self.config_file.name)
        self.assertEqual(app.settings["gzip_enabled"], True)

    @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_gzip_setting_off_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_GZIP_ENABLE = false".encode())
        config_file.close()

        app = TabPyApp(self.config_file.name)
        self.assertEqual(app.settings["gzip_enabled"], False)

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

tableau / TabPy

Pull Request — master (#543)

TestPartialConfigFile.test_gzip_setting_off_valid() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like