support subrepos for get/list/api/import/ls (#4465)

* support subrepos

This adds support for get/list/api/import/ls on a subrepo.

Also fixes #3180, by adding a granular url support for `dvc.api.get_url`
And, of course it fixes #3369.

* Add tests

* Fix tests, started to get lock_error

* Fix tests for windows

* try to fix mac builds

* Fix comments, readd removed comments

* Fix test

* address review suggestions:

* use `spy` instead of `wraps`
* remove dvc fixture in a test
* set `cache_types` on Git repo
This commit is contained in:
Saugat Pachhai
2020-09-03 13:47:14 +03:00
committed by GitHub
parent d460d18942
commit 4564e8be30
21 changed files with 721 additions and 215 deletions
+20 -15
View File
@@ -2,35 +2,40 @@ import os
from contextlib import _GeneratorContextManager as GCM
from contextlib import contextmanager
from dvc.exceptions import DvcException, NotDvcRepoError
from funcy import reraise
from dvc.exceptions import (
NotDvcRepoError,
OutputNotFoundError,
PathMissingError,
)
from dvc.external_repo import external_repo
from dvc.path_info import PathInfo
from dvc.repo import Repo
class UrlNotDvcRepoError(DvcException):
"""Thrown if the given URL is not a DVC repository."""
def __init__(self, url):
super().__init__(f"'{url}' is not a DVC repository.")
def get_url(path, repo=None, rev=None, remote=None):
"""
Returns the URL to the storage location of a data file or directory tracked
in a DVC repo. For Git repos, HEAD is used unless a rev argument is
supplied. The default remote is tried unless a remote argument is supplied.
Raises UrlNotDvcRepoError if repo is not a DVC project.
Raises OutputNotFoundError if the file is not a dvc-tracked file.
NOTE: This function does not check for the actual existence of the file or
directory in the remote storage.
"""
with _make_repo(repo, rev=rev) as _repo:
if not isinstance(_repo, Repo):
raise UrlNotDvcRepoError(_repo.url) # pylint: disable=no-member
out = _repo.find_out_by_relpath(path)
remote_obj = _repo.cloud.get_remote(remote)
return str(remote_obj.tree.hash_to_path_info(out.hash_info.value))
path_info = PathInfo(_repo.root_dir) / path
with reraise(FileNotFoundError, PathMissingError(path, repo)):
metadata = _repo.repo_tree.metadata(path_info)
if not metadata.is_dvc:
raise OutputNotFoundError(path, repo)
cloud = metadata.repo.cloud
hash_info = _repo.repo_tree.get_hash(path_info)
return cloud.get_url_for(remote, checksum=hash_info.value)
def open( # noqa, pylint: disable=redefined-builtin
@@ -97,7 +102,7 @@ def _make_repo(repo_url=None, rev=None):
repo_url = repo_url or os.getcwd()
if rev is None and os.path.exists(repo_url):
try:
yield Repo(repo_url)
yield Repo(repo_url, subrepos=True)
return
except NotDvcRepoError:
pass # fallthrough to external_repo
+4
View File
@@ -114,3 +114,7 @@ class DataCloud:
show_checksums=show_checksums,
log_missing=log_missing,
)
def get_url_for(self, remote, checksum):
remote = self.get_remote(remote)
return str(remote.tree.hash_to_path_info(checksum))
+12 -28
View File
@@ -1,10 +1,5 @@
import os
from voluptuous import Required
from dvc.exceptions import OutputNotFoundError
from dvc.path_info import PathInfo
from .local import LocalDependency
@@ -42,30 +37,18 @@ class RepoDependency(LocalDependency):
def __str__(self):
return "{} ({})".format(self.def_path, self.def_repo[self.PARAM_URL])
def _make_repo(self, *, locked=True):
def _make_repo(self, *, locked=True, **kwargs):
from dvc.external_repo import external_repo
d = self.def_repo
rev = (d.get("rev_lock") if locked else None) or d.get("rev")
return external_repo(d["url"], rev=rev)
return external_repo(d["url"], rev=rev, **kwargs)
def _get_hash(self, locked=True):
from dvc.tree.repo import RepoTree
with self._make_repo(locked=locked) as repo:
try:
return repo.find_out_by_relpath(self.def_path).hash_info
except OutputNotFoundError:
path = PathInfo(os.path.join(repo.root_dir, self.def_path))
# we want stream but not fetch, so DVC out directories are
# walked, but dir contents is not fetched
tree = RepoTree(repo, stream=True)
# We are polluting our repo cache with some dir listing here
if tree.isdir(path):
return self.repo.cache.local.tree.get_hash(path, tree=tree)
return tree.get_file_hash(path)
# we want stream but not fetch, so DVC out directories are
# walked, but dir contents is not fetched
with self._make_repo(locked=locked, fetch=False, stream=True) as repo:
return repo.get_checksum(self.def_path)
def workspace_status(self):
current = self._get_hash(locked=True)
@@ -86,14 +69,15 @@ class RepoDependency(LocalDependency):
return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo}
def download(self, to):
with self._make_repo() as repo:
cache = self.repo.cache.local
with self._make_repo(cache_dir=cache.cache_dir) as repo:
if self.def_repo.get(self.PARAM_REV_LOCK) is None:
self.def_repo[self.PARAM_REV_LOCK] = repo.get_rev()
cache = self.repo.cache.local
with repo.use_cache(cache):
_, _, cache_infos = repo.fetch_external([self.def_path])
cache.checkout(to.path_info, cache_infos[0])
_, _, cache_infos = repo.fetch_external([self.def_path])
cache.checkout(to.path_info, cache_infos[0])
def update(self, rev=None):
if rev:
+208 -127
View File
@@ -6,10 +6,12 @@ from contextlib import contextmanager
from distutils.dir_util import copy_tree
from typing import Iterable
from funcy import cached_property, retry, wrap_with
from funcy import cached_property, reraise, retry, wrap_with
from dvc.cache import Cache
from dvc.config import NoRemoteError, NotDvcRepoError
from dvc.exceptions import (
DvcException,
FileMissingError,
NoOutputInExternalRepoError,
NoRemoteInExternalRepoError,
@@ -20,6 +22,7 @@ from dvc.path_info import PathInfo
from dvc.repo import Repo
from dvc.scm.base import CloneError
from dvc.scm.git import Git
from dvc.state import StateNoop
from dvc.tree.local import LocalTree
from dvc.tree.repo import RepoTree
from dvc.utils import relpath
@@ -28,30 +31,45 @@ from dvc.utils.fs import remove
logger = logging.getLogger(__name__)
class IsADVCRepoError(DvcException):
"""Raised when it is not expected to be a dvc repo."""
@contextmanager
def external_repo(url, rev=None, for_write=False):
def external_repo(url, rev=None, for_write=False, **kwargs):
logger.debug("Creating external repo %s@%s", url, rev)
path = _cached_clone(url, rev, for_write=for_write)
if not rev:
# Local HEAD points to the tip of whatever branch we first cloned from
# (which may not be the default branch), use origin/HEAD here to get
# the tip of the default branch
rev = "refs/remotes/origin/HEAD"
# Local HEAD points to the tip of whatever branch we first cloned from
# (which may not be the default branch), use origin/HEAD here to get
# the tip of the default branch
rev = rev or "refs/remotes/origin/HEAD"
root_dir = path if for_write else os.path.realpath(path)
conf = dict(
root_dir=root_dir,
url=url,
scm=None if for_write else Git(root_dir),
rev=None if for_write else rev,
for_write=for_write,
**kwargs,
)
try:
repo = ExternalRepo(path, url, rev, for_write=for_write)
repo = ExternalRepo(**conf)
except NotDvcRepoError:
repo = ExternalGitRepo(path, url, rev)
repo = ExternalGitRepo(**conf)
try:
yield repo
except NoRemoteError:
raise NoRemoteInExternalRepoError(url)
except NoRemoteError as exc:
raise NoRemoteInExternalRepoError(url) from exc
except OutputNotFoundError as exc:
if exc.repo is repo:
raise NoOutputInExternalRepoError(exc.output, repo.root_dir, url)
raise NoOutputInExternalRepoError(
exc.output, repo.root_dir, url
) from exc
raise
except FileMissingError as exc:
raise PathMissingError(exc.path, url)
raise PathMissingError(exc.path, url) from exc
finally:
repo.close()
if for_write:
@@ -75,53 +93,37 @@ def clean_repos():
class BaseExternalRepo:
# pylint: disable=no-member
_local_cache = None
def __str__(self):
return self.url
@property
def local_cache(self):
if hasattr(self, "cache"):
return self.cache.local
return self._local_cache
@contextmanager
def use_cache(self, cache):
"""Use the specified cache in place of default tmpdir cache for
download operations.
"""
has_cache = hasattr(self, "cache")
if has_cache:
save_cache = self.cache.local # pylint: disable=E0203
self.cache.local = cache # pylint: disable=E0203
else:
from collections import namedtuple
mock_cache = namedtuple("MockCache", ["local"])(local=cache)
self.cache = mock_cache # pylint: disable=W0201
self._local_cache = cache
yield
if has_cache:
self.cache.local = save_cache
else:
del self.cache
self._local_cache = None
@cached_property
def repo_tree(self):
return RepoTree(self, fetch=True)
return self._get_tree_for(
self, subrepos=not self.for_write, repo_factory=self.make_repo
)
def get_rev(self):
assert self.scm
if isinstance(self.tree, LocalTree):
return self.scm.get_rev()
return self.tree.rev
def _fetch_to_cache(self, path_info, repo, callback):
# don't support subrepo traversal as it might fail due to difference
# in remotes
tree = self._get_tree_for(repo)
cache = repo.cache.local
hash_info = tree.get_hash(path_info, download_callback=callback)
cache.save(
path_info,
tree,
hash_info,
save_link=False,
download_callback=callback,
)
return hash_info
def fetch_external(self, paths: Iterable, **kwargs):
"""Fetch specified external repo paths into cache.
@@ -132,117 +134,196 @@ class BaseExternalRepo:
"""
download_results = []
failed = 0
root = PathInfo(self.root_dir)
paths = [PathInfo(self.root_dir) / path for path in paths]
paths = [root / path for path in paths]
def download_update(result):
download_results.append(result)
hash_infos = []
for path in paths:
if not self.repo_tree.exists(path):
raise PathMissingError(path, self.url)
hash_info = self.repo_tree.get_hash(
path, download_callback=download_update
)
self.local_cache.save(
path,
self.repo_tree,
hash_info,
save_link=False,
download_callback=download_update,
)
with reraise(FileNotFoundError, PathMissingError(path, self.url)):
metadata = self.repo_tree.metadata(path)
self._check_repo(path, metadata.repo)
repo = metadata.repo
hash_info = self._fetch_to_cache(path, repo, download_update)
hash_infos.append(hash_info)
return sum(download_results), failed, hash_infos
def _check_repo(self, path_info, repo):
if not repo:
return
repo_path = PathInfo(repo.root_dir)
if path_info == repo_path and isinstance(repo, Repo):
message = "Cannot fetch a complete DVC repository"
if repo.root_dir != self.root_dir:
rel = relpath(repo.root_dir, self.root_dir)
message += f" '{rel}'"
raise IsADVCRepoError(message)
def get_external(self, path, dest):
"""Convenience wrapper for fetch_external and checkout."""
if self.local_cache:
path_info = PathInfo(self.root_dir) / path
with reraise(FileNotFoundError, PathMissingError(path, self.url)):
metadata = self.repo_tree.metadata(path_info)
self._check_repo(path_info, metadata.repo)
if metadata.output_exists:
repo = metadata.repo
cache = repo.cache.local
# fetch DVC and git files to tmpdir cache, then checkout
_, _, save_infos = self.fetch_external([path])
self.local_cache.checkout(PathInfo(dest), save_infos[0])
save_info = self._fetch_to_cache(path_info, repo, None)
cache.checkout(PathInfo(dest), save_info)
else:
# git-only erepo with no cache, just copy files directly
# to dest
path = PathInfo(self.root_dir) / path
if not self.repo_tree.exists(path):
raise PathMissingError(path, self.url)
self.repo_tree.copytree(path, dest)
# git-only folder, just copy files directly to dest
tree = self._get_tree_for(metadata.repo) # ignore subrepos
tree.copytree(path_info, dest)
def _get_tree_for(self, repo, **kwargs):
"""
Provides a combined tree of a single repo with dvc + git/local tree.
"""
kw = {**self.tree_confs, **kwargs}
if "fetch" not in kw:
kw["fetch"] = True
return RepoTree(repo, **kw)
class ExternalRepo(Repo, BaseExternalRepo):
def __init__(self, root_dir, url, rev, for_write=False):
if for_write:
super().__init__(root_dir)
else:
root_dir = os.path.realpath(root_dir)
super().__init__(root_dir, scm=Git(root_dir), rev=rev)
self.url = url
self._set_cache_dir()
self._fix_upstream()
def get_checksum(self, path):
path_info = PathInfo(self.root_dir) / path
with reraise(FileNotFoundError, PathMissingError(path, self.url)):
metadata = self.repo_tree.metadata(path_info)
@wrap_with(threading.Lock())
def _set_cache_dir(self):
try:
cache_dir = CACHE_DIRS[self.url]
except KeyError:
cache_dir = CACHE_DIRS[self.url] = tempfile.mkdtemp("dvc-cache")
# skip subrepos to check for
tree = self._get_tree_for(metadata.repo)
return tree.get_hash(path_info)
self.cache.local.cache_dir = cache_dir
self._local_cache = self.cache.local
def _fix_upstream(self):
if not os.path.isdir(self.url):
return
try:
src_repo = Repo(self.url)
except NotDvcRepoError:
# If ExternalRepo does not throw NotDvcRepoError and Repo does,
# the self.url might be a bare git repo.
# NOTE: This will fail to resolve remote with relative path,
# same as if it was a remote DVC repo.
return
try:
remote_name = self.config["core"].get("remote")
if remote_name:
self._fix_local_remote(src_repo, remote_name)
else:
self._add_upstream(src_repo)
finally:
src_repo.close()
def _fix_local_remote(self, src_repo, remote_name):
@staticmethod
def _fix_local_remote(orig_repo, src_repo, remote_name):
# If a remote URL is relative to the source repo,
# it will have changed upon config load and made
# relative to this new repo. Restore the old one here.
new_remote = self.config["remote"][remote_name]
new_remote = orig_repo.config["remote"][remote_name]
old_remote = src_repo.config["remote"][remote_name]
if new_remote["url"] != old_remote["url"]:
new_remote["url"] = old_remote["url"]
def _add_upstream(self, src_repo):
@staticmethod
def _add_upstream(orig_repo, src_repo):
# Fill the empty upstream entry with a new remote pointing to the
# original repo's cache location.
cache_dir = src_repo.cache.local.cache_dir
self.config["remote"]["auto-generated-upstream"] = {"url": cache_dir}
self.config["core"]["remote"] = "auto-generated-upstream"
orig_repo.config["remote"]["auto-generated-upstream"] = {
"url": cache_dir
}
orig_repo.config["core"]["remote"] = "auto-generated-upstream"
def make_repo(self, path):
repo = Repo(path, scm=self.scm, rev=self.get_rev())
self._setup_cache(repo)
self._fix_upstream(repo)
return repo
def _setup_cache(self, repo):
repo.cache.local.cache_dir = self.cache_dir
if self.cache_types:
repo.cache.local.cache_types = self.cache_types
def _fix_upstream(self, repo):
if not os.path.isdir(self.url):
return
try:
rel_path = os.path.relpath(repo.root_dir, self.root_dir)
src_repo = Repo(PathInfo(self.url) / rel_path)
except NotDvcRepoError:
return
try:
remote_name = repo.config["core"].get("remote")
if remote_name:
self._fix_local_remote(repo, src_repo, remote_name)
else:
self._add_upstream(repo, src_repo)
finally:
src_repo.close()
@wrap_with(threading.Lock())
def _get_cache_dir(self):
try:
cache_dir = CACHE_DIRS[self.url]
except KeyError:
cache_dir = CACHE_DIRS[self.url] = tempfile.mkdtemp("dvc-cache")
return cache_dir
class ExternalRepo(BaseExternalRepo, Repo):
def __init__(
self,
root_dir,
url,
scm=None,
rev=None,
for_write=False,
cache_dir=None,
cache_types=None,
**kwargs,
):
super().__init__(root_dir, scm=scm, rev=rev)
self.url = url
self.for_write = for_write
self.cache_dir = cache_dir or self._get_cache_dir()
self.cache_types = cache_types
self._setup_cache(self)
self._fix_upstream(self)
self.tree_confs = kwargs
class ExternalGitRepo(BaseExternalRepo):
def __init__(self, root_dir, url, rev):
def __init__(
self,
root_dir,
url,
scm=None,
rev=None,
for_write=False,
cache_dir=None,
cache_types=None,
**kwargs,
):
self.root_dir = os.path.realpath(root_dir)
self.scm = scm
self.url = url
self.tree = self.scm.get_tree(rev)
self.for_write = for_write
self.cache_dir = cache_dir or self._get_cache_dir()
self.cache_types = cache_types
self.rev = rev
self.tree_confs = kwargs
self.config = {"cache": {"dir": self.cache_dir}}
self.cache = Cache(self)
if cache_types:
self.cache.local.cache_types = cache_types
self.state = StateNoop()
@cached_property
def scm(self):
return Git(self.root_dir)
def tree(self):
if self.scm:
return self.scm.get_tree(self.rev)
return LocalTree(self, {"url": self.root_dir})
def close(self):
if "scm" in self.__dict__:
if self.scm:
self.scm.close()
def find_out_by_relpath(self, path):
@@ -251,14 +332,14 @@ class ExternalGitRepo(BaseExternalRepo):
@contextmanager
def open_by_relpath(self, path, mode="r", encoding=None, **kwargs):
"""Opens a specified resource as a file object."""
tree = RepoTree(self)
path_info = PathInfo(self.root_dir) / path
try:
with tree.open(
path, mode=mode, encoding=encoding, **kwargs
with self.repo_tree.open(
path_info, mode=mode, encoding=encoding, **kwargs
) as fobj:
yield fobj
except FileNotFoundError:
raise PathMissingError(path, self.url)
except FileNotFoundError as exc:
raise PathMissingError(path, self.url) from exc
def _cached_clone(url, rev, for_write=False):
+9 -6
View File
@@ -66,7 +66,7 @@ class Repo:
from dvc.repo.status import status
from dvc.repo.update import update
def __init__(self, root_dir=None, scm=None, rev=None):
def __init__(self, root_dir=None, scm=None, rev=None, subrepos=False):
from dvc.cache import Cache
from dvc.data_cloud import DataCloud
from dvc.lock import make_lock
@@ -116,6 +116,8 @@ class Repo:
hardlink_lock=hardlink_lock,
friendly=True,
)
# used by RepoTree to determine if it should traverse subrepos
self.subrepos = subrepos
self.cache = Cache(self)
self.cloud = DataCloud(self)
@@ -581,19 +583,20 @@ class Repo:
path_parts = os.path.normpath(path).split(os.path.sep)
return self.DVC_DIR in path_parts
@cached_property
def repo_tree(self):
return RepoTree(self, subrepos=self.subrepos, fetch=True)
@contextmanager
def open_by_relpath(self, path, remote=None, mode="r", encoding=None):
"""Opens a specified resource as a file descriptor"""
tree = RepoTree(self, stream=True)
tree = RepoTree(self, stream=True, subrepos=True)
path = os.path.join(self.root_dir, path)
try:
with self.state:
with tree.open(
os.path.join(self.root_dir, path),
mode=mode,
encoding=encoding,
remote=remote,
path, mode=mode, encoding=encoding, remote=remote,
) as fobj:
yield fobj
except FileNotFoundError as exc:
+7 -5
View File
@@ -79,12 +79,14 @@ def _fetch_external(self, repo_url, repo_rev, files, jobs):
from dvc.external_repo import external_repo
failed, downloaded = 0, 0
cache = self.cache.local
try:
with external_repo(repo_url, repo_rev) as repo:
with repo.use_cache(self.cache.local):
d, f, _ = repo.fetch_external(files, jobs=jobs)
downloaded += d
failed += f
with external_repo(
repo_url, repo_rev, cache_dir=cache.cache_dir
) as repo:
d, f, _ = repo.fetch_external(files, jobs=jobs)
downloaded += d
failed += f
except CloneError:
failed += 1
logger.exception(
+14 -14
View File
@@ -35,21 +35,21 @@ def get(url, path, out=None, rev=None):
# and won't work with reflink/hardlink.
dpath = os.path.dirname(os.path.abspath(out))
tmp_dir = os.path.join(dpath, "." + str(shortuuid.uuid()))
try:
with external_repo(url=url, rev=rev) as repo:
if hasattr(repo, "cache"):
repo.cache.local.cache_dir = tmp_dir
# Try any links possible to avoid data duplication.
#
# Not using symlink, because we need to remove cache after we
# are done, and to make that work we would have to copy data
# over anyway before removing the cache, so we might just copy
# it right away.
#
# Also, we can't use theoretical "move" link type here, because
# the same cache file might be used a few times in a directory.
repo.cache.local.cache_types = ["reflink", "hardlink", "copy"]
# Try any links possible to avoid data duplication.
#
# Not using symlink, because we need to remove cache after we
# are done, and to make that work we would have to copy data
# over anyway before removing the cache, so we might just copy
# it right away.
#
# Also, we can't use theoretical "move" link type here, because
# the same cache file might be used a few times in a directory.
cache_types = ["reflink", "hardlink", "copy"]
try:
with external_repo(
url=url, rev=rev, cache_dir=tmp_dir, cache_types=cache_types
) as repo:
repo.get_external(path, out)
finally:
remove(tmp_dir)
+4 -6
View File
@@ -31,7 +31,9 @@ def ls(
"""
from dvc.external_repo import external_repo
with external_repo(url, rev) as repo:
# use our own RepoTree instance instead of repo.repo_tree since we want to
# fetch directory listings, but don't want to fetch file contents.
with external_repo(url, rev, fetch=False, stream=True) as repo:
path_info = PathInfo(repo.root_dir)
if path:
path_info /= path
@@ -50,14 +52,10 @@ def ls(
def _ls(repo, path_info, recursive=None, dvc_only=False):
from dvc.tree.repo import RepoTree
def onerror(exc):
raise exc
# use our own RepoTree instance instead of repo.repo_tree since we want to
# fetch directory listings, but don't want to fetch file contents.
tree = RepoTree(repo, stream=True)
tree = repo.repo_tree
infos = []
try:
+5 -1
View File
@@ -1,9 +1,12 @@
from dataclasses import dataclass, field
from typing import List
from typing import TYPE_CHECKING, List
from dvc.output import BaseOutput
from dvc.path_info import PathInfo
if TYPE_CHECKING:
from dvc.repo import Repo
@dataclass
class Metadata:
@@ -14,6 +17,7 @@ class Metadata:
# required field
path_info: PathInfo
repo: "Repo"
# computed fields
is_output: bool = field(init=False, default=False) # is it an output?
+1 -1
View File
@@ -268,6 +268,6 @@ class DvcTree(BaseTree): # pylint:disable=abstract-method
path_info = PathInfo(os.path.abspath(path_info))
outs = self._find_outs(path_info, strict=False, recursive=True)
meta = Metadata(path_info=path_info, outs=outs)
meta = Metadata(path_info=path_info, outs=outs, repo=self.repo)
meta.isdir = meta.isdir or self.check_isdir(meta.path_info, meta.outs)
return meta
+2 -1
View File
@@ -74,9 +74,10 @@ class GitTree(BaseTree): # pylint:disable=abstract-method
return cls(self, root)
def open(
self, path, mode="r", encoding="utf-8"
self, path, mode="r", encoding=None
): # pylint: disable=arguments-differ
assert mode in {"r", "rb"}
encoding = encoding or "utf-8"
relative_path = relpath(path, self.git.working_dir)
+6 -2
View File
@@ -364,7 +364,8 @@ class RepoTree(BaseTree): # pylint:disable=abstract-method
return self._main_repo.tree.hash_jobs
def metadata(self, path):
path_info = PathInfo(os.path.abspath(path))
abspath = os.path.abspath(path)
path_info = PathInfo(abspath)
tree, dvc_tree = self._get_tree_pair(path_info)
dvc_meta = None
@@ -379,7 +380,10 @@ class RepoTree(BaseTree): # pylint:disable=abstract-method
if not stat_result and not dvc_meta:
raise FileNotFoundError
meta = dvc_meta or Metadata(path_info=path_info)
meta = dvc_meta or Metadata(
path_info=path_info,
repo=self._get_repo(abspath) or self._main_repo,
)
isdir = bool(stat_result) and stat.S_ISDIR(stat_result.st_mode)
meta.isdir = meta.isdir or isdir
+67 -4
View File
@@ -3,10 +3,10 @@ import os
import pytest
from dvc import api
from dvc.api import UrlNotDvcRepoError
from dvc.exceptions import FileMissingError
from dvc.exceptions import FileMissingError, OutputNotFoundError
from dvc.path_info import URLInfo
from dvc.utils.fs import remove
from tests.unit.tree.test_repo import make_subrepo
cloud_names = [
"s3",
@@ -47,10 +47,10 @@ def test_get_url_external(erepo_dir, cloud):
def test_get_url_requires_dvc(tmp_dir, scm):
tmp_dir.scm_gen({"foo": "foo"}, commit="initial")
with pytest.raises(UrlNotDvcRepoError, match="not a DVC repository"):
with pytest.raises(OutputNotFoundError, match="output 'foo'"):
api.get_url("foo", repo=os.fspath(tmp_dir))
with pytest.raises(UrlNotDvcRepoError):
with pytest.raises(OutputNotFoundError, match="output 'foo'"):
api.get_url("foo", repo=f"file://{tmp_dir}")
@@ -173,3 +173,66 @@ def test_open_not_cached(dvc):
os.remove(metric_file)
with pytest.raises(FileMissingError):
api.read(metric_file)
@pytest.mark.parametrize("local_repo", [False, True])
def test_read_with_subrepos(tmp_dir, scm, local_cloud, local_repo):
tmp_dir.scm_gen("foo.txt", "foo.txt", commit="add foo.txt")
subrepo = tmp_dir / "dir" / "subrepo"
make_subrepo(subrepo, scm, config=local_cloud.config)
with subrepo.chdir():
subrepo.scm_gen({"lorem": "lorem"}, commit="add lorem")
subrepo.dvc_gen({"dir": {"file.txt": "file.txt"}}, commit="add dir")
subrepo.dvc_gen("dvc-file", "dvc-file", commit="add dir")
subrepo.dvc.push()
repo_path = None if local_repo else f"file:///{tmp_dir}"
subrepo_path = os.path.join("dir", "subrepo")
assert api.read("foo.txt", repo=repo_path) == "foo.txt"
assert (
api.read(os.path.join(subrepo_path, "lorem"), repo=repo_path)
== "lorem"
)
assert (
api.read(os.path.join(subrepo_path, "dvc-file"), repo=repo_path)
== "dvc-file"
)
assert (
api.read(os.path.join(subrepo_path, "dir", "file.txt"), repo=repo_path)
== "file.txt"
)
def test_get_url_granular(tmp_dir, dvc, s3):
tmp_dir.add_remote(config=s3.config)
tmp_dir.dvc_gen(
{"dir": {"foo": "foo", "bar": "bar", "nested": {"file": "file"}}}
)
expected_url = URLInfo(s3.url) / "ac/bd18db4cc2f85cedef654fccc4a4d8"
assert api.get_url("dir/foo") == expected_url
expected_url = URLInfo(s3.url) / "37/b51d194a7513e45b56f6524f2d51f2"
assert api.get_url("dir/bar") == expected_url
expected_url = URLInfo(s3.url) / "8c/7dd922ad47494fc02c388e12c00eac"
assert api.get_url(os.path.join("dir", "nested", "file")) == expected_url
def test_get_url_subrepos(tmp_dir, scm, local_cloud):
subrepo = tmp_dir / "subrepo"
make_subrepo(subrepo, scm, config=local_cloud.config)
with subrepo.chdir():
subrepo.dvc_gen(
{"dir": {"foo": "foo"}, "bar": "bar"}, commit="add files"
)
subrepo.dvc.push()
path = os.path.relpath(local_cloud.config["url"])
expected_url = os.path.join(path, "ac", "bd18db4cc2f85cedef654fccc4a4d8")
assert api.get_url(os.path.join("subrepo", "dir", "foo")) == expected_url
expected_url = os.path.join(path, "37", "b51d194a7513e45b56f6524f2d51f2")
assert api.get_url("subrepo/bar") == expected_url
+52 -1
View File
@@ -3,11 +3,13 @@ import os
from mock import ANY, patch
from dvc.external_repo import CLONES, external_repo
from dvc.hash_info import HashInfo
from dvc.path_info import PathInfo
from dvc.scm.git import Git
from dvc.tree.local import LocalTree
from dvc.utils import relpath
from dvc.utils.fs import remove
from dvc.utils.fs import makedirs, remove
from tests.unit.tree.test_repo import make_subrepo
def test_external_repo(erepo_dir):
@@ -173,3 +175,52 @@ def test_shallow_clone_tag(erepo_dir):
assert mock_clone.call_count == 1
_, shallow = CLONES[url]
assert not shallow
def test_subrepos_are_ignored(tmp_dir, erepo_dir):
subrepo = erepo_dir / "dir" / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
with erepo_dir.chdir():
erepo_dir.dvc_gen("dir/foo", "foo", commit="foo")
erepo_dir.scm_gen("dir/bar", "bar", commit="bar")
with subrepo.chdir():
subrepo.dvc_gen({"file": "file"}, commit="add files on subrepo")
with external_repo(os.fspath(erepo_dir)) as repo:
repo.get_external("dir", "out")
expected_files = {"foo": "foo", "bar": "bar", ".gitignore": "/foo\n"}
assert (tmp_dir / "out").read_text() == expected_files
expected_hash = HashInfo("md5", "e1d9e8eae5374860ae025ec84cfd85c7.dir")
assert (
repo.get_checksum(os.path.join(repo.root_dir, "dir"))
== expected_hash
)
# clear cache to test `fetch_external` again
cache_dir = tmp_dir / repo.cache.local.cache_dir
remove(cache_dir)
makedirs(cache_dir)
assert repo.fetch_external(["dir"]) == (
len(expected_files),
0,
[expected_hash],
)
def test_subrepos_are_ignored_for_git_tracked_dirs(tmp_dir, erepo_dir):
subrepo = erepo_dir / "dir" / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
with erepo_dir.chdir():
scm_files = {"foo": "foo", "bar": "bar", "subdir": {"lorem": "lorem"}}
erepo_dir.scm_gen({"dir": scm_files}, commit="add scm dir")
with subrepo.chdir():
subrepo.dvc_gen({"file": "file"}, commit="add files on subrepo")
with external_repo(os.fspath(erepo_dir)) as repo:
repo.get_external("dir", "out")
# subrepo files should not be here
assert (tmp_dir / "out").read_text() == scm_files
+44
View File
@@ -5,11 +5,13 @@ import pytest
from dvc.cache import Cache
from dvc.exceptions import PathMissingError
from dvc.external_repo import IsADVCRepoError
from dvc.main import main
from dvc.repo import Repo
from dvc.repo.get import GetDVCFileError
from dvc.system import System
from dvc.utils.fs import makedirs
from tests.unit.tree.test_repo import make_subrepo
def test_get_repo_file(tmp_dir, erepo_dir):
@@ -265,3 +267,45 @@ def test_get_mixed_dir(tmp_dir, erepo_dir):
"foo": "foo",
"bar": "bar",
}
@pytest.mark.parametrize("is_dvc", [True, False])
@pytest.mark.parametrize("files", [{"foo": "foo"}, {"dir": {"bar": "bar"}}])
def test_get_from_subrepos(tmp_dir, erepo_dir, is_dvc, files):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
gen = subrepo.dvc_gen if is_dvc else subrepo.scm_gen
with subrepo.chdir():
gen(files, commit="add files in subrepo")
key = next(iter(files))
Repo.get(os.fspath(erepo_dir), f"subrepo/{key}", out="out")
assert (tmp_dir / "out").read_text() == files[key]
def test_granular_get_from_subrepos(tmp_dir, erepo_dir):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
with subrepo.chdir():
subrepo.dvc_gen({"dir": {"bar": "bar"}}, commit="files in subrepo")
path = os.path.join("subrepo", "dir", "bar")
Repo.get(os.fspath(erepo_dir), path, out="out")
assert (tmp_dir / "out").read_text() == "bar"
def test_try_to_get_complete_repo(tmp_dir, dvc, erepo_dir):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
with subrepo.chdir():
subrepo.dvc_gen({"dir": {"bar": "bar"}}, commit="files in subrepo")
expected_message = "Cannot fetch a complete DVC repository"
with pytest.raises(IsADVCRepoError) as exc_info:
Repo.get(os.fspath(erepo_dir), "subrepo", out="out")
assert f"{expected_message} 'subrepo'" == str(exc_info.value)
with pytest.raises(IsADVCRepoError) as exc_info:
Repo.get(os.fspath(erepo_dir), ".", out="out")
assert expected_message == str(exc_info.value)
+84
View File
@@ -2,6 +2,7 @@ import filecmp
import os
import pytest
from funcy import first
from mock import patch
import dvc.data_cloud as cloud
@@ -9,9 +10,11 @@ from dvc.cache import Cache
from dvc.config import NoRemoteError
from dvc.dvcfile import Dvcfile
from dvc.exceptions import DownloadError, PathMissingError
from dvc.external_repo import IsADVCRepoError
from dvc.stage.exceptions import StagePathNotFoundError
from dvc.system import System
from dvc.utils.fs import makedirs, remove
from tests.unit.tree.test_repo import make_subrepo
def test_import(tmp_dir, scm, dvc, erepo_dir):
@@ -381,3 +384,84 @@ def test_import_mixed_dir(tmp_dir, dvc, erepo_dir):
"foo": "foo",
"bar": "bar",
}
@pytest.mark.parametrize("is_dvc", [True, False])
@pytest.mark.parametrize("files", [{"foo": "foo"}, {"dir": {"bar": "bar"}}])
def test_import_subrepos(tmp_dir, erepo_dir, dvc, scm, is_dvc, files):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
gen = subrepo.dvc_gen if is_dvc else subrepo.scm_gen
with subrepo.chdir():
gen(files, commit="add files in subrepo")
key = next(iter(files))
path = str((subrepo / key).relative_to(erepo_dir))
stage = dvc.imp(os.fspath(erepo_dir), path, out="out",)
assert (tmp_dir / "out").read_text() == files[key]
assert stage.deps[0].def_path == path
assert stage.deps[0].def_repo == {
"url": os.fspath(erepo_dir),
"rev_lock": erepo_dir.scm.get_rev(),
}
def test_granular_import_from_subrepos(tmp_dir, dvc, erepo_dir):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
with subrepo.chdir():
subrepo.dvc_gen({"dir": {"bar": "bar"}}, commit="files in subrepo")
path = os.path.join("subrepo", "dir", "bar")
stage = dvc.imp(os.fspath(erepo_dir), path, out="out")
assert (tmp_dir / "out").read_text() == "bar"
assert stage.deps[0].def_path == path
assert stage.deps[0].def_repo == {
"url": os.fspath(erepo_dir),
"rev_lock": erepo_dir.scm.get_rev(),
}
@pytest.mark.parametrize("is_dvc", [True, False])
@pytest.mark.parametrize("files", [{"foo": "foo"}, {"dir": {"bar": "bar"}}])
def test_pull_imported_stage_from_subrepos(
tmp_dir, dvc, erepo_dir, is_dvc, files
):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
gen = subrepo.dvc_gen if is_dvc else subrepo.scm_gen
with subrepo.chdir():
gen(files, commit="files in subrepo")
key = first(files)
path = os.path.join("subrepo", key)
dvc.imp(os.fspath(erepo_dir), path, out="out")
# clean everything
remove(dvc.cache.local.cache_dir)
remove("out")
makedirs(dvc.cache.local.cache_dir)
stats = dvc.pull(["out.dvc"])
expected = [f"out{os.sep}"] if isinstance(files[key], dict) else ["out"]
assert stats["added"] == expected
assert (tmp_dir / "out").read_text() == files[key]
def test_try_import_complete_repo(tmp_dir, dvc, erepo_dir):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
with subrepo.chdir():
subrepo.dvc_gen({"dir": {"bar": "bar"}}, commit="files in subrepo")
expected_message = "Cannot fetch a complete DVC repository"
with pytest.raises(IsADVCRepoError) as exc_info:
dvc.imp(os.fspath(erepo_dir), "subrepo", out="out")
assert f"{expected_message} 'subrepo'" == str(exc_info.value)
with pytest.raises(IsADVCRepoError) as exc_info:
dvc.imp(os.fspath(erepo_dir), os.curdir, out="out")
assert expected_message == str(exc_info.value)
+44
View File
@@ -1,6 +1,7 @@
import os
import shutil
import textwrap
from operator import itemgetter
import pytest
@@ -512,3 +513,46 @@ def test_ls_target(erepo_dir, use_scm):
{"isdir": False, "isexec": 0, "isout": False, "path": "bar"},
{"isdir": False, "isexec": 0, "isout": False, "path": "foo"},
]
@pytest.mark.parametrize(
"dvc_top_level, erepo",
[
(True, pytest.lazy_fixture("erepo_dir")),
(False, pytest.lazy_fixture("git_dir")),
],
)
def test_subrepo(dvc_top_level, erepo):
from tests.func.test_get import make_subrepo
dvc_files = {"foo.txt": "foo.txt", "dvc_dir": {"lorem": "lorem"}}
scm_files = {"bar.txt": "bar.txt", "scm_dir": {"ipsum": "ipsum"}}
subrepo = erepo / "subrepo"
make_subrepo(subrepo, erepo.scm)
for repo in [erepo, subrepo]:
with repo.chdir():
repo.scm_gen(scm_files, commit=f"scm track for top {repo}")
if hasattr(repo, "dvc"):
repo.dvc_gen(dvc_files, commit=f"dvc track for {repo}")
def _list_files(path=None):
return set(map(itemgetter("path"), Repo.ls(os.fspath(erepo), path)))
extras = {".dvcignore", ".gitignore"}
git_tracked_outputs = {"bar.txt", "scm_dir"}
dvc_files = {"dvc_dir", "foo.txt", "foo.txt.dvc", "dvc_dir.dvc"}
common_outputs = git_tracked_outputs | extras | dvc_files
top_level_outputs = (
common_outputs if dvc_top_level else git_tracked_outputs
)
assert _list_files() == top_level_outputs | {"subrepo"}
assert _list_files("subrepo") == common_outputs
assert _list_files("scm_dir") == {"ipsum"}
assert _list_files("subrepo/scm_dir") == {"ipsum"}
if dvc_top_level:
assert _list_files("dvc_dir") == {"lorem"}
assert _list_files("subrepo/dvc_dir") == {"lorem"}
+2 -3
View File
@@ -4,7 +4,6 @@ import textwrap
import pytest
from dvc.exceptions import InvalidArgumentError
from dvc.repo import Repo
from dvc.stage.exceptions import DuplicateStageName, InvalidStageName
from dvc.utils.serialize import dump_yaml, parse_yaml_for_update
@@ -306,9 +305,9 @@ def test_run_params_no_exec(tmp_dir, dvc):
{"outs": ["foo"], "deps": ["bar"], "name": "copy-foo-bar"},
],
)
def test_run_without_cmd(kwargs):
def test_run_without_cmd(tmp_dir, dvc, kwargs):
with pytest.raises(InvalidArgumentError) as exc:
Repo().run(**kwargs)
dvc.run(**kwargs)
assert "command is not specified" == str(exc.value)
+30
View File
@@ -3,6 +3,7 @@ import os
import pytest
from dvc.dvcfile import Dvcfile
from tests.unit.tree.test_repo import make_subrepo
@pytest.mark.parametrize("cached", [True, False])
@@ -282,3 +283,32 @@ def test_update_recursive(tmp_dir, dvc, erepo_dir):
assert stage1.deps[0].def_repo["rev_lock"] == new_rev
assert stage2.deps[0].def_repo["rev_lock"] == new_rev
assert stage3.deps[0].def_repo["rev_lock"] == new_rev
@pytest.mark.parametrize("is_dvc", [True, False])
def test_update_from_subrepos(tmp_dir, dvc, erepo_dir, is_dvc):
subrepo = erepo_dir / "subrepo"
make_subrepo(subrepo, erepo_dir.scm)
gen = subrepo.dvc_gen if is_dvc else subrepo.scm_gen
with subrepo.chdir():
gen("foo", "foo", commit="subrepo initial")
path = os.path.join("subrepo", "foo")
repo_path = os.fspath(erepo_dir)
dvc.imp(repo_path, path, out="out")
assert dvc.status() == {}
with subrepo.chdir():
gen("foo", "foobar", commit="subrepo second commit")
assert dvc.status()["out.dvc"][0]["changed deps"] == {
f"{path} ({repo_path})": "update available"
}
(stage,) = dvc.update(["out.dvc"])
assert (tmp_dir / "out").read_text() == "foobar"
assert stage.deps[0].def_path == os.path.join("subrepo", "foo")
assert stage.deps[0].def_repo == {
"url": repo_path,
"rev_lock": erepo_dir.scm.get_rev(),
}
+79
View File
@@ -0,0 +1,79 @@
import os
from unittest.mock import call
import pytest
from dvc.external_repo import external_repo
from tests.unit.tree.test_repo import make_subrepo
def test_hook_is_called(tmp_dir, erepo_dir, mocker):
subrepo_paths = [
"subrepo1",
"subrepo2",
os.path.join("dir", "subrepo3"),
os.path.join("dir", "subrepo4"),
"subrepo5",
os.path.join("subrepo5", "subrepo6"),
]
subrepos = [erepo_dir / path for path in subrepo_paths]
for repo in subrepos:
make_subrepo(repo, erepo_dir.scm)
for repo in subrepos + [erepo_dir]:
with repo.chdir():
repo.scm_gen("foo", "foo", commit=f"git add {repo}/foo")
repo.dvc_gen("bar", "bar", commit=f"dvc add {repo}/bar")
with external_repo(str(erepo_dir)) as repo:
spy = mocker.spy(repo, "make_repo")
list(repo.repo_tree.walk(repo.root_dir)) # drain
assert spy.call_count == len(subrepos)
paths = [os.path.join(repo.root_dir, path) for path in subrepo_paths]
spy.assert_has_calls([call(path) for path in paths], any_order=True)
@pytest.mark.parametrize("root_is_dvc", [False, True])
def test_subrepo_is_constructed_properly(
tmp_dir, scm, mocker, make_tmp_dir, root_is_dvc
):
if root_is_dvc:
make_subrepo(tmp_dir, scm)
subrepo = tmp_dir / "subrepo"
make_subrepo(subrepo, scm)
local_cache = subrepo.dvc.cache.local.cache_dir
tmp_dir.scm_gen("bar", "bar", commit="add bar")
subrepo.dvc_gen("foo", "foo", commit="add foo")
cache_dir = make_tmp_dir("temp-cache")
with external_repo(
str(tmp_dir), cache_dir=str(cache_dir), cache_types=["symlink"]
) as repo:
spy = mocker.spy(repo, "make_repo")
list(repo.repo_tree.walk(repo.root_dir)) # drain
assert spy.call_count == 1
subrepo = spy.return_value
assert repo.url == str(tmp_dir)
assert repo.cache_dir == str(cache_dir)
assert repo.cache.local.cache_dir == str(cache_dir)
assert subrepo.cache.local.cache_dir == str(cache_dir)
assert repo.cache_types == ["symlink"]
assert repo.cache.local.cache_types == ["symlink"]
assert subrepo.cache.local.cache_types == ["symlink"]
assert (
subrepo.config["remote"]["auto-generated-upstream"]["url"]
== local_cache
)
if root_is_dvc:
main_cache = tmp_dir.dvc.cache.local.cache_dir
assert repo.config["remote"]["auto-generated-upstream"][
"url"
] == str(main_cache)
+27 -1
View File
@@ -2,6 +2,7 @@ import pytest
from dvc.path_info import PathInfo
from dvc.tree.repo import RepoTree
from tests.unit.tree.test_repo import make_subrepo
@pytest.fixture(scope="module")
@@ -53,7 +54,7 @@ def repo_tree(temp_repo):
temp_repo.scm_gen(fs_structure, commit="repo init")
temp_repo.dvc_gen(dvc_structure, commit="use dvc")
yield RepoTree(temp_repo.dvc, fetch=True)
yield RepoTree(temp_repo.dvc, fetch=True, subrepos=True)
def test_metadata_not_existing(repo_tree):
@@ -78,6 +79,7 @@ def test_metadata_git_tracked_file(repo_tree, path):
meta = repo_tree.metadata(path)
assert meta.path_info == root / path
assert meta.repo.root_dir == repo_tree.root_dir
assert not meta.is_output
assert not meta.part_of_output
assert not meta.contains_outputs
@@ -116,6 +118,7 @@ def test_metadata_dvc_tracked_file(repo_tree, path, outs, is_output):
meta = repo_tree.metadata(path)
assert meta.path_info == root / path
assert meta.repo.root_dir == repo_tree.root_dir
assert meta.is_output == is_output
assert meta.part_of_output != is_output
assert not meta.contains_outputs
@@ -133,6 +136,7 @@ def test_metadata_git_only_dirs(repo_tree, path):
meta = repo_tree.metadata(path)
assert meta.path_info == root / path
assert meta.repo.root_dir == repo_tree.root_dir
assert not meta.is_output
assert not meta.part_of_output
assert not meta.contains_outputs
@@ -156,6 +160,7 @@ def test_metadata_git_dvc_mixed_dirs(repo_tree, path, expected_outs):
meta = repo_tree.metadata(root / path)
assert meta.path_info == root / path
assert meta.repo.root_dir == repo_tree.root_dir
assert not meta.is_output
assert not meta.part_of_output
assert meta.contains_outputs
@@ -184,6 +189,7 @@ def test_metadata_dvc_only_dirs(repo_tree, path, is_output):
meta = repo_tree.metadata(root / path)
assert meta.path_info == root / path
assert meta.repo.root_dir == repo_tree.root_dir
assert meta.is_output == is_output
assert meta.part_of_output != is_output
assert not meta.contains_outputs
@@ -193,3 +199,23 @@ def test_metadata_dvc_only_dirs(repo_tree, path, is_output):
assert not meta.is_exec
assert not meta.isfile
assert {out.path_info for out in meta.outs} == {data}
def test_metadata_on_subrepos(make_tmp_dir, temp_repo, repo_tree):
subrepo = temp_repo / "subrepo"
make_subrepo(subrepo, temp_repo.scm)
subrepo.scm_gen("foo", "foo", commit="add foo on subrepo")
subrepo.dvc_gen("foobar", "foobar", commit="add foobar on subrepo")
for path in ["subrepo", "subrepo/foo", "subrepo/foobar"]:
meta = repo_tree.metadata(temp_repo / path)
assert meta.repo.root_dir == str(
subrepo
), f"repo root didn't match for {path}"
# supports external outputs on top-level DVC repo
external_dir = make_tmp_dir("external-output")
external_dir.gen("bar", "bar")
temp_repo.dvc.add(str(external_dir / "bar"), external=True)
meta = repo_tree.metadata(external_dir / "bar")
assert meta.repo.root_dir == str(temp_repo)