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