xref: /unit/test/test_static_mount.py (revision 1902:17547cf0ed39)
1import os
2import subprocess
3from pathlib import Path
4
5import pytest
6
7from unit.applications.proto import TestApplicationProto
8
9
10class TestStaticMount(TestApplicationProto):
11    prerequisites = {'features': ['chroot']}
12
13    @pytest.fixture(autouse=True)
14    def setup_method_fixture(self, is_su, temp_dir):
15        if not is_su:
16            pytest.skip('requires root')
17
18        os.makedirs(temp_dir + '/assets/dir/mount')
19        os.makedirs(temp_dir + '/assets/dir/dir')
20        os.makedirs(temp_dir + '/assets/mount')
21        Path(temp_dir + '/assets/index.html').write_text('index')
22        Path(temp_dir + '/assets/dir/dir/file').write_text('file')
23        Path(temp_dir + '/assets/mount/index.html').write_text('mount')
24
25        try:
26            process = subprocess.Popen(
27                [
28                    "mount",
29                    "--bind",
30                    temp_dir + "/assets/mount",
31                    temp_dir + "/assets/dir/mount",
32                ],
33                stderr=subprocess.STDOUT,
34            )
35
36            process.communicate()
37
38        except KeyboardInterrupt:
39            raise
40
41        except:
42            pytest.fail('Can\'t run mount process.')
43
44        self._load_conf(
45            {
46                "listeners": {"*:7080": {"pass": "routes"}},
47                "routes": [{"action": {"share": temp_dir + "/assets/dir"}}],
48            }
49        )
50
51        yield
52
53        try:
54            process = subprocess.Popen(
55                ["umount", "--lazy", temp_dir + "/assets/dir/mount"],
56                stderr=subprocess.STDOUT,
57            )
58
59            process.communicate()
60
61        except KeyboardInterrupt:
62            raise
63
64        except:
65            pytest.fail('Can\'t run umount process.')
66
67    def test_static_mount(self, temp_dir, skip_alert):
68        skip_alert(r'opening.*failed')
69
70        resp = self.get(url='/mount/')
71        assert resp['status'] == 200
72        assert resp['body'] == 'mount'
73
74        assert 'success' in self.conf(
75            {"share": temp_dir + "/assets/dir", "traverse_mounts": False},
76            'routes/0/action',
77        ), 'configure mount disable'
78
79        assert self.get(url='/mount/')['status'] == 403
80
81        assert 'success' in self.conf(
82            {"share": temp_dir + "/assets/dir", "traverse_mounts": True},
83            'routes/0/action',
84        ), 'configure mount enable'
85
86        resp = self.get(url='/mount/')
87        assert resp['status'] == 200
88        assert resp['body'] == 'mount'
89
90    def test_static_mount_two_blocks(self, temp_dir, skip_alert):
91        skip_alert(r'opening.*failed')
92
93        os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link')
94
95        assert 'success' in self.conf(
96            [
97                {
98                    "match": {"method": "HEAD"},
99                    "action": {
100                        "share": temp_dir + "/assets/dir",
101                        "traverse_mounts": False,
102                    },
103                },
104                {
105                    "match": {"method": "GET"},
106                    "action": {
107                        "share": temp_dir + "/assets/dir",
108                        "traverse_mounts": True,
109                    },
110                },
111            ],
112            'routes',
113        ), 'configure two options'
114
115        assert self.get(url='/mount/')['status'] == 200, 'block enabled'
116        assert self.head(url='/mount/')['status'] == 403, 'block disabled'
117
118    def test_static_mount_chroot(self, temp_dir, skip_alert):
119        skip_alert(r'opening.*failed')
120
121        assert 'success' in self.conf(
122            {
123                "share": temp_dir + "/assets/dir",
124                "chroot": temp_dir + "/assets",
125            },
126            'routes/0/action',
127        ), 'configure chroot mount default'
128
129        assert self.get(url='/mount/')['status'] == 200, 'chroot'
130
131        assert 'success' in self.conf(
132            {
133                "share": temp_dir + "/assets/dir",
134                "chroot": temp_dir + "/assets",
135                "traverse_mounts": False,
136            },
137            'routes/0/action',
138        ), 'configure chroot mount disable'
139
140        assert self.get(url='/mount/')['status'] == 403, 'chroot mount'
141