xref: /unit/test/test_python_application.py (revision 1261:0150f7bd061d)
1import time
2import unittest
3from unit.applications.lang.python import TestApplicationPython
4
5
6class TestPythonApplication(TestApplicationPython):
7    prerequisites = {'modules': ['python']}
8
9    def test_python_application_variables(self):
10        self.load('variables')
11
12        body = 'Test body string.'
13
14        resp = self.post(
15            headers={
16                'Host': 'localhost',
17                'Content-Type': 'text/html',
18                'Custom-Header': 'blah',
19                'Connection': 'close',
20            },
21            body=body,
22        )
23
24        self.assertEqual(resp['status'], 200, 'status')
25        headers = resp['headers']
26        header_server = headers.pop('Server')
27        self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header')
28        self.assertEqual(
29            headers.pop('Server-Software'),
30            header_server,
31            'server software header',
32        )
33
34        date = headers.pop('Date')
35        self.assertEqual(date[-4:], ' GMT', 'date header timezone')
36        self.assertLess(
37            abs(self.date_to_sec_epoch(date) - self.sec_epoch()),
38            5,
39            'date header',
40        )
41
42        self.assertDictEqual(
43            headers,
44            {
45                'Connection': 'close',
46                'Content-Length': str(len(body)),
47                'Content-Type': 'text/html',
48                'Request-Method': 'POST',
49                'Request-Uri': '/',
50                'Http-Host': 'localhost',
51                'Server-Protocol': 'HTTP/1.1',
52                'Custom-Header': 'blah',
53                'Wsgi-Version': '(1, 0)',
54                'Wsgi-Url-Scheme': 'http',
55                'Wsgi-Multithread': 'False',
56                'Wsgi-Multiprocess': 'True',
57                'Wsgi-Run-Once': 'False',
58            },
59            'headers',
60        )
61        self.assertEqual(resp['body'], body, 'body')
62
63    def test_python_application_query_string(self):
64        self.load('query_string')
65
66        resp = self.get(url='/?var1=val1&var2=val2')
67
68        self.assertEqual(
69            resp['headers']['Query-String'],
70            'var1=val1&var2=val2',
71            'Query-String header',
72        )
73
74    def test_python_application_query_string_space(self):
75        self.load('query_string')
76
77        resp = self.get(url='/ ?var1=val1&var2=val2')
78        self.assertEqual(
79            resp['headers']['Query-String'],
80            'var1=val1&var2=val2',
81            'Query-String space',
82        )
83
84        resp = self.get(url='/ %20?var1=val1&var2=val2')
85        self.assertEqual(
86            resp['headers']['Query-String'],
87            'var1=val1&var2=val2',
88            'Query-String space 2',
89        )
90
91        resp = self.get(url='/ %20 ?var1=val1&var2=val2')
92        self.assertEqual(
93            resp['headers']['Query-String'],
94            'var1=val1&var2=val2',
95            'Query-String space 3',
96        )
97
98        resp = self.get(url='/blah %20 blah? var1= val1 & var2=val2')
99        self.assertEqual(
100            resp['headers']['Query-String'],
101            ' var1= val1 & var2=val2',
102            'Query-String space 4',
103        )
104
105    def test_python_application_query_string_empty(self):
106        self.load('query_string')
107
108        resp = self.get(url='/?')
109
110        self.assertEqual(resp['status'], 200, 'query string empty status')
111        self.assertEqual(
112            resp['headers']['Query-String'], '', 'query string empty'
113        )
114
115    def test_python_application_query_string_absent(self):
116        self.load('query_string')
117
118        resp = self.get()
119
120        self.assertEqual(resp['status'], 200, 'query string absent status')
121        self.assertEqual(
122            resp['headers']['Query-String'], '', 'query string absent'
123        )
124
125    @unittest.skip('not yet')
126    def test_python_application_server_port(self):
127        self.load('server_port')
128
129        self.assertEqual(
130            self.get()['headers']['Server-Port'], '7080', 'Server-Port header'
131        )
132
133    @unittest.skip('not yet')
134    def test_python_application_working_directory_invalid(self):
135        self.load('empty')
136
137        self.assertIn(
138            'success',
139            self.conf('"/blah"', 'applications/empty/working_directory'),
140            'configure invalid working_directory',
141        )
142
143        self.assertEqual(self.get()['status'], 500, 'status')
144
145    def test_python_application_204_transfer_encoding(self):
146        self.load('204_no_content')
147
148        self.assertNotIn(
149            'Transfer-Encoding',
150            self.get()['headers'],
151            '204 header transfer encoding',
152        )
153
154    def test_python_application_ctx_iter_atexit(self):
155        self.load('ctx_iter_atexit')
156
157        resp = self.post(
158            headers={
159                'Host': 'localhost',
160                'Connection': 'close',
161                'Content-Type': 'text/html',
162            },
163            body='0123456789',
164        )
165
166        self.assertEqual(resp['status'], 200, 'ctx iter status')
167        self.assertEqual(resp['body'], '0123456789', 'ctx iter body')
168
169        self.conf({"listeners": {}, "applications": {}})
170
171        self.stop()
172
173        self.assertIsNotNone(
174            self.wait_for_record(r'RuntimeError'), 'ctx iter atexit'
175        )
176
177    def test_python_keepalive_body(self):
178        self.load('mirror')
179
180        self.assertEqual(self.get()['status'], 200, 'init')
181
182        (resp, sock) = self.post(
183            headers={
184                'Host': 'localhost',
185                'Connection': 'keep-alive',
186                'Content-Type': 'text/html',
187            },
188            start=True,
189            body='0123456789' * 500,
190            read_timeout=1,
191        )
192
193        self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1')
194
195        resp = self.post(
196            headers={
197                'Host': 'localhost',
198                'Connection': 'close',
199                'Content-Type': 'text/html',
200            },
201            sock=sock,
202            body='0123456789',
203        )
204
205        self.assertEqual(resp['body'], '0123456789', 'keep-alive 2')
206
207    def test_python_keepalive_reconfigure(self):
208        self.skip_alerts.extend(
209            [
210                r'pthread_mutex.+failed',
211                r'failed to apply',
212                r'process \d+ exited on signal',
213            ]
214        )
215        self.load('mirror')
216
217        self.assertEqual(self.get()['status'], 200, 'init')
218
219        body = '0123456789'
220        conns = 3
221        socks = []
222
223        for i in range(conns):
224            (resp, sock) = self.post(
225                headers={
226                    'Host': 'localhost',
227                    'Connection': 'keep-alive',
228                    'Content-Type': 'text/html',
229                },
230                start=True,
231                body=body,
232                read_timeout=1,
233            )
234
235            self.assertEqual(resp['body'], body, 'keep-alive open')
236            self.assertIn(
237                'success',
238                self.conf(str(i + 1), 'applications/mirror/processes'),
239                'reconfigure',
240            )
241
242            socks.append(sock)
243
244        for i in range(conns):
245            (resp, sock) = self.post(
246                headers={
247                    'Host': 'localhost',
248                    'Connection': 'keep-alive',
249                    'Content-Type': 'text/html',
250                },
251                start=True,
252                sock=socks[i],
253                body=body,
254                read_timeout=1,
255            )
256
257            self.assertEqual(resp['body'], body, 'keep-alive request')
258            self.assertIn(
259                'success',
260                self.conf(str(i + 1), 'applications/mirror/processes'),
261                'reconfigure 2',
262            )
263
264        for i in range(conns):
265            resp = self.post(
266                headers={
267                    'Host': 'localhost',
268                    'Connection': 'close',
269                    'Content-Type': 'text/html',
270                },
271                sock=socks[i],
272                body=body,
273            )
274
275            self.assertEqual(resp['body'], body, 'keep-alive close')
276            self.assertIn(
277                'success',
278                self.conf(str(i + 1), 'applications/mirror/processes'),
279                'reconfigure 3',
280            )
281
282    def test_python_keepalive_reconfigure_2(self):
283        self.load('mirror')
284
285        self.assertEqual(self.get()['status'], 200, 'init')
286
287        body = '0123456789'
288
289        (resp, sock) = self.post(
290            headers={
291                'Host': 'localhost',
292                'Connection': 'keep-alive',
293                'Content-Type': 'text/html',
294            },
295            start=True,
296            body=body,
297            read_timeout=1,
298        )
299
300        self.assertEqual(resp['body'], body, 'reconfigure 2 keep-alive 1')
301
302        self.load('empty')
303
304        self.assertEqual(self.get()['status'], 200, 'init')
305
306        (resp, sock) = self.post(
307            headers={
308                'Host': 'localhost',
309                'Connection': 'close',
310                'Content-Type': 'text/html',
311            },
312            start=True,
313            sock=sock,
314            body=body,
315        )
316
317        self.assertEqual(resp['status'], 200, 'reconfigure 2 keep-alive 2')
318        self.assertEqual(resp['body'], '', 'reconfigure 2 keep-alive 2 body')
319
320        self.assertIn(
321            'success',
322            self.conf({"listeners": {}, "applications": {}}),
323            'reconfigure 2 clear configuration',
324        )
325
326        resp = self.get(sock=sock)
327
328        self.assertEqual(resp, {}, 'reconfigure 2 keep-alive 3')
329
330    def test_python_keepalive_reconfigure_3(self):
331        self.load('empty')
332
333        self.assertEqual(self.get()['status'], 200, 'init')
334
335        (resp, sock) = self.http(
336            b"""GET / HTTP/1.1
337""",
338            start=True,
339            raw=True,
340            read_timeout=5,
341        )
342
343        self.assertIn(
344            'success',
345            self.conf({"listeners": {}, "applications": {}}),
346            'reconfigure 3 clear configuration',
347        )
348
349        resp = self.http(
350            b"""Host: localhost
351Connection: close
352
353""",
354            sock=sock,
355            raw=True,
356        )
357
358        self.assertEqual(resp['status'], 200, 'reconfigure 3')
359
360    def test_python_atexit(self):
361        self.load('atexit')
362
363        self.get()
364
365        self.conf({"listeners": {}, "applications": {}})
366
367        self.stop()
368
369        self.assertIsNotNone(
370            self.wait_for_record(r'At exit called\.'), 'atexit'
371        )
372
373    @unittest.skip('not yet')
374    def test_python_application_start_response_exit(self):
375        self.load('start_response_exit')
376
377        self.assertEqual(self.get()['status'], 500, 'start response exit')
378
379    @unittest.skip('not yet')
380    def test_python_application_input_iter(self):
381        self.load('input_iter')
382
383        body = '0123456789'
384
385        self.assertEqual(self.post(body=body)['body'], body, 'input iter')
386
387    def test_python_application_input_read_length(self):
388        self.load('input_read_length')
389
390        body = '0123456789'
391
392        resp = self.post(
393            headers={
394                'Host': 'localhost',
395                'Input-Length': '5',
396                'Connection': 'close',
397            },
398            body=body,
399        )
400
401        self.assertEqual(resp['body'], body[:5], 'input read length lt body')
402
403        resp = self.post(
404            headers={
405                'Host': 'localhost',
406                'Input-Length': '15',
407                'Connection': 'close',
408            },
409            body=body,
410        )
411
412        self.assertEqual(resp['body'], body, 'input read length gt body')
413
414        resp = self.post(
415            headers={
416                'Host': 'localhost',
417                'Input-Length': '0',
418                'Connection': 'close',
419            },
420            body=body,
421        )
422
423        self.assertEqual(resp['body'], '', 'input read length zero')
424
425        resp = self.post(
426            headers={
427                'Host': 'localhost',
428                'Input-Length': '-1',
429                'Connection': 'close',
430            },
431            body=body,
432        )
433
434        self.assertEqual(resp['body'], body, 'input read length negative')
435
436    @unittest.skip('not yet')
437    def test_python_application_errors_write(self):
438        self.load('errors_write')
439
440        self.get()
441
442        self.stop()
443
444        self.assertIsNotNone(
445            self.wait_for_record(r'\[error\].+Error in application\.'),
446            'errors write',
447        )
448
449    def test_python_application_body_array(self):
450        self.load('body_array')
451
452        self.assertEqual(self.get()['body'], '0123456789', 'body array')
453
454    def test_python_application_body_io(self):
455        self.load('body_io')
456
457        self.assertEqual(self.get()['body'], '0123456789', 'body io')
458
459    def test_python_application_body_io_file(self):
460        self.load('body_io_file')
461
462        self.assertEqual(self.get()['body'], 'body\n', 'body io file')
463
464    @unittest.skip('not yet')
465    def test_python_application_syntax_error(self):
466        self.skip_alerts.append(r'Python failed to import module "wsgi"')
467        self.load('syntax_error')
468
469        self.assertEqual(self.get()['status'], 500, 'syntax error')
470
471    def test_python_application_close(self):
472        self.load('close')
473
474        self.get()
475
476        self.stop()
477
478        self.assertIsNotNone(self.wait_for_record(r'Close called\.'), 'close')
479
480    def test_python_application_close_error(self):
481        self.load('close_error')
482
483        self.get()
484
485        self.stop()
486
487        self.assertIsNotNone(
488            self.wait_for_record(r'Close called\.'), 'close error'
489        )
490
491    def test_python_application_not_iterable(self):
492        self.load('not_iterable')
493
494        self.get()
495
496        self.stop()
497
498        self.assertIsNotNone(
499            self.wait_for_record(
500                r'\[error\].+the application returned not an iterable object'
501            ),
502            'not iterable',
503        )
504
505    def test_python_application_write(self):
506        self.load('write')
507
508        self.assertEqual(self.get()['body'], '0123456789', 'write')
509
510    def test_python_application_threading(self):
511        """wait_for_record() timeouts after 5s while every thread works at
512        least 3s.  So without releasing GIL test should fail.
513        """
514
515        self.load('threading')
516
517        for _ in range(10):
518            self.get(no_recv=True)
519
520        self.assertIsNotNone(
521            self.wait_for_record(r'\(5\) Thread: 100'), 'last thread finished'
522        )
523
524if __name__ == '__main__':
525    TestPythonApplication.main()
526