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