xref: /unit/test/unit/utils.py (revision 1877:e09738cfc6be)
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 sysctl():
65    try:
66        out = subprocess.check_output(
67            ['sysctl', '-a'], stderr=subprocess.STDOUT
68        ).decode()
69    except FileNotFoundError:
70        pytest.skip('requires sysctl')
71
72    return out
73
74
75def waitformount(template, wait=50):
76    for i in range(wait):
77        if findmnt().find(template) != -1:
78            return True
79
80        time.sleep(0.1)
81
82    return False
83
84
85def waitforunmount(template, wait=50):
86    for i in range(wait):
87        if findmnt().find(template) == -1:
88            return True
89
90        time.sleep(0.1)
91
92    return False
93
94
95def getns(nstype):
96    # read namespace id from symlink file:
97    # it points to: '<nstype>:[<ns id>]'
98    # # eg.: 'pid:[4026531836]'
99    nspath = '/proc/self/ns/' + nstype
100    data = None
101
102    if os.path.exists(nspath):
103        data = int(os.readlink(nspath)[len(nstype) + 2 : -1])
104
105    return data
106