tests.unit.server_tests.test_evaluation_plane_handler.TestEvaluationPlaneHandlerWithAuth.tearDownClass() - Code Metrics - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

TestEvaluationPlaneHandlerWithAuth.tearDownClass() A
last analyzed 2024-11-25 22:00 UTC

↳ 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 json
import os
import tempfile
import string

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 TestEvaluationPlaneHandlerMaxRequestSize(AsyncHTTPTestCase):
    @classmethod
    def setUpClass(cls):
        prefix = "__TestEvaluationPlaneHandlerMaxRequestSize_"

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

    @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 create_large_payload(self):
        num_chars = 2 * 1024 * 1024 # 2MB Size
        large_string = string.printable * (num_chars // len(string.printable))
        large_string += string.printable[:num_chars % len(string.printable)]
        payload = {
            "data": { "_arg1": [1, large_string] },
            "script": "return _arg1"
        }
        return json.dumps(payload).encode('utf-8')

    def test_evaluation_payload_exceeds_max_request_size(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.create_large_payload()
        )
        self.assertEqual(413, response.code)

    def test_evaluation_max_request_size_not_applied(self):
        self.app.max_request_size = None
        response = self.fetch(
            "/evaluate",
            method="POST",
            body=self.create_large_payload()
        )
        self.assertEqual(200, response.code)
        self.assertEqual(1, json.loads(response.body)[0])

    def test_no_content_length_header_present(self):
        response = self.fetch(
            "/evaluate",
            method="POST",
            allow_nonstandard_methods=True
        )
        message = json.loads(response.body)["message"]
        # Ensure it reaches script processing stage in EvaluationPlaneHandler.post
        self.assertEqual("Error processing script", message)


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

tableau / TabPy

TestEvaluationPlaneHandlerWithAuth.tearDownClass() A last analyzed 2024-11-25 22:00 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

TestEvaluationPlaneHandlerWithAuth.tearDownClass() A
last analyzed 2024-11-25 22:00 UTC