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