xref: /unit/test/unit/utils.py (revision 1912:942e252e3f4c)
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