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