tests.unit.server_tests.test_config.TestPartialConfigFile.setUp() - Code Metrics - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

TestPartialConfigFile.setUp() A
last analyzed 2024-11-25 22:00 UTC

↳ Parent: tests.unit.server_tests.test_config

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

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)

    @patch('builtins.input', return_value='y')
    @patch("tabpy.tabpy_server.app.app.os")
    @patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=False)
    @patch("tabpy.tabpy_server.app.app.PythonServiceHandler")
    @patch("tabpy.tabpy_server.app.app._get_state_from_file")
    @patch("tabpy.tabpy_server.app.app.TabPyState")
    def test_handle_configuration_without_authentication(
        self,
        mock_tabpy_state,
        mock_get_state_from_file,
        mock_psws,
        mock_os_path_exists,
        mock_os,
        mock_input,
    ):
        TabPyApp(None)
        mock_input.assert_not_called()

        TabPyApp(None, False)
        mock_input.assert_called()

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)

    @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_min_tls_setting_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_MINIMUM_TLS_VERSION = TLSv1_3".encode())
        config_file.close()

        app = TabPyApp(self.config_file.name)
        self.assertEqual(app.settings["minimum_tls_version"], "TLSv1_3")

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			@patch('builtins.input', return_value='y')
72			@patch("tabpy.tabpy_server.app.app.os")
73			@patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=False)
74			@patch("tabpy.tabpy_server.app.app.PythonServiceHandler")
75			@patch("tabpy.tabpy_server.app.app._get_state_from_file")
76			@patch("tabpy.tabpy_server.app.app.TabPyState")
77			def test_handle_configuration_without_authentication(
78			self,
79			mock_tabpy_state,
80			mock_get_state_from_file,
81			mock_psws,
82			mock_os_path_exists,
83			mock_os,
84			mock_input,
85			):
86			TabPyApp(None)
87			mock_input.assert_not_called()
88
89			TabPyApp(None, False)
90			mock_input.assert_called()
91
92			class TestPartialConfigFile(unittest.TestCase):
93			def setUp(self):
94			self.config_file = NamedTemporaryFile(delete=False)
95
96			def tearDown(self):
97			os.remove(self.config_file.name)
98			self.config_file = None
99
100			@patch("tabpy.tabpy_server.app.app.TabPyState")
101			@patch("tabpy.tabpy_server.app.app._get_state_from_file")
102			@patch("tabpy.tabpy_server.app.app.PythonServiceHandler")
103			@patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=True)
104			@patch("tabpy.tabpy_server.app.app.os")
105			def test_config_file_present(
106			self,
107			mock_os,
108			mock_path_exists,
109			mock_psws,
110			mock_management_util,
111			mock_tabpy_state,
112			):
113			self.assertTrue(self.config_file is not None)
114			config_file = self.config_file
115			config_file.write(
116			"[TabPy]\n"
117			"TABPY_QUERY_OBJECT_PATH = foo\n"
118			"TABPY_STATE_PATH = bar\n".encode()
119			)
120			config_file.close()
121
122			mock_os.path.realpath.return_value = "bar"
123			mock_os.environ = {"TABPY_PORT": "1234"}
124
125			app = TabPyApp(config_file.name)
126
127			self.assertEqual(app.settings["port"], "1234")
128			self.assertEqual(
129			app.settings["server_version"], open("tabpy/VERSION").read().strip()
130			)
131			self.assertEqual(app.settings["upload_dir"], "foo")
132			self.assertEqual(app.settings["state_file_path"], "bar")
133			self.assertEqual(app.settings["transfer_protocol"], "http")
134			self.assertTrue("certificate_file" not in app.settings)
135			self.assertTrue("key_file" not in app.settings)
136			self.assertEqual(app.settings["log_request_context"], False)
137			self.assertEqual(app.settings["evaluate_timeout"], 30)
138
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_custom_evaluate_timeout_valid(
143			self, mock_state, mock_get_state_from_file, mock_path_exists
144			):
145			self.assertTrue(self.config_file is not None)
146			config_file = self.config_file
147			config_file.write("[TabPy]\n" "TABPY_EVALUATE_TIMEOUT = 1996".encode())
148			config_file.close()
149
150			app = TabPyApp(self.config_file.name)
151			self.assertEqual(app.settings["evaluate_timeout"], 1996.0)
152
153			@patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=True)
154			@patch("tabpy.tabpy_server.app.app._get_state_from_file")
155			@patch("tabpy.tabpy_server.app.app.TabPyState")
156			def test_custom_evaluate_timeout_invalid(
157			self, mock_state, mock_get_state_from_file, mock_path_exists
158			):
159			self.assertTrue(self.config_file is not None)
160			config_file = self.config_file
161			config_file.write(
162			"[TabPy]\n" 'TABPY_EVALUATE_TIMEOUT = "im not a float"'.encode()
163			)
164			config_file.close()
165
166			with self.assertRaises(ValueError):
167			TabPyApp(self.config_file.name)
168
169			@patch("tabpy.tabpy_server.app.app.os")
170			@patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=True)
171			@patch("tabpy.tabpy_server.app.app._get_state_from_file")
172			@patch("tabpy.tabpy_server.app.app.TabPyState")
173			def test_env_variables_in_config(
174			self, mock_state, mock_get_state, mock_path_exists, mock_os
175			):
176			mock_os.environ = {"foo": "baz"}
177			config_file = self.config_file
178			config_file.write("[TabPy]\n" "TABPY_PORT = %(foo)sbar".encode())
179			config_file.close()
180
181			app = TabPyApp(self.config_file.name)
182			self.assertEqual(app.settings["port"], "bazbar")
183
184			@patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=True)
185			@patch("tabpy.tabpy_server.app.app._get_state_from_file")
186			@patch("tabpy.tabpy_server.app.app.TabPyState")
187			def test_gzip_setting_on_valid(
188			self, mock_state, mock_get_state_from_file, mock_path_exists
189			):
190			self.assertTrue(self.config_file is not None)
191			config_file = self.config_file
192			config_file.write("[TabPy]\n" "TABPY_GZIP_ENABLE = true".encode())
193			config_file.close()
194
195			app = TabPyApp(self.config_file.name)
196			self.assertEqual(app.settings["gzip_enabled"], True)
197
198			@patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=True)
199			@patch("tabpy.tabpy_server.app.app._get_state_from_file")
200			@patch("tabpy.tabpy_server.app.app.TabPyState")
201			def test_gzip_setting_off_valid(
202			self, mock_state, mock_get_state_from_file, mock_path_exists
203			):
204			self.assertTrue(self.config_file is not None)
205			config_file = self.config_file
206			config_file.write("[TabPy]\n" "TABPY_GZIP_ENABLE = false".encode())
207			config_file.close()
208
209			app = TabPyApp(self.config_file.name)
210			self.assertEqual(app.settings["gzip_enabled"], False)
211
212			@patch("tabpy.tabpy_server.app.app.os.path.exists", return_value=True)
213			@patch("tabpy.tabpy_server.app.app._get_state_from_file")
214			@patch("tabpy.tabpy_server.app.app.TabPyState")
215			def test_min_tls_setting_valid(
216			self, mock_state, mock_get_state_from_file, mock_path_exists
217			):
218			self.assertTrue(self.config_file is not None)
219			config_file = self.config_file
220			config_file.write("[TabPy]\n" "TABPY_MINIMUM_TLS_VERSION = TLSv1_3".encode())
221			config_file.close()
222
223			app = TabPyApp(self.config_file.name)
224			self.assertEqual(app.settings["minimum_tls_version"], "TLSv1_3")
225
226			class TestTransferProtocolValidation(unittest.TestCase):
227			def assertTabPyAppRaisesRuntimeError(self, expected_message):
228			with self.assertRaises(RuntimeError) as err:
229			TabPyApp(self.fp.name)
230			self.assertEqual(err.exception.args[0], expected_message)
231
232			@staticmethod
233			def mock_isfile(target_file, existing_files):
234			if target_file in existing_files:
235			return True
236			return False
237
238			@staticmethod
239			def raise_attribute_error():
240			raise AttributeError()
241
242			def __init__(self, args, *kwargs):
243			super(TestTransferProtocolValidation, self).__init__(args, *kwargs)
244			self.fp = None
245
246			def setUp(self):
247			self.fp = NamedTemporaryFile(mode="w+t", delete=False)
248
249			def tearDown(self):
250			os.remove(self.fp.name)
251			self.fp = None
252
253			def test_invalid_protocol(self):
254			self.fp.write("[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = gopher")
255			self.fp.close()
256
257			self.assertTabPyAppRaisesRuntimeError("Unsupported transfer protocol: gopher")
258
259			def test_http(self):
260			self.fp.write("[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = http")
261			self.fp.close()
262
263			app = TabPyApp(self.fp.name)
264			self.assertEqual(app.settings["transfer_protocol"], "http")
265
266			def test_https_without_cert_and_key(self):
267			self.fp.write("[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = https")
268			self.fp.close()
269
270			self.assertTabPyAppRaisesRuntimeError(
271			"Error using HTTPS: The paramete"
272			"r(s) TABPY_CERTIFICATE_FILE and"
273			" TABPY_KEY_FILE must be set."
274			)
275
276			def test_https_without_cert(self):
277			self.fp.write(
278			"[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = https\n" "TABPY_KEY_FILE = foo"
279			)
280			self.fp.close()
281
282			self.assertTabPyAppRaisesRuntimeError(
283			"Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE must " "be set."
284			)
285
286			def test_https_without_key(self):
287			self.fp.write(
288			"[TabPy]\n"
289			"TABPY_TRANSFER_PROTOCOL = https\n"
290			"TABPY_CERTIFICATE_FILE = foo"
291			)
292			self.fp.close()
293
294			self.assertTabPyAppRaisesRuntimeError(
295			"Error using HTTPS: The parameter(s) TABPY_KEY_FILE must be set."
296			)
297
298			@patch("tabpy.tabpy_server.app.app.os.path")
299			def test_https_cert_and_key_file_not_found(self, mock_path):
300			self.fp.write(
301			"[TabPy]\n"
302			"TABPY_TRANSFER_PROTOCOL = https\n"
303			"TABPY_CERTIFICATE_FILE = foo\n"
304			"TABPY_KEY_FILE = bar"
305			)
306			self.fp.close()
307
308			mock_path.isfile.side_effect = lambda x: self.mock_isfile(x, {self.fp.name})
309
310			self.assertTabPyAppRaisesRuntimeError(
311			"Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE and "
312			"TABPY_KEY_FILE must point to an existing file."
313			)
314
315			@patch("tabpy.tabpy_server.app.app.os.path")
316			def test_https_cert_file_not_found(self, mock_path):
317			self.fp.write(
318			"[TabPy]\n"
319			"TABPY_TRANSFER_PROTOCOL = https\n"
320			"TABPY_CERTIFICATE_FILE = foo\n"
321			"TABPY_KEY_FILE = bar"
322			)
323			self.fp.close()
324
325			mock_path.isfile.side_effect = lambda x: self.mock_isfile(
326			x, {self.fp.name, "bar"}
327			)
328
329			self.assertTabPyAppRaisesRuntimeError(
330			"Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE "
331			"must point to an existing file."
332			)
333
334			@patch("tabpy.tabpy_server.app.app.os.path")
335			def test_https_key_file_not_found(self, mock_path):
336			self.fp.write(
337			"[TabPy]\n"
338			"TABPY_TRANSFER_PROTOCOL = https\n"
339			"TABPY_CERTIFICATE_FILE = foo\n"
340			"TABPY_KEY_FILE = bar"
341			)
342			self.fp.close()
343
344			mock_path.isfile.side_effect = lambda x: self.mock_isfile(
345			x, {self.fp.name, "foo"}
346			)
347
348			self.assertTabPyAppRaisesRuntimeError(
349			"Error using HTTPS: The parameter(s) TABPY_KEY_FILE "
350			"must point to an existing file."
351			)
352
353			@patch("tabpy.tabpy_server.app.app.os.path.isfile", return_value=True)
354			@patch("tabpy.tabpy_server.app.util.validate_cert")
355			def test_https_success(self, mock_isfile, mock_validate_cert):
356			self.fp.write(
357			"[TabPy]\n"
358			"TABPY_TRANSFER_PROTOCOL = HtTpS\n"
359			"TABPY_CERTIFICATE_FILE = foo\n"
360			"TABPY_KEY_FILE = bar"
361			)
362			self.fp.close()
363
364			app = TabPyApp(self.fp.name)
365
366			self.assertEqual(app.settings["transfer_protocol"], "https")
367			self.assertEqual(app.settings["certificate_file"], "foo")
368			self.assertEqual(app.settings["key_file"], "bar")
369
370
371			class TestCertificateValidation(unittest.TestCase):
372			def assertValidateCertRaisesRuntimeError(self, expected_message, path):
373			with self.assertRaises(RuntimeError) as err:
374			validate_cert(path)
375			self.assertEqual(err.exception.args[0], expected_message)
376
377			def __init__(self, args, *kwargs):
378			super(TestCertificateValidation, self).__init__(args, *kwargs)
379			self.resources_path = os.path.join(os.path.dirname(__file__), "resources")
380
381			def test_expired_cert(self):
382			path = os.path.join(self.resources_path, "expired.crt")
383			message = (
384			"Error using HTTPS: The certificate provided expired "
385			"on 2018-08-18 19:47:18."
386			)
387			self.assertValidateCertRaisesRuntimeError(message, path)
388
389			def test_future_cert(self):
390			path = os.path.join(self.resources_path, "future.crt")
391			message = (
392			"Error using HTTPS: The certificate provided is not valid "
393			"until 3001-01-01 00:00:00."
394			)
395			self.assertValidateCertRaisesRuntimeError(message, path)
396
397			def test_valid_cert(self):
398			path = os.path.join(self.resources_path, "valid.crt")
399			validate_cert(path)
400

tableau / TabPy

TestPartialConfigFile.setUp() A last analyzed 2024-11-25 22:00 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestPartialConfigFile.setUp() A
last analyzed 2024-11-25 22:00 UTC