1import glob 2import os 3import socket 4import subprocess 5import time 6 7import pytest 8 9 10def public_dir(path): 11 os.chmod(path, 0o777) 12 13 for root, dirs, files in os.walk(path): 14 for d in dirs: 15 try: 16 os.chmod(os.path.join(root, d), 0o777) 17 except FileNotFoundError: 18 pass 19 for f in files: 20 try: 21 os.chmod(os.path.join(root, f), 0o777) 22 except FileNotFoundError: 23 pass 24 25 26def waitforfiles(*files, timeout=50): 27 for _ in range(timeout): 28 wait = False 29 30 for f in files: 31 if not os.path.exists(f): 32 wait = True 33 break 34 35 if not wait: 36 return True 37 38 time.sleep(0.1) 39 40 return False 41 42 43def waitforglob(pattern, count=1, timeout=50): 44 for _ in range(timeout): 45 n = 0 46 47 for _ in glob.glob(pattern): 48 n += 1 49 50 if n == count: 51 return True 52 53 time.sleep(0.1) 54 55 return False 56 57 58def waitforsocket(port): 59 for _ in range(50): 60 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 61 try: 62 sock.settimeout(5) 63 sock.connect(('127.0.0.1', port)) 64 return 65 66 except ConnectionRefusedError: 67 time.sleep(0.1) 68 69 except KeyboardInterrupt: 70 raise 71 72 pytest.fail(f"Can't connect to the 127.0.0.1:{port}") 73 74 75def check_findmnt(): 76 try: 77 return subprocess.check_output( 78 ['findmnt', '--raw'], stderr=subprocess.STDOUT 79 ).decode() 80 except FileNotFoundError: 81 return False 82 83 84def findmnt(): 85 out = check_findmnt() 86 87 if not out: 88 pytest.skip('requires findmnt') 89 90 return out 91 92 93def sysctl(): 94 try: 95 out = subprocess.check_output( 96 ['sysctl', '-a'], stderr=subprocess.STDOUT 97 ).decode() 98 except FileNotFoundError: 99 pytest.skip('requires sysctl') 100 101 return out 102 103 104def waitformount(template, timeout=50): 105 for _ in range(timeout): 106 if findmnt().find(template) != -1: 107 return True 108 109 time.sleep(0.1) 110 111 return False 112 113 114def waitforunmount(template, timeout=50): 115 for _ in range(timeout): 116 if findmnt().find(template) == -1: 117 return True 118 119 time.sleep(0.1) 120 121 return False 122 123 124def getns(nstype): 125 # read namespace id from symlink file: 126 # it points to: '<nstype>:[<ns id>]' 127 # # eg.: 'pid:[4026531836]' 128 nspath = f'/proc/self/ns/{nstype}' 129 data = None 130 131 if os.path.exists(nspath): 132 data = int(os.readlink(nspath)[len(nstype) + 2 : -1]) 133 134 return data 135