tests.unit.server_tests.test_config.TestCertificateValidation.test_expired_cert() - Code Metrics - Inspection of "Merge pull request #612 from tableau/jichikawa-dev..." - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 5716a6...96aa26 )

unknown

created 2023-06-21 21:30 UTC

TestCertificateValidation.test_expired_cert() A

↳ Parent: tests.unit.server_tests.test_config

Complexity

Conditions

Size

Total Lines	7
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	5
dl	0
loc	7
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)

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			class TestTransferProtocolValidation(unittest.TestCase):
213			def assertTabPyAppRaisesRuntimeError(self, expected_message):
214			with self.assertRaises(RuntimeError) as err:
215			TabPyApp(self.fp.name)
216			self.assertEqual(err.exception.args[0], expected_message)
217
218			@staticmethod
219			def mock_isfile(target_file, existing_files):
220			if target_file in existing_files:
221			return True
222			return False
223
224			@staticmethod
225			def raise_attribute_error():
226			raise AttributeError()
227
228			def __init__(self, args, *kwargs):
229			super(TestTransferProtocolValidation, self).__init__(args, *kwargs)
230			self.fp = None
231
232			def setUp(self):
233			self.fp = NamedTemporaryFile(mode="w+t", delete=False)
234
235			def tearDown(self):
236			os.remove(self.fp.name)
237			self.fp = None
238
239			def test_invalid_protocol(self):
240			self.fp.write("[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = gopher")
241			self.fp.close()
242
243			self.assertTabPyAppRaisesRuntimeError("Unsupported transfer protocol: gopher")
244
245			def test_http(self):
246			self.fp.write("[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = http")
247			self.fp.close()
248
249			app = TabPyApp(self.fp.name)
250			self.assertEqual(app.settings["transfer_protocol"], "http")
251
252			def test_https_without_cert_and_key(self):
253			self.fp.write("[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = https")
254			self.fp.close()
255
256			self.assertTabPyAppRaisesRuntimeError(
257			"Error using HTTPS: The paramete"
258			"r(s) TABPY_CERTIFICATE_FILE and"
259			" TABPY_KEY_FILE must be set."
260			)
261
262			def test_https_without_cert(self):
263			self.fp.write(
264			"[TabPy]\n" "TABPY_TRANSFER_PROTOCOL = https\n" "TABPY_KEY_FILE = foo"
265			)
266			self.fp.close()
267
268			self.assertTabPyAppRaisesRuntimeError(
269			"Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE must " "be set."
270			)
271
272			def test_https_without_key(self):
273			self.fp.write(
274			"[TabPy]\n"
275			"TABPY_TRANSFER_PROTOCOL = https\n"
276			"TABPY_CERTIFICATE_FILE = foo"
277			)
278			self.fp.close()
279
280			self.assertTabPyAppRaisesRuntimeError(
281			"Error using HTTPS: The parameter(s) TABPY_KEY_FILE must be set."
282			)
283
284			@patch("tabpy.tabpy_server.app.app.os.path")
285			def test_https_cert_and_key_file_not_found(self, mock_path):
286			self.fp.write(
287			"[TabPy]\n"
288			"TABPY_TRANSFER_PROTOCOL = https\n"
289			"TABPY_CERTIFICATE_FILE = foo\n"
290			"TABPY_KEY_FILE = bar"
291			)
292			self.fp.close()
293
294			mock_path.isfile.side_effect = lambda x: self.mock_isfile(x, {self.fp.name})
295
296			self.assertTabPyAppRaisesRuntimeError(
297			"Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE and "
298			"TABPY_KEY_FILE must point to an existing file."
299			)
300
301			@patch("tabpy.tabpy_server.app.app.os.path")
302			def test_https_cert_file_not_found(self, mock_path):
303			self.fp.write(
304			"[TabPy]\n"
305			"TABPY_TRANSFER_PROTOCOL = https\n"
306			"TABPY_CERTIFICATE_FILE = foo\n"
307			"TABPY_KEY_FILE = bar"
308			)
309			self.fp.close()
310
311			mock_path.isfile.side_effect = lambda x: self.mock_isfile(
312			x, {self.fp.name, "bar"}
313			)
314
315			self.assertTabPyAppRaisesRuntimeError(
316			"Error using HTTPS: The parameter(s) TABPY_CERTIFICATE_FILE "
317			"must point to an existing file."
318			)
319
320			@patch("tabpy.tabpy_server.app.app.os.path")
321			def test_https_key_file_not_found(self, mock_path):
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			mock_path.isfile.side_effect = lambda x: self.mock_isfile(
331			x, {self.fp.name, "foo"}
332			)
333
334			self.assertTabPyAppRaisesRuntimeError(
335			"Error using HTTPS: The parameter(s) TABPY_KEY_FILE "
336			"must point to an existing file."
337			)
338
339			@patch("tabpy.tabpy_server.app.app.os.path.isfile", return_value=True)
340			@patch("tabpy.tabpy_server.app.util.validate_cert")
341			def test_https_success(self, mock_isfile, mock_validate_cert):
342			self.fp.write(
343			"[TabPy]\n"
344			"TABPY_TRANSFER_PROTOCOL = HtTpS\n"
345			"TABPY_CERTIFICATE_FILE = foo\n"
346			"TABPY_KEY_FILE = bar"
347			)
348			self.fp.close()
349
350			app = TabPyApp(self.fp.name)
351
352			self.assertEqual(app.settings["transfer_protocol"], "https")
353			self.assertEqual(app.settings["certificate_file"], "foo")
354			self.assertEqual(app.settings["key_file"], "bar")
355
356
357			class TestCertificateValidation(unittest.TestCase):
358			def assertValidateCertRaisesRuntimeError(self, expected_message, path):
359			with self.assertRaises(RuntimeError) as err:
360			validate_cert(path)
361			self.assertEqual(err.exception.args[0], expected_message)
362
363			def __init__(self, args, *kwargs):
364			super(TestCertificateValidation, self).__init__(args, *kwargs)
365			self.resources_path = os.path.join(os.path.dirname(__file__), "resources")
366
367			def test_expired_cert(self):
368			path = os.path.join(self.resources_path, "expired.crt")
369			message = (
370			"Error using HTTPS: The certificate provided expired "
371			"on 2018-08-18 19:47:18."
372			)
373			self.assertValidateCertRaisesRuntimeError(message, path)
374
375			def test_future_cert(self):
376			path = os.path.join(self.resources_path, "future.crt")
377			message = (
378			"Error using HTTPS: The certificate provided is not valid "
379			"until 3001-01-01 00:00:00."
380			)
381			self.assertValidateCertRaisesRuntimeError(message, path)
382
383			def test_valid_cert(self):
384			path = os.path.join(self.resources_path, "valid.crt")
385			validate_cert(path)
386

tableau / TabPy

Push — master ( 5716a6...96aa26 )

TestCertificateValidation.test_expired_cert() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like