xref: /unit/test/test_static_mount.py (revision 2004:ca4c5855f139)
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            subprocess.check_output(
26                [
27                    "mount",
28                    "--bind",
29                    temp_dir + "/assets/mount",
30                    temp_dir + "/assets/dir/mount",
31                ],
32                stderr=subprocess.STDOUT,
33            )
34
35        except KeyboardInterrupt:
36            raise
37
38        except subprocess.CalledProcessError:
39            pytest.fail('Can\'t run mount process.')
40
41        self._load_conf(
42            {
43                "listeners": {"*:7080": {"pass": "routes"}},
44                "routes": [
45                    {"action": {"share": temp_dir + "/assets/dir$uri"}}
46                ],
47            }
48        )
49
50        yield
51
52        try:
53            subprocess.check_output(
54                ["umount", "--lazy", temp_dir + "/assets/dir/mount"],
55                stderr=subprocess.STDOUT,
56            )
57
58        except KeyboardInterrupt:
59            raise
60
61        except subprocess.CalledProcessError:
62            pytest.fail('Can\'t run umount process.')
63
64    def test_static_mount(self, temp_dir, skip_alert):
65        skip_alert(r'opening.*failed')
66
67        resp = self.get(url='/mount/')
68        assert resp['status'] == 200
69        assert resp['body'] == 'mount'
70
71        assert 'success' in self.conf(
72            {"share": temp_dir + "/assets/dir$uri", "traverse_mounts": False},
73            'routes/0/action',
74        ), 'configure mount disable'
75
76        assert self.get(url='/mount/')['status'] == 403
77
78        assert 'success' in self.conf(
79            {"share": temp_dir + "/assets/dir$uri", "traverse_mounts": True},
80            'routes/0/action',
81        ), 'configure mount enable'
82
83        resp = self.get(url='/mount/')
84        assert resp['status'] == 200
85        assert resp['body'] == 'mount'
86
87    def test_static_mount_two_blocks(self, temp_dir, skip_alert):
88        skip_alert(r'opening.*failed')
89
90        os.symlink(temp_dir + '/assets/dir', temp_dir + '/assets/link')
91
92        assert 'success' in self.conf(
93            [
94                {
95                    "match": {"method": "HEAD"},
96                    "action": {
97                        "share": temp_dir + "/assets/dir$uri",
98                        "traverse_mounts": False,
99                    },
100                },
101                {
102                    "match": {"method": "GET"},
103                    "action": {
104                        "share": temp_dir + "/assets/dir$uri",
105                        "traverse_mounts": True,
106                    },
107                },
108            ],
109            'routes',
110        ), 'configure two options'
111
112        assert self.get(url='/mount/')['status'] == 200, 'block enabled'
113        assert self.head(url='/mount/')['status'] == 403, 'block disabled'
114
115    def test_static_mount_chroot(self, temp_dir, skip_alert):
116        skip_alert(r'opening.*failed')
117
118        assert 'success' in self.conf(
119            {
120                "share": temp_dir + "/assets/dir$uri",
121                "chroot": temp_dir + "/assets",
122            },
123            'routes/0/action',
124        ), 'configure chroot mount default'
125
126        assert self.get(url='/mount/')['status'] == 200, 'chroot'
127
128        assert 'success' in self.conf(
129            {
130                "share": temp_dir + "/assets/dir$uri",
131                "chroot": temp_dir + "/assets",
132                "traverse_mounts": False,
133            },
134            'routes/0/action',
135        ), 'configure chroot mount disable'
136
137        assert self.get(url='/mount/')['status'] == 403, 'chroot mount'
138