tests.unit.server_tests.test_evaluation_plane_handler.TestEvaluationPlaneHandlerWithAuth.tearDownClass() - Code Metrics - Inspection of "Merge pull request #536 from tableau/hxi" - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b4f401...16d7f4 )

unknown

created 2022-01-20 01:34 UTC

TestEvaluationPlaneHandlerWithAuth.tearDownClass() A

↳ Parent: tests.unit.server_tests.test_evaluation_plane_handler

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

import base64
import os
import tempfile

from tornado.testing import AsyncHTTPTestCase

from tabpy.tabpy_server.app.app import TabPyApp
from tabpy.tabpy_server.handlers.util import hash_password


class TestEvaluationPlaneHandlerWithAuth(AsyncHTTPTestCase):
    @classmethod
    def setUpClass(cls):
        prefix = "__TestEvaluationPlaneHandlerWithAuth_"
        # create password file
        cls.pwd_file = tempfile.NamedTemporaryFile(
            mode="w+t", prefix=prefix, suffix=".txt", delete=False
        )
        username = "username"
        password = "password"
        cls.pwd_file.write(f"{username} {hash_password(username, password)}\n")
        cls.pwd_file.close()

        # create state.ini dir and file
        cls.state_dir = tempfile.mkdtemp(prefix=prefix)
        cls.state_file = open(os.path.join(cls.state_dir, "state.ini"), "w+")
        cls.state_file.write(
            "[Service Info]\n"
            "Name = TabPy Serve\n"
            "Description = \n"
            "Creation Time = 0\n"
            "Access-Control-Allow-Origin = \n"
            "Access-Control-Allow-Headers = \n"
            "Access-Control-Allow-Methods = \n"
            "\n"
            "[Query Objects Service Versions]\n"
            "\n"
            "[Query Objects Docstrings]\n"
            "\n"
            "[Meta]\n"
            "Revision Number = 1\n"
        )
        cls.state_file.close()

        # create config file
        cls.config_file = tempfile.NamedTemporaryFile(
            mode="w+t", prefix=prefix, suffix=".conf", delete=False
        )
        cls.config_file.write(
            "[TabPy]\n"
            f"TABPY_PWD_FILE = {cls.pwd_file.name}\n"
            f"TABPY_STATE_PATH = {cls.state_dir}"
        )
        cls.config_file.close()

        cls.script = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

        cls.script_not_present = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

        cls.args_not_present = (
            '{"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

        cls.args_not_sequential = (
            '{"data":{"_arg1":[2,3],"_arg3":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg3[i])\\nreturn res"}'
        )

        cls.nan_coverts_to_null =\
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'\
            '"script":"return [float(1), float(\\"NaN\\"), float(2)]"}'

        cls.script_returns_none = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"return None"}'
        )

    @classmethod
    def tearDownClass(cls):
        os.remove(cls.pwd_file.name)
        os.remove(cls.state_file.name)
        os.remove(cls.config_file.name)
        os.rmdir(cls.state_dir)

    def get_app(self):
        self.app = TabPyApp(self.config_file.name)
        return self.app._create_tornado_web_app()

    def test_no_creds_required_auth_fails(self):
        response = self.fetch("/evaluate", method="POST", body=self.script)
        self.assertEqual(401, response.code)

    def test_invalid_creds_fails(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("user:wrong_password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )
            },
        )
        self.assertEqual(401, response.code)

    def test_valid_creds_pass(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("username:password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )
            },
        )
        self.assertEqual(200, response.code)

    def test_script_not_present(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script_not_present,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("username:password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )
            },
        )
        self.assertEqual(400, response.code)

    def test_arguments_not_present(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.args_not_present,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("username:password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )

            },
        )
        self.assertEqual(500, response.code)

    def test_arguments_not_sequential(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.args_not_sequential,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("username:password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )
            },
        )
        self.assertEqual(400, response.code)

    def test_nan_converts_to_null(self):
        response = self.fetch(
            '/evaluate',
            method='POST',
            body=self.nan_coverts_to_null,
            headers={
                'Authorization': 'Basic {}'.
                format(
                    base64.b64encode('username:password'.encode('utf-8')).
                    decode('utf-8'))
            })
        self.assertEqual(200, response.code)
        self.assertEqual(b'[1.0, null, 2.0]', response.body)

    def test_script_returns_none(self):
        response = self.fetch(
            '/evaluate',
            method='POST',
            body=self.script_returns_none,
            headers={
                'Authorization': 'Basic {}'.
                format(
                    base64.b64encode('username:password'.encode('utf-8')).
                    decode('utf-8'))
            })
        self.assertEqual(200, response.code)
        self.assertEqual(b'null', response.body)


