From b86891c4ef848a2da05abd1350af5f0b8e4335fa Mon Sep 17 00:00:00 2001 From: Andrei Zeliankou Date: Mon, 28 Jun 2021 22:05:40 +0100 Subject: Tests: renamed share to static. Also minor style changes. --- test/conftest.py | 2 +- test/test_configuration.py | 3 +- test/test_node_es_modules.py | 3 +- test/test_share_chroot.py | 111 ---------------------------- test/test_share_fallback.py | 148 ------------------------------------- test/test_share_mount.py | 142 ------------------------------------ test/test_share_symlink.py | 96 ------------------------ test/test_share_types.py | 170 ------------------------------------------- test/test_static_chroot.py | 108 +++++++++++++++++++++++++++ test/test_static_fallback.py | 149 +++++++++++++++++++++++++++++++++++++ test/test_static_mount.py | 140 +++++++++++++++++++++++++++++++++++ test/test_static_symlink.py | 94 ++++++++++++++++++++++++ test/test_static_types.py | 169 ++++++++++++++++++++++++++++++++++++++++++ 13 files changed, 665 insertions(+), 670 deletions(-) delete mode 100644 test/test_share_chroot.py delete mode 100644 test/test_share_fallback.py delete mode 100644 test/test_share_mount.py delete mode 100644 test/test_share_symlink.py delete mode 100644 test/test_share_types.py create mode 100644 test/test_static_chroot.py create mode 100644 test/test_static_fallback.py create mode 100644 test/test_static_mount.py create mode 100644 test/test_static_symlink.py create mode 100644 test/test_static_types.py (limited to 'test') diff --git a/test/conftest.py b/test/conftest.py index 5ea4e49d..db34984f 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -22,8 +22,8 @@ from unit.check.node import check_node from unit.check.regex import check_regex from unit.check.tls import check_openssl from unit.http import TestHTTP -from unit.option import option from unit.log import Log +from unit.option import option from unit.utils import public_dir from unit.utils import waitforfiles diff --git a/test/test_configuration.py b/test/test_configuration.py index 880aef6c..c149658c 100644 --- a/test/test_configuration.py +++ b/test/test_configuration.py @@ -1,6 +1,7 @@ +import socket + import pytest -import socket from unit.control import TestControl diff --git a/test/test_node_es_modules.py b/test/test_node_es_modules.py index 0945a967..5464d4a6 100644 --- a/test/test_node_es_modules.py +++ b/test/test_node_es_modules.py @@ -1,6 +1,7 @@ +from distutils.version import LooseVersion + import pytest -from distutils.version import LooseVersion from unit.applications.lang.node import TestApplicationNode from unit.applications.websockets import TestApplicationWebsocket diff --git a/test/test_share_chroot.py b/test/test_share_chroot.py deleted file mode 100644 index 02b3657d..00000000 --- a/test/test_share_chroot.py +++ /dev/null @@ -1,111 +0,0 @@ -import os -from pathlib import Path - -import pytest - -from unit.applications.proto import TestApplicationProto - - -class TestShareChroot(TestApplicationProto): - prerequisites = {'features': ['chroot']} - - @pytest.fixture(autouse=True) - def setup_method_fixture(self, temp_dir): - os.makedirs(temp_dir + '/assets/dir') - with open(temp_dir + '/assets/index.html', 'w') as index, open( - temp_dir + '/assets/dir/file', 'w' - ) as file: - index.write('0123456789') - file.write('blah') - - test = Path(__file__) - self.test_path = '/' + test.parent.name + '/' + test.name - - self._load_conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [{"action": {"share": temp_dir + "/assets"}}], - } - ) - - def test_share_chroot(self, temp_dir): - assert self.get(url='/dir/file')['status'] == 200, 'default chroot' - assert self.get(url='/index.html')['status'] == 200, 'default chroot 2' - - assert 'success' in self.conf( - { - "share": temp_dir + "/assets", - "chroot": temp_dir + "/assets/dir", - }, - 'routes/0/action', - ), 'configure chroot' - - assert self.get(url='/dir/file')['status'] == 200, 'chroot' - assert self.get(url='/index.html')['status'] == 403, 'chroot 403 2' - assert self.get(url='/file')['status'] == 403, 'chroot 403' - - def test_share_chroot_permission(self, is_su, temp_dir): - if is_su: - pytest.skip('does\'t work under root') - - os.chmod(temp_dir + '/assets/dir', 0o100) - - assert 'success' in self.conf( - { - "share": temp_dir + "/assets", - "chroot": temp_dir + "/assets/dir", - }, - 'routes/0/action', - ), 'configure chroot' - - assert self.get(url='/dir/file')['status'] == 200, 'chroot' - - def test_share_chroot_empty(self, temp_dir): - assert 'success' in self.conf( - {"share": temp_dir + "/assets", "chroot": ""}, 'routes/0/action', - ), 'configure chroot empty absolute' - - assert ( - self.get(url='/dir/file')['status'] == 200 - ), 'chroot empty absolute' - - assert 'success' in self.conf( - {"share": ".", "chroot": ""}, 'routes/0/action', - ), 'configure chroot empty relative' - - assert ( - self.get(url=self.test_path)['status'] == 200 - ), 'chroot empty relative' - - def test_share_chroot_relative(self, is_su, temp_dir): - if is_su: - pytest.skip('does\'t work under root') - - assert 'success' in self.conf( - {"share": temp_dir + "/assets", "chroot": "."}, 'routes/0/action', - ), 'configure relative chroot' - - assert self.get(url='/dir/file')['status'] == 403, 'relative chroot' - - assert 'success' in self.conf( - {"share": "."}, 'routes/0/action', - ), 'configure relative share' - - assert self.get(url=self.test_path)['status'] == 200, 'relative share' - - assert 'success' in self.conf( - {"share": ".", "chroot": "."}, 'routes/0/action', - ), 'configure relative' - - assert self.get(url=self.test_path)['status'] == 200, 'relative' - - def test_share_chroot_invalid(self, temp_dir): - assert 'error' in self.conf( - {"share": temp_dir, "chroot": True}, 'routes/0/action', - ), 'configure chroot error' - assert 'error' in self.conf( - {"share": temp_dir, "symlinks": "True"}, 'routes/0/action', - ), 'configure symlink error' - assert 'error' in self.conf( - {"share": temp_dir, "mount": "True"}, 'routes/0/action', - ), 'configure mount error' diff --git a/test/test_share_fallback.py b/test/test_share_fallback.py deleted file mode 100644 index 0b1c270e..00000000 --- a/test/test_share_fallback.py +++ /dev/null @@ -1,148 +0,0 @@ -import os - -import pytest - -from unit.applications.proto import TestApplicationProto -from unit.option import option - - -class TestStatic(TestApplicationProto): - prerequisites = {} - - def setup_method(self): - os.makedirs(option.temp_dir + '/assets/dir') - with open(option.temp_dir + '/assets/index.html', 'w') as index: - index.write('0123456789') - - os.makedirs(option.temp_dir + '/assets/403') - os.chmod(option.temp_dir + '/assets/403', 0o000) - - self._load_conf( - { - "listeners": { - "*:7080": {"pass": "routes"}, - "*:7081": {"pass": "routes"}, - }, - "routes": [{"action": {"share": option.temp_dir + "/assets"}}], - "applications": {}, - } - ) - - def teardown_method(self): - try: - os.chmod(option.temp_dir + '/assets/403', 0o777) - except FileNotFoundError: - pass - - def action_update(self, conf): - assert 'success' in self.conf(conf, 'routes/0/action') - - def test_fallback(self): - self.action_update({"share": "/blah"}) - assert self.get()['status'] == 404, 'bad path no fallback' - - self.action_update({"share": "/blah", "fallback": {"return": 200}}) - - resp = self.get() - assert resp['status'] == 200, 'bad path fallback status' - assert resp['body'] == '', 'bad path fallback' - - def test_fallback_valid_path(self, temp_dir): - self.action_update( - {"share": temp_dir + "/assets", "fallback": {"return": 200}} - ) - resp = self.get() - assert resp['status'] == 200, 'fallback status' - assert resp['body'] == '0123456789', 'fallback' - - resp = self.get(url='/403/') - assert resp['status'] == 200, 'fallback status 403' - assert resp['body'] == '', 'fallback 403' - - resp = self.post() - assert resp['status'] == 200, 'fallback status 405' - assert resp['body'] == '', 'fallback 405' - - assert self.get(url='/dir')['status'] == 301, 'fallback status 301' - - def test_fallback_nested(self): - self.action_update( - { - "share": "/blah", - "fallback": { - "share": "/blah/blah", - "fallback": {"return": 200}, - }, - } - ) - - resp = self.get() - assert resp['status'] == 200, 'fallback nested status' - assert resp['body'] == '', 'fallback nested' - - def test_fallback_share(self, temp_dir): - self.action_update( - {"share": "/blah", "fallback": {"share": temp_dir + "/assets"},} - ) - - resp = self.get() - assert resp['status'] == 200, 'fallback share status' - assert resp['body'] == '0123456789', 'fallback share' - - resp = self.head() - assert resp['status'] == 200, 'fallback share status HEAD' - assert resp['body'] == '', 'fallback share HEAD' - - assert ( - self.get(url='/dir')['status'] == 301 - ), 'fallback share status 301' - - def test_fallback_proxy(self): - assert 'success' in self.conf( - [ - { - "match": {"destination": "*:7081"}, - "action": {"return": 200}, - }, - { - "action": { - "share": "/blah", - "fallback": {"proxy": "http://127.0.0.1:7081"}, - } - }, - ], - 'routes', - ), 'configure fallback proxy route' - - resp = self.get() - assert resp['status'] == 200, 'fallback proxy status' - assert resp['body'] == '', 'fallback proxy' - - @pytest.mark.skip('not yet') - def test_fallback_proxy_loop(self, skip_alert): - skip_alert( - r'open.*/blah/index.html.*failed', - r'accept.*failed', - r'socket.*failed', - r'new connections are not accepted', - ) - - self.action_update( - {"share": "/blah", "fallback": {"proxy": "http://127.0.0.1:7080"}} - ) - self.get(no_recv=True) - - assert 'success' in self.conf_delete('listeners/*:7081') - self.get(read_timeout=1) - - def test_fallback_invalid(self): - def check_error(conf): - assert 'error' in self.conf(conf, 'routes/0/action') - - check_error({"share": "/blah", "fallback": {}}) - check_error({"share": "/blah", "fallback": ""}) - check_error({"return": 200, "fallback": {"share": "/blah"}}) - check_error( - {"proxy": "http://127.0.0.1:7081", "fallback": {"share": "/blah"}} - ) - check_error({"fallback": {"share": "/blah"}}) diff --git a/test/test_share_mount.py b/test/test_share_mount.py deleted file mode 100644 index f22fbe75..00000000 --- a/test/test_share_mount.py +++ /dev/null @@ -1,142 +0,0 @@ -import os -import subprocess - -import pytest - -from unit.applications.proto import TestApplicationProto - - -class TestShareMount(TestApplicationProto): - prerequisites = {'features': ['chroot']} - - @pytest.fixture(autouse=True) - def setup_method_fixture(self, is_su, temp_dir): - if not is_su: - pytest.skip('requires root') - - os.makedirs(temp_dir + '/assets/dir/mount') - os.makedirs(temp_dir + '/assets/dir/dir') - os.makedirs(temp_dir + '/assets/mount') - with open(temp_dir + '/assets/index.html', 'w') as index, open( - temp_dir + '/assets/dir/dir/file', 'w' - ) as file, open(temp_dir + '/assets/mount/index.html', 'w') as mount: - index.write('index') - file.write('file') - mount.write('mount') - - try: - process = subprocess.Popen( - [ - "mount", - "--bind", - temp_dir + "/assets/mount", - temp_dir + "/assets/dir/mount", - ], - stderr=subprocess.STDOUT, - ) - - process.communicate() - - except KeyboardInterrupt: - raise - - except: - pytest.fail('Can\'t run mount process.') - - self._load_conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [{"action": {"share": temp_dir + "/assets/dir"}}], - } - ) - - yield - - try: - process = subprocess.Popen( - ["umount", "--lazy", temp_dir + "/assets/dir/mount"], - stderr=subprocess.STDOUT, - ) - - process.communicate() - - except KeyboardInterrupt: - raise - - except: - pytest.fail('Can\'t run umount process.') - - def test_share_mount(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') - - resp = self.get(url='/mount/') - assert resp['status'] == 200 - assert resp['body'] == 'mount' - - assert 'success' in self.conf( - {"share": temp_dir + "/assets/dir", "traverse_mounts": False}, - 'routes/0/action', - ), 'configure mount disable' - - assert self.get(url='/mount/')['status'] == 403 - - assert 'success' in self.conf( - {"share": temp_dir + "/assets/dir", "traverse_mounts": True}, - 'routes/0/action', - ), 'configure mount enable' - - resp = self.get(url='/mount/') - assert resp['status'] == 200 - assert resp['body'] == 'mount' - - def test_share_mount_two_blocks(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') - - os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link') - - assert 'success' in self.conf( - [ - { - "match": {"method": "HEAD"}, - "action": { - "share": temp_dir + "/assets/dir", - "traverse_mounts": False, - }, - }, - { - "match": {"method": "GET"}, - "action": { - "share": temp_dir + "/assets/dir", - "traverse_mounts": True, - }, - }, - ], - 'routes', - ), 'configure two options' - - assert self.get(url='/mount/')['status'] == 200, 'block enabled' - assert self.head(url='/mount/')['status'] == 403, 'block disabled' - - def test_share_mount_chroot(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') - - assert 'success' in self.conf( - { - "share": temp_dir + "/assets/dir", - "chroot": temp_dir + "/assets", - }, - 'routes/0/action', - ), 'configure chroot mount default' - - assert self.get(url='/mount/')['status'] == 200, 'chroot' - - assert 'success' in self.conf( - { - "share": temp_dir + "/assets/dir", - "chroot": temp_dir + "/assets", - "traverse_mounts": False, - }, - 'routes/0/action', - ), 'configure chroot mount disable' - - assert self.get(url='/mount/')['status'] == 403, 'chroot mount' diff --git a/test/test_share_symlink.py b/test/test_share_symlink.py deleted file mode 100644 index 3970b605..00000000 --- a/test/test_share_symlink.py +++ /dev/null @@ -1,96 +0,0 @@ -import os - -import pytest - -from unit.applications.proto import TestApplicationProto - - -class TestShareSymlink(TestApplicationProto): - prerequisites = {'features': ['chroot']} - - @pytest.fixture(autouse=True) - def setup_method_fixture(self, temp_dir): - os.makedirs(temp_dir + '/assets/dir/dir') - with open(temp_dir + '/assets/index.html', 'w') as index, open( - temp_dir + '/assets/dir/file', 'w' - ) as file: - index.write('0123456789') - file.write('blah') - - self._load_conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [{"action": {"share": temp_dir + "/assets"}}], - } - ) - - def test_share_symlink(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') - - os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link') - - assert self.get(url='/dir')['status'] == 301, 'dir' - assert self.get(url='/dir/file')['status'] == 200, 'file' - assert self.get(url='/link')['status'] == 301, 'symlink dir' - assert self.get(url='/link/file')['status'] == 200, 'symlink file' - - assert 'success' in self.conf( - {"share": temp_dir + "/assets", "follow_symlinks": False}, - 'routes/0/action', - ), 'configure symlink disable' - - assert self.get(url='/link/file')['status'] == 403, 'symlink disabled' - - assert 'success' in self.conf( - {"share": temp_dir + "/assets", "follow_symlinks": True}, - 'routes/0/action', - ), 'configure symlink enable' - - assert self.get(url='/link/file')['status'] == 200, 'symlink enabled' - - def test_share_symlink_two_blocks(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') - - os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link') - - assert 'success' in self.conf( - [ - { - "match": {"method": "HEAD"}, - "action": { - "share": temp_dir + "/assets", - "follow_symlinks": False, - }, - }, - { - "match": {"method": "GET"}, - "action": { - "share": temp_dir + "/assets", - "follow_symlinks": True, - }, - }, - ], - 'routes', - ), 'configure two options' - - assert self.get(url='/link/file')['status'] == 200, 'block enabled' - assert self.head(url='/link/file')['status'] == 403, 'block disabled' - - def test_share_symlink_chroot(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') - - os.symlink( - temp_dir + '/assets/dir/file', temp_dir + '/assets/dir/dir/link' - ) - - assert self.get(url='/dir/dir/link')['status'] == 200, 'default chroot' - - assert 'success' in self.conf( - { - "share": temp_dir + "/assets", - "chroot": temp_dir + "/assets/dir/dir", - }, - 'routes/0/action', - ), 'configure chroot' - - assert self.get(url='/dir/dir/link')['status'] == 404, 'chroot' diff --git a/test/test_share_types.py b/test/test_share_types.py deleted file mode 100644 index b5ed97a0..00000000 --- a/test/test_share_types.py +++ /dev/null @@ -1,170 +0,0 @@ -import os -from pathlib import Path - -import pytest -from unit.applications.proto import TestApplicationProto -from unit.option import option - - -class TestShareTypes(TestApplicationProto): - prerequisites = {} - - @pytest.fixture(autouse=True) - def setup_method_fixture(self, temp_dir): - Path(temp_dir + '/assets').mkdir() - for ext in ['.xml', '.mp4', '.php', '', '.txt', '.html', '.png']: - Path(temp_dir + '/assets/file' + ext).write_text(ext) - - Path(temp_dir + '/assets/index.html').write_text('index') - - self._load_conf( - { - "listeners": { - "*:7080": {"pass": "routes"}, - "*:7081": {"pass": "routes"}, - }, - "routes": [{"action": {"share": temp_dir + "/assets"}}], - "applications": {}, - } - ) - - def action_update(self, conf): - assert 'success' in self.conf(conf, 'routes/0/action') - - def check_body(self, http_url, body): - resp = self.get(url=http_url) - assert resp['status'] == 200, 'status' - assert resp['body'] == body, 'body' - - def test_share_types_basic(self, temp_dir): - self.action_update({"share": temp_dir + "/assets"}) - self.check_body('/index.html', 'index') - self.check_body('/file.xml', '.xml') - - self.action_update( - {"share": temp_dir + "/assets", "types": "application/xml"} - ) - self.check_body('/file.xml', '.xml') - - self.action_update( - {"share": temp_dir + "/assets", "types": ["application/xml"]} - ) - self.check_body('/file.xml', '.xml') - - self.action_update({"share": temp_dir + "/assets", "types": [""]}) - assert self.get(url='/file.xml')['status'] == 403, 'no mtype' - - def test_share_types_wildcard(self, temp_dir): - self.action_update( - {"share": temp_dir + "/assets", "types": ["application/*"]} - ) - self.check_body('/file.xml', '.xml') - assert self.get(url='/file.mp4')['status'] == 403, 'app * mtype mp4' - - self.action_update( - {"share": temp_dir + "/assets", "types": ["video/*"]} - ) - assert self.get(url='/file.xml')['status'] == 403, 'video * mtype xml' - self.check_body('/file.mp4', '.mp4') - - def test_share_types_negation(self, temp_dir): - self.action_update( - {"share": temp_dir + "/assets", "types": ["!application/xml"]} - ) - assert self.get(url='/file.xml')['status'] == 403, 'forbidden negation' - self.check_body('/file.mp4', '.mp4') - - # sorting negation - self.action_update( - { - "share": temp_dir + "/assets", - "types": ["!video/*", "image/png", "!image/jpg"], - } - ) - assert self.get(url='/file.mp4')['status'] == 403, 'negation sort mp4' - self.check_body('/file.png', '.png') - assert self.get(url='/file.jpg')['status'] == 403, 'negation sort jpg' - - def test_share_types_regex(self, temp_dir): - self.action_update( - {"share": temp_dir + "/assets", "types": ["~text/(html|plain)"]} - ) - assert self.get(url='/file.php')['status'] == 403, 'regex fail' - self.check_body('/file.html', '.html') - self.check_body('/file.txt', '.txt') - - def test_share_types_case(self, temp_dir): - self.action_update( - {"share": temp_dir + "/assets", "types": ["!APpliCaTiOn/xMl"]} - ) - self.check_body('/file.mp4', '.mp4') - assert ( - self.get(url='/file.xml')['status'] == 403 - ), 'mixed case xml negation' - - self.action_update( - {"share": temp_dir + "/assets", "types": ["vIdEo/mp4"]} - ) - assert self.get(url='/file.mp4')['status'] == 200, 'mixed case' - assert ( - self.get(url='/file.xml')['status'] == 403 - ), 'mixed case video negation' - - self.action_update( - {"share": temp_dir + "/assets", "types": ["vIdEo/*"]} - ) - self.check_body('/file.mp4', '.mp4') - assert ( - self.get(url='/file.xml')['status'] == 403 - ), 'mixed case video * negation' - - def test_share_types_fallback(self, temp_dir): - assert 'success' in self.conf( - [ - { - "match": {"destination": "*:7081"}, - "action": {"return": 200}, - }, - { - "action": { - "share": temp_dir + "/assets", - "types": ["!application/x-httpd-php"], - "fallback": {"proxy": "http://127.0.0.1:7081"}, - } - }, - ], - 'routes', - ), 'configure fallback proxy route' - - self.check_body('/file.php', '') - self.check_body('/file.mp4', '.mp4') - - def test_share_types_index(self, temp_dir): - self.action_update( - {"share": temp_dir + "/assets", "types": "application/xml"} - ) - self.check_body('/', 'index') - self.check_body('/file.xml', '.xml') - assert self.get(url='/file.mp4')['status'] == 403, 'forbidden mtype' - - def test_share_types_custom_mime(self, temp_dir): - self._load_conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [{"action": {"share": temp_dir + "/assets"}}], - "applications": {}, - "settings": { - "http": { - "static": {"mime_types": {"test/mime-type": ["file"]}} - } - }, - } - ) - - self.action_update({"share": temp_dir + "/assets", "types": [""]}) - assert self.get(url='/file')['status'] == 403, 'forbidden custom mime' - - self.action_update( - {"share": temp_dir + "/assets", "types": ["test/mime-type"]} - ) - self.check_body('/file', '') diff --git a/test/test_static_chroot.py b/test/test_static_chroot.py new file mode 100644 index 00000000..f9bc93a8 --- /dev/null +++ b/test/test_static_chroot.py @@ -0,0 +1,108 @@ +import os +from pathlib import Path + +import pytest + +from unit.applications.proto import TestApplicationProto + + +class TestStaticChroot(TestApplicationProto): + prerequisites = {'features': ['chroot']} + + @pytest.fixture(autouse=True) + def setup_method_fixture(self, temp_dir): + os.makedirs(temp_dir + '/assets/dir') + Path(temp_dir + '/assets/index.html').write_text('0123456789') + Path(temp_dir + '/assets/dir/file').write_text('blah') + + test = Path(__file__) + self.test_path = '/' + test.parent.name + '/' + test.name + + self._load_conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [{"action": {"share": temp_dir + "/assets"}}], + } + ) + + def test_static_chroot(self, temp_dir): + assert self.get(url='/dir/file')['status'] == 200, 'default chroot' + assert self.get(url='/index.html')['status'] == 200, 'default chroot 2' + + assert 'success' in self.conf( + { + "share": temp_dir + "/assets", + "chroot": temp_dir + "/assets/dir", + }, + 'routes/0/action', + ), 'configure chroot' + + assert self.get(url='/dir/file')['status'] == 200, 'chroot' + assert self.get(url='/index.html')['status'] == 403, 'chroot 403 2' + assert self.get(url='/file')['status'] == 403, 'chroot 403' + + def test_static_chroot_permission(self, is_su, temp_dir): + if is_su: + pytest.skip('does\'t work under root') + + os.chmod(temp_dir + '/assets/dir', 0o100) + + assert 'success' in self.conf( + { + "share": temp_dir + "/assets", + "chroot": temp_dir + "/assets/dir", + }, + 'routes/0/action', + ), 'configure chroot' + + assert self.get(url='/dir/file')['status'] == 200, 'chroot' + + def test_static_chroot_empty(self, temp_dir): + assert 'success' in self.conf( + {"share": temp_dir + "/assets", "chroot": ""}, 'routes/0/action', + ), 'configure chroot empty absolute' + + assert ( + self.get(url='/dir/file')['status'] == 200 + ), 'chroot empty absolute' + + assert 'success' in self.conf( + {"share": ".", "chroot": ""}, 'routes/0/action', + ), 'configure chroot empty relative' + + assert ( + self.get(url=self.test_path)['status'] == 200 + ), 'chroot empty relative' + + def test_static_chroot_relative(self, is_su, temp_dir): + if is_su: + pytest.skip('does\'t work under root') + + assert 'success' in self.conf( + {"share": temp_dir + "/assets", "chroot": "."}, 'routes/0/action', + ), 'configure relative chroot' + + assert self.get(url='/dir/file')['status'] == 403, 'relative chroot' + + assert 'success' in self.conf( + {"share": "."}, 'routes/0/action', + ), 'configure relative share' + + assert self.get(url=self.test_path)['status'] == 200, 'relative share' + + assert 'success' in self.conf( + {"share": ".", "chroot": "."}, 'routes/0/action', + ), 'configure relative' + + assert self.get(url=self.test_path)['status'] == 200, 'relative' + + def test_static_chroot_invalid(self, temp_dir): + assert 'error' in self.conf( + {"share": temp_dir, "chroot": True}, 'routes/0/action', + ), 'configure chroot error' + assert 'error' in self.conf( + {"share": temp_dir, "symlinks": "True"}, 'routes/0/action', + ), 'configure symlink error' + assert 'error' in self.conf( + {"share": temp_dir, "mount": "True"}, 'routes/0/action', + ), 'configure mount error' diff --git a/test/test_static_fallback.py b/test/test_static_fallback.py new file mode 100644 index 00000000..dc9056b9 --- /dev/null +++ b/test/test_static_fallback.py @@ -0,0 +1,149 @@ +import os +from pathlib import Path + +import pytest + +from unit.applications.proto import TestApplicationProto + + +class TestStaticFallback(TestApplicationProto): + prerequisites = {} + + @pytest.fixture(autouse=True) + def setup_method_fixture(self, temp_dir): + os.makedirs(temp_dir + '/assets/dir') + Path(temp_dir + '/assets/index.html').write_text('0123456789') + + os.makedirs(temp_dir + '/assets/403') + os.chmod(temp_dir + '/assets/403', 0o000) + + self._load_conf( + { + "listeners": { + "*:7080": {"pass": "routes"}, + "*:7081": {"pass": "routes"}, + }, + "routes": [{"action": {"share": temp_dir + "/assets"}}], + "applications": {}, + } + ) + + yield + + try: + os.chmod(temp_dir + '/assets/403', 0o777) + except FileNotFoundError: + pass + + def action_update(self, conf): + assert 'success' in self.conf(conf, 'routes/0/action') + + def test_static_fallback(self): + self.action_update({"share": "/blah"}) + assert self.get()['status'] == 404, 'bad path no fallback' + + self.action_update({"share": "/blah", "fallback": {"return": 200}}) + + resp = self.get() + assert resp['status'] == 200, 'bad path fallback status' + assert resp['body'] == '', 'bad path fallback' + + def test_static_fallback_valid_path(self, temp_dir): + self.action_update( + {"share": temp_dir + "/assets", "fallback": {"return": 200}} + ) + resp = self.get() + assert resp['status'] == 200, 'fallback status' + assert resp['body'] == '0123456789', 'fallback' + + resp = self.get(url='/403/') + assert resp['status'] == 200, 'fallback status 403' + assert resp['body'] == '', 'fallback 403' + + resp = self.post() + assert resp['status'] == 200, 'fallback status 405' + assert resp['body'] == '', 'fallback 405' + + assert self.get(url='/dir')['status'] == 301, 'fallback status 301' + + def test_static_fallback_nested(self): + self.action_update( + { + "share": "/blah", + "fallback": { + "share": "/blah/blah", + "fallback": {"return": 200}, + }, + } + ) + + resp = self.get() + assert resp['status'] == 200, 'fallback nested status' + assert resp['body'] == '', 'fallback nested' + + def test_static_fallback_share(self, temp_dir): + self.action_update( + {"share": "/blah", "fallback": {"share": temp_dir + "/assets"},} + ) + + resp = self.get() + assert resp['status'] == 200, 'fallback share status' + assert resp['body'] == '0123456789', 'fallback share' + + resp = self.head() + assert resp['status'] == 200, 'fallback share status HEAD' + assert resp['body'] == '', 'fallback share HEAD' + + assert ( + self.get(url='/dir')['status'] == 301 + ), 'fallback share status 301' + + def test_static_fallback_proxy(self): + assert 'success' in self.conf( + [ + { + "match": {"destination": "*:7081"}, + "action": {"return": 200}, + }, + { + "action": { + "share": "/blah", + "fallback": {"proxy": "http://127.0.0.1:7081"}, + } + }, + ], + 'routes', + ), 'configure fallback proxy route' + + resp = self.get() + assert resp['status'] == 200, 'fallback proxy status' + assert resp['body'] == '', 'fallback proxy' + + @pytest.mark.skip('not yet') + def test_static_fallback_proxy_loop(self, skip_alert): + skip_alert( + r'open.*/blah/index.html.*failed', + r'accept.*failed', + r'socket.*failed', + r'new connections are not accepted', + ) + + self.action_update( + {"share": "/blah", "fallback": {"proxy": "http://127.0.0.1:7080"}} + ) + self.get(no_recv=True) + + assert 'success' in self.conf_delete('listeners/*:7081') + self.get(read_timeout=1) + + def test_static_fallback_invalid(self): + def check_error(conf): + assert 'error' in self.conf(conf, 'routes/0/action') + + check_error({"share": "/blah", "fallback": {}}) + check_error({"share": "/blah", "fallback": ""}) + check_error({"return": 200, "fallback": {"share": "/blah"}}) + check_error( + {"proxy": "http://127.0.0.1:7081", "fallback": {"share": "/blah"}} + ) + check_error({"fallback": {"share": "/blah"}}) diff --git a/test/test_static_mount.py b/test/test_static_mount.py new file mode 100644 index 00000000..570f6439 --- /dev/null +++ b/test/test_static_mount.py @@ -0,0 +1,140 @@ +import os +import subprocess +from pathlib import Path + +import pytest + +from unit.applications.proto import TestApplicationProto + + +class TestStaticMount(TestApplicationProto): + prerequisites = {'features': ['chroot']} + + @pytest.fixture(autouse=True) + def setup_method_fixture(self, is_su, temp_dir): + if not is_su: + pytest.skip('requires root') + + os.makedirs(temp_dir + '/assets/dir/mount') + os.makedirs(temp_dir + '/assets/dir/dir') + os.makedirs(temp_dir + '/assets/mount') + Path(temp_dir + '/assets/index.html').write_text('index') + Path(temp_dir + '/assets/dir/dir/file').write_text('file') + Path(temp_dir + '/assets/mount/index.html').write_text('mount') + + try: + process = subprocess.Popen( + [ + "mount", + "--bind", + temp_dir + "/assets/mount", + temp_dir + "/assets/dir/mount", + ], + stderr=subprocess.STDOUT, + ) + + process.communicate() + + except KeyboardInterrupt: + raise + + except: + pytest.fail('Can\'t run mount process.') + + self._load_conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [{"action": {"share": temp_dir + "/assets/dir"}}], + } + ) + + yield + + try: + process = subprocess.Popen( + ["umount", "--lazy", temp_dir + "/assets/dir/mount"], + stderr=subprocess.STDOUT, + ) + + process.communicate() + + except KeyboardInterrupt: + raise + + except: + pytest.fail('Can\'t run umount process.') + + def test_static_mount(self, temp_dir, skip_alert): + skip_alert(r'opening.*failed') + + resp = self.get(url='/mount/') + assert resp['status'] == 200 + assert resp['body'] == 'mount' + + assert 'success' in self.conf( + {"share": temp_dir + "/assets/dir", "traverse_mounts": False}, + 'routes/0/action', + ), 'configure mount disable' + + assert self.get(url='/mount/')['status'] == 403 + + assert 'success' in self.conf( + {"share": temp_dir + "/assets/dir", "traverse_mounts": True}, + 'routes/0/action', + ), 'configure mount enable' + + resp = self.get(url='/mount/') + assert resp['status'] == 200 + assert resp['body'] == 'mount' + + def test_static_mount_two_blocks(self, temp_dir, skip_alert): + skip_alert(r'opening.*failed') + + os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link') + + assert 'success' in self.conf( + [ + { + "match": {"method": "HEAD"}, + "action": { + "share": temp_dir + "/assets/dir", + "traverse_mounts": False, + }, + }, + { + "match": {"method": "GET"}, + "action": { + "share": temp_dir + "/assets/dir", + "traverse_mounts": True, + }, + }, + ], + 'routes', + ), 'configure two options' + + assert self.get(url='/mount/')['status'] == 200, 'block enabled' + assert self.head(url='/mount/')['status'] == 403, 'block disabled' + + def test_static_mount_chroot(self, temp_dir, skip_alert): + skip_alert(r'opening.*failed') + + assert 'success' in self.conf( + { + "share": temp_dir + "/assets/dir", + "chroot": temp_dir + "/assets", + }, + 'routes/0/action', + ), 'configure chroot mount default' + + assert self.get(url='/mount/')['status'] == 200, 'chroot' + + assert 'success' in self.conf( + { + "share": temp_dir + "/assets/dir", + "chroot": temp_dir + "/assets", + "traverse_mounts": False, + }, + 'routes/0/action', + ), 'configure chroot mount disable' + + assert self.get(url='/mount/')['status'] == 403, 'chroot mount' diff --git a/test/test_static_symlink.py b/test/test_static_symlink.py new file mode 100644 index 00000000..35eb402a --- /dev/null +++ b/test/test_static_symlink.py @@ -0,0 +1,94 @@ +import os +from pathlib import Path + +import pytest + +from unit.applications.proto import TestApplicationProto + + +class TestStaticSymlink(TestApplicationProto): + prerequisites = {'features': ['chroot']} + + @pytest.fixture(autouse=True) + def setup_method_fixture(self, temp_dir): + os.makedirs(temp_dir + '/assets/dir/dir') + Path(temp_dir + '/assets/index.html').write_text('0123456789') + Path(temp_dir + '/assets/dir/file').write_text('blah') + + self._load_conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [{"action": {"share": temp_dir + "/assets"}}], + } + ) + + def test_static_symlink(self, temp_dir, skip_alert): + skip_alert(r'opening.*failed') + + os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link') + + assert self.get(url='/dir')['status'] == 301, 'dir' + assert self.get(url='/dir/file')['status'] == 200, 'file' + assert self.get(url='/link')['status'] == 301, 'symlink dir' + assert self.get(url='/link/file')['status'] == 200, 'symlink file' + + assert 'success' in self.conf( + {"share": temp_dir + "/assets", "follow_symlinks": False}, + 'routes/0/action', + ), 'configure symlink disable' + + assert self.get(url='/link/file')['status'] == 403, 'symlink disabled' + + assert 'success' in self.conf( + {"share": temp_dir + "/assets", "follow_symlinks": True}, + 'routes/0/action', + ), 'configure symlink enable' + + assert self.get(url='/link/file')['status'] == 200, 'symlink enabled' + + def test_static_symlink_two_blocks(self, temp_dir, skip_alert): + skip_alert(r'opening.*failed') + + os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link') + + assert 'success' in self.conf( + [ + { + "match": {"method": "HEAD"}, + "action": { + "share": temp_dir + "/assets", + "follow_symlinks": False, + }, + }, + { + "match": {"method": "GET"}, + "action": { + "share": temp_dir + "/assets", + "follow_symlinks": True, + }, + }, + ], + 'routes', + ), 'configure two options' + + assert self.get(url='/link/file')['status'] == 200, 'block enabled' + assert self.head(url='/link/file')['status'] == 403, 'block disabled' + + def test_static_symlink_chroot(self, temp_dir, skip_alert): + skip_alert(r'opening.*failed') + + os.symlink( + temp_dir + '/assets/dir/file', temp_dir + '/assets/dir/dir/link' + ) + + assert self.get(url='/dir/dir/link')['status'] == 200, 'default chroot' + + assert 'success' in self.conf( + { + "share": temp_dir + "/assets", + "chroot": temp_dir + "/assets/dir/dir", + }, + 'routes/0/action', + ), 'configure chroot' + + assert self.get(url='/dir/dir/link')['status'] == 404, 'chroot' diff --git a/test/test_static_types.py b/test/test_static_types.py new file mode 100644 index 00000000..20defddf --- /dev/null +++ b/test/test_static_types.py @@ -0,0 +1,169 @@ +from pathlib import Path + +import pytest + +from unit.applications.proto import TestApplicationProto + + +class TestStaticTypes(TestApplicationProto): + prerequisites = {} + + @pytest.fixture(autouse=True) + def setup_method_fixture(self, temp_dir): + Path(temp_dir + '/assets').mkdir() + for ext in ['.xml', '.mp4', '.php', '', '.txt', '.html', '.png']: + Path(temp_dir + '/assets/file' + ext).write_text(ext) + + Path(temp_dir + '/assets/index.html').write_text('index') + + self._load_conf( + { + "listeners": { + "*:7080": {"pass": "routes"}, + "*:7081": {"pass": "routes"}, + }, + "routes": [{"action": {"share": temp_dir + "/assets"}}], + "applications": {}, + } + ) + + def action_update(self, conf): + assert 'success' in self.conf(conf, 'routes/0/action') + + def check_body(self, http_url, body): + resp = self.get(url=http_url) + assert resp['status'] == 200, 'status' + assert resp['body'] == body, 'body' + + def test_static_types_basic(self, temp_dir): + self.action_update({"share": temp_dir + "/assets"}) + self.check_body('/index.html', 'index') + self.check_body('/file.xml', '.xml') + + self.action_update( + {"share": temp_dir + "/assets", "types": "application/xml"} + ) + self.check_body('/file.xml', '.xml') + + self.action_update( + {"share": temp_dir + "/assets", "types": ["application/xml"]} + ) + self.check_body('/file.xml', '.xml') + + self.action_update({"share": temp_dir + "/assets", "types": [""]}) + assert self.get(url='/file.xml')['status'] == 403, 'no mtype' + + def test_static_types_wildcard(self, temp_dir): + self.action_update( + {"share": temp_dir + "/assets", "types": ["application/*"]} + ) + self.check_body('/file.xml', '.xml') + assert self.get(url='/file.mp4')['status'] == 403, 'app * mtype mp4' + + self.action_update( + {"share": temp_dir + "/assets", "types": ["video/*"]} + ) + assert self.get(url='/file.xml')['status'] == 403, 'video * mtype xml' + self.check_body('/file.mp4', '.mp4') + + def test_static_types_negation(self, temp_dir): + self.action_update( + {"share": temp_dir + "/assets", "types": ["!application/xml"]} + ) + assert self.get(url='/file.xml')['status'] == 403, 'forbidden negation' + self.check_body('/file.mp4', '.mp4') + + # sorting negation + self.action_update( + { + "share": temp_dir + "/assets", + "types": ["!video/*", "image/png", "!image/jpg"], + } + ) + assert self.get(url='/file.mp4')['status'] == 403, 'negation sort mp4' + self.check_body('/file.png', '.png') + assert self.get(url='/file.jpg')['status'] == 403, 'negation sort jpg' + + def test_static_types_regex(self, temp_dir): + self.action_update( + {"share": temp_dir + "/assets", "types": ["~text/(html|plain)"]} + ) + assert self.get(url='/file.php')['status'] == 403, 'regex fail' + self.check_body('/file.html', '.html') + self.check_body('/file.txt', '.txt') + + def test_static_types_case(self, temp_dir): + self.action_update( + {"share": temp_dir + "/assets", "types": ["!APpliCaTiOn/xMl"]} + ) + self.check_body('/file.mp4', '.mp4') + assert ( + self.get(url='/file.xml')['status'] == 403 + ), 'mixed case xml negation' + + self.action_update( + {"share": temp_dir + "/assets", "types": ["vIdEo/mp4"]} + ) + assert self.get(url='/file.mp4')['status'] == 200, 'mixed case' + assert ( + self.get(url='/file.xml')['status'] == 403 + ), 'mixed case video negation' + + self.action_update( + {"share": temp_dir + "/assets", "types": ["vIdEo/*"]} + ) + self.check_body('/file.mp4', '.mp4') + assert ( + self.get(url='/file.xml')['status'] == 403 + ), 'mixed case video * negation' + + def test_static_types_fallback(self, temp_dir): + assert 'success' in self.conf( + [ + { + "match": {"destination": "*:7081"}, + "action": {"return": 200}, + }, + { + "action": { + "share": temp_dir + "/assets", + "types": ["!application/x-httpd-php"], + "fallback": {"proxy": "http://127.0.0.1:7081"}, + } + }, + ], + 'routes', + ), 'configure fallback proxy route' + + self.check_body('/file.php', '') + self.check_body('/file.mp4', '.mp4') + + def test_static_types_index(self, temp_dir): + self.action_update( + {"share": temp_dir + "/assets", "types": "application/xml"} + ) + self.check_body('/', 'index') + self.check_body('/file.xml', '.xml') + assert self.get(url='/file.mp4')['status'] == 403, 'forbidden mtype' + + def test_static_types_custom_mime(self, temp_dir): + self._load_conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [{"action": {"share": temp_dir + "/assets"}}], + "applications": {}, + "settings": { + "http": { + "static": {"mime_types": {"test/mime-type": ["file"]}} + } + }, + } + ) + + self.action_update({"share": temp_dir + "/assets", "types": [""]}) + assert self.get(url='/file')['status'] == 403, 'forbidden custom mime' + + self.action_update( + {"share": temp_dir + "/assets", "types": ["test/mime-type"]} + ) + self.check_body('/file', '') -- cgit