Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 07:18:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

repofs: migrate to fsspec

Browse Source
This commit is contained in:
Ruslan Kuprieiev
2022-04-18 14:17:20 +03:00
parent 53c7f44480
commit 517f45d45f
20 changed files with 300 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/api.py
+5 -8
View File
@@ -22,17 +22,14 @@ def get_url(path, repo=None, rev=None, remote=None):
with reraise(FileNotFoundError, PathMissingError(path, repo)):
info = _repo.repo_fs.info(path)
if not info["isdvc"]:
dvc_info = info.get("dvc_info")
if not dvc_info:
raise OutputNotFoundError(path, repo)
cloud = info["repo"].cloud
_, _, dvc_fs, dvc_path = _repo.repo_fs._get_fs_pair(path)
dvc_repo = info["repo"]
md5 = dvc_info["md5"]
if not os.path.isabs(path):
dvc_path = dvc_path.replace("\\", "/")
md5 = info["repo"].dvcfs.info(dvc_path)["md5"]
return cloud.get_url_for(remote, checksum=md5)
return dvc_repo.cloud.get_url_for(remote, checksum=md5)
def open( # noqa, pylint: disable=redefined-builtin
dvc/data/reference.py
+2 -5
View File
@@ -30,7 +30,6 @@ class ReferenceHashFile(HashFile):
checksum: Optional[str] = None,
**kwargs,
):
from dvc.fs.repo import RepoFileSystem
super().__init__(fs_path, fs, hash_info, **kwargs)
self.checksum = checksum or fs.checksum(fs_path)
@@ -53,8 +52,6 @@ class ReferenceHashFile(HashFile):
return self.fs.checksum(self.fs_path)
def to_bytes(self):
from dvc.fs.repo import RepoFileSystem
# NOTE: dumping reference FS's this way is insecure, as the
# fully parsed remote FS config will include credentials
#
@@ -75,7 +72,7 @@ class ReferenceHashFile(HashFile):
@classmethod
def from_bytes(cls, data: bytes, fs_cache: Optional[dict] = None):
from dvc.fs import get_fs_cls
from dvc.fs.repo import RepoFileSystem
from dvc.fs.repo import RepoFileSystem, _RepoFileSystem
try:
dict_ = pickle.loads(data)
@@ -92,7 +89,7 @@ class ReferenceHashFile(HashFile):
fs = fs_cache.get((scheme, config_pairs)) if fs_cache else None
if not fs:
config = dict(config_pairs)
if RepoFileSystem.PARAM_REPO_URL in config:
if _RepoFileSystem.PARAM_REPO_URL in config:
fs_cls = RepoFileSystem
else:
fs_cls = get_fs_cls(config, scheme=scheme)
dvc/dependency/repo.py
+6 -9
View File
@@ -101,6 +101,7 @@ class RepoDependency(Dependency):
from dvc.data.stage import stage
from dvc.data.tree import Tree, TreeError
from dvc.exceptions import NoOutputOrStageError, PathMissingError
from dvc.utils import as_posix
local_odb = self.repo.odb.local
locked = kwargs.pop("locked", True)
@@ -115,11 +116,7 @@ class RepoDependency(Dependency):
if not obj_only:
try:
for odb, obj_ids in repo.used_objs(
[
os.path.abspath(
os.path.join(repo.root_dir, self.def_path)
)
],
[os.path.join(repo.root_dir, self.def_path)],
force=True,
jobs=kwargs.get("jobs"),
recursive=True,
@@ -135,7 +132,7 @@ class RepoDependency(Dependency):
try:
staging, _, staged_obj = stage(
local_odb,
self.def_path,
as_posix(self.def_path),
repo.repo_fs,
local_odb.fs.PARAM_CHECKSUM,
)
@@ -175,17 +172,17 @@ class RepoDependency(Dependency):
continue
if (
obj.fs.repo_url in checked_urls
or obj.fs._root_dir in checked_urls
or obj.fs.repo.root_dir in checked_urls
):
continue
self_url = self.repo.url or self.repo.root_dir
if (
obj.fs.repo_url is not None
and obj.fs.repo_url == self_url
or obj.fs._root_dir == self.repo.root_dir
or obj.fs.repo.root_dir == self.repo.root_dir
):
raise CircularImportError(self, obj.fs.repo_url, self_url)
checked_urls.update([obj.fs.repo_url, obj.fs._root_dir])
checked_urls.update([obj.fs.repo_url, obj.fs.repo.root_dir])
def get_obj(self, filter_info=None, **kwargs):
locked = kwargs.get("locked", True)
dvc/fs/git.py
+2 -2
View File
@@ -51,5 +51,5 @@ class GitFileSystem(FSSpecWrapper): # pylint:disable=abstract-method
def rev(self) -> str:
return self.fs.rev
def ls(self, path, **kwargs):
return self.fs.ls(path, **kwargs) or []
def ls(self, path, detail=True, **kwargs):
return self.fs.ls(path, detail=detail, **kwargs) or []
dvc/fs/path.py
-2
View File
@@ -81,8 +81,6 @@ class Path:
return self.isin_or_eq(left, right) or self.isin(right, left)
def relpath(self, path, start):
if not start:
return path
return self.flavour.relpath(path, start=start)
def relparts(self, path, base):
dvc/fs/repo.py
+89 -110
View File
@@ -5,11 +5,14 @@ from contextlib import suppress
from itertools import takewhile
from typing import TYPE_CHECKING, Callable, Optional, Tuple, Type, Union
from funcy import wrap_with
from fsspec.spec import AbstractFileSystem
from funcy import cached_property, wrap_prop, wrap_with
from ._callback import DEFAULT_CALLBACK
from .base import FileSystem
from .dvc import DvcFileSystem
from .fsspec_wrapper import FSSpecWrapper
from .path import Path
if TYPE_CHECKING:
from dvc.repo import Repo
@@ -41,27 +44,27 @@ def _ls(fs, path):
def _merge_info(repo, fs_info, dvc_info):
from dvc.utils import is_exec
if not fs_info:
dvc_info["repo"] = repo
dvc_info["isdvc"] = True
return dvc_info
ret = {"repo": repo}
fs_info["repo"] = repo
fs_info["isout"] = dvc_info.get("isout", False) if dvc_info else False
fs_info["outs"] = dvc_info["outs"] if dvc_info else None
fs_info["isdvc"] = dvc_info["isdvc"] if dvc_info else False
fs_info["meta"] = dvc_info.get("meta") if dvc_info else None
isexec = False
if dvc_info:
isexec = dvc_info["isexec"]
elif fs_info["type"] == "file":
isexec = is_exec(fs_info["mode"])
fs_info["isexec"] = isexec
return fs_info
ret["dvc_info"] = dvc_info
ret["type"] = dvc_info["type"]
ret["size"] = dvc_info["size"]
if not fs_info and "md5" in dvc_info:
ret["md5"] = dvc_info["md5"]
if fs_info:
ret["type"] = fs_info["type"]
ret["size"] = fs_info["size"]
isexec = False
if fs_info["type"] == "file":
isexec = is_exec(fs_info["mode"])
ret["isexec"] = isexec
return ret
class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
class _RepoFileSystem(AbstractFileSystem): # pylint:disable=abstract-method
"""DVC + git-tracked files fs.
Args:
@@ -71,10 +74,6 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
kwargs: Additional keyword arguments passed to the `DvcFileSystem()`.
"""
sep = os.sep
scheme = "local"
PARAM_CHECKSUM = "md5"
PARAM_REPO_URL = "repo_url"
PARAM_REPO_ROOT = "repo_root"
PARAM_REV = "rev"
@@ -105,6 +104,7 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
else:
self.repo_factory = repo_factory
self.path = Path(self.sep)
self.repo = repo
self.hash_jobs = repo.fs.hash_jobs
self._root_dir: str = repo.root_dir
@@ -131,7 +131,7 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
def config(self):
return {
self.PARAM_REPO_URL: self.repo_url,
self.PARAM_REPO_ROOT: self._root_dir,
self.PARAM_REPO_ROOT: self.repo.root_dir,
self.PARAM_REV: getattr(self.repo.fs, "rev", None),
self.PARAM_CACHE_DIR: os.path.abspath(
self.repo.odb.local.cache_dir
@@ -188,7 +188,7 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
) as repo:
return repo, factory
def _get_repo(self, path: str) -> Optional["Repo"]:
def _get_repo(self, path: str) -> "Repo":
"""Returns repo that the path falls in, using prefix.
If the path is already tracked/collected, it just returns the repo.
@@ -204,7 +204,7 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
if not prefix:
return self.repo
parents = (parent for parent in self.path.parents(path))
parents = (parent for parent in self.repo.fs.path.parents(path))
dirs = [path] + list(takewhile(lambda p: p != prefix, parents))
dirs.reverse()
self._update(dirs, starting_repo=repo)
@@ -236,12 +236,24 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
def _get_fs_pair(
self, path
) -> Tuple[FileSystem, Optional[DvcFileSystem], str]:
) -> Tuple[
Optional[FileSystem],
Optional[str],
Optional[DvcFileSystem],
Optional[str],
]:
"""
Returns a pair of fss based on repo the path falls in, using prefix.
"""
from dvc.utils import as_posix
if os.path.isabs(path):
return None, None, self.repo.dvcfs, path
if self.repo.fs.path.isin_or_eq(path, self.repo.root_dir):
path = self.repo.fs.path.relpath(path, self.repo.root_dir)
else:
return None, None, self.repo.dvcfs, path
path = as_posix(path)
parts = self.path.parts(path)
if parts and parts[0] == os.curdir:
@@ -260,13 +272,16 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
dvc_parts = dvc_parts[1:]
dvc_fs = self._dvcfss.get(repo.root_dir)
dvc_path = dvc_fs.path.join(*dvc_parts) if dvc_parts else ""
if dvc_fs:
dvc_path = dvc_fs.path.join(*dvc_parts) if dvc_parts else ""
else:
dvc_path = None
return fs, fs_path, dvc_fs, dvc_path
def open(
self, path, mode="r", encoding="utf-8", **kwargs
): # pylint: disable=arguments-renamed
): # pylint: disable=arguments-renamed, arguments-differ
if "b" in mode:
encoding = None
@@ -279,29 +294,10 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
return dvc_fs.open(dvc_path, mode=mode, encoding=encoding, **kwargs)
def exists(self, path, **kwargs) -> bool:
try:
self.info(path, **kwargs)
return True
except FileNotFoundError:
return False
def isdir(self, path, **kwargs): # pylint: disable=arguments-renamed
try:
return self.info(path, **kwargs)["type"] == "directory"
except FileNotFoundError:
return False
def isdvc(self, path, **kwargs):
_, _, dvc_fs, dvc_path = self._get_fs_pair(path)
return dvc_fs is not None and dvc_fs.isdvc(dvc_path, **kwargs)
def isfile(self, path, **kwargs): # pylint: disable=arguments-renamed
try:
return self.info(path, **kwargs)["type"] == "file"
except FileNotFoundError:
return False
def ls(self, path, detail=True, **kwargs):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
@@ -315,13 +311,14 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
for entry in dvc_fs.ls(dvc_path, detail=False):
names.add(dvc_fs.path.name(entry))
try:
for entry in dvcignore.ls(
fs, fs_path, detail=False, ignore_subrepos=ignore_subrepos
):
names.add(fs.path.name(entry))
except (FileNotFoundError, NotADirectoryError):
pass
if fs:
try:
for entry in dvcignore.ls(
fs, fs_path, detail=False, ignore_subrepos=ignore_subrepos
):
names.add(fs.path.name(entry))
except (FileNotFoundError, NotADirectoryError):
pass
dvcfiles = kwargs.get("dvcfiles", False)
@@ -354,61 +351,13 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
return infos
def walk(self, top, topdown=True, **kwargs):
"""Walk and merge both DVC and repo fss.
Args:
top: path to walk from
topdown: if True, fs will be walked from top down.
dvcfiles: if True, dvcfiles will be included in the files list
for walked directories.
Any kwargs will be passed into methods used for fetching and/or
streaming DVC outs from remotes.
"""
dirs = []
nondirs = []
ignore_subrepos = kwargs.get("ignore_subrepos", True)
if not self.exists(top, ignore_subrepos=ignore_subrepos):
return
if not self.isdir(top, ignore_subrepos=ignore_subrepos):
return
for entry in self.ls(top, **kwargs):
name = self.path.name(entry["name"])
if entry["type"] == "directory":
dirs.append(name)
else:
nondirs.append(name)
if topdown:
yield top, dirs, nondirs
for dname in dirs:
yield from self.walk(
self.path.join(top, dname), topdown=topdown, **kwargs
)
if not topdown:
yield top, dirs, nondirs
def find(self, path, prefix=None):
for root, _, files in self.walk(path):
for fname in files:
yield self.path.join(root, fname)
def get_file(
self, from_info, to_file, callback=DEFAULT_CALLBACK, **kwargs
):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(from_info)
def get_file(self, rpath, lpath, callback=DEFAULT_CALLBACK, **kwargs):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(rpath)
if fs:
try:
fs.get_file( # pylint: disable=protected-access
fs_path, to_file, callback=callback, **kwargs
fs_path, lpath, callback=callback, **kwargs
)
return
except FileNotFoundError:
@@ -416,7 +365,7 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
raise
dvc_fs.get_file( # pylint: disable=protected-access
dvc_path, to_file, callback=callback, **kwargs
dvc_path, lpath, callback=callback, **kwargs
)
def info(self, path, **kwargs):
@@ -446,9 +395,9 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
raise
# NOTE: if some parent in fs_path turns out to be a file, it means
# that that whole repofs branch doesn't exist.
# that the whole repofs branch doesn't exist.
if fs and not fs_info and dvc_info:
for parent in self.path.parents(fs_path):
for parent in fs.path.parents(fs_path):
try:
if fs.info(parent)["type"] != "directory":
dvc_info = None
@@ -459,7 +408,9 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
if not dvc_info and not fs_info:
raise FileNotFoundError
return _merge_info(dvc_fs.repo, fs_info, dvc_info)
info = _merge_info(dvc_fs.repo, fs_info, dvc_info)
info["name"] = path
return info
def checksum(self, path):
fs, fs_path, dvc_fs, dvc_path = self._get_fs_pair(path)
@@ -468,3 +419,31 @@ class RepoFileSystem(FileSystem): # pylint:disable=abstract-method
return fs.checksum(fs_path)
except FileNotFoundError:
return dvc_fs.checksum(dvc_path)
class RepoFileSystem(FSSpecWrapper):
scheme = "local"
PARAM_CHECKSUM = "md5"
def _prepare_credentials(self, **config):
return config
@wrap_prop(threading.Lock())
@cached_property
def fs(self):
return _RepoFileSystem(**self.fs_args)
def isdvc(self, path, **kwargs):
return self.fs.isdvc(path, **kwargs)
@property
def repo(self):
return self.fs.repo
@property
def repo_url(self):
return self.fs.repo_url
@property
def config(self):
return self.fs.config
dvc/ignore.py
+1 -1
View File
@@ -303,7 +303,7 @@ class DvcIgnoreFilter:
for root, _, files in self.walk(fs, path, **kwargs):
for file in files:
# NOTE: os.path.join is ~5.5 times slower
yield f"{root}{os.sep}{file}"
yield f"{root}{fs.sep}{file}"
else:
yield from fs.find(path)
dvc/output.py
-5
View File
@@ -386,11 +386,6 @@ class Output:
os.path.realpath(self.fs_path), self.repo.root_dir
)
@property
def repo_path(self):
assert self.is_in_repo
return relpath(self.fs_path, self.repo.root_dir)
@property
def use_scm_ignore(self):
if not self.is_in_repo:
dvc/repo/__init__.py
+4 -2
View File
@@ -482,7 +482,9 @@ class Repo:
def repo_fs(self):
from dvc.fs.repo import RepoFileSystem
return RepoFileSystem(self, subrepos=self.subrepos, **self._fs_conf)
return RepoFileSystem(
repo=self, subrepos=self.subrepos, **self._fs_conf
)
@cached_property
def index_db_dir(self):
@@ -493,7 +495,7 @@ class Repo:
"""Opens a specified resource as a file descriptor"""
from dvc.fs.repo import RepoFileSystem
fs = RepoFileSystem(self, subrepos=True)
fs = RepoFileSystem(repo=self, subrepos=True)
try:
with fs.open(
path, mode=mode, encoding=encoding, remote=remote
dvc/repo/collect.py
+13 -15
View File
@@ -34,22 +34,22 @@ def _collect_paths(
from dvc.fs.repo import RepoFileSystem
from dvc.utils import relpath
fs = RepoFileSystem(repo)
fs_paths = [os.path.abspath(target) for target in targets]
fs = RepoFileSystem(repo=repo)
target_paths = []
for target in targets:
if os.path.isabs(target):
target = relpath(target, repo.root_dir)
for fs_path in fs_paths:
if recursive and fs.isdir(fs_path):
target_paths.extend(fs.find(fs_path))
if recursive and fs.isdir(target):
target_paths.extend(repo.dvcignore.find(fs, target))
if not fs.exists(target):
rel = relpath(fs_path)
if not fs.exists(fs_path):
if rev == "workspace" or rev == "":
logger.warning("'%s' was not found in current workspace.", target)
logger.warning("'%s' was not found in current workspace.", rel)
else:
logger.warning("'%s' was not found at: '%s'.", target, rev)
target_paths.append(target)
logger.warning("'%s' was not found at: '%s'.", rel, rev)
target_paths.append(fs_path)
return target_paths
@@ -58,14 +58,12 @@ def _filter_duplicates(
) -> Tuple[Outputs, StrPaths]:
res_outs: Outputs = []
fs_res_paths = fs_paths
from dvc.utils import relpath
for out in outs:
rel = relpath(out.fs_path, out.stage.repo.root_dir)
if rel in fs_paths:
if out.fs_path in fs_paths:
res_outs.append(out)
# MUTATING THE SAME LIST!!
fs_res_paths.remove(rel)
fs_res_paths.remove(out.fs_path)
return res_outs, fs_res_paths
dvc/repo/diff.py
+4 -3
View File
@@ -24,7 +24,7 @@ def diff(self, a_rev="HEAD", b_rev=None, targets=None):
from dvc.fs.repo import RepoFileSystem
repo_fs = RepoFileSystem(self)
repo_fs = RepoFileSystem(repo=self)
b_rev = b_rev if b_rev else "workspace"
results = {}
@@ -188,10 +188,11 @@ def _filter_missing(repo_fs, paths):
for path in paths:
try:
info = repo_fs.info(path)
dvc_info = info.get("dvc_info")
if (
info["isdvc"]
dvc_info
and info["type"] == "directory"
and not info["meta"].obj
and not dvc_info["meta"].obj
):
yield path
except FileNotFoundError:
dvc/repo/ls.py
+25 -20
View File
@@ -29,9 +29,9 @@ def ls(url, path=None, rev=None, recursive=None, dvc_only=False):
from . import Repo
with Repo.open(url, rev=rev, subrepos=True, uninitialized=True) as repo:
fs_path = path or ""
path = path or ""
ret = _ls(repo, fs_path, recursive, dvc_only)
ret = _ls(repo.repo_fs, path, recursive, dvc_only)
if path and not ret:
raise PathMissingError(path, repo, dvc_only=dvc_only)
@@ -44,31 +44,36 @@ def ls(url, path=None, rev=None, recursive=None, dvc_only=False):
return ret_list
def _ls(repo, fs_path, recursive=None, dvc_only=False):
fs = repo.repo_fs
infos = []
def _ls(fs, path, recursive=None, dvc_only=False):
try:
fs_path = fs.info(path)["name"]
except FileNotFoundError:
return {}
infos = {}
for root, dirs, files in fs.walk(fs_path, dvcfiles=True):
entries = chain(files, dirs) if not recursive else files
infos.extend(fs.path.join(root, entry) for entry in entries)
for entry in entries:
entry_fs_path = fs.path.join(root, entry)
relparts = fs.path.relparts(entry_fs_path, fs_path)
name = os.path.join(*relparts)
infos[name] = fs.info(entry_fs_path)
if not recursive:
break
if not infos and fs.isfile(fs_path):
infos.append(fs_path)
infos[os.path.basename(path)] = fs.info(fs_path)
ret = {}
for info in infos:
_info = fs.info(info)
if _info.get("outs") or not dvc_only:
path = (
fs.path.name(fs_path)
if fs_path == info
else fs.path.relpath(info, fs_path)
)
ret[path] = {
"isout": _info.get("isout", False),
"isdir": _info["type"] == "directory",
"isexec": _info["isexec"],
for name, info in infos.items():
dvc_info = info.get("dvc_info", {})
if dvc_info.get("outs") or not dvc_only:
ret[name] = {
"isout": dvc_info.get("isout", False),
"isdir": info["type"] == "directory",
"isexec": info.get("isexec", False),
}
return ret
dvc/repo/metrics/show.py
+4 -5
View File
@@ -76,16 +76,15 @@ def _read_metric(path, fs, rev, **kwargs):
def _read_metrics(repo, metrics, rev, onerror=None):
fs = RepoFileSystem(repo)
fs = RepoFileSystem(repo=repo)
res = {}
for metric in metrics:
fs_path = repo.fs.path.relpath(metric, repo.root_dir)
if not fs.isfile(fs_path):
if not fs.isfile(metric):
continue
res[fs.path.relpath(metric, os.getcwd())] = _read_metric(
fs_path, fs, rev, onerror=onerror
res[os.path.relpath(metric, os.getcwd())] = _read_metric(
metric, fs, rev, onerror=onerror
)
return res
dvc/repo/plots/__init__.py
+4 -4
View File
@@ -106,7 +106,7 @@ class Plots:
):
from dvc.fs.repo import RepoFileSystem
fs = RepoFileSystem(self.repo)
fs = RepoFileSystem(repo=self.repo)
plots = _collect_plots(self.repo, targets, revision, recursive)
res: Dict[str, Any] = {}
for fs_path, rev_props in plots.items():
@@ -240,7 +240,7 @@ def _collect_plots(
targets: List[str] = None,
rev: str = None,
recursive: bool = False,
) -> Dict["List[str]", Dict]:
) -> Dict[str, Dict]:
from dvc.repo.collect import collect
plots, fs_paths = collect(
@@ -251,8 +251,8 @@ def _collect_plots(
recursive=recursive,
)
result = {relpath(plot.fs_path, repo.root_dir): _plot_props(plot) for plot in plots}
result.update({relpath(fs_path, repo.root_dir): {} for fs_path in fs_paths})
result = {plot.fs_path: _plot_props(plot) for plot in plots}
result.update({fs_path: {} for fs_path in fs_paths})
return result
dvc/repo/reproduce.py
+1 -1
View File
@@ -63,7 +63,7 @@ def _get_stage_files(stage: "Stage") -> typing.Iterator[str]:
if (
not dep.use_scm_ignore
and dep.is_in_repo
and not stage.repo.repo_fs.isdvc(dep.fs.path.relpath(dep.fs_path, stage.repo.root_dir))
and not stage.repo.repo_fs.isdvc(dep.fs_path)
):
yield dep.fs_path
for out in stage.outs:
dvc/utils/__init__.py
+7
View File
@@ -332,6 +332,13 @@ def relpath(path, start=os.curdir):
return os.path.relpath(path, start)
def as_posix(path):
import ntpath
import posixpath
return path.replace(ntpath.sep, posixpath.sep)
def env2bool(var, undefined=False):
"""
undefined: return value if env var is unset
tests/func/test_external_repo.py
+1 -1
View File
@@ -96,7 +96,7 @@ def test_pull_subdir_file(tmp_dir, erepo_dir):
dest = tmp_dir / "file"
with external_repo(os.fspath(erepo_dir)) as repo:
repo.repo_fs.download(
os.path.join("subdir", "file"),
"subdir/file",
os.fspath(dest),
)
tests/unit/fs/test_repo.py
+103 -101
View File
@@ -1,4 +1,5 @@
import os
import posixpath
import shutil
from unittest import mock
@@ -116,8 +117,8 @@ def test_isdir_isfile(tmp_dir, dvc):
assert fs.isdir("subdir")
assert not fs.isfile("subdir")
assert not fs.isdvc("subdir")
assert fs.isfile(os.path.join("subdir", "baz"))
assert fs.isdir(os.path.join("subdir", "data"))
assert fs.isfile("subdir/baz")
assert fs.isdir("subdir/data")
def test_exists_isdir_isfile_dirty(tmp_dir, dvc):
@@ -129,31 +130,30 @@ def test_exists_isdir_isfile_dirty(tmp_dir, dvc):
shutil.rmtree(tmp_dir / "datadir")
(tmp_dir / "datafile").unlink()
root = tmp_dir
assert fs.exists("datafile")
assert fs.exists("datadir")
assert fs.exists(os.path.join("datadir", "foo"))
assert fs.exists("datadir/foo")
assert fs.isfile("datafile")
assert not fs.isfile("datadir")
assert fs.isfile(os.path.join("datadir", "foo"))
assert fs.isfile("datadir/foo")
assert not fs.isdir("datafile")
assert fs.isdir("datadir")
assert not fs.isdir(os.path.join("datadir", "foo"))
assert not fs.isdir("datadir/foo")
# NOTE: creating file instead of dir and dir instead of file
tmp_dir.gen({"datadir": "data", "datafile": {"foo": "foo", "bar": "bar"}})
assert fs.exists("datafile")
assert fs.exists("datadir")
assert not fs.exists(os.path.join("datadir", "foo"))
assert fs.exists(os.path.join("datafile", "foo"))
assert not fs.exists("datadir/foo")
assert fs.exists("datafile/foo")
assert not fs.isfile("datafile")
assert fs.isfile("datadir")
assert not fs.isfile(os.path.join("datadir", "foo"))
assert fs.isfile(os.path.join("datafile", "foo"))
assert not fs.isfile("datadir/foo")
assert fs.isfile("datafile/foo")
assert fs.isdir("datafile")
assert not fs.isdir("datadir")
assert not fs.isdir(os.path.join("datadir", "foo"))
assert not fs.isdir(os.path.join("datafile", "foo"))
assert not fs.isdir("datadir/foo")
assert not fs.isdir("datafile/foo")
def test_isdir_mixed(tmp_dir, dvc):
@@ -173,9 +173,9 @@ def test_isdir_mixed(tmp_dir, dvc):
(
True,
[
os.path.join("dir", "subdir1", "foo1.dvc"),
os.path.join("dir", "subdir1", "bar1.dvc"),
os.path.join("dir", "subdir2", "foo2.dvc"),
"dir/subdir1/foo1.dvc",
"dir/subdir1/bar1.dvc",
"dir/subdir2/foo2.dvc",
],
),
],
@@ -194,21 +194,21 @@ def test_walk(tmp_dir, dvc, dvcfiles, extra_expected):
fs = RepoFileSystem(repo=dvc)
expected = [
os.path.join("dir", "subdir1"),
os.path.join("dir", "subdir2"),
os.path.join("dir", "subdir1", "foo1"),
os.path.join("dir", "subdir1", "bar1"),
os.path.join("dir", "subdir2", "foo2"),
os.path.join("dir", "foo"),
os.path.join("dir", "bar"),
"dir/subdir1",
"dir/subdir2",
"dir/subdir1/foo1",
"dir/subdir1/bar1",
"dir/subdir2/foo2",
"dir/foo",
"dir/bar",
]
actual = []
for root, dirs, files in fs.walk("dir", dvcfiles=dvcfiles):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
expected += [os.path.join(path) for path in extra_expected]
expected += extra_expected
assert set(actual) == set(expected)
assert len(actual) == len(expected)
@@ -228,21 +228,21 @@ def test_walk_dirty(tmp_dir, dvc):
fs = RepoFileSystem(repo=dvc)
expected = [
os.path.join("dir", "subdir1"),
os.path.join("dir", "subdir2"),
os.path.join("dir", "subdir3"),
os.path.join("dir", "subdir1", "foo1"),
os.path.join("dir", "subdir1", "bar1"),
os.path.join("dir", "subdir2", "foo2"),
os.path.join("dir", "subdir3", "foo3"),
os.path.join("dir", "bar"),
os.path.join("dir", "foo"),
"dir/subdir1",
"dir/subdir2",
"dir/subdir3",
"dir/subdir1/foo1",
"dir/subdir1/bar1",
"dir/subdir2/foo2",
"dir/subdir3/foo3",
"dir/bar",
"dir/foo",
]
actual = []
for root, dirs, files in fs.walk("dir"):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
assert set(actual) == set(expected)
assert len(actual) == len(expected)
@@ -254,14 +254,12 @@ def test_walk_dirty_cached_dir(tmp_dir, scm, dvc):
fs = RepoFileSystem(repo=dvc)
data = tmp_dir / "data"
actual = []
for root, dirs, files in fs.walk("data"):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
expected = [os.path.join("data", "foo"), os.path.join("data", "bar")]
expected = ["data/foo", "data/bar"]
assert set(actual) == set(expected)
assert len(actual) == len(expected)
@@ -281,14 +279,14 @@ def test_walk_mixed_dir(tmp_dir, scm, dvc):
fs = RepoFileSystem(repo=dvc)
expected = [
os.path.join("dir", "foo"),
os.path.join("dir", "bar"),
os.path.join("dir", ".gitignore"),
"dir/foo",
"dir/bar",
"dir/.gitignore",
]
actual = []
for root, dirs, files in fs.walk("dir"):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
assert set(actual) == set(expected)
assert len(actual) == len(expected)
@@ -317,8 +315,8 @@ def test_isdvc(tmp_dir, dvc):
assert fs.isdvc("foo")
assert not fs.isdvc("bar")
assert fs.isdvc("dir")
assert fs.isdvc(os.path.join("dir", "baz"))
assert fs.isdvc(os.path.join("dir", "baz"), recursive=True)
assert fs.isdvc("dir/baz")
assert fs.isdvc("dir/baz", recursive=True)
def make_subrepo(dir_, scm, config=None):
@@ -353,7 +351,7 @@ def test_subrepos(tmp_dir, scm, dvc, mocker):
fs = RepoFileSystem(repo=dvc, subrepos=True)
def assert_fs_belongs_to_repo(ret_val):
method = fs._get_repo
method = fs.fs._get_repo
def f(*args, **kwargs):
r = method(*args, **kwargs)
@@ -363,32 +361,32 @@ def test_subrepos(tmp_dir, scm, dvc, mocker):
return f
with mock.patch.object(
fs, "_get_repo", side_effect=assert_fs_belongs_to_repo(subrepo1.dvc)
fs.fs, "_get_repo", side_effect=assert_fs_belongs_to_repo(subrepo1.dvc)
):
assert fs.exists(os.path.join("dir", "repo", "foo")) is True
assert fs.exists(os.path.join("dir", "repo", "bar")) is False
assert fs.exists("dir/repo/foo") is True
assert fs.exists("dir/repo/bar") is False
assert fs.isfile(os.path.join("dir", "repo", "foo")) is True
assert fs.isfile(os.path.join("dir", "repo", "dir1", "bar")) is True
assert fs.isfile(os.path.join("dir", "repo", "dir1")) is False
assert fs.isfile("dir/repo/foo") is True
assert fs.isfile("dir/repo/dir1/bar") is True
assert fs.isfile("dir/repo/dir1") is False
assert fs.isdir(os.path.join("dir", "repo", "dir1")) is True
assert fs.isdir(os.path.join("dir", "repo", "dir1", "bar")) is False
assert fs.isdvc(os.path.join("dir", "repo", "foo")) is True
assert fs.isdir("dir/repo/dir1") is True
assert fs.isdir("dir/repo/dir1/bar") is False
assert fs.isdvc("dir/repo/foo") is True
with mock.patch.object(
fs, "_get_repo", side_effect=assert_fs_belongs_to_repo(subrepo2.dvc)
fs.fs, "_get_repo", side_effect=assert_fs_belongs_to_repo(subrepo2.dvc)
):
assert fs.exists(os.path.join("dir", "repo2", "lorem")) is True
assert fs.exists(os.path.join("dir", "repo2", "ipsum")) is False
assert fs.exists("dir/repo2/lorem") is True
assert fs.exists("dir/repo2/ipsum") is False
assert fs.isfile(os.path.join("dir", "repo2", "lorem")) is True
assert fs.isfile(os.path.join("dir", "repo2", "dir2", "ipsum")) is True
assert fs.isfile(os.path.join("dir", "repo2", "dir2")) is False
assert fs.isfile("dir/repo2/lorem") is True
assert fs.isfile("dir/repo2/dir2/ipsum") is True
assert fs.isfile("dir/repo2/dir2") is False
assert fs.isdir(os.path.join("dir", "repo2", "dir2")) is True
assert fs.isdir(os.path.join("dir", "repo2", "dir2", "ipsum")) is False
assert fs.isdvc(os.path.join("dir", "repo2", "lorem")) is True
assert fs.isdir("dir/repo2/dir2") is True
assert fs.isdir("dir/repo2/dir2/ipsum") is False
assert fs.isdvc("dir/repo2/lorem") is True
@pytest.mark.parametrize(
@@ -398,12 +396,12 @@ def test_subrepos(tmp_dir, scm, dvc, mocker):
(
True,
[
os.path.join("dir", "repo", "foo.dvc"),
os.path.join("dir", "repo", ".dvcignore"),
os.path.join("dir", "repo", "dir1.dvc"),
os.path.join("dir", "repo2", ".dvcignore"),
os.path.join("dir", "repo2", "lorem.dvc"),
os.path.join("dir", "repo2", "dir2.dvc"),
"dir/repo/foo.dvc",
"dir/repo/.dvcignore",
"dir/repo/dir1.dvc",
"dir/repo2/.dvcignore",
"dir/repo2/lorem.dvc",
"dir/repo2/dir2.dvc",
],
),
],
@@ -432,17 +430,17 @@ def test_subrepo_walk(tmp_dir, scm, dvc, dvcfiles, extra_expected):
dvc._reset()
fs = RepoFileSystem(repo=dvc)
expected = [
os.path.join("dir", "repo"),
os.path.join("dir", "repo.txt"),
os.path.join("dir", "repo2"),
os.path.join("dir", "repo", ".gitignore"),
os.path.join("dir", "repo", "foo"),
os.path.join("dir", "repo", "dir1"),
os.path.join("dir", "repo", "dir1", "bar"),
os.path.join("dir", "repo2", ".gitignore"),
os.path.join("dir", "repo2", "lorem"),
os.path.join("dir", "repo2", "dir2"),
os.path.join("dir", "repo2", "dir2", "ipsum"),
"dir/repo",
"dir/repo.txt",
"dir/repo2",
"dir/repo/.gitignore",
"dir/repo/foo",
"dir/repo/dir1",
"dir/repo/dir1/bar",
"dir/repo2/.gitignore",
"dir/repo2/lorem",
"dir/repo2/dir2",
"dir/repo2/dir2/ipsum",
]
actual = []
@@ -452,7 +450,7 @@ def test_subrepo_walk(tmp_dir, scm, dvc, dvcfiles, extra_expected):
ignore_subrepos=False,
):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
expected += extra_expected
assert set(actual) == set(expected)
@@ -481,27 +479,27 @@ def test_repo_fs_no_subrepos(tmp_dir, dvc, scm):
"lorem",
"lorem.dvc",
"dir",
os.path.join("dir", "repo.txt"),
"dir/repo.txt",
]
actual = []
for root, dirs, files in fs.walk("", dvcfiles=True):
for entry in dirs + files:
actual.append(os.path.join(root, entry))
actual.append(posixpath.join(root, entry))
assert set(actual) == set(expected)
assert len(actual) == len(expected)
assert fs.isfile("lorem") is True
assert fs.isfile(os.path.join("dir", "repo", "foo")) is False
assert fs.isdir(os.path.join("dir", "repo")) is False
assert fs.isfile("dir/repo/foo") is False
assert fs.isdir("dir/repo") is False
assert fs.isdir("dir") is True
assert fs.isdvc("lorem") is True
assert fs.isdvc(os.path.join("dir", "repo", "dir1")) is False
assert fs.isdvc("dir/repo/dir1") is False
assert fs.exists(os.path.join("dir", "repo.txt")) is True
assert fs.exists(os.path.join("repo", "ipsum")) is False
assert fs.exists("dir/repo.txt") is True
assert fs.exists("repo/ipsum") is False
def test_get_hash_cached_file(tmp_dir, dvc, mocker):
@@ -540,18 +538,18 @@ def test_get_hash_cached_granular(tmp_dir, dvc, mocker):
{"dir": {"foo": "foo", "bar": "bar", "subdir": {"data": "data"}}}
)
fs = RepoFileSystem(repo=dvc)
subdir = os.path.join("dir", "subdir")
subdir = "dir/subdir"
assert fs.info(subdir).get("md5") is None
_, _, obj = stage(dvc.odb.local, subdir, fs, "md5")
assert obj.hash_info == HashInfo(
"md5", "af314506f1622d107e0ed3f14ec1a3b5.dir"
)
assert fs.info(os.path.join(subdir, "data")).get("md5") is None
_, _, obj = stage(dvc.odb.local, os.path.join(subdir, "data"), fs, "md5")
assert fs.info(posixpath.join(subdir, "data")).get("md5") is None
_, _, obj = stage(dvc.odb.local, posixpath.join(subdir, "data"), fs, "md5")
assert obj.hash_info == HashInfo("md5", "8d777f385d3dfec8815d20f7496026dc")
(tmp_dir / "dir" / "subdir" / "data").unlink()
assert (
fs.info(os.path.join(subdir, "data"))["md5"]
fs.info(posixpath.join(subdir, "data"))["md5"]
== "8d777f385d3dfec8815d20f7496026dc"
)
@@ -593,9 +591,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
# get_file_hash(file) should return workspace hash, not DVC cached hash
fs = RepoFileSystem(repo=dvc)
assert fs.info("file").get("md5") is None
staging, _, obj = stage(
dvc.odb.local, "file", fs, "md5"
)
staging, _, obj = stage(dvc.odb.local, "file", fs, "md5")
assert obj.hash_info == something_hash_info
check(staging, obj)
@@ -607,9 +603,7 @@ def test_get_hash_dirty_file(tmp_dir, dvc):
# get_file_hash(file) should return DVC cached hash
assert fs.info("file")["md5"] == file_hash_info.value
_, hash_info = get_file_hash(
"file", fs, "md5", state=dvc.state
)
_, hash_info = get_file_hash("file", fs, "md5", state=dvc.state)
assert hash_info == file_hash_info
# tmp_dir/file can be staged even though it is missing in workspace since
@@ -662,14 +656,22 @@ def test_walk_nested_subrepos(tmp_dir, dvc, scm, traverse_subrepos):
repo_dir.dvc_gen(dvc_files, commit=f"dvc add in {repo_dir}")
if traverse_subrepos or repo_dir == tmp_dir:
repo_dir_path = repo_dir.relative_to(tmp_dir).fs_path if repo_dir != tmp_dir else ""
repo_dir_path = (
repo_dir.relative_to(tmp_dir).as_posix()
if repo_dir != tmp_dir
else ""
)
expected[repo_dir_path] = set(
scm_files.keys() | dvc_files.keys() | extras
)
# files inside a dvc directory
expected[os.path.join(repo_dir_path, f"dvc-{base}")] = {f"ipsum-{base}"}
expected[posixpath.join(repo_dir_path, f"dvc-{base}")] = {
f"ipsum-{base}"
}
# files inside a git directory
expected[os.path.join(repo_dir_path, f"dir-{base}")] = {f"bar-{base}"}
expected[posixpath.join(repo_dir_path, f"dir-{base}")] = {
f"bar-{base}"
}
if traverse_subrepos:
# update subrepos
tests/unit/fs/test_repo_info.py
+26 -28
View File
@@ -38,31 +38,29 @@ def repo_fs(tmp_dir, dvc, scm):
tmp_dir.scm_gen(fs_structure, commit="repo init")
tmp_dir.dvc_gen(dvc_structure, commit="use dvc")
yield RepoFileSystem(dvc, subrepos=True)
yield RepoFileSystem(repo=dvc, subrepos=True)
def test_info_not_existing(repo_fs):
path = os.path.join("path", "that", "does", "not", "exist")
with pytest.raises(FileNotFoundError):
repo_fs.info(path)
repo_fs.info("path/that/does/not/exist")
@pytest.mark.parametrize(
"path",
[
"README.md",
os.path.join("models", "train.py"),
os.path.join("models", "test.py"),
os.path.join("src", "utils", "__init__.py"),
os.path.join("src", "utils", "serve_model.py"),
"models/train.py",
"models/test.py",
"src/utils/__init__.py",
"src/utils/serve_model.py",
],
)
def test_info_git_tracked_file(repo_fs, path):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs._root_dir
assert not info["isdvc"]
assert info["repo"].root_dir == repo_fs.repo.root_dir
assert "dvc_info" not in info
assert info["type"] == "file"
assert not info["isexec"]
@@ -70,28 +68,28 @@ def test_info_git_tracked_file(repo_fs, path):
@pytest.mark.parametrize(
"path",
[
os.path.join("data", "raw", "raw-1.csv"),
os.path.join("data", "raw", "raw-2.csv"),
os.path.join("data", "processed", "processed-1.csv"),
os.path.join("data", "processed", "processed-2.csv"),
os.path.join("models", "transform.pickle"),
"data/raw/raw-1.csv",
"data/raw/raw-2.csv",
"data/processed/processed-1.csv",
"data/processed/processed-2.csv",
"models/transform.pickle",
],
)
def test_info_dvc_tracked_file(repo_fs, path):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs._root_dir
assert info["isdvc"]
assert info["repo"].root_dir == repo_fs.repo.root_dir
assert info["dvc_info"]["isdvc"]
assert info["type"] == "file"
assert not info["isexec"]
@pytest.mark.parametrize("path", ["src", os.path.join("src", "utils")])
@pytest.mark.parametrize("path", ["src", "src/utils"])
def test_info_git_only_dirs(repo_fs, path):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs._root_dir
assert not info["isdvc"]
assert info["repo"].root_dir == repo_fs.repo.root_dir
assert "dvc_info" not in info
assert info["type"] == "directory"
assert not info["isexec"]
@@ -100,8 +98,8 @@ def test_info_git_only_dirs(repo_fs, path):
def test_info_git_dvc_mixed_dirs(repo_fs, path):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs._root_dir
assert not info["isdvc"]
assert info["repo"].root_dir == repo_fs.repo.root_dir
assert not info["dvc_info"]["isdvc"]
assert info["type"] == "directory"
assert not info["isexec"]
@@ -110,15 +108,15 @@ def test_info_git_dvc_mixed_dirs(repo_fs, path):
"path",
[
"data",
os.path.join("data", "raw"),
os.path.join("data", "processed"),
"data/raw",
"data/processed",
],
)
def test_info_dvc_only_dirs(repo_fs, path):
info = repo_fs.info(path)
assert info["repo"].root_dir == repo_fs._root_dir
assert info["isdvc"]
assert info["repo"].root_dir == repo_fs.repo.root_dir
assert info["dvc_info"]["isdvc"]
assert info["type"] == "directory"
assert not info["isexec"]
@@ -132,8 +130,8 @@ def test_info_on_subrepos(make_tmp_dir, tmp_dir, dvc, scm, repo_fs):
for path in [
"subrepo",
os.path.join("subrepo", "foo"),
os.path.join("subrepo", "foobar"),
"subrepo/foo",
"subrepo/foobar",
]:
info = repo_fs.info(path)
assert info["repo"].root_dir == str(
tests/unit/test_external_repo.py
+3 -3
View File
@@ -26,7 +26,7 @@ def test_hook_is_called(tmp_dir, erepo_dir, mocker):
repo.dvc_gen("bar", "bar", commit=f"dvc add {repo}/bar")
with external_repo(str(erepo_dir)) as repo:
spy = mocker.spy(repo.repo_fs, "repo_factory")
spy = mocker.spy(repo.repo_fs.fs, "repo_factory")
list(repo.repo_fs.walk("", ignore_subrepos=False)) # drain
assert spy.call_count == len(subrepos)
@@ -37,7 +37,7 @@ def test_hook_is_called(tmp_dir, erepo_dir, mocker):
call(
path,
fs=repo.fs,
repo_factory=repo.repo_fs.repo_factory,
repo_factory=repo.repo_fs.fs.repo_factory,
)
for path in paths
],
@@ -63,7 +63,7 @@ def test_subrepo_is_constructed_properly(
with external_repo(
str(tmp_dir), cache_dir=str(cache_dir), cache_types=["symlink"]
) as repo:
spy = mocker.spy(repo.repo_fs, "repo_factory")
spy = mocker.spy(repo.repo_fs.fs, "repo_factory")
list(repo.repo_fs.walk("", ignore_subrepos=False)) # drain
assert spy.call_count == 1