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