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