xref: /unit/test/unit/utils.py (revision 2482:88df458ead6d)
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 waitformount(template, timeout=50):
94    for _ in range(timeout):
95        if findmnt().find(template) != -1:
96            return True
97
98        time.sleep(0.1)
99
100    return False
101
102
103def waitforunmount(template, timeout=50):
104    for _ in range(timeout):
105        if findmnt().find(template) == -1:
106            return True
107
108        time.sleep(0.1)
109
110    return False
111
112
113def getns(nstype):
114    # read namespace id from symlink file:
115    # it points to: '<nstype>:[<ns id>]'
116    # # eg.: 'pid:[4026531836]'
117    nspath = f'/proc/self/ns/{nstype}'
118    data = None
119
120    if os.path.exists(nspath):
121        data = int(os.readlink(nspath)[len(nstype) + 2 : -1])
122
123    return data
124