xref: /unit/test/unit/utils.py (revision 2477:4633c7db7fb5)
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