class TestEvaluationPlaneHandlerWithoutAuth(AsyncHTTPTestCase):
    @classmethod
    def setUpClass(cls):
        prefix = "__TestEvaluationPlaneHandlerWithoutAuth_"

        # create state.ini dir and file
        cls.state_dir = tempfile.mkdtemp(prefix=prefix)
        cls.state_file = open(os.path.join(cls.state_dir, "state.ini"), "w+")
        cls.state_file.write(
            "[Service Info]\n"
            "Name = TabPy Serve\n"
            "Description = \n"
            "Creation Time = 0\n"
            "Access-Control-Allow-Origin = \n"
            "Access-Control-Allow-Headers = \n"
            "Access-Control-Allow-Methods = \n"
            "\n"
            "[Query Objects Service Versions]\n"
            "\n"
            "[Query Objects Docstrings]\n"
            "\n"
            "[Meta]\n"
            "Revision Number = 1\n"
        )
        cls.state_file.close()

        cls.script = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

        cls.script_not_present = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

        cls.args_not_present = (
            '{"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

        cls.args_not_sequential = (
            '{"data":{"_arg1":[2,3],"_arg3":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg3[i])\\nreturn res"}'
        )

        cls.nan_coverts_to_null =\
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'\
            '"script":"return [float(1), float(\\"NaN\\"), float(2)]"}'

        cls.script_returns_none = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"return None"}'
        )

    @classmethod
    def tearDownClass(cls):
        os.remove(cls.state_file.name)
        os.rmdir(cls.state_dir)

    def get_app(self):
        self.app = TabPyApp(None)
        return self.app._create_tornado_web_app()

    def test_creds_no_auth_fails(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("username:password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )
            },
        )
        self.assertEqual(406, response.code)


class TestEvaluationPlaneHandlerDisabledWithoutAuth(AsyncHTTPTestCase):

    @classmethod
    def setUpClass(cls):
        prefix = "__TestEvaluationPlaneHandlerDisabledWithoutAuth_"

        # create config file
        cls.config_file = tempfile.NamedTemporaryFile(
            mode="w+t", prefix=prefix, suffix=".conf", delete=False
        )
        cls.config_file.write(
            "[TabPy]\n"
            f"TABPY_EVALUATE_ENABLE = false"
        )
        cls.config_file.close()

        cls.script = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

    @classmethod
    def tearDownClass(cls):
        os.remove(cls.config_file.name)

    def get_app(self):
        self.app = TabPyApp(self.config_file.name)
        return self.app._create_tornado_web_app()

    def test_evaluation_disabled_fails(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script
        )
        self.assertEqual(404, response.code)


class TestEvaluationPlaneHandlerDisabledWithAuth(AsyncHTTPTestCase):
    @classmethod

    def setUpClass(cls):
        prefix = "__TestEvaluationPlaneHandlerDisabledWithAuth_"

        # create password file
        cls.pwd_file = tempfile.NamedTemporaryFile(
            mode="w+t", prefix=prefix, suffix=".txt", delete=False
        )
        username = "username"
        password = "password"
        cls.pwd_file.write(f"{username} {hash_password(username, password)}\n")
        cls.pwd_file.close()

        # create state.ini dir and file
        cls.state_dir = tempfile.mkdtemp(prefix=prefix)
        cls.state_file = open(os.path.join(cls.state_dir, "state.ini"), "w+")
        cls.state_file.write(
            "[Service Info]\n"
            "Name = TabPy Serve\n"
            "Description = \n"
            "Creation Time = 0\n"
            "Access-Control-Allow-Origin = \n"
            "Access-Control-Allow-Headers = \n"
            "Access-Control-Allow-Methods = \n"
            "\n"
            "[Query Objects Service Versions]\n"
            "\n"
            "[Query Objects Docstrings]\n"
            "\n"
            "[Meta]\n"
            "Revision Number = 1\n"
        )
        cls.state_file.close()

        # create config file
        cls.config_file = tempfile.NamedTemporaryFile(
            mode="w+t", prefix=prefix, suffix=".conf", delete=False
        )
        cls.config_file.write(
            "[TabPy]\n"
            f"TABPY_PWD_FILE = {cls.pwd_file.name}\n"
            f"TABPY_STATE_PATH = {cls.state_dir}\n"
            f"TABPY_EVALUATE_ENABLE = false"
        )
        cls.config_file.close()

        cls.script = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

    @classmethod
    def tearDownClass(cls):
        os.remove(cls.pwd_file.name)
        os.remove(cls.state_file.name)
        os.remove(cls.config_file.name)
        os.rmdir(cls.state_dir)

    def get_app(self):
        self.app = TabPyApp(self.config_file.name)
        return self.app._create_tornado_web_app()

    def test_evaluation_disabled_fails_with_invalid_creds(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("user:wrong_password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )
            },
        )
        self.assertEqual(401, response.code)

    def test_evaluation_disabled_fails_with_valid_creds(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script,
            headers={
                "Authorization": "Basic {}".format(
                    base64.b64encode("username:password".encode("utf-8")).decode(
                        "utf-8"
                    )
                )
            },
        )
        self.assertEqual(404, response.code)


