diff options
Diffstat (limited to 'tests/internals')
-rw-r--r-- | tests/internals/storage_vdir_import.py | 73 |
1 files changed, 66 insertions, 7 deletions
diff --git a/tests/internals/storage_vdir_import.py b/tests/internals/storage_vdir_import.py index 7c6cbe4fb..92aa7aae7 100644 --- a/tests/internals/storage_vdir_import.py +++ b/tests/internals/storage_vdir_import.py @@ -14,11 +14,14 @@ # You should have received a copy of the GNU Lesser General Public # License along with this library. If not, see <http://www.gnu.org/licenses/>. from hashlib import sha256 +import multiprocessing import os import random +import signal import pytest +from buildstream import utils, _signals from buildstream.storage._casbaseddirectory import CasBasedDirectory from buildstream.storage._filebaseddirectory import FileBasedDirectory from buildstream._cas import CASCache @@ -55,6 +58,34 @@ RANDOM_SEED = 69105 NUM_RANDOM_TESTS = 10 +# 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 _run_test_in_subprocess(func, *args): + queue = multiprocessing.Queue() + process = multiprocessing.Process(target=_queue_wrapper, args=(func, queue, *args)) + 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 generate_import_roots(rootno, directory): rootname = "root{}".format(rootno) rootdir = os.path.join(directory, "content", rootname) @@ -182,7 +213,7 @@ def directory_not_empty(path): return os.listdir(path) -def _import_test(tmpdir, original, overlay, generator_function, verify_contents=False): +def _import_test_subprocess(tmpdir, original, overlay, generator_function, verify_contents=False): cas_cache = CASCache(tmpdir) # Create some fake content generator_function(original, tmpdir) @@ -234,6 +265,10 @@ def _import_test(tmpdir, original, overlay, generator_function, verify_contents= assert duplicate_cas._get_digest().hash == d._get_digest().hash +def _import_test(tmpdir, original, overlay, generator_function, verify_contents=False): + _run_test_in_subprocess(_import_test_subprocess, tmpdir, original, overlay, generator_function, verify_contents) + + # 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. @@ -249,7 +284,7 @@ def test_random_cas_import(tmpdir, original): _import_test(str(tmpdir), original, overlay, generate_random_root, verify_contents=False) -def _listing_test(tmpdir, root, generator_function): +def _listing_test_subprocess(tmpdir, root, generator_function): cas_cache = CASCache(tmpdir) # Create some fake content generator_function(root, tmpdir) @@ -263,6 +298,10 @@ def _listing_test(tmpdir, root, generator_function): assert filelist == filelist2 +def _listing_test(tmpdir, root, generator_function): + _run_test_in_subprocess(_listing_test_subprocess, tmpdir, root, generator_function) + + @pytest.mark.parametrize("root", range(1, 11)) def test_random_directory_listing(tmpdir, root): _listing_test(str(tmpdir), root, generate_random_root) @@ -274,7 +313,7 @@ def test_fixed_directory_listing(tmpdir, root): # Check that the vdir is decending and readable -def test_descend(tmpdir): +def _test_descend_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) d = CasBasedDirectory(cas_cache) @@ -294,10 +333,14 @@ def test_descend(tmpdir): assert Content_to_check == open(cas_cache.objpath(digest)).read() +def test_descend(tmpdir): + _run_test_in_subprocess(_test_descend_subprocess, tmpdir) + + # Check symlink logic for edgecases # Make sure the correct erros are raised when trying # to decend in to files or links to files -def test_bad_symlinks(tmpdir): +def _test_bad_symlinks_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) d = CasBasedDirectory(cas_cache) @@ -325,9 +368,13 @@ def test_bad_symlinks(tmpdir): assert error.reason == exp_reason +def test_bad_symlinks(tmpdir): + _run_test_in_subprocess(_test_bad_symlinks_subprocess, tmpdir) + + # Check symlink logic for edgecases # Check decend accross relitive link -def test_relitive_symlink(tmpdir): +def _test_relative_symlink_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) d = CasBasedDirectory(cas_cache) @@ -347,9 +394,13 @@ def test_relitive_symlink(tmpdir): assert Content_to_check == open(cas_cache.objpath(digest)).read() +def test_relative_symlink(tmpdir): + _run_test_in_subprocess(_test_relative_symlink_subprocess, tmpdir) + + # Check symlink logic for edgecases # Check deccend accross abs link -def test_abs_symlink(tmpdir): +def _test_abs_symlink_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) d = CasBasedDirectory(cas_cache) @@ -370,9 +421,13 @@ def test_abs_symlink(tmpdir): assert Content_to_check == open(cas_cache.objpath(digest)).read() +def test_abs_symlink(tmpdir): + _run_test_in_subprocess(_test_abs_symlink_subprocess, tmpdir) + + # Check symlink logic for edgecases # Check symlink can not escape root -def test_bad_sym_escape(tmpdir): +def _test_bad_sym_escape_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) d = CasBasedDirectory(cas_cache) @@ -391,3 +446,7 @@ def test_bad_sym_escape(tmpdir): with pytest.raises(VirtualDirectoryError) as error: d.descend('a', 'l', follow_symlinks=True) assert error.reason == "directory-not-found" + + +def test_bad_sym_escape(tmpdir): + _run_test_in_subprocess(_test_bad_sym_escape_subprocess, tmpdir) |