diff options
author | Andrei Zeliankou <zelenkov@nginx.com> | 2023-06-14 18:20:09 +0100 |
---|---|---|
committer | Andrei Zeliankou <zelenkov@nginx.com> | 2023-06-14 18:20:09 +0100 |
commit | c183bd8749a19477390f8cb77efe5f6d223f0905 (patch) | |
tree | 4e821e9cb07be9a86bf2d442acb3ea6740ba5a99 /test/test_asgi_application.py | |
parent | c6d05191a069ac150cc8eb2bece75cf79c0a465a (diff) | |
download | unit-c183bd8749a19477390f8cb77efe5f6d223f0905.tar.gz unit-c183bd8749a19477390f8cb77efe5f6d223f0905.tar.bz2 |
Tests: get rid of classes in test files.
Class usage came from the unittest framework and it was always redundant
after migration to the pytest. This commit removes classes from files
containing tests to make them more readable and understandable.
Diffstat (limited to 'test/test_asgi_application.py')
-rw-r--r-- | test/test_asgi_application.py | 630 |
1 files changed, 325 insertions, 305 deletions
diff --git a/test/test_asgi_application.py b/test/test_asgi_application.py index 1f98b170..98d4bcd5 100644 --- a/test/test_asgi_application.py +++ b/test/test_asgi_application.py @@ -3,23 +3,22 @@ import time import pytest from packaging import version -from unit.applications.lang.python import TestApplicationPython +from unit.applications.lang.python import ApplicationPython prerequisites = { 'modules': {'python': lambda v: version.parse(v) >= version.parse('3.5')} } +client = ApplicationPython(load_module='asgi') -class TestASGIApplication(TestApplicationPython): - load_module = 'asgi' - def test_asgi_application_variables(self, date_to_sec_epoch, sec_epoch): - self.load('variables') +def test_asgi_application_variables(date_to_sec_epoch, sec_epoch): + client.load('variables') - body = 'Test body string.' + body = 'Test body string.' - resp = self.http( - f"""POST / HTTP/1.1 + resp = client.http( + f"""POST / HTTP/1.1 Host: localhost Content-Length: {len(body)} Custom-Header: blah @@ -29,263 +28,230 @@ Connection: close custom-header: BLAH {body}""".encode(), - raw=True, - ) + raw=True, + ) - assert resp['status'] == 200, 'status' - headers = resp['headers'] - header_server = headers.pop('Server') - assert re.search(r'Unit/[\d\.]+', header_server), 'server header' + assert resp['status'] == 200, 'status' + headers = resp['headers'] + header_server = headers.pop('Server') + assert re.search(r'Unit/[\d\.]+', header_server), 'server header' - date = headers.pop('Date') - assert date[-4:] == ' GMT', 'date header timezone' - assert abs(date_to_sec_epoch(date) - sec_epoch) < 5, 'date header' + date = headers.pop('Date') + assert date[-4:] == ' GMT', 'date header timezone' + assert abs(date_to_sec_epoch(date) - sec_epoch) < 5, 'date header' - assert headers == { - 'Connection': 'close', - 'content-length': str(len(body)), - 'content-type': 'text/html', - 'request-method': 'POST', - 'request-uri': '/', - 'http-host': 'localhost', - 'http-version': '1.1', - 'custom-header': 'blah, Blah, BLAH', - 'asgi-version': '3.0', - 'asgi-spec-version': '2.1', - 'scheme': 'http', - }, 'headers' - assert resp['body'] == body, 'body' - - def test_asgi_application_ipv6(self): - self.load('empty') - - assert 'success' in self.conf( - {"[::1]:7080": {"pass": "applications/empty"}}, 'listeners' - ) + assert headers == { + 'Connection': 'close', + 'content-length': str(len(body)), + 'content-type': 'text/html', + 'request-method': 'POST', + 'request-uri': '/', + 'http-host': 'localhost', + 'http-version': '1.1', + 'custom-header': 'blah, Blah, BLAH', + 'asgi-version': '3.0', + 'asgi-spec-version': '2.1', + 'scheme': 'http', + }, 'headers' + assert resp['body'] == body, 'body' - assert self.get(sock_type='ipv6')['status'] == 200 - def test_asgi_application_unix(self, temp_dir): - self.load('empty') +def test_asgi_application_ipv6(): + client.load('empty') - addr = f'{temp_dir}/sock' - assert 'success' in self.conf( - {f"unix:{addr}": {"pass": "applications/empty"}}, 'listeners' - ) + assert 'success' in client.conf( + {"[::1]:7080": {"pass": "applications/empty"}}, 'listeners' + ) - assert self.get(sock_type='unix', addr=addr)['status'] == 200 + assert client.get(sock_type='ipv6')['status'] == 200 - def test_asgi_application_query_string(self): - self.load('query_string') - resp = self.get(url='/?var1=val1&var2=val2') +def test_asgi_application_unix(temp_dir): + client.load('empty') - assert ( - resp['headers']['query-string'] == 'var1=val1&var2=val2' - ), 'query-string header' + addr = f'{temp_dir}/sock' + assert 'success' in client.conf( + {f"unix:{addr}": {"pass": "applications/empty"}}, 'listeners' + ) - def test_asgi_application_prefix(self): - self.load('prefix', prefix='/api/rest') + assert client.get(sock_type='unix', addr=addr)['status'] == 200 - def set_prefix(prefix): - self.conf(f'"{prefix}"', 'applications/prefix/prefix') - def check_prefix(url, prefix): - resp = self.get(url=url) - assert resp['status'] == 200 - assert resp['headers']['prefix'] == prefix +def test_asgi_application_query_string(): + client.load('query_string') - check_prefix('/ap', 'NULL') - check_prefix('/api', 'NULL') - check_prefix('/api/', 'NULL') - check_prefix('/api/res', 'NULL') - check_prefix('/api/restful', 'NULL') - check_prefix('/api/rest', '/api/rest') - check_prefix('/api/rest/', '/api/rest') - check_prefix('/api/rest/get', '/api/rest') - check_prefix('/api/rest/get/blah', '/api/rest') + resp = client.get(url='/?var1=val1&var2=val2') - set_prefix('/api/rest/') - check_prefix('/api/rest', '/api/rest') - check_prefix('/api/restful', 'NULL') - check_prefix('/api/rest/', '/api/rest') - check_prefix('/api/rest/blah', '/api/rest') + assert ( + resp['headers']['query-string'] == 'var1=val1&var2=val2' + ), 'query-string header' - set_prefix('/app') - check_prefix('/ap', 'NULL') - check_prefix('/app', '/app') - check_prefix('/app/', '/app') - check_prefix('/application/', 'NULL') - set_prefix('/') - check_prefix('/', 'NULL') - check_prefix('/app', 'NULL') +def test_asgi_application_prefix(): + client.load('prefix', prefix='/api/rest') - def test_asgi_application_query_string_space(self): - self.load('query_string') + def set_prefix(prefix): + client.conf(f'"{prefix}"', 'applications/prefix/prefix') - resp = self.get(url='/ ?var1=val1&var2=val2') - assert ( - resp['headers']['query-string'] == 'var1=val1&var2=val2' - ), 'query-string space' + def check_prefix(url, prefix): + resp = client.get(url=url) + assert resp['status'] == 200 + assert resp['headers']['prefix'] == prefix - resp = self.get(url='/ %20?var1=val1&var2=val2') - assert ( - resp['headers']['query-string'] == 'var1=val1&var2=val2' - ), 'query-string space 2' + check_prefix('/ap', 'NULL') + check_prefix('/api', 'NULL') + check_prefix('/api/', 'NULL') + check_prefix('/api/res', 'NULL') + check_prefix('/api/restful', 'NULL') + check_prefix('/api/rest', '/api/rest') + check_prefix('/api/rest/', '/api/rest') + check_prefix('/api/rest/get', '/api/rest') + check_prefix('/api/rest/get/blah', '/api/rest') - resp = self.get(url='/ %20 ?var1=val1&var2=val2') - assert ( - resp['headers']['query-string'] == 'var1=val1&var2=val2' - ), 'query-string space 3' + set_prefix('/api/rest/') + check_prefix('/api/rest', '/api/rest') + check_prefix('/api/restful', 'NULL') + check_prefix('/api/rest/', '/api/rest') + check_prefix('/api/rest/blah', '/api/rest') - resp = self.get(url='/blah %20 blah? var1= val1 & var2=val2') - assert ( - resp['headers']['query-string'] == ' var1= val1 & var2=val2' - ), 'query-string space 4' + set_prefix('/app') + check_prefix('/ap', 'NULL') + check_prefix('/app', '/app') + check_prefix('/app/', '/app') + check_prefix('/application/', 'NULL') - def test_asgi_application_query_string_empty(self): - self.load('query_string') + set_prefix('/') + check_prefix('/', 'NULL') + check_prefix('/app', 'NULL') - resp = self.get(url='/?') - assert resp['status'] == 200, 'query string empty status' - assert resp['headers']['query-string'] == '', 'query string empty' +def test_asgi_application_query_string_space(): + client.load('query_string') - def test_asgi_application_query_string_absent(self): - self.load('query_string') + resp = client.get(url='/ ?var1=val1&var2=val2') + assert ( + resp['headers']['query-string'] == 'var1=val1&var2=val2' + ), 'query-string space' - resp = self.get() + resp = client.get(url='/ %20?var1=val1&var2=val2') + assert ( + resp['headers']['query-string'] == 'var1=val1&var2=val2' + ), 'query-string space 2' - assert resp['status'] == 200, 'query string absent status' - assert resp['headers']['query-string'] == '', 'query string absent' + resp = client.get(url='/ %20 ?var1=val1&var2=val2') + assert ( + resp['headers']['query-string'] == 'var1=val1&var2=val2' + ), 'query-string space 3' - @pytest.mark.skip('not yet') - def test_asgi_application_server_port(self): - self.load('server_port') + resp = client.get(url='/blah %20 blah? var1= val1 & var2=val2') + assert ( + resp['headers']['query-string'] == ' var1= val1 & var2=val2' + ), 'query-string space 4' - assert ( - self.get()['headers']['Server-Port'] == '7080' - ), 'Server-Port header' - @pytest.mark.skip('not yet') - def test_asgi_application_working_directory_invalid(self): - self.load('empty') +def test_asgi_application_query_string_empty(): + client.load('query_string') - assert 'success' in self.conf( - '"/blah"', 'applications/empty/working_directory' - ), 'configure invalid working_directory' + resp = client.get(url='/?') - assert self.get()['status'] == 500, 'status' + assert resp['status'] == 200, 'query string empty status' + assert resp['headers']['query-string'] == '', 'query string empty' - def test_asgi_application_204_transfer_encoding(self): - self.load('204_no_content') - assert ( - 'Transfer-Encoding' not in self.get()['headers'] - ), '204 header transfer encoding' +def test_asgi_application_query_string_absent(): + client.load('query_string') - def test_asgi_application_shm_ack_handle(self): - # Minimum possible limit - shm_limit = 10 * 1024 * 1024 + resp = client.get() - self.load('mirror', limits={"shm": shm_limit}) + assert resp['status'] == 200, 'query string absent status' + assert resp['headers']['query-string'] == '', 'query string absent' - # Should exceed shm_limit - max_body_size = 12 * 1024 * 1024 - assert 'success' in self.conf( - f'{{"http":{{"max_body_size": {max_body_size} }}}}', - 'settings', - ) +@pytest.mark.skip('not yet') +def test_asgi_application_server_port(): + client.load('server_port') - assert self.get()['status'] == 200, 'init' + assert ( + client.get()['headers']['Server-Port'] == '7080' + ), 'Server-Port header' - body = '0123456789AB' * 1024 * 1024 # 12 Mb - resp = self.post(body=body, read_buffer_size=1024 * 1024) - assert resp['body'] == body, 'keep-alive 1' +@pytest.mark.skip('not yet') +def test_asgi_application_working_directory_invalid(): + client.load('empty') - def test_asgi_keepalive_body(self): - self.load('mirror') + assert 'success' in client.conf( + '"/blah"', 'applications/empty/working_directory' + ), 'configure invalid working_directory' - assert self.get()['status'] == 200, 'init' + assert client.get()['status'] == 500, 'status' - body = '0123456789' * 500 - (resp, sock) = self.post( - headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - }, - start=True, - body=body, - read_timeout=1, - ) - assert resp['body'] == body, 'keep-alive 1' +def test_asgi_application_204_transfer_encoding(): + client.load('204_no_content') + + assert ( + 'Transfer-Encoding' not in client.get()['headers'] + ), '204 header transfer encoding' + - body = '0123456789' - resp = self.post(sock=sock, body=body) +def test_asgi_application_shm_ack_handle(): + # Minimum possible limit + shm_limit = 10 * 1024 * 1024 - assert resp['body'] == body, 'keep-alive 2' + client.load('mirror', limits={"shm": shm_limit}) - def test_asgi_keepalive_reconfigure(self): - self.load('mirror') + # Should exceed shm_limit + max_body_size = 12 * 1024 * 1024 - assert self.get()['status'] == 200, 'init' + assert 'success' in client.conf( + f'{{"http":{{"max_body_size": {max_body_size} }}}}', + 'settings', + ) - body = '0123456789' - conns = 3 - socks = [] + assert client.get()['status'] == 200, 'init' - for i in range(conns): - (resp, sock) = self.post( - headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - }, - start=True, - body=body, - read_timeout=1, - ) + body = '0123456789AB' * 1024 * 1024 # 12 Mb + resp = client.post(body=body, read_buffer_size=1024 * 1024) - assert resp['body'] == body, 'keep-alive open' + assert resp['body'] == body, 'keep-alive 1' - self.load('mirror', processes=i + 1) - socks.append(sock) +def test_asgi_keepalive_body(): + client.load('mirror') - for i in range(conns): - (resp, sock) = self.post( - headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - }, - start=True, - sock=socks[i], - body=body, - read_timeout=1, - ) + assert client.get()['status'] == 200, 'init' - assert resp['body'] == body, 'keep-alive request' + body = '0123456789' * 500 + (resp, sock) = client.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + }, + start=True, + body=body, + read_timeout=1, + ) - self.load('mirror', processes=i + 1) + assert resp['body'] == body, 'keep-alive 1' - for i in range(conns): - resp = self.post(sock=socks[i], body=body) + body = '0123456789' + resp = client.post(sock=sock, body=body) - assert resp['body'] == body, 'keep-alive close' + assert resp['body'] == body, 'keep-alive 2' - self.load('mirror', processes=i + 1) - def test_asgi_keepalive_reconfigure_2(self): - self.load('mirror') +def test_asgi_keepalive_reconfigure(): + client.load('mirror') - assert self.get()['status'] == 200, 'init' + assert client.get()['status'] == 200, 'init' - body = '0123456789' + body = '0123456789' + conns = 3 + socks = [] - (resp, sock) = self.post( + for i in range(conns): + (resp, sock) = client.post( headers={ 'Host': 'localhost', 'Connection': 'keep-alive', @@ -295,162 +261,216 @@ custom-header: BLAH read_timeout=1, ) - assert resp['body'] == body, 'reconfigure 2 keep-alive 1' + assert resp['body'] == body, 'keep-alive open' - self.load('empty') + client.load('mirror', processes=i + 1) - assert self.get()['status'] == 200, 'init' + socks.append(sock) - (resp, sock) = self.post(start=True, sock=sock, body=body) + for i in range(conns): + (resp, sock) = client.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + }, + start=True, + sock=socks[i], + body=body, + read_timeout=1, + ) - assert resp['status'] == 200, 'reconfigure 2 keep-alive 2' - assert resp['body'] == '', 'reconfigure 2 keep-alive 2 body' + assert resp['body'] == body, 'keep-alive request' - assert 'success' in self.conf( - {"listeners": {}, "applications": {}} - ), 'reconfigure 2 clear configuration' + client.load('mirror', processes=i + 1) - resp = self.get(sock=sock) + for i in range(conns): + resp = client.post(sock=socks[i], body=body) - assert resp == {}, 'reconfigure 2 keep-alive 3' + assert resp['body'] == body, 'keep-alive close' - def test_asgi_keepalive_reconfigure_3(self): - self.load('empty') + client.load('mirror', processes=i + 1) - assert self.get()['status'] == 200, 'init' - sock = self.http( - b"""GET / HTTP/1.1 -""", - raw=True, - no_recv=True, - ) +def test_asgi_keepalive_reconfigure_2(): + client.load('mirror') - assert self.get()['status'] == 200 + assert client.get()['status'] == 200, 'init' - assert 'success' in self.conf( - {"listeners": {}, "applications": {}} - ), 'reconfigure 3 clear configuration' + body = '0123456789' - resp = self.http( - b"""Host: localhost -Connection: close + (resp, sock) = client.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + }, + start=True, + body=body, + read_timeout=1, + ) -""", - sock=sock, - raw=True, - ) + assert resp['body'] == body, 'reconfigure 2 keep-alive 1' - assert resp['status'] == 200, 'reconfigure 3' + client.load('empty') - def test_asgi_process_switch(self): - self.load('delayed', processes=2) + assert client.get()['status'] == 200, 'init' - self.get( - headers={ - 'Host': 'localhost', - 'Content-Length': '0', - 'X-Delay': '5', - 'Connection': 'close', - }, - no_recv=True, - ) + (resp, sock) = client.post(start=True, sock=sock, body=body) - headers_delay_1 = { - 'Connection': 'close', - 'Host': 'localhost', - 'Content-Length': '0', - 'X-Delay': '1', - } + assert resp['status'] == 200, 'reconfigure 2 keep-alive 2' + assert resp['body'] == '', 'reconfigure 2 keep-alive 2 body' - self.get(headers=headers_delay_1, no_recv=True) + assert 'success' in client.conf( + {"listeners": {}, "applications": {}} + ), 'reconfigure 2 clear configuration' - time.sleep(0.5) + resp = client.get(sock=sock) - for _ in range(10): - self.get(headers=headers_delay_1, no_recv=True) + assert resp == {}, 'reconfigure 2 keep-alive 3' - self.get(headers=headers_delay_1) - def test_asgi_application_loading_error(self, skip_alert): - skip_alert(r'Python failed to import module "blah"') +def test_asgi_keepalive_reconfigure_3(): + client.load('empty') - self.load('empty', module="blah") + assert client.get()['status'] == 200, 'init' - assert self.get()['status'] == 503, 'loading error' + sock = client.http( + b"""GET / HTTP/1.1 +""", + raw=True, + no_recv=True, + ) - def test_asgi_application_threading(self, wait_for_record): - """wait_for_record() timeouts after 5s while every thread works at - least 3s. So without releasing GIL test should fail. - """ + assert client.get()['status'] == 200 - self.load('threading') + assert 'success' in client.conf( + {"listeners": {}, "applications": {}} + ), 'reconfigure 3 clear configuration' - for _ in range(10): - self.get(no_recv=True) + resp = client.http( + b"""Host: localhost +Connection: close - assert ( - wait_for_record(r'\(5\) Thread: 100', wait=50) is not None - ), 'last thread finished' +""", + sock=sock, + raw=True, + ) - def test_asgi_application_threads(self): - self.load('threads', threads=2) + assert resp['status'] == 200, 'reconfigure 3' - socks = [] - for _ in range(2): - sock = self.get( - headers={ - 'Host': 'localhost', - 'X-Delay': '3', - 'Connection': 'close', - }, - no_recv=True, - ) +def test_asgi_process_switch(): + client.load('delayed', processes=2) - socks.append(sock) + client.get( + headers={ + 'Host': 'localhost', + 'Content-Length': '0', + 'X-Delay': '5', + 'Connection': 'close', + }, + no_recv=True, + ) - time.sleep(1.0) # required to avoid greedy request reading + headers_delay_1 = { + 'Connection': 'close', + 'Host': 'localhost', + 'Content-Length': '0', + 'X-Delay': '1', + } - threads = set() + client.get(headers=headers_delay_1, no_recv=True) - for sock in socks: - resp = self.recvall(sock).decode('utf-8') + time.sleep(0.5) - self.log_in(resp) + for _ in range(10): + client.get(headers=headers_delay_1, no_recv=True) - resp = self._resp_to_dict(resp) + client.get(headers=headers_delay_1) - assert resp['status'] == 200, 'status' - threads.add(resp['headers']['x-thread']) +def test_asgi_application_loading_error(skip_alert): + skip_alert(r'Python failed to import module "blah"') - sock.close() + client.load('empty', module="blah") - assert len(socks) == len(threads), 'threads differs' + assert client.get()['status'] == 503, 'loading error' - def test_asgi_application_legacy(self): - self.load('legacy') - resp = self.get( - headers={ - 'Host': 'localhost', - 'Content-Length': '0', - 'Connection': 'close', - }, - ) +def test_asgi_application_threading(wait_for_record): + """wait_for_record() timeouts after 5s while every thread works at + least 3s. So without releasing GIL test should fail. + """ + + client.load('threading') + + for _ in range(10): + client.get(no_recv=True) + + assert ( + wait_for_record(r'\(5\) Thread: 100', wait=50) is not None + ), 'last thread finished' - assert resp['status'] == 200, 'status' - def test_asgi_application_legacy_force(self): - self.load('legacy_force', protocol='asgi') +def test_asgi_application_threads(): + client.load('threads', threads=2) - resp = self.get( + socks = [] + + for _ in range(2): + sock = client.get( headers={ 'Host': 'localhost', - 'Content-Length': '0', + 'X-Delay': '3', 'Connection': 'close', }, + no_recv=True, ) + socks.append(sock) + + time.sleep(1.0) # required to avoid greedy request reading + + threads = set() + + for sock in socks: + resp = client.recvall(sock).decode('utf-8') + + client.log_in(resp) + + resp = client._resp_to_dict(resp) + assert resp['status'] == 200, 'status' + + threads.add(resp['headers']['x-thread']) + + sock.close() + + assert len(socks) == len(threads), 'threads differs' + + +def test_asgi_application_legacy(): + client.load('legacy') + + resp = client.get( + headers={ + 'Host': 'localhost', + 'Content-Length': '0', + 'Connection': 'close', + }, + ) + + assert resp['status'] == 200, 'status' + + +def test_asgi_application_legacy_force(): + client.load('legacy_force', protocol='asgi') + + resp = client.get( + headers={ + 'Host': 'localhost', + 'Content-Length': '0', + 'Connection': 'close', + }, + ) + + assert resp['status'] == 200, 'status' |