From 0096e9239482ef0b53c4174196365d20ef3c820f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=BCrg=20Billeter?= Date: Fri, 12 Jul 2019 05:43:04 +0200 Subject: tests/internals/storage_vdir_import.py: Terminate casd at end of test --- tests/internals/storage_vdir_import.py | 278 ++++++++++++++++++--------------- 1 file changed, 150 insertions(+), 128 deletions(-) diff --git a/tests/internals/storage_vdir_import.py b/tests/internals/storage_vdir_import.py index 92aa7aae7..e0165fc13 100644 --- a/tests/internals/storage_vdir_import.py +++ b/tests/internals/storage_vdir_import.py @@ -215,54 +215,58 @@ def directory_not_empty(path): def _import_test_subprocess(tmpdir, original, overlay, generator_function, verify_contents=False): cas_cache = CASCache(tmpdir) - # Create some fake content - generator_function(original, tmpdir) - if original != overlay: - generator_function(overlay, tmpdir) - - d = create_new_casdir(original, cas_cache, tmpdir) - - duplicate_cas = create_new_casdir(original, cas_cache, tmpdir) - - assert duplicate_cas._get_digest().hash == d._get_digest().hash - - d2 = create_new_casdir(overlay, cas_cache, 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 - - duplicate_cas.import_files(roundtrip_dir) - - assert duplicate_cas._get_digest().hash == d._get_digest().hash + try: + # Create some fake content + generator_function(original, tmpdir) + if original != overlay: + generator_function(overlay, tmpdir) + + d = create_new_casdir(original, cas_cache, tmpdir) + + duplicate_cas = create_new_casdir(original, cas_cache, tmpdir) + + assert duplicate_cas._get_digest().hash == d._get_digest().hash + + d2 = create_new_casdir(overlay, cas_cache, 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 + + duplicate_cas.import_files(roundtrip_dir) + + assert duplicate_cas._get_digest().hash == d._get_digest().hash + finally: + cas_cache.release_resources() def _import_test(tmpdir, original, overlay, generator_function, verify_contents=False): @@ -286,16 +290,19 @@ def test_random_cas_import(tmpdir, original): def _listing_test_subprocess(tmpdir, root, generator_function): cas_cache = CASCache(tmpdir) - # Create some fake content - generator_function(root, tmpdir) + try: + # Create some fake content + generator_function(root, tmpdir) - d = create_new_filedir(root, tmpdir) - filelist = list(d.list_relative_paths()) + d = create_new_filedir(root, tmpdir) + filelist = list(d.list_relative_paths()) - d2 = create_new_casdir(root, cas_cache, tmpdir) - filelist2 = list(d2.list_relative_paths()) + d2 = create_new_casdir(root, cas_cache, tmpdir) + filelist2 = list(d2.list_relative_paths()) - assert filelist == filelist2 + assert filelist == filelist2 + finally: + cas_cache.release_resources() def _listing_test(tmpdir, root, generator_function): @@ -316,21 +323,24 @@ def test_fixed_directory_listing(tmpdir, root): def _test_descend_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) - d = CasBasedDirectory(cas_cache) + try: + d = CasBasedDirectory(cas_cache) - Content_to_check = 'You got me' - test_dir = os.path.join(str(tmpdir), 'importfrom') - filesys_discription = [ - ('a', 'D', ''), - ('a/l', 'D', ''), - ('a/l/g', 'F', Content_to_check) - ] - generate_import_root(test_dir, filesys_discription) + Content_to_check = 'You got me' + test_dir = os.path.join(str(tmpdir), 'importfrom') + filesys_discription = [ + ('a', 'D', ''), + ('a/l', 'D', ''), + ('a/l/g', 'F', Content_to_check) + ] + generate_import_root(test_dir, filesys_discription) - d.import_files(test_dir) - digest = d.descend('a', 'l').index['g'].get_digest() + d.import_files(test_dir) + digest = d.descend('a', 'l').index['g'].get_digest() - assert Content_to_check == open(cas_cache.objpath(digest)).read() + assert Content_to_check == open(cas_cache.objpath(digest)).read() + finally: + cas_cache.release_resources() def test_descend(tmpdir): @@ -343,29 +353,32 @@ def test_descend(tmpdir): def _test_bad_symlinks_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) - d = CasBasedDirectory(cas_cache) + try: + d = CasBasedDirectory(cas_cache) - test_dir = os.path.join(str(tmpdir), 'importfrom') - filesys_discription = [ - ('a', 'D', ''), - ('a/l', 'S', '../target'), - ('target', 'F', 'You got me') - ] - generate_import_root(test_dir, filesys_discription) - d.import_files(test_dir) - exp_reason = "not-a-directory" + test_dir = os.path.join(str(tmpdir), 'importfrom') + filesys_discription = [ + ('a', 'D', ''), + ('a/l', 'S', '../target'), + ('target', 'F', 'You got me') + ] + generate_import_root(test_dir, filesys_discription) + d.import_files(test_dir) + exp_reason = "not-a-directory" - with pytest.raises(VirtualDirectoryError) as error: - d.descend('a', 'l', follow_symlinks=True) - assert error.reason == exp_reason + with pytest.raises(VirtualDirectoryError) as error: + d.descend('a', 'l', follow_symlinks=True) + assert error.reason == exp_reason - with pytest.raises(VirtualDirectoryError) as error: - d.descend('a', 'l') - assert error.reason == exp_reason + with pytest.raises(VirtualDirectoryError) as error: + d.descend('a', 'l') + assert error.reason == exp_reason - with pytest.raises(VirtualDirectoryError) as error: - d.descend('a', 'f') - assert error.reason == exp_reason + with pytest.raises(VirtualDirectoryError) as error: + d.descend('a', 'f') + assert error.reason == exp_reason + finally: + cas_cache.release_resources() def test_bad_symlinks(tmpdir): @@ -377,21 +390,24 @@ def test_bad_symlinks(tmpdir): def _test_relative_symlink_subprocess(tmpdir): cas_dir = os.path.join(str(tmpdir), 'cas') cas_cache = CASCache(cas_dir) - d = CasBasedDirectory(cas_cache) + try: + d = CasBasedDirectory(cas_cache) - Content_to_check = 'You got me' - test_dir = os.path.join(str(tmpdir), 'importfrom') - filesys_discription = [ - ('a', 'D', ''), - ('a/l', 'S', '../target'), - ('target', 'D', ''), - ('target/file', 'F', Content_to_check) - ] - generate_import_root(test_dir, filesys_discription) - d.import_files(test_dir) + Content_to_check = 'You got me' + test_dir = os.path.join(str(tmpdir), 'importfrom') + filesys_discription = [ + ('a', 'D', ''), + ('a/l', 'S', '../target'), + ('target', 'D', ''), + ('target/file', 'F', Content_to_check) + ] + generate_import_root(test_dir, filesys_discription) + d.import_files(test_dir) - digest = d.descend('a', 'l', follow_symlinks=True).index['file'].get_digest() - assert Content_to_check == open(cas_cache.objpath(digest)).read() + digest = d.descend('a', 'l', follow_symlinks=True).index['file'].get_digest() + assert Content_to_check == open(cas_cache.objpath(digest)).read() + finally: + cas_cache.release_resources() def test_relative_symlink(tmpdir): @@ -403,22 +419,25 @@ def test_relative_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) + try: + d = CasBasedDirectory(cas_cache) - Content_to_check = 'two step file' - test_dir = os.path.join(str(tmpdir), 'importfrom') - filesys_discription = [ - ('a', 'D', ''), - ('a/l', 'S', '/target'), - ('target', 'D', ''), - ('target/file', 'F', Content_to_check) - ] - generate_import_root(test_dir, filesys_discription) - d.import_files(test_dir) + Content_to_check = 'two step file' + test_dir = os.path.join(str(tmpdir), 'importfrom') + filesys_discription = [ + ('a', 'D', ''), + ('a/l', 'S', '/target'), + ('target', 'D', ''), + ('target/file', 'F', Content_to_check) + ] + generate_import_root(test_dir, filesys_discription) + d.import_files(test_dir) - digest = d.descend('a', 'l', follow_symlinks=True).index['file'].get_digest() + digest = d.descend('a', 'l', follow_symlinks=True).index['file'].get_digest() - assert Content_to_check == open(cas_cache.objpath(digest)).read() + assert Content_to_check == open(cas_cache.objpath(digest)).read() + finally: + cas_cache.release_resources() def test_abs_symlink(tmpdir): @@ -430,22 +449,25 @@ def test_abs_symlink(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) - - test_dir = os.path.join(str(tmpdir), 'importfrom') - filesys_discription = [ - ('jail', 'D', ''), - ('jail/a', 'D', ''), - ('jail/a/l', 'S', '../../target'), - ('target', 'D', ''), - ('target/file', 'F', 'two step file') - ] - generate_import_root(test_dir, filesys_discription) - d.import_files(os.path.join(test_dir, 'jail')) - - with pytest.raises(VirtualDirectoryError) as error: - d.descend('a', 'l', follow_symlinks=True) - assert error.reason == "directory-not-found" + try: + d = CasBasedDirectory(cas_cache) + + test_dir = os.path.join(str(tmpdir), 'importfrom') + filesys_discription = [ + ('jail', 'D', ''), + ('jail/a', 'D', ''), + ('jail/a/l', 'S', '../../target'), + ('target', 'D', ''), + ('target/file', 'F', 'two step file') + ] + generate_import_root(test_dir, filesys_discription) + d.import_files(os.path.join(test_dir, 'jail')) + + with pytest.raises(VirtualDirectoryError) as error: + d.descend('a', 'l', follow_symlinks=True) + assert error.reason == "directory-not-found" + finally: + cas_cache.release_resources() def test_bad_sym_escape(tmpdir): -- cgit v1.2.1