diff options
Diffstat (limited to 'test/test_static.py')
| -rw-r--r-- | test/test_static.py | 638 |
1 files changed, 324 insertions, 314 deletions
diff --git a/test/test_static.py b/test/test_static.py index f7eade7c..d46247d9 100644 --- a/test/test_static.py +++ b/test/test_static.py @@ -2,351 +2,361 @@ import os import socket import pytest -from unit.applications.proto import TestApplicationProto -from unit.option import option +from unit.applications.proto import ApplicationProto from unit.utils import waitforfiles -class TestStatic(TestApplicationProto): - prerequisites = {} - - def setup_method(self): - os.makedirs(f'{option.temp_dir}/assets/dir') - with open(f'{option.temp_dir}/assets/index.html', 'w') as index, open( - f'{option.temp_dir}/assets/README', 'w' - ) as readme, open( - f'{option.temp_dir}/assets/log.log', 'w' - ) as log, open( - f'{option.temp_dir}/assets/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'{option.temp_dir}/assets$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' |