test_asgi_application.py (1626:d20f04158166) test_asgi_application.py (1635:97afbb6c5a15)
1import grp
2import pytest
3import pwd
4import re
5import time
6from distutils.version import LooseVersion
7
1import re
2import time
3from distutils.version import LooseVersion
4
8from unit.applications.lang.python import TestApplicationPython
5import pytest
6
9from conftest import skip_alert
7from conftest import skip_alert
8from unit.applications.lang.python import TestApplicationPython
10
11
12class TestASGIApplication(TestApplicationPython):
13 prerequisites = {'modules': {'python':
14 lambda v: LooseVersion(v) >= LooseVersion('3.5')}}
15 load_module = 'asgi'
16
17 def findall(self, pattern):
18 with open(self.temp_dir + '/unit.log', 'r', errors='ignore') as f:
19 return re.findall(pattern, f.read())
20
9
10
11class TestASGIApplication(TestApplicationPython):
12 prerequisites = {'modules': {'python':
13 lambda v: LooseVersion(v) >= LooseVersion('3.5')}}
14 load_module = 'asgi'
15
16 def findall(self, pattern):
17 with open(self.temp_dir + '/unit.log', 'r', errors='ignore') as f:
18 return re.findall(pattern, f.read())
19
21 def test_asgi_application__variables(self):
20 def test_asgi_application_variables(self):
22 self.load('variables')
23
24 body = 'Test body string.'
25
26 resp = self.http(
27 b"""POST / HTTP/1.1
28Host: localhost
29Content-Length: %d

--- 28 unchanged lines hidden (view full) ---

58 'http-version': '1.1',
59 'custom-header': 'blah, Blah, BLAH',
60 'asgi-version': '3.0',
61 'asgi-spec-version': '2.1',
62 'scheme': 'http',
63 }, 'headers'
64 assert resp['body'] == body, 'body'
65
21 self.load('variables')
22
23 body = 'Test body string.'
24
25 resp = self.http(
26 b"""POST / HTTP/1.1
27Host: localhost
28Content-Length: %d

--- 28 unchanged lines hidden (view full) ---

57 'http-version': '1.1',
58 'custom-header': 'blah, Blah, BLAH',
59 'asgi-version': '3.0',
60 'asgi-spec-version': '2.1',
61 'scheme': 'http',
62 }, 'headers'
63 assert resp['body'] == body, 'body'
64
66 def test_asgi_application__query_string(self):
65 def test_asgi_application_query_string(self):
67 self.load('query_string')
68
69 resp = self.get(url='/?var1=val1&var2=val2')
70
71 assert (
72 resp['headers']['query-string'] == 'var1=val1&var2=val2'
73 ), 'query-string header'
74
66 self.load('query_string')
67
68 resp = self.get(url='/?var1=val1&var2=val2')
69
70 assert (
71 resp['headers']['query-string'] == 'var1=val1&var2=val2'
72 ), 'query-string header'
73
75 def test_asgi_application__query_string_space(self):
74 def test_asgi_application_query_string_space(self):
76 self.load('query_string')
77
78 resp = self.get(url='/ ?var1=val1&var2=val2')
79 assert (
80 resp['headers']['query-string'] == 'var1=val1&var2=val2'
81 ), 'query-string space'
82
83 resp = self.get(url='/ %20?var1=val1&var2=val2')

--- 6 unchanged lines hidden (view full) ---

90 resp['headers']['query-string'] == 'var1=val1&var2=val2'
91 ), 'query-string space 3'
92
93 resp = self.get(url='/blah %20 blah? var1= val1 & var2=val2')
94 assert (
95 resp['headers']['query-string'] == ' var1= val1 & var2=val2'
96 ), 'query-string space 4'
97
75 self.load('query_string')
76
77 resp = self.get(url='/ ?var1=val1&var2=val2')
78 assert (
79 resp['headers']['query-string'] == 'var1=val1&var2=val2'
80 ), 'query-string space'
81
82 resp = self.get(url='/ %20?var1=val1&var2=val2')

--- 6 unchanged lines hidden (view full) ---

