1import os 2import socket 3import subprocess 4import time 5 6import pytest 7 8 9def public_dir(path): 10 os.chmod(path, 0o777) 11 12 for root, dirs, files in os.walk(path): 13 for d in dirs: 14 os.chmod(os.path.join(root, d), 0o777) 15 for f in files: 16 os.chmod(os.path.join(root, f), 0o777) 17 18 19def waitforfiles(*files): 20 for i in range(50): 21 wait = False 22 23 for f in files: 24 if not os.path.exists(f): 25 wait = True 26 break 27 28 if not wait: 29 return True 30 31 time.sleep(0.1) 32 33 return False 34 35 36def waitforsocket(port): 37 for i in range(50): 38 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: 39 try: 40 sock.settimeout(5) 41 sock.connect(('127.0.0.1', port)) 42 return 43 44 except ConnectionRefusedError: 45 time.sleep(0.1) 46 47 except KeyboardInterrupt: 48 raise 49 50 pytest.fail('Can\'t connect to the 127.0.0.1:' + port) 51 52 53def findmnt(): 54 try: 55 out = subprocess.check_output( 56 ['findmnt', '--raw'], stderr=subprocess.STDOUT 57 ).decode() 58 except FileNotFoundError: 59 pytest.skip('requires findmnt') 60 61 return out 62 63 64def waitformount(template, wait=50): 65 for i in range(wait): 66 if findmnt().find(template) != -1: 67 return True 68 69 time.sleep(0.1) 70 71 return False 72 73 74def waitforunmount(template, wait=50): 75 for i in range(wait): 76 if findmnt().find(template) == -1: 77 return True 78 79 time.sleep(0.1) 80 81 return False 82 83 84def getns(nstype): 85 # read namespace id from symlink file: 86 # it points to: '<nstype>:[<ns id>]' 87 # # eg.: 'pid:[4026531836]' 88 nspath = '/proc/self/ns/' + nstype 89 data = None 90 91 if os.path.exists(nspath): 92 data = int(os.readlink(nspath)[len(nstype) + 2 : -1]) 93 94 return data 95