summaryrefslogtreecommitdiffhomepage
path: root/test/test_static_mount.py
diff options
context:
space:
mode:
authorAndrei Zeliankou <zelenkov@nginx.com>2021-06-28 22:05:40 +0100
committerAndrei Zeliankou <zelenkov@nginx.com>2021-06-28 22:05:40 +0100
commitb86891c4ef848a2da05abd1350af5f0b8e4335fa (patch)
tree38dccefa03558d95014a2fd14a6d8cc85e380473 /test/test_static_mount.py
parent72420358be128640dd09a0914a8d88b50c152545 (diff)
downloadunit-b86891c4ef848a2da05abd1350af5f0b8e4335fa.tar.gz
unit-b86891c4ef848a2da05abd1350af5f0b8e4335fa.tar.bz2
Tests: renamed share to static.
Also minor style changes.
Diffstat (limited to 'test/test_static_mount.py')
-rw-r--r--test/test_static_mount.py140
1 files changed, 140 insertions, 0 deletions
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'