xref: /unit/test/unit/utils.py (revision 1803:57b43c5bf726)
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:' + str(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