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_static.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 '')
-rw-r--r-- | test/test_static.py | 634 |
1 files changed, 324 insertions, 310 deletions
diff --git a/test/test_static.py b/test/test_static.py index 60a066a0..d46247d9 100644 --- a/test/test_static.py +++ b/test/test_static.py @@ -2,347 +2,361 @@ import os import socket import pytest -from unit.applications.proto import TestApplicationProto +from unit.applications.proto import ApplicationProto from unit.utils import waitforfiles -class TestStatic(TestApplicationProto): - @pytest.fixture(autouse=True) - def setup_method_fixture(self, temp_dir): - os.makedirs(f'{temp_dir}/assets/dir') - assets_dir = f'{temp_dir}/assets' - - with open(f'{assets_dir}/index.html', 'w') as index, open( - f'{assets_dir}/README', 'w' - ) as readme, open(f'{assets_dir}/log.log', 'w') as log, open( - f'{assets_dir}/dir/file', 'w' - ) as file: - index.write('0123456789') - readme.write('readme') - log.write('[debug]') - file.write('blah') - - self._load_conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [{"action": {"share": f'{assets_dir}$uri'}}], - "settings": { - "http": { - "static": { - "mime_types": {"text/plain": [".log", "README"]} - } - } - }, - } - ) +client = ApplicationProto() + + +@pytest.fixture(autouse=True) +def setup_method_fixture(temp_dir): + os.makedirs(f'{temp_dir}/assets/dir') + assets_dir = f'{temp_dir}/assets' + + with open(f'{assets_dir}/index.html', 'w') as index, open( + f'{assets_dir}/README', 'w' + ) as readme, open(f'{assets_dir}/log.log', 'w') as log, open( + f'{assets_dir}/dir/file', 'w' + ) as file: + index.write('0123456789') + readme.write('readme') + log.write('[debug]') + file.write('blah') + + assert 'success' in client.conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [{"action": {"share": f'{assets_dir}$uri'}}], + "settings": { + "http": { + "static": {"mime_types": {"text/plain": [".log", "README"]}} + } + }, + } + ) - def test_static_index(self, temp_dir): - def set_index(index): - assert 'success' in self.conf( - {"share": f'{temp_dir}/assets$uri', "index": index}, - 'routes/0/action', - ), 'configure index' - - set_index('README') - assert self.get()['body'] == 'readme', 'index' - - self.conf_delete('routes/0/action/index') - assert self.get()['body'] == '0123456789', 'delete index' - - set_index('') - assert self.get()['status'] == 404, 'index empty' - - def test_static_index_default(self): - assert self.get(url='/index.html')['body'] == '0123456789', 'index' - assert self.get(url='/')['body'] == '0123456789', 'index 2' - assert self.get(url='//')['body'] == '0123456789', 'index 3' - assert self.get(url='/.')['body'] == '0123456789', 'index 4' - assert self.get(url='/./')['body'] == '0123456789', 'index 5' - assert self.get(url='/?blah')['body'] == '0123456789', 'index vars' - assert self.get(url='/#blah')['body'] == '0123456789', 'index anchor' - assert self.get(url='/dir/')['status'] == 404, 'index not found' - - resp = self.get(url='/index.html/') - assert resp['status'] == 404, 'index not found 2 status' - assert ( - resp['headers']['Content-Type'] == 'text/html' - ), 'index not found 2 Content-Type' - def test_static_index_invalid(self, skip_alert, temp_dir): - skip_alert(r'failed to apply new conf') +def test_static_index(temp_dir): + def set_index(index): + assert 'success' in client.conf( + {"share": f'{temp_dir}/assets$uri', "index": index}, + 'routes/0/action', + ), 'configure index' - def check_index(index): - assert 'error' in self.conf( - {"share": f'{temp_dir}/assets$uri', "index": index}, - 'routes/0/action', - ) + set_index('README') + assert client.get()['body'] == 'readme', 'index' - check_index({}) - check_index(['index.html', '$blah']) + client.conf_delete('routes/0/action/index') + assert client.get()['body'] == '0123456789', 'delete index' - def test_static_large_file(self, temp_dir): - file_size = 32 * 1024 * 1024 - with open(f'{temp_dir}/assets/large', 'wb') as f: - f.seek(file_size - 1) - f.write(b'\0') + set_index('') + assert client.get()['status'] == 404, 'index empty' - assert ( - len(self.get(url='/large', read_buffer_size=1024 * 1024)['body']) - == file_size - ), 'large file' - def test_static_etag(self, temp_dir): - etag = self.get(url='/')['headers']['ETag'] - etag_2 = self.get(url='/README')['headers']['ETag'] +def test_static_index_default(): + assert client.get(url='/index.html')['body'] == '0123456789', 'index' + assert client.get(url='/')['body'] == '0123456789', 'index 2' + assert client.get(url='//')['body'] == '0123456789', 'index 3' + assert client.get(url='/.')['body'] == '0123456789', 'index 4' + assert client.get(url='/./')['body'] == '0123456789', 'index 5' + assert client.get(url='/?blah')['body'] == '0123456789', 'index vars' + assert client.get(url='/#blah')['body'] == '0123456789', 'index anchor' + assert client.get(url='/dir/')['status'] == 404, 'index not found' - assert etag != etag_2, 'different ETag' - assert etag == self.get(url='/')['headers']['ETag'], 'same ETag' + resp = client.get(url='/index.html/') + assert resp['status'] == 404, 'index not found 2 status' + assert ( + resp['headers']['Content-Type'] == 'text/html' + ), 'index not found 2 Content-Type' - with open(f'{temp_dir}/assets/index.html', 'w') as f: - f.write('blah') - assert etag != self.get(url='/')['headers']['ETag'], 'new ETag' +def test_static_index_invalid(skip_alert, temp_dir): + skip_alert(r'failed to apply new conf') - def test_static_redirect(self): - resp = self.get(url='/dir') - assert resp['status'] == 301, 'redirect status' - assert resp['headers']['Location'] == '/dir/', 'redirect Location' - assert 'Content-Type' not in resp['headers'], 'redirect Content-Type' + def check_index(index): + assert 'error' in client.conf( + {"share": f'{temp_dir}/assets$uri', "index": index}, + 'routes/0/action', + ) - def test_static_space_in_name(self, temp_dir): - assets_dir = f'{temp_dir}/assets' + check_index({}) + check_index(['index.html', '$blah']) - os.rename( - f'{assets_dir}/dir/file', - f'{assets_dir}/dir/fi le', - ) - assert waitforfiles(f'{assets_dir}/dir/fi le') - assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name' - os.rename(f'{assets_dir}/dir', f'{assets_dir}/di r') - assert waitforfiles(f'{assets_dir}/di r/fi le') - assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name' +def test_static_large_file(temp_dir): + file_size = 32 * 1024 * 1024 + with open(f'{temp_dir}/assets/large', 'wb') as f: + f.seek(file_size - 1) + f.write(b'\0') - os.rename(f'{assets_dir}/di r', f'{assets_dir}/ di r ') - assert waitforfiles(f'{assets_dir}/ di r /fi le') - assert ( - self.get(url='/ di r /fi le')['body'] == 'blah' - ), 'dir name enclosing' + assert ( + len(client.get(url='/large', read_buffer_size=1024 * 1024)['body']) + == file_size + ), 'large file' - assert ( - self.get(url='/%20di%20r%20/fi le')['body'] == 'blah' - ), 'dir encoded' - assert ( - self.get(url='/ di r %2Ffi le')['body'] == 'blah' - ), 'slash encoded' - assert self.get(url='/ di r /fi%20le')['body'] == 'blah', 'file encoded' - assert ( - self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah' - ), 'encoded' - assert ( - self.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body'] - == 'blah' - ), 'encoded 2' - os.rename( - f'{assets_dir}/ di r /fi le', - f'{assets_dir}/ di r / fi le ', - ) - assert waitforfiles(f'{assets_dir}/ di r / fi le ') - assert ( - self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah' - ), 'file name enclosing' - - try: - open(f'{temp_dir}/ф а', 'a').close() - utf8 = True - - except KeyboardInterrupt: - raise - - except: - utf8 = False - - if utf8: - os.rename( - f'{assets_dir}/ di r / fi le ', - f'{assets_dir}/ di r /фа йл', - ) - assert waitforfiles(f'{assets_dir}/ di r /фа йл') - assert ( - self.get(url='/ di r /фа йл')['body'] == 'blah' - ), 'file name 2' - - os.rename( - f'{assets_dir}/ di r ', - f'{assets_dir}/ди ректория', - ) - assert waitforfiles(f'{assets_dir}/ди ректория/фа йл') - assert ( - self.get(url='/ди ректория/фа йл')['body'] == 'blah' - ), 'dir name 2' - - def test_static_unix_socket(self, temp_dir): - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.bind(f'{temp_dir}/assets/unix_socket') - - assert self.get(url='/unix_socket')['status'] == 404, 'socket' - - sock.close() - - def test_static_unix_fifo(self, temp_dir): - os.mkfifo(f'{temp_dir}/assets/fifo') - - assert self.get(url='/fifo')['status'] == 404, 'fifo' - - def test_static_method(self): - resp = self.head() - assert resp['status'] == 200, 'HEAD status' - assert resp['body'] == '', 'HEAD empty body' - - assert self.delete()['status'] == 405, 'DELETE' - assert self.post()['status'] == 405, 'POST' - assert self.put()['status'] == 405, 'PUT' - - def test_static_path(self): - assert self.get(url='/dir/../dir/file')['status'] == 200, 'relative' - - assert self.get(url='./')['status'] == 400, 'path invalid' - assert self.get(url='../')['status'] == 400, 'path invalid 2' - assert self.get(url='/..')['status'] == 400, 'path invalid 3' - assert self.get(url='../assets/')['status'] == 400, 'path invalid 4' - assert self.get(url='/../assets/')['status'] == 400, 'path invalid 5' - - def test_static_two_clients(self): - sock = self.get(no_recv=True) - sock2 = self.get(no_recv=True) - - assert sock.recv(1) == b'H', 'client 1' - assert sock2.recv(1) == b'H', 'client 2' - assert sock.recv(1) == b'T', 'client 1 again' - assert sock2.recv(1) == b'T', 'client 2 again' - - sock.close() - sock2.close() - - def test_static_mime_types(self): - assert 'success' in self.conf( - { - "text/x-code/x-blah/x-blah": "readme", - "text/plain": [".html", ".log", "file"], - }, - 'settings/http/static/mime_types', - ), 'configure mime_types' +def test_static_etag(temp_dir): + etag = client.get(url='/')['headers']['ETag'] + etag_2 = client.get(url='/README')['headers']['ETag'] - assert ( - self.get(url='/README')['headers']['Content-Type'] - == 'text/x-code/x-blah/x-blah' - ), 'mime_types string case insensitive' - assert ( - self.get(url='/index.html')['headers']['Content-Type'] - == 'text/plain' - ), 'mime_types html' - assert ( - self.get(url='/')['headers']['Content-Type'] == 'text/plain' - ), 'mime_types index default' - assert ( - self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain' - ), 'mime_types file in dir' + assert etag != etag_2, 'different ETag' + assert etag == client.get(url='/')['headers']['ETag'], 'same ETag' - def test_static_mime_types_partial_match(self): - assert 'success' in self.conf( - { - "text/x-blah": ["ile", "fil", "f", "e", ".file"], - }, - 'settings/http/static/mime_types', - ), 'configure mime_types' - assert 'Content-Type' not in self.get(url='/dir/file'), 'partial match' - - def test_static_mime_types_reconfigure(self): - assert 'success' in self.conf( - { - "text/x-code": "readme", - "text/plain": [".html", ".log", "file"], - }, - 'settings/http/static/mime_types', - ), 'configure mime_types' + with open(f'{temp_dir}/assets/index.html', 'w') as f: + f.write('blah') - assert self.conf_get('settings/http/static/mime_types') == { - 'text/x-code': 'readme', - 'text/plain': ['.html', '.log', 'file'], - }, 'mime_types get' - assert ( - self.conf_get('settings/http/static/mime_types/text%2Fx-code') - == 'readme' - ), 'mime_types get string' - assert self.conf_get( - 'settings/http/static/mime_types/text%2Fplain' - ) == ['.html', '.log', 'file'], 'mime_types get array' - assert ( - self.conf_get('settings/http/static/mime_types/text%2Fplain/1') - == '.log' - ), 'mime_types get array element' + assert etag != client.get(url='/')['headers']['ETag'], 'new ETag' - assert 'success' in self.conf_delete( - 'settings/http/static/mime_types/text%2Fplain/2' - ), 'mime_types remove array element' - assert ( - 'Content-Type' not in self.get(url='/dir/file')['headers'] - ), 'mime_types removed' - assert 'success' in self.conf_post( - '"file"', 'settings/http/static/mime_types/text%2Fplain' - ), 'mime_types add array element' - assert ( - self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain' - ), 'mime_types reverted' +def test_static_redirect(): + resp = client.get(url='/dir') + assert resp['status'] == 301, 'redirect status' + assert resp['headers']['Location'] == '/dir/', 'redirect Location' + assert 'Content-Type' not in resp['headers'], 'redirect Content-Type' - assert 'success' in self.conf( - '"file"', 'settings/http/static/mime_types/text%2Fplain' - ), 'configure mime_types update' - assert ( - self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain' - ), 'mime_types updated' - assert ( - 'Content-Type' not in self.get(url='/log.log')['headers'] - ), 'mime_types updated 2' - assert 'success' in self.conf( - '".log"', 'settings/http/static/mime_types/text%2Fblahblahblah' - ), 'configure mime_types create' +def test_static_space_in_name(temp_dir): + assets_dir = f'{temp_dir}/assets' + + os.rename( + f'{assets_dir}/dir/file', + f'{assets_dir}/dir/fi le', + ) + assert waitforfiles(f'{assets_dir}/dir/fi le') + assert client.get(url='/dir/fi le')['body'] == 'blah', 'file name' + + os.rename(f'{assets_dir}/dir', f'{assets_dir}/di r') + assert waitforfiles(f'{assets_dir}/di r/fi le') + assert client.get(url='/di r/fi le')['body'] == 'blah', 'dir name' + + os.rename(f'{assets_dir}/di r', f'{assets_dir}/ di r ') + assert waitforfiles(f'{assets_dir}/ di r /fi le') + assert ( + client.get(url='/ di r /fi le')['body'] == 'blah' + ), 'dir name enclosing' + + assert ( + client.get(url='/%20di%20r%20/fi le')['body'] == 'blah' + ), 'dir encoded' + assert client.get(url='/ di r %2Ffi le')['body'] == 'blah', 'slash encoded' + assert client.get(url='/ di r /fi%20le')['body'] == 'blah', 'file encoded' + assert ( + client.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah' + ), 'encoded' + assert ( + client.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body'] + == 'blah' + ), 'encoded 2' + + os.rename( + f'{assets_dir}/ di r /fi le', + f'{assets_dir}/ di r / fi le ', + ) + assert waitforfiles(f'{assets_dir}/ di r / fi le ') + assert ( + client.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah' + ), 'file name enclosing' + + try: + open(f'{temp_dir}/ф а', 'a').close() + utf8 = True + + except KeyboardInterrupt: + raise + + except: + utf8 = False + + if utf8: + os.rename( + f'{assets_dir}/ di r / fi le ', + f'{assets_dir}/ di r /фа йл', + ) + assert waitforfiles(f'{assets_dir}/ di r /фа йл') + assert client.get(url='/ di r /фа йл')['body'] == 'blah' + + os.rename( + f'{assets_dir}/ di r ', + f'{assets_dir}/ди ректория', + ) + assert waitforfiles(f'{assets_dir}/ди ректория/фа йл') assert ( - self.get(url='/log.log')['headers']['Content-Type'] - == 'text/blahblahblah' - ), 'mime_types create' - - def test_static_mime_types_correct(self): - assert 'error' in self.conf( - {"text/x-code": "readme", "text/plain": "readme"}, - 'settings/http/static/mime_types', - ), 'mime_types same extensions' - assert 'error' in self.conf( - {"text/x-code": [".h", ".c"], "text/plain": ".c"}, - 'settings/http/static/mime_types', - ), 'mime_types same extensions array' - assert 'error' in self.conf( - { - "text/x-code": [".h", ".c", "readme"], - "text/plain": "README", - }, - 'settings/http/static/mime_types', - ), 'mime_types same extensions case insensitive' + client.get(url='/ди ректория/фа йл')['body'] == 'blah' + ), 'dir name 2' - @pytest.mark.skip('not yet') - def test_static_mime_types_invalid(self, temp_dir): - assert 'error' in self.http( - b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r + +def test_static_unix_socket(temp_dir): + sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + sock.bind(f'{temp_dir}/assets/unix_socket') + + assert client.get(url='/unix_socket')['status'] == 404, 'socket' + + sock.close() + + +def test_static_unix_fifo(temp_dir): + os.mkfifo(f'{temp_dir}/assets/fifo') + + assert client.get(url='/fifo')['status'] == 404, 'fifo' + + +def test_static_method(): + resp = client.head() + assert resp['status'] == 200, 'HEAD status' + assert resp['body'] == '', 'HEAD empty body' + + assert client.delete()['status'] == 405, 'DELETE' + assert client.post()['status'] == 405, 'POST' + assert client.put()['status'] == 405, 'PUT' + + +def test_static_path(): + assert client.get(url='/dir/../dir/file')['status'] == 200, 'relative' + + assert client.get(url='./')['status'] == 400, 'path invalid' + assert client.get(url='../')['status'] == 400, 'path invalid 2' + assert client.get(url='/..')['status'] == 400, 'path invalid 3' + assert client.get(url='../assets/')['status'] == 400, 'path invalid 4' + assert client.get(url='/../assets/')['status'] == 400, 'path invalid 5' + + +def test_static_two_clients(): + sock = client.get(no_recv=True) + sock2 = client.get(no_recv=True) + + assert sock.recv(1) == b'H', 'client 1' + assert sock2.recv(1) == b'H', 'client 2' + assert sock.recv(1) == b'T', 'client 1 again' + assert sock2.recv(1) == b'T', 'client 2 again' + + sock.close() + sock2.close() + + +def test_static_mime_types(): + assert 'success' in client.conf( + { + "text/x-code/x-blah/x-blah": "readme", + "text/plain": [".html", ".log", "file"], + }, + 'settings/http/static/mime_types', + ), 'configure mime_types' + + assert ( + client.get(url='/README')['headers']['Content-Type'] + == 'text/x-code/x-blah/x-blah' + ), 'mime_types string case insensitive' + assert ( + client.get(url='/index.html')['headers']['Content-Type'] == 'text/plain' + ), 'mime_types html' + assert ( + client.get(url='/')['headers']['Content-Type'] == 'text/plain' + ), 'mime_types index default' + assert ( + client.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain' + ), 'mime_types file in dir' + + +def test_static_mime_types_partial_match(): + assert 'success' in client.conf( + { + "text/x-blah": ["ile", "fil", "f", "e", ".file"], + }, + 'settings/http/static/mime_types', + ), 'configure mime_types' + assert 'Content-Type' not in client.get(url='/dir/file'), 'partial match' + + +def test_static_mime_types_reconfigure(): + assert 'success' in client.conf( + { + "text/x-code": "readme", + "text/plain": [".html", ".log", "file"], + }, + 'settings/http/static/mime_types', + ), 'configure mime_types' + + assert client.conf_get('settings/http/static/mime_types') == { + 'text/x-code': 'readme', + 'text/plain': ['.html', '.log', 'file'], + }, 'mime_types get' + assert ( + client.conf_get('settings/http/static/mime_types/text%2Fx-code') + == 'readme' + ), 'mime_types get string' + assert client.conf_get('settings/http/static/mime_types/text%2Fplain') == [ + '.html', + '.log', + 'file', + ], 'mime_types get array' + assert ( + client.conf_get('settings/http/static/mime_types/text%2Fplain/1') + == '.log' + ), 'mime_types get array element' + + assert 'success' in client.conf_delete( + 'settings/http/static/mime_types/text%2Fplain/2' + ), 'mime_types remove array element' + assert ( + 'Content-Type' not in client.get(url='/dir/file')['headers'] + ), 'mime_types removed' + + assert 'success' in client.conf_post( + '"file"', 'settings/http/static/mime_types/text%2Fplain' + ), 'mime_types add array element' + assert ( + client.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain' + ), 'mime_types reverted' + + assert 'success' in client.conf( + '"file"', 'settings/http/static/mime_types/text%2Fplain' + ), 'configure mime_types update' + assert ( + client.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain' + ), 'mime_types updated' + assert ( + 'Content-Type' not in client.get(url='/log.log')['headers'] + ), 'mime_types updated 2' + + assert 'success' in client.conf( + '".log"', 'settings/http/static/mime_types/text%2Fblahblahblah' + ), 'configure mime_types create' + assert ( + client.get(url='/log.log')['headers']['Content-Type'] + == 'text/blahblahblah' + ), 'mime_types create' + + +def test_static_mime_types_correct(): + assert 'error' in client.conf( + {"text/x-code": "readme", "text/plain": "readme"}, + 'settings/http/static/mime_types', + ), 'mime_types same extensions' + assert 'error' in client.conf( + {"text/x-code": [".h", ".c"], "text/plain": ".c"}, + 'settings/http/static/mime_types', + ), 'mime_types same extensions array' + assert 'error' in client.conf( + { + "text/x-code": [".h", ".c", "readme"], + "text/plain": "README", + }, + 'settings/http/static/mime_types', + ), 'mime_types same extensions case insensitive' + + +@pytest.mark.skip('not yet') +def test_static_mime_types_invalid(temp_dir): + assert 'error' in client.http( + b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r Host: localhost\r Connection: close\r Content-Length: 6\r \r \"blah\"""", - raw_resp=True, - raw=True, - sock_type='unix', - addr=f'{temp_dir}/control.unit.sock', - ), 'mime_types invalid' + raw_resp=True, + raw=True, + sock_type='unix', + addr=f'{temp_dir}/control.unit.sock', + ), 'mime_types invalid' |