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
|
import multiprocessing
import os
import signal
import pytest
from buildstream import utils, _signals
from buildstream._cas import CASCache
from buildstream.storage._casbaseddirectory import CasBasedDirectory
from buildstream.storage._filebaseddirectory import FileBasedDirectory
DATA_DIR = os.path.join(
os.path.dirname(os.path.realpath(__file__)),
"storage"
)
# Since parent processes wait for queue events, we need
# to put something on it if the called process raises an
# exception.
def _queue_wrapper(target, queue, *args):
try:
target(*args)
except Exception as e:
queue.put(str(e))
raise
queue.put(None)
def setup_backend(backend_class, tmpdir):
if backend_class == FileBasedDirectory:
return backend_class(os.path.join(tmpdir, "vdir"))
else:
cas_cache = CASCache(tmpdir)
return backend_class(cas_cache)
def _test_import_subprocess(tmpdir, datafiles, backend):
original = os.path.join(str(datafiles), "original")
c = setup_backend(backend, str(tmpdir))
c.import_files(original)
assert "bin/bash" in c.list_relative_paths()
assert "bin/hello" in c.list_relative_paths()
@pytest.mark.parametrize("backend", [
FileBasedDirectory, CasBasedDirectory])
@pytest.mark.datafiles(DATA_DIR)
def test_import(tmpdir, datafiles, backend):
queue = multiprocessing.Queue()
process = multiprocessing.Process(target=_queue_wrapper,
args=(_test_import_subprocess, queue,
tmpdir, datafiles, backend))
try:
with _signals.blocked([signal.SIGINT], ignore=False):
process.start()
error = queue.get()
process.join()
except KeyboardInterrupt:
utils._kill_process_tree(process.pid)
raise
assert not error
def _test_modified_file_list_subprocess(tmpdir, datafiles, backend):
original = os.path.join(str(datafiles), "original")
overlay = os.path.join(str(datafiles), "overlay")
c = setup_backend(backend, str(tmpdir))
c.import_files(original)
c.mark_unmodified()
c.import_files(overlay)
print("List of all paths in imported results: {}".format(c.list_relative_paths()))
assert "bin/bash" in c.list_relative_paths()
assert "bin/bash" in c.list_modified_paths()
assert "bin/hello" not in c.list_modified_paths()
@pytest.mark.parametrize("backend", [
FileBasedDirectory, CasBasedDirectory])
@pytest.mark.datafiles(DATA_DIR)
def test_modified_file_list(tmpdir, datafiles, backend):
queue = multiprocessing.Queue()
process = multiprocessing.Process(target=_queue_wrapper,
args=(_test_modified_file_list_subprocess, queue,
tmpdir, datafiles, backend))
try:
with _signals.blocked([signal.SIGINT], ignore=False):
process.start()
error = queue.get()
process.join()
except KeyboardInterrupt:
utils._kill_process_tree(process.pid)
raise
assert not error
|