class TestEvaluationPlaneHandlerEnabled(AsyncHTTPTestCase):

    @classmethod
    def setUpClass(cls):
        prefix = "__TestEvaluationPlaneHandlerEnabled_"

        # create config file
        cls.config_file = tempfile.NamedTemporaryFile(
            mode="w+t", prefix=prefix, suffix=".conf", delete=False
        )
        cls.config_file.write(
            "[TabPy]\n"
            f"TABPY_EVALUATE_ENABLE = true"
        )
        cls.config_file.close()

        cls.script = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

    @classmethod
    def tearDownClass(cls):
        os.remove(cls.config_file.name)

    def get_app(self):
        self.app = TabPyApp(self.config_file.name)
        return self.app._create_tornado_web_app()

    def test_evaluation_enabled(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script
        )
        self.assertEqual(200, response.code)


class TestEvaluationPlaneHandlerDefault(AsyncHTTPTestCase):

    @classmethod
    def setUpClass(cls):
        prefix = "__TestEvaluationPlaneHandlerDefault_"

        # create config file
        cls.config_file = tempfile.NamedTemporaryFile(
            mode="w+t", prefix=prefix, suffix=".conf", delete=False
        )
        cls.config_file.write(
            "[TabPy]"
        )
        cls.config_file.close()

        cls.script = (
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
        )

    @classmethod
    def tearDownClass(cls):
        os.remove(cls.config_file.name)

    def get_app(self):
        self.app = TabPyApp(self.config_file.name)
        return self.app._create_tornado_web_app()

    def test_evaluation_default(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.script
        )
        self.assertEqual(200, response.code)


