1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
|
from hashlib import sha256
import os
import pytest
import random
import tempfile
from tests.testutils import cli
from buildstream.storage._casbaseddirectory import CasBasedDirectory
from buildstream.storage._filebaseddirectory import FileBasedDirectory
from buildstream._artifactcache import ArtifactCache
from buildstream._artifactcache.cascache import CASCache
from buildstream import utils
# These are comparitive tests that check that FileBasedDirectory and
# CasBasedDirectory act identically.
class FakeArtifactCache():
def __init__(self):
self.cas = None
class FakeContext():
def __init__(self):
self.artifactdir = ''
self.artifactcache = FakeArtifactCache()
# This is a set of example file system contents. It's a set of trees
# which are either expected to be problematic or were found to be
# problematic during random testing.
# The test attempts to import each on top of each other to test
# importing works consistently. Each tuple is defined as (<filename>,
# <type>, <content>). Type can be 'F' (file), 'S' (symlink) or 'D'
# (directory) with content being the contents for a file or the
# destination for a symlink.
root_filesets = [
[('a/b/c/textfile1', 'F', 'This is textfile 1\n')],
[('a/b/c/textfile1', 'F', 'This is the replacement textfile 1\n')],
[('a/b/d', 'D', '')],
[('a/b/c', 'S', '/a/b/d')],
[('a/b/d', 'S', '/a/b/c')],
[('a/b/d', 'D', ''), ('a/b/c', 'S', '/a/b/d')],
[('a/b/c', 'D', ''), ('a/b/d', 'S', '/a/b/c')],
[('a/b', 'F', 'This is textfile 1\n')],
[('a/b/c', 'F', 'This is textfile 1\n')],
[('a/b/c', 'D', '')]
]
empty_hash_ref = sha256().hexdigest()
RANDOM_SEED = 69105
NUM_RANDOM_TESTS = 10
def generate_import_roots(rootno, directory):
rootname = "root{}".format(rootno)
rootdir = os.path.join(directory, "content", rootname)
if os.path.exists(rootdir):
return
for (path, typesymbol, content) in root_filesets[rootno - 1]:
if typesymbol == 'F':
(dirnames, filename) = os.path.split(path)
os.makedirs(os.path.join(rootdir, dirnames), exist_ok=True)
with open(os.path.join(rootdir, dirnames, filename), "wt") as f:
f.write(content)
elif typesymbol == 'D':
os.makedirs(os.path.join(rootdir, path), exist_ok=True)
elif typesymbol == 'S':
(dirnames, filename) = os.path.split(path)
os.makedirs(os.path.join(rootdir, dirnames), exist_ok=True)
os.symlink(content, os.path.join(rootdir, path))
def generate_random_root(rootno, directory):
# By seeding the random number generator, we ensure these tests
# will be repeatable, at least until Python changes the random
# number algorithm.
random.seed(RANDOM_SEED + rootno)
rootname = "root{}".format(rootno)
rootdir = os.path.join(directory, "content", rootname)
if os.path.exists(rootdir):
return
things = []
locations = ['.']
os.makedirs(rootdir)
for i in range(0, 100):
location = random.choice(locations)
thingname = "node{}".format(i)
thing = random.choice(['dir', 'link', 'file'])
target = os.path.join(rootdir, location, thingname)
if thing == 'dir':
os.makedirs(target)
locations.append(os.path.join(location, thingname))
elif thing == 'file':
with open(target, "wt") as f:
f.write("This is node {}\n".format(i))
elif thing == 'link':
symlink_type = random.choice(['absolute', 'relative', 'broken'])
if symlink_type == 'broken' or not things:
os.symlink("/broken", target)
elif symlink_type == 'absolute':
symlink_destination = random.choice(things)
os.symlink(symlink_destination, target)
else:
symlink_destination = random.choice(things)
relative_link = os.path.relpath(symlink_destination, start=location)
os.symlink(relative_link, target)
things.append(os.path.join(location, thingname))
def file_contents(path):
with open(path, "r") as f:
result = f.read()
return result
def file_contents_are(path, contents):
return file_contents(path) == contents
def create_new_casdir(root_number, fake_context, tmpdir):
d = CasBasedDirectory(fake_context)
d.import_files(os.path.join(tmpdir, "content", "root{}".format(root_number)))
assert d.ref.hash != empty_hash_ref
return d
def create_new_filedir(root_number, tmpdir):
root = os.path.join(tmpdir, "vdir")
os.makedirs(root)
d = FileBasedDirectory(root)
d.import_files(os.path.join(tmpdir, "content", "root{}".format(root_number)))
return d
def combinations(integer_range):
for x in integer_range:
for y in integer_range:
yield (x, y)
def resolve_symlinks(path, root):
""" A function to resolve symlinks inside 'path' components apart from the last one.
For example, resolve_symlinks('/a/b/c/d', '/a/b')
will return '/a/b/f/d' if /a/b/c is a symlink to /a/b/f. The final component of
'path' is not resolved, because we typically want to inspect the symlink found
at that path, not its target.
"""
components = path.split(os.path.sep)
location = root
for i in range(0, len(components) - 1):
location = os.path.join(location, components[i])
if os.path.islink(location):
# Resolve the link, add on all the remaining components
target = os.path.join(os.readlink(location))
tail = os.path.sep.join(components[i + 1:])
if target.startswith(os.path.sep):
# Absolute link - relative to root
location = os.path.join(root, target, tail)
else:
# Relative link - relative to symlink location
location = os.path.join(location, target)
return resolve_symlinks(location, root)
# If we got here, no symlinks were found. Add on the final component and return.
location = os.path.join(location, components[-1])
return location
def directory_not_empty(path):
return os.listdir(path)
def _import_test(tmpdir, original, overlay, generator_function, verify_contents=False):
fake_context = FakeContext()
fake_context.artifactcache.cas = CASCache(tmpdir)
# Create some fake content
generator_function(original, tmpdir)
if original != overlay:
generator_function(overlay, tmpdir)
d = create_new_casdir(original, fake_context, tmpdir)
duplicate_cas = create_new_casdir(original, fake_context, tmpdir)
assert duplicate_cas.ref.hash == d.ref.hash
d2 = create_new_casdir(overlay, fake_context, tmpdir)
d.import_files(d2)
export_dir = os.path.join(tmpdir, "output-{}-{}".format(original, overlay))
roundtrip_dir = os.path.join(tmpdir, "roundtrip-{}-{}".format(original, overlay))
d2.export_files(roundtrip_dir)
d.export_files(export_dir)
if verify_contents:
for item in root_filesets[overlay - 1]:
(path, typename, content) = item
realpath = resolve_symlinks(path, export_dir)
if typename == 'F':
if os.path.isdir(realpath) and directory_not_empty(realpath):
# The file should not have overwritten the directory in this case.
pass
else:
assert os.path.isfile(realpath), "{} did not exist in the combined virtual directory".format(path)
assert file_contents_are(realpath, content)
elif typename == 'S':
if os.path.isdir(realpath) and directory_not_empty(realpath):
# The symlink should not have overwritten the directory in this case.
pass
else:
assert os.path.islink(realpath)
assert os.readlink(realpath) == content
elif typename == 'D':
# We can't do any more tests than this because it
# depends on things present in the original. Blank
# directories here will be ignored and the original
# left in place.
assert os.path.lexists(realpath)
# Now do the same thing with filebaseddirectories and check the contents match
files = list(utils.list_relative_paths(roundtrip_dir))
duplicate_cas._import_files_from_directory(roundtrip_dir, files=files)
duplicate_cas._recalculate_recursing_down()
if duplicate_cas.parent:
duplicate_cas.parent._recalculate_recursing_up(duplicate_cas)
assert duplicate_cas.ref.hash == d.ref.hash
# It's possible to parameterize on both original and overlay values,
# but this leads to more tests being listed in the output than are
# comfortable.
@pytest.mark.parametrize("original", range(1, len(root_filesets) + 1))
def test_fixed_cas_import(cli, tmpdir, original):
for overlay in range(1, len(root_filesets) + 1):
_import_test(str(tmpdir), original, overlay, generate_import_roots, verify_contents=True)
@pytest.mark.parametrize("original", range(1, NUM_RANDOM_TESTS + 1))
def test_random_cas_import(cli, tmpdir, original):
for overlay in range(1, NUM_RANDOM_TESTS + 1):
_import_test(str(tmpdir), original, overlay, generate_random_root, verify_contents=False)
def _listing_test(tmpdir, root, generator_function):
fake_context = FakeContext()
fake_context.artifactcache.cas = CASCache(tmpdir)
# Create some fake content
generator_function(root, tmpdir)
d = create_new_filedir(root, tmpdir)
filelist = list(d.list_relative_paths())
d2 = create_new_casdir(root, fake_context, tmpdir)
filelist2 = list(d2.list_relative_paths())
assert filelist == filelist2
@pytest.mark.parametrize("root", range(1, 11))
def test_random_directory_listing(cli, tmpdir, root):
_listing_test(str(tmpdir), root, generate_random_root)
@pytest.mark.parametrize("root", [1, 2, 3, 4, 5])
def test_fixed_directory_listing(cli, tmpdir, root):
_listing_test(str(tmpdir), root, generate_import_roots)
|
