From 226c68f8ff98343d3bf9a9a4b815112ebca93543 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 9 Feb 2021 04:00:06 +0200 Subject: [PATCH] dvc: introduce File/Tree cache objects (#5424) --- dvc/cache/base.py | 308 +++++-------------------------- dvc/cache/local.py | 28 +-- dvc/checkout.py | 231 ++++++++--------------- dvc/dependency/repo.py | 8 +- dvc/objects.py | 254 +++++++++++++++++++++++++ dvc/output/base.py | 111 +++++++---- dvc/repo/fetch.py | 9 +- dvc/tree/base.py | 16 +- dvc/tree/dvc.py | 7 +- dvc/tree/memory.py | 37 ++++ setup.py | 1 + tests/func/test_add.py | 4 +- tests/func/test_cache.py | 16 +- tests/func/test_checkout.py | 6 +- tests/func/test_commit.py | 1 + tests/func/test_external_repo.py | 17 +- tests/unit/output/test_local.py | 25 ++- tests/unit/test_external_repo.py | 9 +- tests/unit/tree/test_dvc.py | 8 +- tests/unit/tree/test_repo.py | 8 +- 20 files changed, 587 insertions(+), 517 deletions(-) create mode 100644 dvc/objects.py create mode 100644 dvc/tree/memory.py diff --git a/dvc/cache/base.py b/dvc/cache/base.py index ab96f078f..b211278ce 100644 --- a/dvc/cache/base.py +++ b/dvc/cache/base.py @@ -1,8 +1,5 @@ -import errno import itertools -import json import logging -import os from concurrent import futures from concurrent.futures import ThreadPoolExecutor from copy import copy @@ -18,18 +15,11 @@ from dvc.remote.slow_link_detection import ( # type: ignore[attr-defined] slow_link_guard, ) -from ..tree.base import RemoteActionNotImplemented +from ..objects import File, HashFile, ObjectFormatError, Tree logger = logging.getLogger(__name__) -class DirCacheError(DvcException): - def __init__(self, hash_): - super().__init__( - f"Failed to load dir cache for hash value: '{hash_}'." - ) - - @decorator def use_state(call): tree = call._args[0].tree # pylint: disable=protected-access @@ -52,40 +42,6 @@ class CloudCache: self.DEFAULT_CACHE_TYPES ) self.cache_type_confirmed = False - self._dir_info = {} - - def get_dir_cache(self, hash_info): - assert hash_info - - dir_info = self._dir_info.get(hash_info.value) - if dir_info: - return dir_info - - try: - dir_info = self.load_dir_cache(hash_info) - except DirCacheError: - dir_info = DirInfo() - - self._dir_info[hash_info.value] = dir_info - return dir_info - - def load_dir_cache(self, hash_info): - path_info = self.tree.hash_to_path_info(hash_info.value) - - try: - with self.tree.open(path_info, "r") as fobj: - d = json.load(fobj) - except (ValueError, FileNotFoundError) as exc: - raise DirCacheError(hash_info) from exc - - if not isinstance(d, list): - logger.error( - "dir cache file format error '%s' [skipping the file]", - path_info, - ) - d = [] - - return DirInfo.from_list(d) def link(self, from_info, to_info): self._link(from_info, to_info, self.cache_types) @@ -141,31 +97,39 @@ class CloudCache: "Created '%s': %s -> %s", self.cache_types[0], from_info, to_info, ) - def add(self, path_info, tree, hash_info, **kwargs): - if not self.changed_cache_file(hash_info): - return + def get(self, hash_info): + """ get raw object """ + return HashFile( + # Prefer string path over PathInfo when possible due to performance + self.hash_to_path(hash_info.value), + self.tree, + hash_info, + ) - cache_info = self.tree.hash_to_path_info(hash_info.value) + @use_state + def add(self, obj, **kwargs): + try: + self.check(obj.hash_info) + return + except (ObjectFormatError, FileNotFoundError): + pass + + cache_info = self.tree.hash_to_path_info(obj.hash_info.value) # using our makedirs to create dirs with proper permissions self.makedirs(cache_info.parent) - if isinstance(tree, type(self.tree)): - self.tree.move(path_info, cache_info) + if isinstance(obj.tree, type(self.tree)): + self.tree.move(obj.path_info, cache_info) else: - with tree.open(path_info, mode="rb") as fobj: + with obj.tree.open(obj.path_info, mode="rb") as fobj: self.tree.upload_fobj(fobj, cache_info) - tree.state.save(path_info, hash_info) + obj.tree.state.save(obj.path_info, obj.hash_info) self.protect(cache_info) - self.tree.state.save(cache_info, hash_info) + self.tree.state.save(cache_info, obj.hash_info) callback = kwargs.get("download_callback") if callback: callback(1) - def _save_file(self, path_info, tree, hash_info, **kwargs): - assert hash_info - - self.add(path_info, tree, hash_info, **kwargs) - def _transfer_file(self, from_tree, from_info): from dvc.utils import tmp_fname from dvc.utils.stream import HashedStreamReader @@ -225,19 +189,11 @@ class CloudCache: ) in self._transfer_directory_contents( from_tree, from_info, jobs, pbar ): - self.add(entry_tmp_info, self.tree, entry_hash) + obj = File(entry_tmp_info, self.tree, entry_hash) + self.add(obj) dir_info.trie[entry_info.parts] = entry_hash - local_cache = self.repo.cache.local - ( - hash_info, - to_info, - ) = local_cache.get_dir_info_hash( # pylint: disable=protected-access - dir_info - ) - - self.add(to_info, self.repo.cache.local.tree, hash_info) - return hash_info + return Tree.save_dir_info(self, dir_info) def transfer(self, from_tree, from_info, jobs=None, no_progress_bar=False): jobs = jobs or min((from_tree.jobs, self.tree.jobs)) @@ -250,7 +206,8 @@ class CloudCache: no_progress_bar=no_progress_bar, ) tmp_info, hash_info = self._transfer_file(from_tree, from_info) - self.add(tmp_info, self.tree, hash_info) + obj = File(tmp_info, self.tree, hash_info) + self.add(obj) return hash_info def cache_is_copy(self, path_info): @@ -264,6 +221,7 @@ class CloudCache: workspace_file = path_info.with_name("." + uuid()) test_cache_file = self.tree.path_info / ".cache_type_test_file" if not self.tree.exists(test_cache_file): + self.makedirs(test_cache_file.parent) with self.tree.open(test_cache_file, "wb") as fobj: fobj.write(bytes(1)) try: @@ -275,89 +233,9 @@ class CloudCache: self.cache_type_confirmed = True return self.cache_types[0] == "copy" - def get_dir_info_hash(self, dir_info): - import tempfile - - from dvc.path_info import PathInfo - from dvc.utils import tmp_fname - - tmp = tempfile.NamedTemporaryFile(delete=False).name - with open(tmp, "w+") as fobj: - json.dump(dir_info.to_list(), fobj, sort_keys=True) - - from_info = PathInfo(tmp) - to_info = self.tree.path_info / tmp_fname("") - self.tree.upload(from_info, to_info, no_progress_bar=True) - - hash_info = self.tree.get_file_hash(to_info) - hash_info.value += self.tree.CHECKSUM_DIR_SUFFIX - hash_info.dir_info = dir_info - hash_info.nfiles = dir_info.nfiles - - return hash_info, to_info - @use_state - def save_dir_info(self, dir_info, hash_info=None): - if ( - hash_info - and hash_info.name == self.tree.PARAM_CHECKSUM - and not self.changed_cache_file(hash_info) - ): - return hash_info - - hi, tmp_info = self.get_dir_info_hash(dir_info) - self.add(tmp_info, self.tree, hi) - - return hi - - def _save_dir( - self, path_info, tree, hash_info, filter_info=None, **kwargs, - ): - if not hash_info.dir_info: - hash_info.dir_info = tree.cache.get_dir_cache(hash_info) - hi = self.save_dir_info(hash_info.dir_info, hash_info) - for entry_info, entry_hash in Tqdm( - hi.dir_info.items(path_info), - desc="Saving " + path_info.name, - unit="file", - ): - if filter_info and not entry_info.isin_or_eq(filter_info): - continue - - self._save_file(entry_info, tree, entry_hash, **kwargs) - - cache_info = self.tree.hash_to_path_info(hi.value) - self.tree.state.save(cache_info, hi) - tree.state.save(path_info, hi) - - @use_state - def save(self, path_info, tree, hash_info, **kwargs): - if path_info.scheme != self.tree.scheme: - raise RemoteActionNotImplemented( - f"save {path_info.scheme} -> {self.tree.scheme}", - self.tree.scheme, - ) - - if not hash_info: - kw = kwargs.copy() - kw.pop("download_callback", None) - hash_info = tree.get_hash(path_info, **kw) - if not hash_info: - raise FileNotFoundError( - errno.ENOENT, os.strerror(errno.ENOENT), path_info - ) - - self._save(path_info, tree, hash_info, **kwargs) - return hash_info - - def _save(self, path_info, tree, hash_info, **kwargs): - to_info = self.tree.hash_to_path_info(hash_info.value) - logger.debug("Saving '%s' to '%s'.", path_info, to_info) - - if tree.isdir(path_info): - self._save_dir(path_info, tree, hash_info, **kwargs) - else: - self._save_file(path_info, tree, hash_info, **kwargs) + def save(self, obj, **kwargs): + obj.save(self, **kwargs) # Override to return path as a string instead of PathInfo for clouds # which support string paths (see local) @@ -376,7 +254,7 @@ class CloudCache: def set_exec(self, path_info): # pylint: disable=unused-argument pass - def changed_cache_file(self, hash_info): + def check(self, hash_info): """Compare the given hash with the (corresponding) actual one. - Use `State` as a cache for computed hashes @@ -388,119 +266,25 @@ class CloudCache: - Remove the file from cache if it doesn't match the actual hash """ - # Prefer string path over PathInfo when possible due to performance - cache_info = self.hash_to_path(hash_info.value) - if self.is_protected(cache_info): + + obj = self.get(hash_info) + if self.is_protected(obj.path_info): logger.trace( - "Assuming '%s' is unchanged since it is read-only", cache_info + "Assuming '%s' is unchanged since it is read-only", + obj.path_info, ) - return False + return - actual = self.tree.get_hash(cache_info) - - logger.trace( - "cache '%s' expected '%s' actual '%s'", - cache_info, - hash_info, - actual, - ) - - if not hash_info or not actual: - return True - - if actual.value.split(".")[0] == hash_info.value.split(".")[0]: - # making cache file read-only so we don't need to check it - # next time - self.protect(cache_info) - return False - - if self.tree.exists(cache_info): - logger.warning("corrupted cache file '%s'.", cache_info) - self.tree.remove(cache_info) - - return True - - def _changed_dir_cache(self, hash_info, path_info=None, filter_info=None): - if self.changed_cache_file(hash_info): - return True - - dir_info = self.get_dir_cache(hash_info) - for entry_info, entry_hash in dir_info.items(path_info): - if path_info and filter_info: - if not entry_info.isin_or_eq(filter_info): - continue - - if self.changed_cache_file(entry_hash): - return True - - return False - - def changed_cache(self, hash_info, path_info=None, filter_info=None): - if hash_info.isdir: - return self._changed_dir_cache( - hash_info, path_info=path_info, filter_info=filter_info - ) - return self.changed_cache_file(hash_info) - - def get_files_number(self, path_info, hash_info, filter_info): - from funcy.py3 import ilen - - if not hash_info: - return 0 - - if not hash_info.isdir: - return 1 - - if not filter_info: - return self.get_dir_cache(hash_info).nfiles - - return ilen( - filter_info.isin_or_eq(path_info / relpath) - for relpath, _ in self.get_dir_cache(hash_info).items() - ) - - def _get_dir_size(self, dir_info): try: - return sum( - self.tree.getsize(self.tree.hash_to_path_info(hi.value)) - for _, hi in dir_info.items() - ) - except FileNotFoundError: - return None + obj.check(self) + except ObjectFormatError: + logger.warning("corrupted cache file '%s'.", obj.path_info) + self.tree.remove(obj.path_info) + raise - def merge(self, ancestor_info, our_info, their_info): - assert our_info - assert their_info - - if ancestor_info: - ancestor = self.get_dir_cache(ancestor_info) - else: - ancestor = DirInfo() - - our = self.get_dir_cache(our_info) - their = self.get_dir_cache(their_info) - - merged = our.merge(ancestor, their) - hash_info = self.save_dir_info(merged) - hash_info.size = self._get_dir_size(merged) - return hash_info - - @use_state - def get_hash(self, tree, path_info): - hash_info = tree.get_hash(path_info) - if not hash_info.isdir: - assert hash_info.name == self.tree.PARAM_CHECKSUM - return hash_info - - hi = self.save_dir_info(hash_info.dir_info, hash_info) - hi.size = hash_info.size - return hi - - def set_dir_info(self, hash_info): - assert hash_info.isdir - - hash_info.dir_info = self.get_dir_cache(hash_info) - hash_info.nfiles = hash_info.dir_info.nfiles + # making cache file read-only so we don't need to check it + # next time + self.protect(obj.path_info) def _list_paths(self, prefix=None, progress_callback=None): if prefix: diff --git a/dvc/cache/local.py b/dvc/cache/local.py index 75855029e..35c01c97e 100644 --- a/dvc/cache/local.py +++ b/dvc/cache/local.py @@ -6,6 +6,7 @@ from funcy import cached_property from shortuuid import uuid from dvc.hash_info import HashInfo +from dvc.objects import ObjectFormatError from dvc.path_info import PathInfo from dvc.progress import Tqdm @@ -68,18 +69,21 @@ class LocalCache(CloudCache): def hashes_exist( self, hashes, jobs=None, name=None ): # pylint: disable=unused-argument - return [ - hash_ - for hash_ in Tqdm( - hashes, - unit="file", - desc="Querying " - + ("cache in " + name if name else "local cache"), - ) - if not self.changed_cache_file( - HashInfo(self.tree.PARAM_CHECKSUM, hash_) - ) - ] + ret = [] + + for hash_ in Tqdm( + hashes, + unit="file", + desc="Querying " + ("cache in " + name if name else "local cache"), + ): + hash_info = HashInfo(self.tree.PARAM_CHECKSUM, hash_) + try: + self.check(hash_info) + ret.append(hash_) + except (FileNotFoundError, ObjectFormatError): + pass + + return ret def _verify_link(self, path_info, link_type): if link_type == "hardlink" and self.tree.getsize(path_info) == 0: diff --git a/dvc/checkout.py b/dvc/checkout.py index c04238b4d..fb911179d 100644 --- a/dvc/checkout.py +++ b/dvc/checkout.py @@ -1,101 +1,58 @@ import logging import dvc.prompt as prompt -from dvc.dir_info import DirInfo from dvc.exceptions import CheckoutError, ConfirmRemoveError +from dvc.objects import ObjectFormatError, check, load logger = logging.getLogger(__name__) -def _filter_hash_info(cache, hash_info, path_info, filter_info): - if not filter_info or path_info == filter_info: - return hash_info - - dir_info = cache.get_dir_cache(hash_info) - hash_key = filter_info.relative_to(path_info).parts - - # Check whether it is a file that exists on the trie - hi = dir_info.trie.get(hash_key) - if hi: - return hi - - depth = len(hash_key) - filtered_dir_info = DirInfo() - try: - for key, value in dir_info.trie.items(hash_key): - filtered_dir_info.trie[key[depth:]] = value - except KeyError: - return None - - return cache.get_dir_info_hash(filtered_dir_info)[0] - - -def _changed(path_info, tree, hash_info, cache, filter_info=None): - """Checks if data has changed. - - A file is considered changed if: - - It doesn't exist on the working directory (was unlinked) - - Hash value is not computed (saving a new file) - - The hash value stored is different from the given one - - There's no file in the cache - - Args: - path_info: dict with path information. - hash: expected hash value for this data. - filter_info: an optional argument to target a specific path. - - Returns: - bool: True if data has changed, False otherwise. - """ - - path = filter_info or path_info - logger.trace("checking if '%s'('%s') has changed.", path_info, hash_info) - - hi = _filter_hash_info(cache, hash_info, path_info, filter_info) - if not hi: - logger.debug("hash value for '%s' is missing.", path) - return True - - if cache.changed_cache(hi): - logger.debug("cache for '%s'('%s') has changed.", path, hi) - return True +def _changed(path_info, tree, obj, cache): + logger.trace("checking if '%s'('%s') has changed.", path_info, obj) try: - actual = tree.get_hash(path) - except FileNotFoundError: - logger.debug("'%s' doesn't exist.", path) - return True - - if hi != actual: + check(cache, obj) + except (FileNotFoundError, ObjectFormatError): logger.debug( - "hash value '%s' for '%s' has changed (actual '%s').", - hi, - actual, - path, + "cache for '%s'('%s') has changed.", path_info, obj.hash_info ) return True - logger.trace("'%s' hasn't changed.", path) + try: + actual = tree.get_hash(path_info) + except FileNotFoundError: + logger.debug("'%s' doesn't exist.", path_info) + return True + + if obj.hash_info != actual: + logger.debug( + "hash value '%s' for '%s' has changed (actual '%s').", + obj.hash_info, + actual, + path_info, + ) + return True + + logger.trace("'%s' hasn't changed.", path_info) return False -def _is_cached(cache, path_info, tree): - current = tree.get_hash(path_info) - - if not current: - return False - - return not cache.changed_cache(current) - - def _remove(path_info, tree, cache, force=False): if not tree.exists(path_info): return - if not force and not _is_cached(cache, path_info, tree): + if force: + tree.remove(path_info) + return + + current = tree.get_hash(path_info) + try: + obj = load(cache, current) + check(cache, obj) + except (FileNotFoundError, ObjectFormatError): msg = ( - "file '{}' is going to be removed." - " Are you sure you want to proceed?".format(str(path_info)) + f"file/directory '{path_info}' is going to be removed. " + "Are you sure you want to proceed?" ) if not prompt.confirm(msg): @@ -105,26 +62,17 @@ def _remove(path_info, tree, cache, force=False): def _checkout_file( - path_info, - tree, - hash_info, - cache, - force, - progress_callback=None, - relink=False, + path_info, tree, obj, cache, force, progress_callback=None, relink=False, ): """The file is changed we need to checkout a new copy""" - cache_info = cache.tree.hash_to_path_info(hash_info.value) + modified = False + cache_info = cache.tree.hash_to_path_info(obj.hash_info.value) if tree.exists(path_info): - added = False - - if not relink and _changed(path_info, tree, hash_info, cache): + if not relink and _changed(path_info, tree, obj, cache): modified = True _remove(path_info, tree, cache, force=force) cache.link(cache_info, path_info) else: - modified = False - if tree.iscopy(path_info) and cache.cache_is_copy(path_info): cache.unprotect(path_info) else: @@ -132,13 +80,13 @@ def _checkout_file( cache.link(cache_info, path_info) else: cache.link(cache_info, path_info) - added, modified = True, False + modified = True - tree.state.save(path_info, hash_info) + tree.state.save(path_info, obj.hash_info) if progress_callback: progress_callback(str(path_info)) - return added, modified + return modified def _remove_redundant_files(path_info, tree, dir_info, cache, force): @@ -153,77 +101,59 @@ def _remove_redundant_files(path_info, tree, dir_info, cache, force): def _checkout_dir( - path_info, - tree, - hash_info, - cache, - force, - progress_callback=None, - relink=False, - filter_info=None, + path_info, tree, obj, cache, force, progress_callback=None, relink=False, ): - added, modified = False, False + modified = False, False # Create dir separately so that dir is created # even if there are no files in it if not tree.exists(path_info): - added = True + modified = True tree.makedirs(path_info) - dir_info = cache.get_dir_cache(hash_info) - logger.debug("Linking directory '%s'.", path_info) - for entry_info, entry_hash_info in dir_info.items(path_info): - if filter_info and not entry_info.isin_or_eq(filter_info): - continue - - entry_added, entry_modified = _checkout_file( + for entry_info, entry_hash_info in obj.hash_info.dir_info.items(path_info): + entry_modified = _checkout_file( entry_info, tree, - entry_hash_info, + cache.get(entry_hash_info), cache, force, progress_callback, relink, ) - if entry_added or entry_modified: + if entry_modified: modified = True modified = ( - _remove_redundant_files(path_info, tree, dir_info, cache, force) + _remove_redundant_files( + path_info, tree, obj.hash_info.dir_info, cache, force + ) or modified ) - tree.state.save(path_info, hash_info) + tree.state.save(path_info, obj.hash_info) # relink is not modified, assume it as nochange - return added, not added and modified and not relink + return modified and not relink def _checkout( path_info, tree, - hash_info, + obj, cache, force=False, progress_callback=None, relink=False, - filter_info=None, ): - if not hash_info.isdir: + if not obj.hash_info.isdir: ret = _checkout_file( - path_info, tree, hash_info, cache, force, progress_callback, relink + path_info, tree, obj, cache, force, progress_callback, relink ) else: ret = _checkout_dir( - path_info, - tree, - hash_info, - cache, - force, - progress_callback, - relink, - filter_info, + path_info, tree, obj, cache, force, progress_callback, relink, ) tree.state.save_link(path_info) @@ -234,12 +164,11 @@ def _checkout( def checkout( path_info, tree, - hash_info, + obj, cache, force=False, progress_callback=None, relink=False, - filter_info=None, quiet=False, ): if path_info.scheme not in ["local", cache.tree.scheme]: @@ -247,7 +176,7 @@ def checkout( failed = None skip = False - if not hash_info: + if not obj: if not quiet: logger.warning( "No file hash info found for '%s'. It won't be created.", @@ -256,43 +185,33 @@ def checkout( _remove(path_info, tree, cache, force=force) failed = path_info - elif not relink and not _changed( - path_info, tree, hash_info, cache, filter_info=filter_info - ): + elif not relink and not _changed(path_info, tree, obj, cache): logger.trace("Data '%s' didn't change.", path_info) skip = True - - elif cache.changed_cache( - hash_info, path_info=path_info, filter_info=filter_info - ): - if not quiet: - logger.warning( - "Cache '%s' not found. File '%s' won't be created.", - hash_info, - path_info, - ) - _remove(path_info, tree, cache, force=force) - failed = path_info + else: + try: + check(cache, obj) + except (FileNotFoundError, ObjectFormatError): + if not quiet: + logger.warning( + "Cache '%s' not found. File '%s' won't be created.", + obj.hash_info, + path_info, + ) + _remove(path_info, tree, cache, force=force) + failed = path_info if failed or skip: - if progress_callback: + if progress_callback and obj: progress_callback( - str(path_info), - cache.get_files_number(path_info, hash_info, filter_info), + str(path_info), len(obj), ) if failed: raise CheckoutError([failed]) return - logger.debug("Checking out '%s' with cache '%s'.", path_info, hash_info) + logger.debug("Checking out '%s' with cache '%s'.", path_info, obj) return _checkout( - path_info, - tree, - hash_info, - cache, - force, - progress_callback, - relink, - filter_info, + path_info, tree, obj, cache, force, progress_callback, relink, ) diff --git a/dvc/dependency/repo.py b/dvc/dependency/repo.py index dfaf3829d..7f8652c09 100644 --- a/dvc/dependency/repo.py +++ b/dvc/dependency/repo.py @@ -71,6 +71,7 @@ class RepoDependency(LocalDependency): def download(self, to, jobs=None): from dvc.checkout import checkout + from dvc.objects import stage cache = self.repo.cache.local @@ -78,15 +79,16 @@ class RepoDependency(LocalDependency): if self.def_repo.get(self.PARAM_REV_LOCK) is None: self.def_repo[self.PARAM_REV_LOCK] = repo.get_rev() path_info = PathInfo(repo.root_dir) / self.def_path - hash_info = cache.save( + obj = stage( + cache, path_info, repo.repo_tree, - None, jobs=jobs, follow_subrepos=False, ) + cache.save(obj, jobs=jobs) - checkout(to.path_info, to.tree, hash_info, cache) + checkout(to.path_info, to.tree, obj, cache) def update(self, rev=None): if rev: diff --git a/dvc/objects.py b/dvc/objects.py new file mode 100644 index 000000000..ff0838f60 --- /dev/null +++ b/dvc/objects.py @@ -0,0 +1,254 @@ +import json +import logging + +from .dir_info import DirInfo +from .exceptions import DvcException + +logger = logging.getLogger(__name__) + + +class ObjectError(DvcException): + pass + + +class ObjectFormatError(ObjectError): + pass + + +def _get_hash(path_info, tree, name, **kwargs): + if tree.PARAM_CHECKSUM == name: + return tree.get_hash(path_info, **kwargs) + + if name == "md5": + from dvc.hash_info import HashInfo + from dvc.utils import file_md5 + + return HashInfo("md5", file_md5(path_info, tree, **kwargs)) + + raise AssertionError(f"unsupported hash '{name}'") + + +class HashFile: + def __init__(self, path_info, tree, hash_info): + self.path_info = path_info + self.tree = tree + self.hash_info = hash_info + + def __len__(self): + return 1 + + def __str__(self): + return f"object {self.hash_info}" + + def __bool__(self): + return bool(self.hash_info) + + def check(self, odb): + actual = _get_hash(self.path_info, self.tree, odb.tree.PARAM_CHECKSUM) + + logger.trace( + "cache '%s' expected '%s' actual '%s'", + self.path_info, + self.hash_info, + actual, + ) + + assert actual.name == self.hash_info.name + if actual.value.split(".")[0] != self.hash_info.value.split(".")[0]: + raise ObjectFormatError(f"{self} is corrupted") + + @classmethod + def load(cls, odb, hash_info): + return odb.get(hash_info) + + def save(self, odb, **kwargs): + odb.add(self, **kwargs) + + +class File(HashFile): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.src = None + + @classmethod + def stage(cls, odb, path_info, tree, **kwargs): + hash_info = _get_hash( + path_info, tree, odb.tree.PARAM_CHECKSUM, **kwargs + ) + raw = odb.get(hash_info) + obj = cls(raw.path_info, raw.tree, hash_info) + obj.src = HashFile(path_info, tree, hash_info) + return obj + + def save(self, odb, **kwargs): + self.src.save(odb, **kwargs) + + +class Tree(HashFile): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.src_hash_info = None + self.src_path_info = None + self.src_tree = None + + def __len__(self): + return self.hash_info.dir_info.nfiles + + def __iter__(self): + for ( + key, + hash_info, + ) in self.hash_info.dir_info.trie.iteritems(): # noqa: B301 + yield (key, hash_info) + + @classmethod + def save_dir_info(cls, odb, dir_info, hash_info=None): + if hash_info and hash_info.name == odb.tree.PARAM_CHECKSUM: + try: + odb.check(hash_info) + assert hash_info.dir_info == dir_info + return hash_info + except (FileNotFoundError, ObjectFormatError): + pass + + from dvc.path_info import PathInfo + from dvc.tree.memory import MemoryTree + from dvc.utils import tmp_fname + + tree = MemoryTree(None, {}) + path_info = PathInfo(tmp_fname("")) + with tree.open(path_info, "w") as fobj: + json.dump(dir_info.to_list(), fobj, sort_keys=True) + + tmp_info = odb.tree.path_info / tmp_fname("") + with tree.open(path_info, "rb") as fobj: + odb.tree.upload_fobj(fobj, tmp_info) + + hash_info = _get_hash(tmp_info, odb.tree, odb.tree.PARAM_CHECKSUM) + hash_info.value += odb.tree.CHECKSUM_DIR_SUFFIX + hash_info.dir_info = dir_info + hash_info.nfiles = dir_info.nfiles + + obj = HashFile(tmp_info, odb.tree, hash_info) + odb.add(obj) + + return hash_info + + @classmethod + def stage(cls, odb, path_info, tree, **kwargs): + hash_info = _get_hash( + path_info, tree, odb.tree.PARAM_CHECKSUM, **kwargs + ) + hi = cls.save_dir_info(odb, hash_info.dir_info, hash_info) + hi.size = hash_info.size + raw = odb.get(hi) + obj = cls(raw.path_info, raw.tree, hi) + obj.src_hash_info = hash_info + obj.src_path_info = path_info + obj.src_tree = tree + return obj + + @classmethod + def load(cls, odb, hash_info): + + obj = odb.get(hash_info) + + try: + with obj.tree.open(obj.path_info, "r") as fobj: + raw = json.load(fobj) + except ValueError as exc: + raise ObjectFormatError(f"{obj} is corrupted") from exc + + if not isinstance(raw, list): + logger.error( + "dir cache file format error '%s' [skipping the file]", + obj.path_info, + ) + raise ObjectFormatError(f"{obj} is corrupted") + + dir_info = DirInfo.from_list(raw) + hash_info.dir_info = dir_info + hash_info.nfiles = dir_info.nfiles + + return cls(obj.path_info, obj.tree, hash_info) + + def save(self, odb, **kwargs): + from dvc.progress import Tqdm + + assert self.src_hash_info.dir_info + hi = self.save_dir_info( + odb, self.src_hash_info.dir_info, self.hash_info + ) + for entry_info, entry_hash in Tqdm( + hi.dir_info.items(self.src_path_info), + desc="Saving " + self.src_path_info.name, + unit="file", + ): + entry_obj = HashFile(entry_info, self.src_tree, entry_hash) + entry_obj.save(odb, **kwargs) + cache_info = odb.tree.hash_to_path_info(hi.value) + odb.tree.state.save(cache_info, hi) + self.src_tree.state.save(self.src_path_info, hi) + + def filter(self, odb, prefix): + hash_info = self.hash_info.dir_info.trie.get(prefix) + if hash_info: + return load(odb, hash_info) + + depth = len(prefix) + dir_info = DirInfo() + try: + for key, value in self.hash_info.dir_info.trie.items(prefix): + dir_info.trie[key[depth:]] = value + except KeyError: + return None + + return load(odb, self.save_dir_info(odb, dir_info)) + + +def stage(odb, path_info, tree, **kwargs): + if tree.isdir(path_info): + return Tree.stage(odb, path_info, tree, **kwargs) + return File.stage(odb, path_info, tree, **kwargs) + + +def check(odb, obj): + odb.check(obj.hash_info) + + if isinstance(obj, Tree): + for _, hash_info in obj: + odb.check(hash_info) + + +def load(odb, hash_info): + if hash_info.isdir: + return Tree.load(odb, hash_info) + return File.load(odb, hash_info) + + +def _get_dir_size(odb, dir_info): + try: + return sum( + odb.tree.getsize(odb.tree.hash_to_path_info(hi.value)) + for _, hi in dir_info.items() + ) + except FileNotFoundError: + return None + + +def merge(odb, ancestor_info, our_info, their_info): + assert our_info + assert their_info + + if ancestor_info: + ancestor = load(odb, ancestor_info).hash_info.dir_info + else: + ancestor = DirInfo() + + our = load(odb, our_info).hash_info.dir_info + their = load(odb, their_info).hash_info.dir_info + + merged = our.merge(ancestor, their) + hash_info = Tree.save_dir_info(odb, merged) + hash_info.size = _get_dir_size(odb, merged) + return hash_info diff --git a/dvc/output/base.py b/dvc/output/base.py index 06a1c0bb7..a812036bd 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -6,6 +6,7 @@ from urllib.parse import urlparse from voluptuous import Any +import dvc.objects as objects import dvc.prompt as prompt from dvc.cache import NamedCache from dvc.checkout import checkout @@ -144,6 +145,7 @@ class BaseOutput: if self.use_cache and self.cache is None: raise RemoteCacheRequiredError(self.path_info) + self.obj = None self.isexec = False if self.IS_DEPENDENCY else isexec def _parse_path(self, tree, path): @@ -181,7 +183,7 @@ class BaseOutput: @property def dir_cache(self): - return self.cache.get_dir_cache(self.hash_info) + return self.hash_info.dir_info @classmethod def supported(cls, url): @@ -194,7 +196,7 @@ class BaseOutput: def get_hash(self): if not self.use_cache: return self.tree.get_hash(self.path_info) - return self.cache.get_hash(self.tree, self.path_info) + return objects.stage(self.cache, self.path_info, self.tree).hash_info @property def is_dir_checksum(self): @@ -211,9 +213,15 @@ class BaseOutput: if not self.use_cache or not self.hash_info: return True - return self.cache.changed_cache( - self.hash_info, filter_info=filter_info - ) + obj = self.get_obj(filter_info=filter_info) + if not obj: + return True + + try: + objects.check(self.cache, obj) + return False + except (FileNotFoundError, objects.ObjectFormatError): + return True def workspace_status(self): if not self.exists: @@ -296,7 +304,8 @@ class BaseOutput: logger.debug("Output '%s' didn't change. Skipping saving.", self) return - self.hash_info = self.get_hash() + self.obj = objects.stage(self.cache, self.path_info, self.tree) + self.hash_info = self.obj.hash_info self.isexec = self.isfile() and self.tree.isexec(self.path_info) def set_exec(self): @@ -310,20 +319,16 @@ class BaseOutput: assert self.hash_info if self.use_cache: - self.cache.save( - self.path_info, - self.tree, - self.hash_info, - filter_info=filter_info, + obj = objects.stage( + self.cache, filter_info or self.path_info, self.tree ) - + self.cache.save(obj) checkout( - self.path_info, + filter_info or self.path_info, self.tree, - self.hash_info, + obj, self.cache, relink=True, - filter_info=filter_info, ) self.set_exec() @@ -373,6 +378,23 @@ class BaseOutput: def download(self, to, jobs=None): self.tree.download(self.path_info, to.path_info, jobs=jobs) + def get_obj(self, filter_info=None): + if self.obj: + obj = self.obj + elif self.hash_info: + try: + obj = objects.load(self.cache, self.hash_info) + except FileNotFoundError: + return None + else: + return None + + if filter_info and filter_info != self.path_info: + prefix = filter_info.relative_to(self.path_info).parts + obj = obj.filter(self.cache, prefix) + + return obj + def checkout( self, force=False, @@ -389,16 +411,22 @@ class BaseOutput: ) return None + obj = self.get_obj(filter_info=filter_info) + if not obj and (filter_info and filter_info != self.path_info): + # backward compatibility + return None + + added = not self.exists + try: - res = checkout( - self.path_info, + modified = checkout( + filter_info or self.path_info, self.tree, - self.hash_info, + obj, self.cache, force=force, progress_callback=progress_callback, relink=relink, - filter_info=filter_info, **kwargs, ) except CheckoutError: @@ -406,7 +434,7 @@ class BaseOutput: return None raise self.set_exec() - return res + return added, False if added else modified def remove(self, ignore_remove=False): self.tree.remove(self.path_info) @@ -431,11 +459,23 @@ class BaseOutput: self.repo.scm.ignore(self.fspath) def get_files_number(self, filter_info=None): - if not self.use_cache: + from funcy import ilen + + if not self.use_cache or not self.hash_info: return 0 - return self.cache.get_files_number( - self.path_info, self.hash_info, filter_info + if not self.hash_info.isdir: + return 1 + + if not filter_info or filter_info == self.path_info: + return self.hash_info.nfiles + + if not self.hash_info.dir_info: + return 0 + + return ilen( + filter_info.isin_or_eq(self.path_info / relpath) + for relpath, _ in self.hash_info.dir_info.items() ) def unprotect(self): @@ -443,17 +483,25 @@ class BaseOutput: self.cache.unprotect(self.path_info) def get_dir_cache(self, **kwargs): + if not self.is_dir_checksum: raise DvcException("cannot get dir cache for file checksum") - if self.cache.changed_cache_file(self.hash_info): + + try: + objects.check(self.cache, self.cache.get(self.hash_info)) + except (FileNotFoundError, objects.ObjectFormatError): self.repo.cloud.pull( NamedCache.make("local", self.hash_info.value, str(self)), show_checksums=False, **kwargs, ) + + try: + objects.load(self.cache, self.hash_info) + assert self.hash_info.dir_info + except (objects.ObjectFormatError, FileNotFoundError): self.hash_info.dir_info = None - if not self.hash_info.dir_info: - self.cache.set_dir_info(self.hash_info) + return self.dir_cache def collect_used_dir_cache( @@ -485,7 +533,9 @@ class BaseOutput: except DvcException: logger.debug(f"failed to pull cache for '{self}'") - if self.cache.changed_cache_file(self.hash_info): + try: + objects.check(self.cache, self.cache.get(self.hash_info)) + except (FileNotFoundError, objects.ObjectFormatError): msg = ( "Missing cache for directory '{}'. " "Cache for files inside will be lost. " @@ -496,8 +546,7 @@ class BaseOutput: "unable to fully collect used cache" " without cache for directory '{}'".format(self) ) - else: - return cache + return cache path = str(self.path_info) filter_path = str(filter_info) if filter_info else None @@ -615,6 +664,6 @@ class BaseOutput: self._check_can_merge(self) self._check_can_merge(other) - self.hash_info = self.cache.merge( - ancestor_info, self.hash_info, other.hash_info + self.hash_info = objects.merge( + self.cache, ancestor_info, self.hash_info, other.hash_info ) diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py index 345a182ec..7b015424f 100644 --- a/dvc/repo/fetch.py +++ b/dvc/repo/fetch.py @@ -81,6 +81,7 @@ def fetch( def _fetch_external(self, repo_url, repo_rev, files, jobs): from dvc.external_repo import external_repo + from dvc.objects import stage from dvc.path_info import PathInfo from dvc.scm.base import CloneError @@ -109,14 +110,16 @@ def _fetch_external(self, repo_url, repo_rev, files, jobs): cb(repo.cloud.pull(used, jobs)) except (NoOutputOrStageError, NoRemoteError): pass - self.cache.local.save( + obj = stage( + cache, path_info, repo.repo_tree, - None, jobs=jobs, - download_callback=cb, follow_subrepos=False, ) + cache.save( + obj, jobs=jobs, download_callback=cb, + ) except CloneError: failed += 1 logger.exception( diff --git a/dvc/tree/base.py b/dvc/tree/base.py index cebddfcee..ec1e79637 100644 --- a/dvc/tree/base.py +++ b/dvc/tree/base.py @@ -1,4 +1,6 @@ +import errno import logging +import os from concurrent.futures import ThreadPoolExecutor, as_completed from functools import partial from multiprocessing import cpu_count @@ -252,7 +254,9 @@ class BaseTree: ) if not self.exists(path_info): - return None + raise FileNotFoundError( + errno.ENOENT, os.strerror(errno.ENOENT), path_info + ) # pylint: disable=assignment-from-none hash_info = self.state.get(path_info) @@ -272,7 +276,10 @@ class BaseTree: if hash_info: assert hash_info.name == self.PARAM_CHECKSUM if hash_info.isdir: - self.cache.set_dir_info(hash_info) + from dvc.objects import Tree + + # NOTE: loading the tree will restore hash_info.dir_info + Tree.load(self.cache, hash_info) return hash_info if self.isdir(path_info): @@ -347,9 +354,12 @@ class BaseTree: @use_state def get_dir_hash(self, path_info, **kwargs): + from dvc.objects import Tree + dir_info = self._collect_dir(path_info, **kwargs) - hash_info = self.repo.cache.local.save_dir_info(dir_info) + hash_info = Tree.save_dir_info(self.repo.cache.local, dir_info) hash_info.size = dir_info.size + hash_info.dir_info = dir_info return hash_info def upload( diff --git a/dvc/tree/dvc.py b/dvc/tree/dvc.py index 2bade0304..c718894eb 100644 --- a/dvc/tree/dvc.py +++ b/dvc/tree/dvc.py @@ -132,7 +132,12 @@ class DvcTree(BaseTree): # pylint:disable=abstract-method out.get_dir_cache(**kwargs) dir_cache = out.dir_cache - hash_info = out.cache.save_dir_info(dir_cache) + if not dir_cache: + raise FileNotFoundError + + from dvc.objects import Tree + + hash_info = Tree.save_dir_info(out.cache, dir_cache) if hash_info != out.hash_info: raise FileNotFoundError diff --git a/dvc/tree/memory.py b/dvc/tree/memory.py new file mode 100644 index 000000000..11097740f --- /dev/null +++ b/dvc/tree/memory.py @@ -0,0 +1,37 @@ +from dvc.hash_info import HashInfo +from dvc.utils import file_md5 + +from .base import BaseTree + + +class MemoryTree(BaseTree): + scheme = "local" + PARAM_CHECKSUM = "md5" + + def __init__(self, repo, config): + from fsspec.implementations.memory import MemoryFileSystem + + super().__init__(repo, config) + + self.fs = MemoryFileSystem() + + def exists(self, path_info, use_dvcignore=True): + return self.fs.exists(path_info.path) + + def open(self, path_info, mode="r", encoding=None, **kwargs): + return self.fs.open( + path_info.fspath, mode=mode, encoding=encoding, **kwargs + ) + + def stat(self, path_info): + import os + + info = self.fs.info(path_info.fspath) + + return os.stat_result((0, 0, 0, 0, 0, 0, info["size"], 0, 0, 0)) + + def get_file_hash(self, path_info): + return HashInfo(self.PARAM_CHECKSUM, file_md5(path_info, self)) + + def walk_files(self, path_info, **kwargs): + raise NotImplementedError diff --git a/setup.py b/setup.py index b88f1fca8..951c867f2 100644 --- a/setup.py +++ b/setup.py @@ -87,6 +87,7 @@ install_requires = [ "python-benedict>=0.21.1", "pyparsing==2.4.7", "typing_extensions>=3.7.4", + "fsspec>=0.8.5", ] diff --git a/tests/func/test_add.py b/tests/func/test_add.py index 207ac6276..4690877d8 100644 --- a/tests/func/test_add.py +++ b/tests/func/test_add.py @@ -96,6 +96,8 @@ def test_add_unsupported_file(dvc): def test_add_directory(tmp_dir, dvc): + from dvc.objects import load + (stage,) = tmp_dir.dvc_gen({"dir": {"file": "file"}}) assert stage is not None @@ -104,7 +106,7 @@ def test_add_directory(tmp_dir, dvc): hash_info = stage.outs[0].hash_info - dir_info = dvc.cache.local.load_dir_cache(hash_info) + dir_info = load(dvc.cache.local, hash_info).hash_info.dir_info for path, _ in dir_info.trie.items(): assert "\\" not in path diff --git a/tests/func/test_cache.py b/tests/func/test_cache.py index 18822250c..2794fc09f 100644 --- a/tests/func/test_cache.py +++ b/tests/func/test_cache.py @@ -6,10 +6,9 @@ import configobj import pytest from dvc.cache import Cache -from dvc.cache.base import DirCacheError -from dvc.dir_info import DirInfo from dvc.hash_info import HashInfo from dvc.main import main +from dvc.objects import ObjectFormatError from dvc.utils import relpath from tests.basic_env import TestDir, TestDvc @@ -49,24 +48,23 @@ class TestCacheLoadBadDirCache(TestDvc): self.assertEqual(len(ret), 0) def test(self): + from dvc.objects import load + dir_hash = "123.dir" fname = os.fspath( self.dvc.cache.local.tree.hash_to_path_info(dir_hash) ) self.create(fname, "not,json") - with pytest.raises(DirCacheError): - self.dvc.cache.local.load_dir_cache(HashInfo("md5", dir_hash)) + with pytest.raises(ObjectFormatError): + load(self.dvc.cache.local, HashInfo("md5", dir_hash)) dir_hash = "234.dir" fname = os.fspath( self.dvc.cache.local.tree.hash_to_path_info(dir_hash) ) self.create(fname, '{"a": "b"}') - dir_info = self.dvc.cache.local.load_dir_cache( - HashInfo("md5", dir_hash) - ) - self.assertTrue(isinstance(dir_info, DirInfo)) - self.assertEqual(dir_info.nfiles, 0) + with pytest.raises(ObjectFormatError): + load(self.dvc.cache.local, HashInfo("md5", dir_hash)) class TestExternalCacheDir(TestDvc): diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py index 2bdfdfc4d..e3e000334 100644 --- a/tests/func/test_checkout.py +++ b/tests/func/test_checkout.py @@ -85,6 +85,8 @@ class TestCheckoutCorruptedCacheFile(TestRepro): class TestCheckoutCorruptedCacheDir(TestDvc): def test(self): + from dvc.objects import load + # NOTE: using 'copy' so that cache and link don't have same inode ret = main(["config", "cache.type", "copy"]) self.assertEqual(ret, 0) @@ -98,7 +100,9 @@ class TestCheckoutCorruptedCacheDir(TestDvc): # NOTE: modifying cache file for one of the files inside the directory # to check if dvc will detect that the cache is corrupted. _, entry_hash = next( - self.dvc.cache.local.load_dir_cache(out.hash_info).items() + load( + self.dvc.cache.local, out.hash_info + ).hash_info.dir_info.items() ) cache = os.fspath( self.dvc.cache.local.tree.hash_to_path_info(entry_hash.value) diff --git a/tests/func/test_commit.py b/tests/func/test_commit.py index bcd724f96..fb4d4c687 100644 --- a/tests/func/test_commit.py +++ b/tests/func/test_commit.py @@ -148,6 +148,7 @@ def test_commit_granular_dir(tmp_dir, dvc): dvc.commit(os.path.join("data", "subdir")) assert set(cache.glob("*/*")) == { cache / "1a" / "ca2c799df82929bbdd976557975546.dir", + cache / "26" / "d6b64d96a660707412f523e8184b5f.dir", cache / "ac" / "bd18db4cc2f85cedef654fccc4a4d8", cache / "4c" / "e8d2a2cf314a52fa7f315ca37ca445", cache / "68" / "dde2c3c4e7953c2290f176bbdc9a54", diff --git a/tests/func/test_external_repo.py b/tests/func/test_external_repo.py index 7d9ba0804..0471b38ab 100644 --- a/tests/func/test_external_repo.py +++ b/tests/func/test_external_repo.py @@ -3,7 +3,7 @@ import os from mock import ANY, patch from dvc.external_repo import CLONES, external_repo -from dvc.hash_info import HashInfo +from dvc.objects import stage from dvc.path_info import PathInfo from dvc.scm.git import Git from dvc.tree.local import LocalTree @@ -200,20 +200,13 @@ def test_subrepos_are_ignored(tmp_dir, erepo_dir): remove(cache_dir) makedirs(cache_dir) - expected_hash = HashInfo("md5", "e1d9e8eae5374860ae025ec84cfd85c7.dir") - assert ( - repo.repo_tree.get_hash( - os.path.join(repo.root_dir, "dir"), follow_subrepos=False - ) - == expected_hash - ) - - repo.cache.local.save( + obj = stage( + repo.cache.local, PathInfo(repo.root_dir) / "dir", repo.repo_tree, - expected_hash, - link=False, + follow_subrepos=False, ) + repo.cache.local.save(obj) assert set(cache_dir.glob("*/*")) == { cache_dir / "e1" / "d9e8eae5374860ae025ec84cfd85c7.dir", cache_dir / "37" / "b51d194a7513e45b56f6524f2d51f2", diff --git a/tests/unit/output/test_local.py b/tests/unit/output/test_local.py index 3b1c4d1a4..f04c3b973 100644 --- a/tests/unit/output/test_local.py +++ b/tests/unit/output/test_local.py @@ -2,7 +2,6 @@ import os from mock import patch -from dvc.cache.local import LocalCache from dvc.dir_info import DirInfo from dvc.hash_info import HashInfo from dvc.output import LocalOutput @@ -79,19 +78,19 @@ class TestGetFilesNumber(TestDvc): o.use_cache = False self.assertEqual(0, o.get_files_number()) - @patch.object( - LocalCache, - "get_dir_cache", - return_value=DirInfo.from_list( - [ - {"relpath": "foo", "md5": "asdf"}, - {"relpath": "bar", "md5": "qwe"}, - ] - ), - ) - def test_return_multiple_for_dir(self, _mock_get_dir_cache): + def test_return_multiple_for_dir(self): o = self._get_output() - o.hash_info = HashInfo("md5", "12345678.dir") + o.hash_info = HashInfo( + "md5", + "12345678.dir", + nfiles=2, + dir_info=DirInfo.from_list( + [ + {"relpath": "foo", "md5": "asdf"}, + {"relpath": "bar", "md5": "qwe"}, + ] + ), + ) self.assertEqual(2, o.get_files_number()) @patch.object(LocalOutput, "is_dir_checksum", False) diff --git a/tests/unit/test_external_repo.py b/tests/unit/test_external_repo.py index 601946d4c..4bf6f0dd8 100644 --- a/tests/unit/test_external_repo.py +++ b/tests/unit/test_external_repo.py @@ -4,6 +4,7 @@ from unittest.mock import call import pytest from dvc.external_repo import external_repo +from dvc.objects import stage from dvc.path_info import PathInfo from tests.unit.tree.test_repo import make_subrepo @@ -109,12 +110,16 @@ def test_fetch_external_repo_jobs(tmp_dir, scm, mocker, dvc, local_remote): with external_repo(str(tmp_dir)) as repo: spy = mocker.spy(repo.cloud, "pull") - dvc.cache.local.save( + obj = stage( + dvc.cache.local, PathInfo(repo.root_dir) / "dir1", repo.repo_tree, - None, follow_subrepos=False, jobs=3, ) + dvc.cache.local.save( + obj, jobs=3, + ) + run_jobs = tuple(spy.call_args_list[0])[1].get("jobs") assert run_jobs == 3 diff --git a/tests/unit/tree/test_dvc.py b/tests/unit/tree/test_dvc.py index fc9d8ead4..8fae9835d 100644 --- a/tests/unit/tree/test_dvc.py +++ b/tests/unit/tree/test_dvc.py @@ -23,7 +23,7 @@ def test_open(tmp_dir, dvc): (tmp_dir / "foo").unlink() tree = DvcTree(dvc) - with tree.open("foo", "r") as fobj: + with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj: assert fobj.read() == "foo" @@ -32,7 +32,7 @@ def test_open_dirty_hash(tmp_dir, dvc): (tmp_dir / "file").write_text("something") tree = DvcTree(dvc) - with tree.open("file", "r") as fobj: + with tree.open(PathInfo(tmp_dir) / "file", "r") as fobj: # NOTE: Unlike RepoTree, DvcTree should not # be affected by a dirty workspace. assert fobj.read() == "file" @@ -46,7 +46,7 @@ def test_open_dirty_no_hash(tmp_dir, dvc): # NOTE: Unlike RepoTree, DvcTree should not # be affected by a dirty workspace. with pytest.raises(FileNotFoundError): - with tree.open("file", "r"): + with tree.open(PathInfo(tmp_dir) / "file", "r"): pass @@ -66,7 +66,7 @@ def test_open_in_history(tmp_dir, scm, dvc): continue tree = DvcTree(dvc) - with tree.open("foo", "r") as fobj: + with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj: assert fobj.read() == "foo" diff --git a/tests/unit/tree/test_repo.py b/tests/unit/tree/test_repo.py index ced69d6a2..cbaf3cf75 100644 --- a/tests/unit/tree/test_repo.py +++ b/tests/unit/tree/test_repo.py @@ -24,7 +24,7 @@ def test_open(tmp_dir, dvc): (tmp_dir / "foo").unlink() tree = RepoTree(dvc) - with tree.open("foo", "r") as fobj: + with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj: assert fobj.read() == "foo" @@ -33,7 +33,7 @@ def test_open_dirty_hash(tmp_dir, dvc): (tmp_dir / "file").write_text("something") tree = RepoTree(dvc) - with tree.open("file", "r") as fobj: + with tree.open(PathInfo(tmp_dir) / "file", "r") as fobj: assert fobj.read() == "something" @@ -42,7 +42,7 @@ def test_open_dirty_no_hash(tmp_dir, dvc): (tmp_dir / "file.dvc").write_text("outs:\n- path: file\n") tree = RepoTree(dvc) - with tree.open("file", "r") as fobj: + with tree.open(PathInfo(tmp_dir) / "file", "r") as fobj: assert fobj.read() == "file" @@ -62,7 +62,7 @@ def test_open_in_history(tmp_dir, scm, dvc): continue tree = RepoTree(dvc) - with tree.open("foo", "r") as fobj: + with tree.open(PathInfo(tmp_dir) / "foo", "r") as fobj: assert fobj.read() == "foo"