1		import base64
2		import os
3		import tempfile
4
5		from tornado.testing import AsyncHTTPTestCase
6
7		from tabpy.tabpy_server.app.app import TabPyApp
8		from tabpy.tabpy_server.handlers.util import hash_password
9
10
11		class TestEvaluationPlaneHandlerWithAuth(AsyncHTTPTestCase):
12		@classmethod
13		def setUpClass(cls):
14		prefix = "__TestEvaluationPlaneHandlerWithAuth_"
15		# create password file
16		cls.pwd_file = tempfile.NamedTemporaryFile(
17		mode="w+t", prefix=prefix, suffix=".txt", delete=False
18		)
19		username = "username"
20		password = "password"
21		cls.pwd_file.write(f"{username} {hash_password(username, password)}\n")
22		cls.pwd_file.close()
23
24		# create state.ini dir and file
25		cls.state_dir = tempfile.mkdtemp(prefix=prefix)
26		cls.state_file = open(os.path.join(cls.state_dir, "state.ini"), "w+")
27		cls.state_file.write(
28		"[Service Info]\n"
29		"Name = TabPy Serve\n"
30		"Description = \n"
31		"Creation Time = 0\n"
32		"Access-Control-Allow-Origin = \n"
33		"Access-Control-Allow-Headers = \n"
34		"Access-Control-Allow-Methods = \n"
35		"\n"
36		"[Query Objects Service Versions]\n"
37		"\n"
38		"[Query Objects Docstrings]\n"
39		"\n"
40		"[Meta]\n"
41		"Revision Number = 1\n"
42		)
43		cls.state_file.close()
44
45		# create config file
46		cls.config_file = tempfile.NamedTemporaryFile(
47		mode="w+t", prefix=prefix, suffix=".conf", delete=False
48		)
49		cls.config_file.write(
50		"[TabPy]\n"
51		f"TABPY_PWD_FILE = {cls.pwd_file.name}\n"
52		f"TABPY_STATE_PATH = {cls.state_dir}"
53		)
54		cls.config_file.close()
55
56		cls.script = (
57		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
58		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
59		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
60		)
61
62		cls.script_not_present = (
63		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
64		'"":"res=[]\\nfor i in range(len(_arg1)):\\n '
65		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
66		)
67
68		cls.args_not_present = (
69		'{"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
70		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
71		)
72
73		cls.args_not_sequential = (
74		'{"data":{"_arg1":[2,3],"_arg3":[3,-1]},'
75		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
76		'res.append(_arg1[i] * _arg3[i])\\nreturn res"}'
77		)
78
79		cls.nan_coverts_to_null =\
80		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'\
81		'"script":"return [float(1), float(\\"NaN\\"), float(2)]"}'
82
83		cls.script_returns_none = (
84		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
85		'"script":"return None"}'
86		)
87
88		@classmethod
89		def tearDownClass(cls):
90		os.remove(cls.pwd_file.name)
91		os.remove(cls.state_file.name)
92		os.remove(cls.config_file.name)
93		os.rmdir(cls.state_dir)
94
95		def get_app(self):
96		self.app = TabPyApp(self.config_file.name)
97		return self.app._create_tornado_web_app()
98
99		def test_no_creds_required_auth_fails(self):
100		response = self.fetch("/evaluate", method="POST", body=self.script)
101		self.assertEqual(401, response.code)
102
103		def test_invalid_creds_fails(self):
104		response = self.fetch(
105		"/evaluate",
106		method="POST",
107		body=self.script,
108		headers={
109		"Authorization": "Basic {}".format(
110		base64.b64encode("user:wrong_password".encode("utf-8")).decode(
111		"utf-8"
112		)
113		)
114		},
115		)
116		self.assertEqual(401, response.code)
117
118		def test_valid_creds_pass(self):
119		response = self.fetch(
120		"/evaluate",
121		method="POST",
122		body=self.script,
123		headers={
124		"Authorization": "Basic {}".format(
125		base64.b64encode("username:password".encode("utf-8")).decode(
126		"utf-8"
127		)
128		)
129		},
130		)
131		self.assertEqual(200, response.code)
132
133		def test_script_not_present(self):
134		response = self.fetch(
135		"/evaluate",
136		method="POST",
137		body=self.script_not_present,
138		headers={
139		"Authorization": "Basic {}".format(
140		base64.b64encode("username:password".encode("utf-8")).decode(
141		"utf-8"
142		)
143		)
144		},
145		)
146		self.assertEqual(400, response.code)
147
148		def test_arguments_not_present(self):
149		response = self.fetch(
150		"/evaluate",
151		method="POST",
152		body=self.args_not_present,
153		headers={
154		"Authorization": "Basic {}".format(
155		base64.b64encode("username:password".encode("utf-8")).decode(
156		"utf-8"
157		)
158		)
159
160		},
161		)
162		self.assertEqual(500, response.code)
163
164		def test_arguments_not_sequential(self):
165		response = self.fetch(
166		"/evaluate",
167		method="POST",
168		body=self.args_not_sequential,
169		headers={
170		"Authorization": "Basic {}".format(
171		base64.b64encode("username:password".encode("utf-8")).decode(
172		"utf-8"
173		)
174		)
175		},
176		)
177		self.assertEqual(400, response.code)
178
179		def test_nan_converts_to_null(self):
180		response = self.fetch(
181		'/evaluate',
182		method='POST',
183		body=self.nan_coverts_to_null,
184		headers={
185		'Authorization': 'Basic {}'.
186		format(
187		base64.b64encode('username:password'.encode('utf-8')).
188		decode('utf-8'))
189		})
190		self.assertEqual(200, response.code)
191		self.assertEqual(b'[1.0, null, 2.0]', response.body)
192
193		def test_script_returns_none(self):
194		response = self.fetch(
195		'/evaluate',
196		method='POST',
197		body=self.script_returns_none,
198		headers={
199		'Authorization': 'Basic {}'.
200		format(
201		base64.b64encode('username:password'.encode('utf-8')).
202		decode('utf-8'))
203		})
204		self.assertEqual(200, response.code)
205		self.assertEqual(b'null', response.body)
206
207
208		class TestEvaluationPlaneHandlerWithoutAuth(AsyncHTTPTestCase):
209		@classmethod
210		def setUpClass(cls):
211		prefix = "__TestEvaluationPlaneHandlerWithoutAuth_"
212
213		# create state.ini dir and file
214		cls.state_dir = tempfile.mkdtemp(prefix=prefix)
215		cls.state_file = open(os.path.join(cls.state_dir, "state.ini"), "w+")
216		cls.state_file.write(
217		"[Service Info]\n"
218		"Name = TabPy Serve\n"
219		"Description = \n"
220		"Creation Time = 0\n"
221		"Access-Control-Allow-Origin = \n"
222		"Access-Control-Allow-Headers = \n"
223		"Access-Control-Allow-Methods = \n"
224		"\n"
225		"[Query Objects Service Versions]\n"
226		"\n"
227		"[Query Objects Docstrings]\n"
228		"\n"
229		"[Meta]\n"
230		"Revision Number = 1\n"
231		)
232		cls.state_file.close()
233
234		cls.script = (
235		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
236		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
237		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
238		)
239
240		cls.script_not_present = (
241		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
242		'"":"res=[]\\nfor i in range(len(_arg1)):\\n '
243		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
244		)
245
246		cls.args_not_present = (
247		'{"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
248		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
249		)
250
251		cls.args_not_sequential = (
252		'{"data":{"_arg1":[2,3],"_arg3":[3,-1]},'
253		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
254		'res.append(_arg1[i] * _arg3[i])\\nreturn res"}'
255		)
256
257		cls.nan_coverts_to_null =\
258		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'\
259		'"script":"return [float(1), float(\\"NaN\\"), float(2)]"}'
260
261		cls.script_returns_none = (
262		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
263		'"script":"return None"}'
264		)
265
266		@classmethod
267		def tearDownClass(cls):
268		os.remove(cls.state_file.name)
269		os.rmdir(cls.state_dir)
270
271		def get_app(self):
272		self.app = TabPyApp(None)
273		return self.app._create_tornado_web_app()
274
275		def test_creds_no_auth_fails(self):
276		response = self.fetch(
277		"/evaluate",
278		method="POST",
279		body=self.script,
280		headers={
281		"Authorization": "Basic {}".format(
282		base64.b64encode("username:password".encode("utf-8")).decode(
283		"utf-8"
284		)
285		)
286		},
287		)
288		self.assertEqual(406, response.code)
289
290
291	View Code Duplication	class TestEvaluationPlaneHandlerDisabledWithoutAuth(AsyncHTTPTestCase):
		0 ignored issues – show Duplication introduced 2021-07-08 22:55 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
