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