import hashlib import os import shutil import tempfile from time import sleep, time import pytest from scantree import SymlinkRecursionError from dirhash import ( Filter, Protocol, _get_hasher_factory, _parmap, algorithms_available, algorithms_guaranteed, dirhash, dirhash_impl, get_match_patterns, included_paths, ) def osp(path: str) -> str: """Normalize path for OS.""" if os.name == "nt": # pragma: no cover return path.replace("/", "\\") return path def map_osp(paths): return [osp(path) for path in paths] class TestGetHasherFactory: def test_get_guaranteed(self): algorithm_and_hasher_factory = [ ("md5", hashlib.md5), ("sha1", hashlib.sha1), ("sha224", hashlib.sha224), ("sha256", hashlib.sha256), ("sha384", hashlib.sha384), ("sha512", hashlib.sha512), ] assert algorithms_guaranteed == {a for a, _ in algorithm_and_hasher_factory} for algorithm, expected_hasher_factory in algorithm_and_hasher_factory: hasher_factory = _get_hasher_factory(algorithm) assert hasher_factory == expected_hasher_factory def test_get_available(self): for algorithm in algorithms_available: hasher_factory = _get_hasher_factory(algorithm) try: hasher = hasher_factory() except ValueError as exc: # Some "available" algorithms are not necessarily available # (fails for e.g. 'ripemd160' in github actions for python 3.8). # See: https://stackoverflow.com/questions/72409563/unsupported-hash-type-ripemd160-with-hashlib-in-python # noqa: E501 print(f"Failed to create hasher for {algorithm}: {exc}") assert exc.args[0] == f"unsupported hash type {algorithm}" hasher = None if hasher is not None: assert hasattr(hasher, "update") assert hasattr(hasher, "hexdigest") def test_not_available(self): with pytest.raises(ValueError): _get_hasher_factory("not available") def test_bypass_hasher_factory(self): # test standard hasher hasher_factory = _get_hasher_factory(hashlib.sha256) assert hasher_factory is hashlib.sha256 # test raise on custom hasher with bad interface class IncompleteMockHasher: def __init__(self, *args, **kwargs): pass def update(self, *args, **kwargs): pass with pytest.raises(ValueError): _get_hasher_factory(IncompleteMockHasher) # test custom hasher with ok interface class MockHasher(IncompleteMockHasher): def hexdigest(self): return "" hasher_factory = _get_hasher_factory(MockHasher) assert hasher_factory is MockHasher class TestGetMatchPatterns: def test_default_match_all(self): ms = get_match_patterns() assert ms == ["*"] def test_only_match(self): ms = get_match_patterns(match=["a*", "b*"]) assert ms == ["a*", "b*"] def test_only_ignore(self): ms = get_match_patterns(ignore=["a*", "b*"]) assert ms == ["*", "!a*", "!b*"] def test_match_and_ignore(self): ms = get_match_patterns(match=["a*"], ignore=["*.ext"]) assert ms == ["a*", "!*.ext"] def test_ignore_hidden(self): ms = get_match_patterns(ignore_hidden=True) assert ms == ["*", "!.*", "!.*/"] # should not duplicate if present in (general) ignore ms = get_match_patterns(ignore=[".*"], ignore_hidden=True) assert ms == ["*", "!.*", "!.*/"] ms = get_match_patterns(ignore=[".*/"], ignore_hidden=True) assert ms == ["*", "!.*/", "!.*"] ms = get_match_patterns(ignore=[".*", ".*/"], ignore_hidden=True) assert ms == ["*", "!.*", "!.*/"] def test_ignore_extensions(self): ms = get_match_patterns(ignore_extensions=[".ext"]) assert ms == ["*", "!*.ext"] # automatically adds '.' ms = get_match_patterns(ignore_extensions=["ext"]) assert ms == ["*", "!*.ext"] # mixed also works ms = get_match_patterns(ignore_extensions=["ext1", ".ext2"]) assert ms == ["*", "!*.ext1", "!*.ext2"] # should not duplicate if present in (general) ignore ms = get_match_patterns(ignore=["*.ext"], ignore_extensions=[".ext"]) assert ms == ["*", "!*.ext"] ms = get_match_patterns(ignore=["*.ext"], ignore_extensions=["ext"]) assert ms == ["*", "!*.ext"] class TempDirTest: def setup_method(self): self.dir = tempfile.mkdtemp() def teardown_method(self): if os.path.exists(self.dir): shutil.rmtree(self.dir) def path_to(self, relpath): return os.path.join(self.dir, osp(relpath)) def mkdirs(self, dirpath): os.makedirs(self.path_to(dirpath)) def mkfile(self, relpath, content=None): with open(self.path_to(relpath), "w") as f: if content: f.write(content) def symlink(self, src, dst): os.symlink(self.path_to(src), self.path_to(dst)) def remove(self, relpath): if os.path.isdir(self.path_to(relpath)): shutil.rmtree(self.path_to(relpath)) os.remove(self.path_to(relpath)) class TestGetIncludedPaths(TempDirTest): # Integration tests with `pathspec` for basic use cases. def test_basic(self): self.mkdirs("root/d1/d11") self.mkdirs("root/d2") self.mkfile("root/f1") self.mkfile("root/d1/f1") self.mkfile("root/d1/d11/f1") self.mkfile("root/d2/f1") expected_filepaths = map_osp(["d1/d11/f1", "d1/f1", "d2/f1", "f1"]) filepaths = included_paths(self.path_to("root")) assert filepaths == expected_filepaths # end with '/' or not should not matter filepaths = included_paths(self.path_to("root/")) assert filepaths == expected_filepaths def test_not_a_directory(self): self.mkdirs("root") self.mkfile("root/f1") # does not exist with pytest.raises(ValueError): included_paths(self.path_to("wrong_root")) with pytest.raises(ValueError): included_paths(self.path_to("root/f1")) def test_symlinked_file(self): self.mkdirs("root") self.mkfile("root/f1") self.mkfile("linked_file") self.symlink("linked_file", "root/f2") filepaths = included_paths(self.path_to("root"), linked_files=True) assert filepaths == ["f1", "f2"] filepaths = included_paths(self.path_to("root"), linked_files=False) assert filepaths == ["f1"] # default is 'linked_files': True filepaths = included_paths( self.path_to("root"), ) assert filepaths == ["f1", "f2"] def test_symlinked_dir(self): self.mkdirs("root") self.mkfile("root/f1") self.mkdirs("linked_dir") self.mkfile("linked_dir/f1") self.mkfile("linked_dir/f2") self.symlink("linked_dir", "root/d1") filepaths = included_paths(self.path_to("root"), linked_dirs=False) assert filepaths == ["f1"] filepaths = included_paths(self.path_to("root"), linked_dirs=True) assert filepaths == map_osp(["d1/f1", "d1/f2", "f1"]) # default is 'linked_dirs': True filepaths = included_paths(self.path_to("root")) assert filepaths == map_osp(["d1/f1", "d1/f2", "f1"]) def test_cyclic_link(self): self.mkdirs("root/d1") self.symlink("root", "root/d1/link_back") with pytest.raises(SymlinkRecursionError) as exc_info: included_paths(self.path_to("root"), allow_cyclic_links=False) assert exc_info.value.real_path == os.path.realpath(self.path_to("root")) assert exc_info.value.first_path == self.path_to("root/") assert exc_info.value.second_path == self.path_to("root/d1/link_back") assert str(exc_info.value).startswith("Symlink recursion:") filepaths = included_paths(self.path_to("root"), allow_cyclic_links=True) assert filepaths == map_osp(["d1/link_back/."]) # default is 'allow_cyclic_links': False with pytest.raises(SymlinkRecursionError): filepaths = included_paths(self.path_to("root")) def test_ignore_hidden(self): self.mkdirs("root/d1") self.mkdirs("root/.d2") self.mkfile("root/f1") self.mkfile("root/.f2") self.mkfile("root/d1/f1") self.mkfile("root/d1/.f2") self.mkfile("root/.d2/f1") # no ignore filepaths = included_paths(self.path_to("root")) assert filepaths == map_osp([".d2/f1", ".f2", "d1/.f2", "d1/f1", "f1"]) # with ignore filepaths = included_paths(self.path_to("root"), match=["*", "!.*"]) assert filepaths == map_osp(["d1/f1", "f1"]) def test_ignore_hidden_files_only(self): self.mkdirs("root/d1") self.mkdirs("root/.d2") self.mkfile("root/f1") self.mkfile("root/.f2") self.mkfile("root/d1/f1") self.mkfile("root/d1/.f2") self.mkfile("root/.d2/f1") # no ignore filepaths = included_paths(self.path_to("root")) assert filepaths == map_osp([".d2/f1", ".f2", "d1/.f2", "d1/f1", "f1"]) # with ignore filepaths = included_paths( self.path_to("root"), match=["**/*", "!**/.*", "**/.*/*", "!**/.*/.*"] ) assert filepaths == map_osp([".d2/f1", "d1/f1", "f1"]) def test_ignore_hidden_explicitly_recursive(self): self.mkdirs("root/d1") self.mkdirs("root/.d2") self.mkfile("root/f1") self.mkfile("root/.f2") self.mkfile("root/d1/f1") self.mkfile("root/d1/.f2") self.mkfile("root/.d2/f1") # no ignore filepaths = included_paths(self.path_to("root")) assert filepaths == map_osp([".d2/f1", ".f2", "d1/.f2", "d1/f1", "f1"]) # with ignore filepaths = included_paths(self.path_to("root"), match=["*", "!**/.*"]) assert filepaths == map_osp(["d1/f1", "f1"]) def test_exclude_hidden_dirs(self): self.mkdirs("root/d1") self.mkdirs("root/.d2") self.mkdirs("root/d1/.d1") self.mkfile("root/f1") self.mkfile("root/.f2") self.mkfile("root/d1/f1") self.mkfile("root/d1/.f2") self.mkfile("root/.d2/f1") # no ignore filepaths = included_paths(self.path_to("root"), empty_dirs=True) assert filepaths == map_osp( [".d2/f1", ".f2", "d1/.d1/.", "d1/.f2", "d1/f1", "f1"] ) # with ignore filepaths = included_paths(self.path_to("root"), match=["*", "!.*/"]) assert filepaths == map_osp([".f2", "d1/.f2", "d1/f1", "f1"]) def test_exclude_hidden_dirs_and_files(self): self.mkdirs("root/d1") self.mkdirs("root/.d2") self.mkfile("root/f1") self.mkfile("root/.f2") self.mkfile("root/d1/f1") self.mkfile("root/d1/.f2") self.mkfile("root/.d2/f1") # no ignore filepaths = included_paths(self.path_to("root")) assert filepaths == map_osp([".d2/f1", ".f2", "d1/.f2", "d1/f1", "f1"]) # using ignore filepaths = included_paths(self.path_to("root"), match=["*", "!.*/", "!.*"]) assert filepaths == map_osp(["d1/f1", "f1"]) def test_exclude_extensions(self): self.mkdirs("root/d1") self.mkfile("root/f") self.mkfile("root/f.txt") self.mkfile("root/f.skip1") self.mkfile("root/fskip1") self.mkfile("root/f.skip2") self.mkfile("root/f.skip1.txt") self.mkfile("root/f.skip1.skip2") self.mkfile("root/f.skip1skip2") self.mkfile("root/d1/f.txt") self.mkfile("root/d1/f.skip1") filepaths = included_paths( self.path_to("root"), match=["*", "!*.skip1", "!*.skip2"] ) assert filepaths == map_osp( [ "d1/f.txt", "f", "f.skip1.txt", "f.skip1skip2", "f.txt", "fskip1", ] ) def test_empty_dirs_include_vs_exclude(self): self.mkdirs("root/d1") self.mkdirs("root/d2") self.mkdirs("root/d3/d31") self.mkdirs("root/d4/d41") self.mkfile("root/d1/f") self.mkfile("root/d3/d31/f") filepaths = included_paths(self.path_to("root"), empty_dirs=False) assert filepaths == map_osp(["d1/f", "d3/d31/f"]) # `include_empty=False` is default filepaths = included_paths(self.path_to("root")) assert filepaths == map_osp(["d1/f", "d3/d31/f"]) filepaths = included_paths(self.path_to("root"), empty_dirs=True) assert filepaths == map_osp(["d1/f", "d2/.", "d3/d31/f", "d4/d41/."]) def test_empty_dirs_because_of_filter_include_vs_exclude(self): self.mkdirs("root/d1") self.mkdirs("root/d2") self.mkfile("root/d1/f") self.mkfile("root/d2/.f") filepaths = included_paths( self.path_to("root"), match=["*", "!.*"], empty_dirs=False ) assert filepaths == map_osp(["d1/f"]) # `include_empty=False` is default filepaths = included_paths( self.path_to("root"), match=["*", "!.*"], ) assert filepaths == map_osp(["d1/f"]) filepaths = included_paths( self.path_to("root"), match=["*", "!.*"], empty_dirs=True ) assert filepaths == map_osp(["d1/f", "d2/."]) def test_empty_dir_inclusion_not_affected_by_match(self): self.mkdirs("root/d1") self.mkdirs("root/.d2") # NOTE that empty dirs are not excluded by match_patterns: filepaths = included_paths( self.path_to("root"), match=["*", "!.*"], empty_dirs=True ) assert filepaths == map_osp([".d2/.", "d1/."]) filepaths = included_paths( self.path_to("root"), match=["*", "!.*/"], empty_dirs=True ) assert filepaths == map_osp([".d2/.", "d1/."]) filepaths = included_paths( self.path_to("root"), match=["*", "!d1"], empty_dirs=True ) assert filepaths == map_osp([".d2/.", "d1/."]) def dirhash_mp_comp(*args, **kwargs): res = dirhash(*args, **kwargs) res_mp = dirhash(*args, **{**kwargs, "jobs": 2}) assert res == res_mp return res class TestDirhash(TempDirTest): def test_guaranteed_algorithms(self): self.mkdirs("root/d1/d11") self.mkdirs("root/d2") self.mkfile("root/f1", "a") self.mkfile("root/d1/f1", "b") self.mkfile("root/d1/d11/f1", "c") self.mkfile("root/d2/f1", "d") for algorithm, expected_hash in [ ("md5", "3c631c7f5771468a2187494f802fad8f"), ("sha1", "992aa2d00d2ed94f0c19eff7f151f5c6a7e0cc41"), ("sha224", "18013e1df933d5781b2eddb94aceeb7ab689643f1df24060fb478999"), ( "sha256", "ef7e95269fbc0e3478ad31fddd1c7d08" "907d189c61725332e8a2fd14448fe175", ), ( "sha384", "64ef4360c172bc68250f9326ea231cd1" "46a7fa1afe9d386cee0cae0e9f1b4ad2" "1df050d1df436cff792bbe81d6698026", ), ( "sha512", "7854226eb0278bc136056998890a8399" "f85ca383f7c54665026358d28b5dc716" "0ec654d2bcebf5d60974f82ed820600d" "8e807ea53d57578d076ec1c82f501208", ), ]: hash_value = dirhash_mp_comp(self.path_to("root"), algorithm) assert hash_value == expected_hash def test_recursive_descriptor(self): self.mkdirs("root/d1") self.mkdirs("root/d2") self.mkfile("root/f1", "a") self.mkfile("root/d1/f12", "b") f1_desc = "data:a\000name:f1" f12_desc = "data:b\000name:f12" d1_desc = f"dirhash:{f12_desc}\000name:d1" d2_desc = "dirhash:\000name:d2" empty_dirs_false_expected = "\000\000".join([f1_desc, d1_desc]) empty_dirs_true_expected = "\000\000".join([f1_desc, d2_desc, d1_desc]) empty_dirs_false = dirhash(self.path_to("root"), algorithm=IdentityHasher) assert empty_dirs_false == empty_dirs_false_expected empty_dirs_true = dirhash( self.path_to("root"), algorithm=IdentityHasher, empty_dirs=True ) assert empty_dirs_true == empty_dirs_true_expected def test_symlinked_file(self): self.mkdirs("root1") self.mkfile("root1/f1", "a") self.mkfile("linked_file", "b") self.symlink("linked_file", "root1/f2") self.mkdirs("root2") self.mkfile("root2/f1", "a") self.mkfile("root2/f2", "b") root1_linked_files_true = dirhash_mp_comp( self.path_to("root1"), algorithm="md5" ) root1_linked_files_false = dirhash_mp_comp( self.path_to("root1"), algorithm="md5", linked_files=False ) root2 = dirhash_mp_comp(self.path_to("root2"), algorithm="md5") assert root1_linked_files_false != root1_linked_files_true assert root1_linked_files_true == root2 def test_symlinked_dir(self): self.mkdirs("root1") self.mkfile("root1/f1", "a") self.mkdirs("linked_dir") self.mkfile("linked_dir/f1", "b") self.mkfile("linked_dir/f2", "c") self.symlink("linked_dir", "root1/d1") self.mkdirs("root2") self.mkfile("root2/f1", "a") self.mkdirs("root2/d1") self.mkfile("root2/d1/f1", "b") self.mkfile("root2/d1/f2", "c") root1_linked_dirs_true = dirhash_mp_comp( self.path_to("root1"), algorithm="md5", linked_dirs=True ) root1_linked_dirs_false = dirhash_mp_comp( self.path_to("root1"), algorithm="md5", linked_dirs=False ) root2 = dirhash_mp_comp(self.path_to("root2"), algorithm="md5") assert root1_linked_dirs_false != root1_linked_dirs_true assert root1_linked_dirs_true == root2 def test_cache_used_for_symlinks(self): self.mkdirs("root/dir") self.mkfile("root/file", "< one chunk content") for i in range(10): self.symlink("root/file", f"root/link_{i}") for i in range(10): self.symlink("root/file", f"root/dir/link_{i}") start = time() dirhash(self.path_to("root"), algorithm=SlowHasher) end = time() elapsed = end - start assert elapsed < SlowHasher.wait_time * 2 def test_raise_on_empty_root_without_include_empty(self): self.mkdirs("root") with pytest.raises(ValueError): dirhash_mp_comp(self.path_to("root"), "sha256") def test_empty_root_include_empty(self): self.mkdirs("root") dirhash_ = dirhash_mp_comp(self.path_to("root"), "sha256", empty_dirs=True) expected_dirhash = hashlib.sha256(b"").hexdigest() assert dirhash_ == expected_dirhash def test_include_empty(self): self.mkdirs("root/d1") self.mkdirs("root/d2") self.mkfile("root/d1/f") args = (self.path_to("root"), "sha256") dirhash_ = dirhash_mp_comp(*args, empty_dirs=False) dirhash_empty = dirhash_mp_comp(*args, empty_dirs=True) assert dirhash_ != dirhash_empty def test_chunksize(self): self.mkdirs("root") self.mkfile("root/numbers.txt", str(range(1000))) hash_value = dirhash_mp_comp(self.path_to("root"), "sha256") for chunk_size in [2**4, 2**8, 2**16]: assert ( dirhash_mp_comp(self.path_to("root"), "sha256", chunk_size=chunk_size) == hash_value ) def test_data_only(self): self.mkdirs("root1") self.mkfile("root1/a.txt", "abc") self.mkfile("root1/b.txt", "def") self.mkdirs("root2") self.mkfile("root2/a.txt", "abc") self.mkfile("root2/c.txt", "def") hash1 = dirhash_mp_comp(self.path_to("root1"), "sha256") hash2 = dirhash_mp_comp(self.path_to("root2"), "sha256") assert hash1 != hash2 # with entry hash remains the same as long as order of files is the # same [dhash1, dhash2] = [ dirhash_mp_comp(self.path_to(root), "sha256", entry_properties=["data"]) for root in ["root1", "root2"] ] assert dhash1 == dhash2 def test_name_only(self): self.mkdirs("root1") self.mkfile("root1/a.txt", "abc") self.mkfile("root1/b.txt", "def") self.mkdirs("root2") self.mkfile("root2/a.txt", "abc") self.mkfile("root2/b.txt", "___") hash1 = dirhash_mp_comp(self.path_to("root1"), "sha256") hash2 = dirhash_mp_comp(self.path_to("root2"), "sha256") assert hash1 != hash2 [dhash1, dhash2] = [ dirhash_mp_comp(self.path_to(root), "sha256", entry_properties=["name"]) for root in ["root1", "root2"] ] assert dhash1 == dhash2 def test_is_link_property(self): self.mkdirs("root1") self.mkfile("root1/a.txt", "abc") self.mkfile("root1/b.txt", "def") self.mkdirs("root2") self.mkfile("b_target", "def") self.mkfile("root2/a.txt", "abc") self.symlink("b_target", "root2/b.txt") hash1 = dirhash_mp_comp(self.path_to("root1"), "sha256") hash2 = dirhash_mp_comp(self.path_to("root2"), "sha256") assert hash1 == hash2 for entry_properties in [ ["name", "data", "is_link"], ["name", "is_link"], ["data", "is_link"], ]: [hash1, hash2] = [ dirhash_mp_comp( self.path_to(root), "sha256", entry_properties=entry_properties ) for root in ["root1", "root2"] ] assert hash1 != hash2 def test_raise_on_not_at_least_one_of_name_and_data(self): self.mkdirs("root1") self.mkfile("root1/a.txt", "abc") dirhash_mp_comp(self.path_to("root1"), "sha256") # check ok with pytest.raises(ValueError): dirhash_mp_comp(self.path_to("root1"), "sha256", entry_properties=[]) with pytest.raises(ValueError): dirhash_mp_comp( self.path_to("root1"), "sha256", entry_properties=["is_link"] ) @pytest.mark.skipif( os.name == "nt", reason="TODO: not getting expected speedup on Windows.", # TODO: see https://github.com/andhus/scantree/issues/25 ) def test_multiproc_speedup(self): self.mkdirs("root/dir") num_files = 10 for i in range(num_files): self.mkfile(f"root/file_{i}", "< one chunk content") expected_min_elapsed_sequential = SlowHasher.wait_time * num_files start = time() dirhash(self.path_to("root"), algorithm=SlowHasher) end = time() elapsed_sequential = end - start assert elapsed_sequential > expected_min_elapsed_sequential start = time() dirhash(self.path_to("root"), algorithm=SlowHasher, jobs=num_files) end = time() elapsed_muliproc = end - start assert elapsed_muliproc < 0.9 * expected_min_elapsed_sequential # just check "any speedup", the overhead varies (and is high on Travis) def test_cache_by_real_path_speedup(self, tmpdir): num_links = 10 # reference run without links root1 = tmpdir.join("root1") root1.ensure(dir=True) for i in range(num_links): file_i = root1.join(f"file_{i}") file_i.write("< one chunk content", ensure=True) wait_time = SlowHasher.wait_time expected_min_elapsed_no_links = wait_time * num_links start = time() dirhash(root1, algorithm=SlowHasher) end = time() elapsed_no_links = end - start assert elapsed_no_links > expected_min_elapsed_no_links overhead = elapsed_no_links - expected_min_elapsed_no_links # all links to same file root2 = tmpdir.join("root2") root2.ensure(dir=True) target_file = tmpdir.join("target_file") target_file.ensure() for i in range(num_links): os.symlink(target_file, root2.join(f"link_{i}")) overhead_margin_factor = 1.5 expected_max_elapsed_with_links = overhead * overhead_margin_factor + wait_time assert expected_max_elapsed_with_links < expected_min_elapsed_no_links start = time() dirhash(root2, algorithm=SlowHasher) end = time() elapsed_with_links = end - start assert elapsed_with_links < expected_max_elapsed_with_links def test_cache_together_with_multiprocess_speedup(self, tmpdir): target_file_names = ["target_file_1", "target_file_2"] num_links_per_file = 10 num_links = num_links_per_file * len(target_file_names) # reference run without links root1 = tmpdir.join("root1") root1.ensure(dir=True) for i in range(num_links): file_i = root1.join(f"file_{i}") file_i.write("< one chunk content", ensure=True) jobs = 2 wait_time = SlowHasher.wait_time expected_min_elapsed_no_links = wait_time * num_links / jobs start = time() dirhash(root1, algorithm=SlowHasher, jobs=jobs) end = time() elapsed_no_links = end - start assert elapsed_no_links > expected_min_elapsed_no_links overhead = elapsed_no_links - expected_min_elapsed_no_links root2 = tmpdir.join("root2") root2.ensure(dir=True) for i, target_file_name in enumerate(target_file_names): target_file = tmpdir.join(target_file_name) target_file.write("< one chunk content", ensure=True) for j in range(num_links_per_file): os.symlink(target_file, root2.join(f"link_{i}_{j}")) overhead_margin_factor = 1.5 expected_max_elapsed_with_links = ( overhead * overhead_margin_factor + wait_time * 2 ) assert expected_max_elapsed_with_links < expected_min_elapsed_no_links start = time() dirhash(root2, algorithm=SlowHasher, jobs=jobs) end = time() elapsed_mp_with_links = end - start assert elapsed_mp_with_links < expected_max_elapsed_with_links def test_hash_cyclic_link_to_root(self): self.mkdirs("root/d1") self.symlink("root", "root/d1/link_back") dirhash(self.path_to("root"), "sha256", allow_cyclic_links=True) def test_hash_cyclic_link(self): self.mkdirs("root/d1/d2") self.symlink("root/d1", "root/d1/d2/link_back") dirhash(self.path_to("root"), "sha256", allow_cyclic_links=True) def test_pass_filtering_instance(self): self.mkdirs("root") self.mkfile("root/f1", "") dirhash_impl(self.path_to("root"), "sha256", filter_=Filter()) def test_pass_protocol_instance(self): self.mkdirs("root") self.mkfile("root/f1", "") dirhash_impl(self.path_to("root"), "sha256", protocol=Protocol()) def test_raise_on_wrong_type(self): self.mkdirs("root") self.mkfile("root/f1", "") with pytest.raises(TypeError): dirhash_impl(self.path_to("root"), "sha256", filter_="") with pytest.raises(TypeError): dirhash_impl(self.path_to("root"), "sha256", protocol="") class SlowHasher: wait_time = 0.25 def __init__(self, *args, **kwargs): pass def update(self, data): if data != b"": sleep(self.wait_time) def hexdigest(self): return "" class IdentityHasher: def __init__(self, initial_data=b""): self.datas = [initial_data.decode("utf-8")] def update(self, data): self.datas.append(data.decode("utf-8")) def hexdigest(self): return "".join(self.datas) class TestProtocol: def test_raise_for_invalid_entry_properties(self): with pytest.raises(ValueError): Protocol(entry_properties=["not-valid"]) def test_raise_for_invalid_allow_cyclic_links(self): with pytest.raises(ValueError): Protocol(allow_cyclic_links="not-valid") def mock_func(x): return x * 2 @pytest.mark.parametrize("jobs", [1, 2, 4]) def test_parmap(jobs): inputs = [1, 2, 3, 4] assert _parmap(mock_func, inputs, jobs=jobs) == [2, 4, 6, 8]