292		@classmethod
293		def setUpClass(cls):
294		prefix = "__TestEvaluationPlaneHandlerDisabledWithoutAuth_"
295
296		# create config file
297		cls.config_file = tempfile.NamedTemporaryFile(
298		mode="w+t", prefix=prefix, suffix=".conf", delete=False
299		)
300		cls.config_file.write(
301		"[TabPy]\n"
302		f"TABPY_EVALUATE_ENABLE = false"
303		)
304		cls.config_file.close()
305
306		cls.script = (
307		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
308		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
309		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
310		)
311
312		@classmethod
313		def tearDownClass(cls):
314		os.remove(cls.config_file.name)
315
316		def get_app(self):
317		self.app = TabPyApp(self.config_file.name)
318		return self.app._create_tornado_web_app()
319
320		def test_evaluation_disabled_fails(self):
321		response = self.fetch(
322		"/evaluate",
323		method="POST",
324		body=self.script
325		)
326		self.assertEqual(404, response.code)
327
328
329		class TestEvaluationPlaneHandlerDisabledWithAuth(AsyncHTTPTestCase):
330	View Code Duplication	@classmethod
		0 ignored issues – show Duplication introduced 2021-07-08 22:55 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
331		def setUpClass(cls):
332		prefix = "__TestEvaluationPlaneHandlerDisabledWithAuth_"
333
334		# create password file
335		cls.pwd_file = tempfile.NamedTemporaryFile(
336		mode="w+t", prefix=prefix, suffix=".txt", delete=False
337		)
338		username = "username"
339		password = "password"
340		cls.pwd_file.write(f"{username} {hash_password(username, password)}\n")
341		cls.pwd_file.close()
342
343		# create state.ini dir and file
344		cls.state_dir = tempfile.mkdtemp(prefix=prefix)
345		cls.state_file = open(os.path.join(cls.state_dir, "state.ini"), "w+")
346		cls.state_file.write(
347		"[Service Info]\n"
348		"Name = TabPy Serve\n"
349		"Description = \n"
350		"Creation Time = 0\n"
351		"Access-Control-Allow-Origin = \n"
352		"Access-Control-Allow-Headers = \n"
353		"Access-Control-Allow-Methods = \n"
354		"\n"
355		"[Query Objects Service Versions]\n"
356		"\n"
357		"[Query Objects Docstrings]\n"
358		"\n"
359		"[Meta]\n"
360		"Revision Number = 1\n"
361		)
362		cls.state_file.close()
363
364		# create config file
365		cls.config_file = tempfile.NamedTemporaryFile(
366		mode="w+t", prefix=prefix, suffix=".conf", delete=False
367		)
368		cls.config_file.write(
369		"[TabPy]\n"
370		f"TABPY_PWD_FILE = {cls.pwd_file.name}\n"
371		f"TABPY_STATE_PATH = {cls.state_dir}\n"
372		f"TABPY_EVALUATE_ENABLE = false"
373		)
374		cls.config_file.close()
375
376		cls.script = (
377		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
378		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
379		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
380		)
381
382		@classmethod
383		def tearDownClass(cls):
384		os.remove(cls.pwd_file.name)
385		os.remove(cls.state_file.name)
386		os.remove(cls.config_file.name)
387		os.rmdir(cls.state_dir)
388
389		def get_app(self):
390		self.app = TabPyApp(self.config_file.name)
391		return self.app._create_tornado_web_app()
392
393		def test_evaluation_disabled_fails_with_invalid_creds(self):
394		response = self.fetch(
395		"/evaluate",
396		method="POST",
397		body=self.script,
398		headers={
399		"Authorization": "Basic {}".format(
400		base64.b64encode("user:wrong_password".encode("utf-8")).decode(
401		"utf-8"
402		)
403		)
404		},
405		)
406		self.assertEqual(401, response.code)
407
408		def test_evaluation_disabled_fails_with_valid_creds(self):
409		response = self.fetch(
410		"/evaluate",
411		method="POST",
412		body=self.script,
413		headers={
414		"Authorization": "Basic {}".format(
415		base64.b64encode("username:password".encode("utf-8")).decode(
416		"utf-8"
417		)
418		)
419		},
420		)
421		self.assertEqual(404, response.code)
422
423
424	View Code Duplication	class TestEvaluationPlaneHandlerEnabled(AsyncHTTPTestCase):
		0 ignored issues – show Duplication introduced 2021-07-08 22:55 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
