• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1import sys
2import os
3import socket
4import stat
5
6# Ensure that this is being run on a specific platform
7assert sys.platform.startswith('linux') or sys.platform.startswith('darwin') \
8    or sys.platform.startswith('cygwin') or sys.platform.startswith('freebsd') \
9    or sys.platform.startswith('netbsd')
10
11def env_path():
12    ep = os.environ.get('LIBCXX_FILESYSTEM_DYNAMIC_TEST_ROOT')
13    assert ep is not None
14    ep = os.path.realpath(ep)
15    assert os.path.isdir(ep)
16    return ep
17
18env_path_global = env_path()
19
20# Make sure we don't try and write outside of env_path.
21# All paths used should be sanitized
22def sanitize(p):
23    p = os.path.realpath(p)
24    if os.path.commonprefix([env_path_global, p]):
25        return p
26    assert False
27
28"""
29Some of the tests restrict permissions to induce failures.
30Before we delete the test environment, we have to walk it and re-raise the
31permissions.
32"""
33def clean_recursive(root_p):
34    if not os.path.islink(root_p):
35        os.chmod(root_p, 0o777)
36    for ent in os.listdir(root_p):
37        p = os.path.join(root_p, ent)
38        if os.path.islink(p) or not os.path.isdir(p):
39            os.remove(p)
40        else:
41            assert os.path.isdir(p)
42            clean_recursive(p)
43            os.rmdir(p)
44
45
46def init_test_directory(root_p):
47    root_p = sanitize(root_p)
48    assert not os.path.exists(root_p)
49    os.makedirs(root_p)
50
51
52def destroy_test_directory(root_p):
53    root_p = sanitize(root_p)
54    clean_recursive(root_p)
55    os.rmdir(root_p)
56
57
58def create_file(fname, size):
59    with open(sanitize(fname), 'w') as f:
60        f.write('c' * size)
61
62
63def create_dir(dname):
64    os.mkdir(sanitize(dname))
65
66
67def create_symlink(source, link):
68    os.symlink(sanitize(source), sanitize(link))
69
70
71def create_hardlink(source, link):
72    os.link(sanitize(source), sanitize(link))
73
74
75def create_fifo(source):
76    os.mkfifo(sanitize(source))
77
78
79def create_socket(source):
80    sock = socket.socket(socket.AF_UNIX)
81    sanitized_source = sanitize(source)
82    # AF_UNIX sockets may have very limited path length, so split it
83    # into chdir call (with technically unlimited length) followed
84    # by bind() relative to the directory
85    os.chdir(os.path.dirname(sanitized_source))
86    sock.bind(os.path.basename(sanitized_source))
87
88
89if __name__ == '__main__':
90    command = " ".join(sys.argv[1:])
91    eval(command)
92    sys.exit(0)
93