89 resp['headers']['query-string'] == 'var1=val1&var2=val2'
90 ), 'query-string space 3'
91
92 resp = self.get(url='/blah %20 blah? var1= val1 & var2=val2')
93 assert (
94 resp['headers']['query-string'] == ' var1= val1 & var2=val2'
95 ), 'query-string space 4'
96
98 def test_asgi_application__query_string_empty(self):
97 def test_asgi_application_query_string_empty(self):
99 self.load('query_string')
100
101 resp = self.get(url='/?')
102
103 assert resp['status'] == 200, 'query string empty status'
104 assert resp['headers']['query-string'] == '', 'query string empty'
105
98 self.load('query_string')
99
100 resp = self.get(url='/?')
101
102 assert resp['status'] == 200, 'query string empty status'
103 assert resp['headers']['query-string'] == '', 'query string empty'
104
106 def test_asgi_application__query_string_absent(self):
105 def test_asgi_application_query_string_absent(self):
107 self.load('query_string')
108
109 resp = self.get()
110
111 assert resp['status'] == 200, 'query string absent status'
112 assert resp['headers']['query-string'] == '', 'query string absent'
113
114 @pytest.mark.skip('not yet')
106 self.load('query_string')
107
108 resp = self.get()
109
110 assert resp['status'] == 200, 'query string absent status'
111 assert resp['headers']['query-string'] == '', 'query string absent'
112
113 @pytest.mark.skip('not yet')
115 def test_asgi_application__server_port(self):
114 def test_asgi_application_server_port(self):
116 self.load('server_port')
117
118 assert (
119 self.get()['headers']['Server-Port'] == '7080'
120 ), 'Server-Port header'
121
122 @pytest.mark.skip('not yet')
115 self.load('server_port')
116
117 assert (
118 self.get()['headers']['Server-Port'] == '7080'
119 ), 'Server-Port header'
120
121 @pytest.mark.skip('not yet')
123 def test_asgi_application__working_directory_invalid(self):
122 def test_asgi_application_working_directory_invalid(self):
124 self.load('empty')
125
126 assert 'success' in self.conf(
127 '"/blah"', 'applications/empty/working_directory'
128 ), 'configure invalid working_directory'
129
130 assert self.get()['status'] == 500, 'status'
131
123 self.load('empty')
124
125 assert 'success' in self.conf(
126 '"/blah"', 'applications/empty/working_directory'
127 ), 'configure invalid working_directory'
128
129 assert self.get()['status'] == 500, 'status'
130
132 def test_asgi_application__204_transfer_encoding(self):
131 def test_asgi_application_204_transfer_encoding(self):
133 self.load('204_no_content')
134
135 assert (
136 'Transfer-Encoding' not in self.get()['headers']
137 ), '204 header transfer encoding'
138
132 self.load('204_no_content')
133
134 assert (
135 'Transfer-Encoding' not in self.get()['headers']
136 ), '204 header transfer encoding'
137
139 def test_asgi_application__shm_ack_handle(self):
138 def test_asgi_application_shm_ack_handle(self):
140 self.load('mirror')
141
142 # Minimum possible limit
143 shm_limit = 10 * 1024 * 1024
144
145 assert (
146 'success' in self.conf('{"shm": ' + str(shm_limit) + '}',
147 'applications/mirror/limits')

--- 226 unchanged lines hidden (view full) ---

374
375 time.sleep(0.5)
376
377 for _ in range(10):
378 self.get(headers=headers_delay_1, no_recv=True)
379
380 self.get(headers=headers_delay_1)
381
139 self.load('mirror')
140
141 # Minimum possible limit
142 shm_limit = 10 * 1024 * 1024
143
144 assert (
145 'success' in self.conf('{"shm": ' + str(shm_limit) + '}',
146 'applications/mirror/limits')

--- 226 unchanged lines hidden (view full) ---

373
374 time.sleep(0.5)
375
376 for _ in range(10):
377 self.get(headers=headers_delay_1, no_recv=True)
378
379 self.get(headers=headers_delay_1)
380
382 def test_asgi_application__loading_error(self):
381 def test_asgi_application_loading_error(self):
383 skip_alert(r'Python failed to import module "blah"')
384
385 self.load('empty')
386
387 assert 'success' in self.conf('"blah"', 'applications/empty/module')
388
389 assert self.get()['status'] == 503, 'loading error'
390
382 skip_alert(r'Python failed to import module "blah"')
383
384 self.load('empty')
385
386 assert 'success' in self.conf('"blah"', 'applications/empty/module')
387
388 assert self.get()['status'] == 503, 'loading error'
389
391 def test_asgi_application__threading(self):
390 def test_asgi_application_threading(self):
392 """wait_for_record() timeouts after 5s while every thread works at
393 least 3s. So without releasing GIL test should fail.
394 """
395
396 self.load('threading')
397
398 for _ in range(10):
399 self.get(no_recv=True)
400
401 assert (
402 self.wait_for_record(r'\(5\) Thread: 100') is not None
403 ), 'last thread finished'
391 """wait_for_record() timeouts after 5s while every thread works at
392 least 3s. So without releasing GIL test should fail.
393 """
394
395 self.load('threading')
396
397 for _ in range(10):
398 self.get(no_recv=True)
399
400 assert (
401 self.wait_for_record(r'\(5\) Thread: 100') is not None
402 ), 'last thread finished'