425		@classmethod
426		def setUpClass(cls):
427		prefix = "__TestEvaluationPlaneHandlerEnabled_"
428
429		# create config file
430		cls.config_file = tempfile.NamedTemporaryFile(
431		mode="w+t", prefix=prefix, suffix=".conf", delete=False
432		)
433		cls.config_file.write(
434		"[TabPy]\n"
435		f"TABPY_EVALUATE_ENABLE = true"
436		)
437		cls.config_file.close()
438
439		cls.script = (
440		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
441		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
442		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
443		)
444
445		@classmethod
446		def tearDownClass(cls):
447		os.remove(cls.config_file.name)
448
449		def get_app(self):
450		self.app = TabPyApp(self.config_file.name)
451		return self.app._create_tornado_web_app()
452
453		def test_evaluation_enabled(self):
454		response = self.fetch(
455		"/evaluate",
456		method="POST",
457		body=self.script
458		)
459		self.assertEqual(200, response.code)
460
461
462	View Code Duplication	class TestEvaluationPlaneHandlerDefault(AsyncHTTPTestCase):
		0 ignored issues – show Duplication introduced 2022-01-18 23:04 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
463		@classmethod
464		def setUpClass(cls):
465		prefix = "__TestEvaluationPlaneHandlerDefault_"
466
467		# create config file
468		cls.config_file = tempfile.NamedTemporaryFile(
469		mode="w+t", prefix=prefix, suffix=".conf", delete=False
470		)
471		cls.config_file.write(
472		"[TabPy]"
473		)
474		cls.config_file.close()
475
476		cls.script = (
477		'{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
478		'"script":"res=[]\\nfor i in range(len(_arg1)):\\n '
479		'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
480		)
481
482		@classmethod
483		def tearDownClass(cls):
484		os.remove(cls.config_file.name)
485
486		def get_app(self):
487		self.app = TabPyApp(self.config_file.name)
488		return self.app._create_tornado_web_app()
489
490		def test_evaluation_default(self):
491		response = self.fetch(
492		"/evaluate",
493		method="POST",
494		body=self.script
495		)
496		self.assertEqual(200, response.code)
497

tableau / TabPy

Push — master ( b4f401...16d7f4 )

TestEvaluationPlaneHandlerWithAuth.tearDownClass() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like