diff options
Diffstat (limited to '')
| -rw-r--r-- | test/test_static_mount.py | 217 |
1 files changed, 109 insertions, 108 deletions
diff --git a/test/test_static_mount.py b/test/test_static_mount.py index 406922b1..ccd18919 100644 --- a/test/test_static_mount.py +++ b/test/test_static_mount.py @@ -3,133 +3,134 @@ 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(f'{temp_dir}/assets/dir/mount') - os.makedirs(f'{temp_dir}/assets/dir/dir') - os.makedirs(f'{temp_dir}/assets/mount') - Path(f'{temp_dir}/assets/index.html').write_text('index') - Path(f'{temp_dir}/assets/dir/dir/file').write_text('file') - Path(f'{temp_dir}/assets/mount/index.html').write_text('mount') - - try: - subprocess.check_output( - [ - "mount", - "--bind", - f'{temp_dir}/assets/mount', - f'{temp_dir}/assets/dir/mount', - ], - stderr=subprocess.STDOUT, - ) - - except KeyboardInterrupt: - raise - - except subprocess.CalledProcessError: - pytest.fail("Can't run mount process.") - - self._load_conf( - { - "listeners": {"*:7080": {"pass": "routes"}}, - "routes": [{"action": {"share": f'{temp_dir}/assets/dir$uri'}}], - } +from unit.applications.proto import ApplicationProto + +prerequisites = {'features': {'chroot': True}, 'privileged_user': True} + +client = ApplicationProto() + + +@pytest.fixture(autouse=True) +def setup_method_fixture(temp_dir): + os.makedirs(f'{temp_dir}/assets/dir/mount') + os.makedirs(f'{temp_dir}/assets/dir/dir') + os.makedirs(f'{temp_dir}/assets/mount') + Path(f'{temp_dir}/assets/index.html').write_text('index') + Path(f'{temp_dir}/assets/dir/dir/file').write_text('file') + Path(f'{temp_dir}/assets/mount/index.html').write_text('mount') + + try: + subprocess.check_output( + [ + "mount", + "--bind", + f'{temp_dir}/assets/mount', + f'{temp_dir}/assets/dir/mount', + ], + stderr=subprocess.STDOUT, ) - yield + except KeyboardInterrupt: + raise - try: - subprocess.check_output( - ["umount", "--lazy", f'{temp_dir}/assets/dir/mount'], - stderr=subprocess.STDOUT, - ) + except subprocess.CalledProcessError: + pytest.fail("Can't run mount process.") - except KeyboardInterrupt: - raise + assert 'success' in client.conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [{"action": {"share": f'{temp_dir}/assets/dir$uri'}}], + } + ) - except subprocess.CalledProcessError: - pytest.fail("Can't run umount process.") + yield - def test_static_mount(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') + try: + subprocess.check_output( + ["umount", "--lazy", f'{temp_dir}/assets/dir/mount'], + stderr=subprocess.STDOUT, + ) - resp = self.get(url='/mount/') - assert resp['status'] == 200 - assert resp['body'] == 'mount' + except KeyboardInterrupt: + raise - assert 'success' in self.conf( - {"share": f'{temp_dir}/assets/dir$uri', "traverse_mounts": False}, - 'routes/0/action', - ), 'configure mount disable' + except subprocess.CalledProcessError: + pytest.fail("Can't run umount process.") - assert self.get(url='/mount/')['status'] == 403 - assert 'success' in self.conf( - {"share": f'{temp_dir}/assets/dir$uri', "traverse_mounts": True}, - 'routes/0/action', - ), 'configure mount enable' +def test_static_mount(temp_dir, skip_alert): + skip_alert(r'opening.*failed') - resp = self.get(url='/mount/') - assert resp['status'] == 200 - assert resp['body'] == 'mount' + resp = client.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') + assert 'success' in client.conf( + {"share": f'{temp_dir}/assets/dir$uri', "traverse_mounts": False}, + 'routes/0/action', + ), 'configure mount disable' - os.symlink(f'{temp_dir}/assets/dir', f'{temp_dir}/assets/link') + assert client.get(url='/mount/')['status'] == 403 - assert 'success' in self.conf( - [ - { - "match": {"method": "HEAD"}, - "action": { - "share": f'{temp_dir}/assets/dir$uri', - "traverse_mounts": False, - }, - }, - { - "match": {"method": "GET"}, - "action": { - "share": f'{temp_dir}/assets/dir$uri', - "traverse_mounts": True, - }, - }, - ], - 'routes', - ), 'configure two options' + assert 'success' in client.conf( + {"share": f'{temp_dir}/assets/dir$uri', "traverse_mounts": True}, + 'routes/0/action', + ), 'configure mount enable' - assert self.get(url='/mount/')['status'] == 200, 'block enabled' - assert self.head(url='/mount/')['status'] == 403, 'block disabled' + resp = client.get(url='/mount/') + assert resp['status'] == 200 + assert resp['body'] == 'mount' - def test_static_mount_chroot(self, temp_dir, skip_alert): - skip_alert(r'opening.*failed') - assert 'success' in self.conf( - { - "share": f'{temp_dir}/assets/dir$uri', - "chroot": f'{temp_dir}/assets', - }, - 'routes/0/action', - ), 'configure chroot mount default' +def test_static_mount_two_blocks(temp_dir, skip_alert): + skip_alert(r'opening.*failed') - assert self.get(url='/mount/')['status'] == 200, 'chroot' + os.symlink(f'{temp_dir}/assets/dir', f'{temp_dir}/assets/link') - assert 'success' in self.conf( + assert 'success' in client.conf( + [ + { + "match": {"method": "HEAD"}, + "action": { + "share": f'{temp_dir}/assets/dir$uri', + "traverse_mounts": False, + }, + }, { - "share": f'{temp_dir}/assets/dir$uri', - "chroot": f'{temp_dir}/assets', - "traverse_mounts": False, + "match": {"method": "GET"}, + "action": { + "share": f'{temp_dir}/assets/dir$uri', + "traverse_mounts": True, + }, }, - 'routes/0/action', - ), 'configure chroot mount disable' + ], + 'routes', + ), 'configure two options' + + assert client.get(url='/mount/')['status'] == 200, 'block enabled' + assert client.head(url='/mount/')['status'] == 403, 'block disabled' + + +def test_static_mount_chroot(temp_dir, skip_alert): + skip_alert(r'opening.*failed') + + assert 'success' in client.conf( + { + "share": f'{temp_dir}/assets/dir$uri', + "chroot": f'{temp_dir}/assets', + }, + 'routes/0/action', + ), 'configure chroot mount default' + + assert client.get(url='/mount/')['status'] == 200, 'chroot' + + assert 'success' in client.conf( + { + "share": f'{temp_dir}/assets/dir$uri', + "chroot": f'{temp_dir}/assets', + "traverse_mounts": False, + }, + 'routes/0/action', + ), 'configure chroot mount disable' - assert self.get(url='/mount/')['status'] == 403, 'chroot mount' + assert client.get(url='/mount/')['status'] == 403, 'chroot mount' |