Passed
Push — master ( 938ce3...bc2c02 )
by
unknown
12:15
created

TestEvaluationPlainHandlerDefault.test_evaluation_default()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 6

Duplication

Lines 7
Ratio 100 %

Importance

Changes 0
Metric Value
eloc 6
dl 7
loc 7
rs 10
c 0
b 0
f 0
cc 1
nop 1
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 TestEvaluationPlainHandlerWithAuth(AsyncHTTPTestCase):
12
    @classmethod
13
    def setUpClass(cls):
14
        prefix = "__TestEvaluationPlainHandlerWithAuth_"
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 TestEvaluationPlainHandlerWithoutAuth(AsyncHTTPTestCase):
209
    @classmethod
210
    def setUpClass(cls):
211
        prefix = "__TestEvaluationPlainHandlerWithoutAuth_"
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(400, response.code)
289
290
291 View Code Duplication
class TestEvaluationPlainHandlerDisabled(AsyncHTTPTestCase):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
292
    @classmethod
293
    def setUpClass(cls):
294
        prefix = "__TestEvaluationPlainHandlerDisabled_"
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 View Code Duplication
class TestEvaluationPlainHandlerEnabled(AsyncHTTPTestCase):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
330
    @classmethod
331
    def setUpClass(cls):
332
        prefix = "__TestEvaluationPlainHandlerEnabled_"
333
334
        # create config file
335
        cls.config_file = tempfile.NamedTemporaryFile(
336
            mode="w+t", prefix=prefix, suffix=".conf", delete=False
337
        )
338
        cls.config_file.write(
339
            "[TabPy]\n"
340
            f"TABPY_EVALUATE_ENABLE = true"
341
        )
342
        cls.config_file.close()
343
344
        cls.script = (
345
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
346
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
347
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
348
        )
349
350
    @classmethod
351
    def tearDownClass(cls):
352
        os.remove(cls.config_file.name)
353
354
    def get_app(self):
355
        self.app = TabPyApp(self.config_file.name)
356
        return self.app._create_tornado_web_app()
357
358
    def test_evaluation_enabled(self):
359
        response = self.fetch(
360
            "/evaluate",
361
            method="POST",
362
            body=self.script
363
        )
364
        self.assertEqual(200, response.code)
365
366
367 View Code Duplication
class TestEvaluationPlainHandlerDefault(AsyncHTTPTestCase):
0 ignored issues
show
Duplication introduced by
This code seems to be duplicated in your project.
Loading history...
368
    @classmethod
369
    def setUpClass(cls):
370
        prefix = "__TestEvaluationPlainHandlerDefault_"
371
372
        # create config file
373
        cls.config_file = tempfile.NamedTemporaryFile(
374
            mode="w+t", prefix=prefix, suffix=".conf", delete=False
375
        )
376
        cls.config_file.write(
377
            "[TabPy]"
378
        )
379
        cls.config_file.close()
380
381
        cls.script = (
382
            '{"data":{"_arg1":[2,3],"_arg2":[3,-1]},'
383
            '"script":"res=[]\\nfor i in range(len(_arg1)):\\n  '
384
            'res.append(_arg1[i] * _arg2[i])\\nreturn res"}'
385
        )
386
387
    @classmethod
388
    def tearDownClass(cls):
389
        os.remove(cls.config_file.name)
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_default(self):
396
        response = self.fetch(
397
            "/evaluate",
398
            method="POST",
399
            body=self.script
400
        )
401
        self.assertEqual(200, response.code)
402