Deleted
Added
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' |