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