dvc: bump to 1.3.0

pytest 6.0.0 had issue with `pylint`, complaining for `pytest.mark.*` being `not-callable`.

dvc/command/experiments.py

@@ -48,6 +48,10 @@ def _collect_rows(
         return val
 
     def _extend(row, names, items):
+        if not items:
+            row.extend(["-"] * len(names))
+            return
+
         for fname, item in items:
             if isinstance(item, dict):
                 item = flatten(item, ".")
@@ -62,13 +66,14 @@ def _collect_rows(
     for i, (rev, exp) in enumerate(experiments.items()):
         row = []
         style = None
+        queued = "*" if exp.get("queued", False) else ""
         if rev == "baseline":
             row.append(f"{base_rev}")
             style = "bold"
         elif i < len(experiments) - 1:
-            row.append(f"├── {rev[:7]}")
+            row.append(f"├── {queued}{rev[:7]}")
         else:
-            row.append(f"└── {rev[:7]}")
+            row.append(f"└── {queued}{rev[:7]}")
 
         _extend(row, metric_names, exp.get("metrics", {}).items())
         _extend(row, param_names, exp.get("params", {}).items())
@@ -229,6 +234,7 @@ def add_parser(subparsers, parent_parser):
     experiments_parser = subparsers.add_parser(
         "experiments",
         parents=[parent_parser],
+        aliases=["exp"],
         description=append_doc_link(EXPERIMENTS_HELP, "experiments"),
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )

dvc/command/repro.py

@@ -41,6 +41,9 @@ class CmdRepro(CmdBase):
                     recursive=self.args.recursive,
                     force_downstream=self.args.force_downstream,
                     experiment=self.args.experiment,
+                    queue=self.args.queue,
+                    run_all=self.args.run_all,
+                    jobs=self.args.jobs,
                 )
 
                 if len(stages) == 0:
@@ -174,4 +177,16 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help=argparse.SUPPRESS,
     )
+    repro_parser.add_argument(
+        "--queue", action="store_true", default=False, help=argparse.SUPPRESS
+    )
+    repro_parser.add_argument(
+        "--run-all",
+        action="store_true",
+        default=False,
+        help=argparse.SUPPRESS,
+    )
+    repro_parser.add_argument(
+        "-j", "--jobs", type=int, help=argparse.SUPPRESS, metavar="<number>"
+    )
     repro_parser.set_defaults(func=CmdRepro)

dvc/command/version.py

@@ -10,7 +10,7 @@ from dvc.command.base import CmdBaseNoRepo, append_doc_link
 from dvc.exceptions import DvcException, NotDvcRepoError
 from dvc.scm.base import SCMError
 from dvc.system import System
-from dvc.utils import is_binary, relpath
+from dvc.utils import relpath
 from dvc.utils.pkg import PKG
 from dvc.version import __version__
 
@@ -27,13 +27,18 @@ class CmdVersion(CmdBaseNoRepo):
     def run(self):
         from dvc.repo import Repo
 
+        package = PKG
+        if PKG is None:
+            package = ""
+        else:
+            package = f"({PKG})"
+
         info = [
-            f"DVC version: {__version__}",
-            f"Python version: {platform.python_version()}",
-            f"Platform: {platform.platform()}",
-            f"Binary: {is_binary()}",
-            f"Package: {PKG}",
-            f"Supported remotes: {self.get_supported_remotes()}",
+            f"DVC version: {__version__} {package}",
+            "---------------------------------",
+            f"Platform: Python {platform.python_version()} on "
+            f"{platform.platform()}",
+            f"Supports: {self.get_supported_remotes()}",
         ]
 
         try:
@@ -46,13 +51,13 @@ class CmdVersion(CmdBaseNoRepo):
             # `dvc config cache.shared group`.
             if os.path.exists(repo.cache.local.cache_dir):
                 info.append(
-                    "Cache: {}".format(self.get_linktype_support_info(repo))
+                    "Cache types: {}".format(
+                        self.get_linktype_support_info(repo)
+                    )
                 )
                 if psutil:
                     fs_type = self.get_fs_type(repo.cache.local.cache_dir)
-                    info.append(
-                        f"Filesystem type (cache directory): {fs_type}"
-                    )
+                    info.append(f"Cache directory: {fs_type}")
             else:
                 logger.warning(
                     "Unable to detect supported link types, as cache "
@@ -68,11 +73,11 @@ class CmdVersion(CmdBaseNoRepo):
             root_directory = os.getcwd()
             info.append("Repo: dvc, git (broken)")
         else:
-            info.append("Repo: {}".format(_get_dvc_repo_info(repo)))
+            if psutil:
+                fs_root = self.get_fs_type(os.path.abspath(root_directory))
+                info.append(f"Workspace directory: {fs_root}")
 
-        if psutil:
-            fs_root = self.get_fs_type(os.path.abspath(root_directory))
-            info.append(f"Filesystem type (workspace): {fs_root}")
+            info.append("Repo: {}".format(_get_dvc_repo_info(repo)))
 
         logger.info("\n".join(info))
         return 0
@@ -80,7 +85,7 @@ class CmdVersion(CmdBaseNoRepo):
     @staticmethod
     def get_fs_type(path):
         partition = {
-            pathlib.Path(part.mountpoint): (part.fstype, part.device)
+            pathlib.Path(part.mountpoint): (part.fstype + " on " + part.device)
             for part in psutil.disk_partitions(all=True)
         }
 
@@ -115,7 +120,9 @@ class CmdVersion(CmdBaseNoRepo):
                 os.unlink(dst)
             except DvcException:
                 status = "not supported"
-            cache.append(f"{name} - {status}")
+
+            if status == "supported":
+                cache.append(name)
         os.remove(src)
 
         return ", ".join(cache)
@@ -129,6 +136,12 @@ class CmdVersion(CmdBaseNoRepo):
             if not tree_cls.get_missing_deps():
                 supported_remotes.append(tree_cls.scheme)
 
+        if len(supported_remotes) == len(TREES):
+            return "All remotes"
+
+        if len(supported_remotes) == 1:
+            return supported_remotes
+
         return ", ".join(supported_remotes)
 
 

dvc/config.py

@@ -119,6 +119,16 @@ HTTP_COMMON = {
     "password": str,
     "ask_password": Bool,
 }
+WEBDAV_COMMON = {
+    "user": str,
+    "password": str,
+    "ask_password": Bool,
+    "token": str,
+    "cert_path": str,
+    "key_path": str,
+    "timeout": Coerce(int),
+}
+
 SCHEMA = {
     "core": {
         "remote": Lower,
@@ -199,6 +209,8 @@ SCHEMA = {
                 },
                 "http": {**HTTP_COMMON, **REMOTE_COMMON},
                 "https": {**HTTP_COMMON, **REMOTE_COMMON},
+                "webdav": {**WEBDAV_COMMON, **REMOTE_COMMON},
+                "webdavs": {**WEBDAV_COMMON, **REMOTE_COMMON},
                 "remote": {str: object},  # Any of the above options are valid
             }
         )

dvc/dependency/repo.py

@@ -67,7 +67,7 @@ class RepoDependency(LocalDependency):
                     return self.repo.cache.local.tree.get_hash(path, tree=tree)
                 return tree.get_file_hash(path)
 
-    def status(self):
+    def workspace_status(self):
         current_checksum = self._get_checksum(locked=True)
         updated_checksum = self._get_checksum(locked=False)
 
@@ -76,6 +76,9 @@ class RepoDependency(LocalDependency):
 
         return {}
 
+    def status(self):
+        return self.workspace_status()
+
     def save(self):
         pass
 

dvc/main.py

@@ -21,7 +21,7 @@ from dvc.utils import format_link
 logger = logging.getLogger("dvc")
 
 
-def main(argv=None):
+def main(argv=None):  # noqa: C901
     """Run dvc CLI command.
 
     Args:
@@ -37,16 +37,17 @@ def main(argv=None):
     try:
         args = parse_args(argv)
 
-        verbosity = args.verbose - args.quiet
-        if verbosity:
-            logger.setLevel(
-                {
-                    -2: logging.CRITICAL,
-                    -1: logging.ERROR,
-                    1: logging.DEBUG,
-                    2: logging.TRACE,
-                }[max(-2, min(verbosity, 2))]
-            )
+        level = None
+        if args.quiet:
+            level = logging.CRITICAL
+        elif args.verbose == 1:
+            level = logging.DEBUG
+        elif args.verbose > 1:
+            level = logging.TRACE
+
+        if level is not None:
+            logger.setLevel(level)
+
         logger.trace(args)
 
         cmd = args.func(args)
@@ -81,6 +82,15 @@ def main(argv=None):
     except Exception:  # noqa, pylint: disable=broad-except
         logger.exception("unexpected error")
         ret = 255
+
+    try:
+        if ret != 0:
+            logger.info(FOOTER)
+
+        if analytics.is_enabled():
+            analytics.collect_and_send_report(args, ret)
+
+        return ret
     finally:
         logger.setLevel(outerLogLevel)
 
@@ -92,11 +102,3 @@ def main(argv=None):
         # Remove cached repos in the end of the call, these are anonymous
         # so won't be reused by any other subsequent run anyway.
         clean_repos()
-
-    if ret != 0:
-        logger.info(FOOTER)
-
-    if analytics.is_enabled():
-        analytics.collect_and_send_report(args, ret)
-
-    return ret

dvc/output/base.py

@@ -199,10 +199,7 @@ class BaseOutput:
 
         return self.cache.changed_cache(self.checksum, filter_info=filter_info)
 
-    def status(self):
-        if self.checksum and self.use_cache and self.changed_cache():
-            return {str(self): "not in cache"}
-
+    def workspace_status(self):
         if not self.exists:
             return {str(self): "deleted"}
 
@@ -214,6 +211,12 @@ class BaseOutput:
 
         return {}
 
+    def status(self):
+        if self.checksum and self.use_cache and self.changed_cache():
+            return {str(self): "not in cache"}
+
+        return self.workspace_status()
+
     def changed(self):
         status = self.status()
         logger.debug(str(status))
@@ -280,6 +283,7 @@ class BaseOutput:
         self.info = self.save_info()
 
     def commit(self):
+        assert self.info
         if self.use_cache:
             self.cache.save(self.path_info, self.cache.tree, self.info)
 

dvc/path_info.py

@@ -315,3 +315,17 @@ class HTTPURLInfo(URLInfo):
             and self._path == other._path
             and self._extra_parts == other._extra_parts
         )
+
+
+# See https://github.com/shizacat/dvc/blob/remote-webdav/dvc/path_info.py
+class WebDAVURLInfo(HTTPURLInfo):
+    @cached_property
+    def url(self):
+        return "{}://{}{}{}{}{}".format(
+            self.scheme.replace("webdav", "http"),
+            self.netloc,
+            self._spath,
+            (";" + self.params) if self.params else "",
+            ("?" + self.query) if self.query else "",
+            ("#" + self.fragment) if self.fragment else "",
+        )

dvc/repo/experiments/__init__.py

@@ -1,21 +1,33 @@
 import logging
 import os
+import re
 import tempfile
-from contextlib import contextmanager
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from typing import Iterable, Optional
 
 from funcy import cached_property
 
 from dvc.exceptions import DvcException
+from dvc.repo.experiments.executor import ExperimentExecutor, LocalExecutor
 from dvc.scm.git import Git
 from dvc.stage.serialize import to_lockfile
 from dvc.utils import dict_sha256, env2bool, relpath
-from dvc.utils.fs import remove
+from dvc.utils.fs import copyfile, remove
 
 logger = logging.getLogger(__name__)
 
 
+def hash_exp(stages):
+    exp_data = {}
+    for stage in stages:
+        exp_data.update(to_lockfile(stage))
+    return dict_sha256(exp_data)
+
+
 class UnchangedExperimentError(DvcException):
-    pass
+    def __init__(self, rev):
+        super().__init__("Experiment identical to baseline '{rev[:7]}'.")
+        self.rev = rev
 
 
 class Experiments:
@@ -26,6 +38,11 @@ class Experiments:
     """
 
     EXPERIMENTS_DIR = "experiments"
+    PACKED_ARGS_FILE = "repro.dat"
+    STASH_MSG_PREFIX = "dvc-exp-"
+    STASH_EXPERIMENT_RE = re.compile(
+        r"(?:On \(.*\): )dvc-exp-(?P<baseline_rev>[0-9a-f]+)$"
+    )
 
     def __init__(self, repo):
         if not (
@@ -47,10 +64,13 @@ class Experiments:
             return Git(self.exp_dir)
         return self._init_clone()
 
+    @cached_property
+    def dvc_dir(self):
+        return relpath(self.repo.dvc_dir, self.repo.scm.root_dir)
+
     @cached_property
     def exp_dvc_dir(self):
-        dvc_dir = relpath(self.repo.dvc_dir, self.repo.scm.root_dir)
-        return os.path.join(self.exp_dir, dvc_dir)
+        return os.path.join(self.exp_dir, self.dvc_dir)
 
     @cached_property
     def exp_dvc(self):
@@ -59,19 +79,24 @@ class Experiments:
 
         return Repo(self.exp_dvc_dir)
 
-    @staticmethod
-    def exp_hash(stages):
-        exp_data = {}
-        for stage in stages:
-            exp_data.update(to_lockfile(stage))
-        return dict_sha256(exp_data)
+    @cached_property
+    def args_file(self):
+        return os.path.join(self.exp_dvc.tmp_dir, self.PACKED_ARGS_FILE)
 
-    @contextmanager
-    def chdir(self):
-        cwd = os.getcwd()
-        os.chdir(self.exp_dvc.root_dir)
-        yield
-        os.chdir(cwd)
+    @property
+    def stash_reflog(self):
+        if "refs/stash" in self.scm.repo.refs:
+            return self.scm.repo.refs["refs/stash"].log()
+        return []
+
+    @property
+    def stash_revs(self):
+        revs = {}
+        for i, entry in enumerate(self.stash_reflog):
+            m = self.STASH_EXPERIMENT_RE.match(entry.message)
+            if m:
+                revs[entry.newhexsha] = (i, m.group("baseline_rev"))
+        return revs
 
     def _init_clone(self):
         src_dir = self.repo.scm.root_dir
@@ -90,15 +115,18 @@ class Experiments:
 
     def _scm_checkout(self, rev):
         self.scm.repo.git.reset(hard=True)
+        if self.scm.repo.head.is_detached:
+            # switch back to default branch
+            self.scm.repo.heads[0].checkout()
         if not Git.is_sha(rev) or not self.scm.has_rev(rev):
-            self.scm.fetch(all=True)
+            self.scm.pull()
         logger.debug("Checking out base experiment commit '%s'", rev)
         self.scm.checkout(rev)
 
-    def _patch_exp(self):
-        """Create a patch based on the current (parent) workspace and apply it
-        to the experiment workspace.
+    def _stash_exp(self, *args, **kwargs):
+        """Stash changes from the current (parent) workspace as an experiment.
         """
+        rev = self.scm.get_rev()
         tmp = tempfile.NamedTemporaryFile(delete=False).name
         try:
             self.repo.scm.repo.git.diff(patch=True, output=tmp)
@@ -106,16 +134,28 @@ class Experiments:
                 logger.debug("Patching experiment workspace")
                 self.scm.repo.git.apply(tmp)
             else:
-                raise UnchangedExperimentError(
-                    "Experiment identical to baseline commit."
-                )
+                raise UnchangedExperimentError(rev)
         finally:
             remove(tmp)
+        self._pack_args(*args, **kwargs)
+        msg = f"{self.STASH_MSG_PREFIX}{rev}"
+        self.scm.repo.git.stash("push", "-m", msg)
+        return self.scm.resolve_rev("stash@{0}")
 
-    def _commit(self, stages, check_exists=True, branch=True, rev=None):
+    def _pack_args(self, *args, **kwargs):
+        ExperimentExecutor.pack_repro_args(self.args_file, *args, **kwargs)
+        self.scm.add(self.args_file)
+
+    def _unpack_args(self, tree=None):
+        args_file = os.path.join(self.exp_dvc.tmp_dir, self.PACKED_ARGS_FILE)
+        return ExperimentExecutor.unpack_repro_args(args_file, tree=tree)
+
+    def _commit(self, exp_hash, check_exists=True, branch=True):
         """Commit stages as an experiment and return the commit SHA."""
-        hash_ = self.exp_hash(stages)
-        exp_name = f"{rev[:7]}-{hash_}"
+        if not self.scm.is_dirty():
+            raise UnchangedExperimentError(self.scm.get_rev())
+        rev = self.scm.get_rev()
+        exp_name = f"{rev[:7]}-{exp_hash}"
         if branch:
             if check_exists and exp_name in self.scm.list_branches():
                 logger.debug("Using existing experiment branch '%s'", exp_name)
@@ -126,10 +166,30 @@ class Experiments:
         self.scm.commit(f"Add experiment {exp_name}")
         return self.scm.get_rev()
 
-    def _reproduce(self, *args, **kwargs):
-        """Run `dvc repro` inside the experiments workspace."""
-        with self.chdir():
-            return self.exp_dvc.reproduce(*args, **kwargs)
+    def reproduce_one(self, queue=False, **kwargs):
+        """Reproduce and checkout a single experiment."""
+        stash_rev = self.new(**kwargs)
+        if queue:
+            logger.info(
+                "Queued experiment '%s' for future execution.", stash_rev[:7]
+            )
+            return [stash_rev]
+        results = self.reproduce([stash_rev], keep_stash=False)
+        for exp_rev in results:
+            self.checkout_exp(exp_rev, force=True)
+        return results
+
+    def reproduce_queued(self, **kwargs):
+        results = self.reproduce(**kwargs)
+        if results:
+            revs = [f"{rev[:7]}" for rev in results]
+            logger.info(
+                "Successfully reproduced experiment(s) '%s'.\n"
+                "Use `dvc exp checkout <exp_rev>` to apply the results of "
+                "a specific experiment to your workspace.",
+                ", ".join(revs),
+            )
+        return results
 
     def new(self, *args, workspace=True, **kwargs):
         """Create a new experiment.
@@ -141,19 +201,142 @@ class Experiments:
         self._scm_checkout(rev)
         if workspace:
             try:
-                self._patch_exp()
+                stash_rev = self._stash_exp(*args, **kwargs)
             except UnchangedExperimentError as exc:
                 logger.info("Reproducing existing experiment '%s'.", rev[:7])
                 raise exc
         else:
             # configure params via command line here
             pass
-        self.exp_dvc.checkout()
-        stages = self._reproduce(*args, **kwargs)
-        exp_rev = self._commit(stages, rev=rev)
-        self.checkout_exp(exp_rev, force=True)
-        logger.info("Generated experiment '%s'.", exp_rev[:7])
-        return stages
+        logger.debug(
+            "Stashed experiment '%s' for future execution.", stash_rev[:7]
+        )
+        return stash_rev
+
+    def reproduce(
+        self,
+        revs: Optional[Iterable] = None,
+        keep_stash: Optional[bool] = True,
+        **kwargs,
+    ):
+        """Reproduce the specified experiments.
+
+        Args:
+            revs: If revs is not specified, all stashed experiments will be
+                reproduced.
+            keep_stash: If True, stashed experiments will be preserved if they
+                fail to reproduce successfully.
+        """
+        stash_revs = self.stash_revs
+
+        # to_run contains mapping of:
+        #   input_rev: (stash_index, baseline_rev)
+        # where input_rev contains the changes to execute (usually a stash
+        # commit) and baseline_rev is the baseline to compare output against.
+        # The final experiment commit will be branched from baseline_rev.
+        if revs is None:
+            to_run = {
+                rev: baseline_rev
+                for rev, (_, baseline_rev) in stash_revs.items()
+            }
+        else:
+            to_run = {
+                rev: stash_revs[rev][1] if rev in stash_revs else rev
+                for rev in revs
+            }
+
+        # setup executors
+        executors = {}
+        for rev, baseline_rev in to_run.items():
+            tree = self.scm.get_tree(rev)
+            repro_args, repro_kwargs = self._unpack_args(tree)
+            executor = LocalExecutor(
+                tree,
+                baseline_rev,
+                repro_args=repro_args,
+                repro_kwargs=repro_kwargs,
+                dvc_dir=self.dvc_dir,
+                cache_dir=self.repo.cache.local.cache_dir,
+            )
+            executors[rev] = executor
+
+        exec_results = self._reproduce(executors, **kwargs)
+
+        if keep_stash:
+            # only drop successfully run stashed experiments
+            to_drop = sorted(
+                (
+                    stash_revs[rev][0]
+                    for rev in exec_results
+                    if rev in stash_revs
+                ),
+                reverse=True,
+            )
+        else:
+            # drop all stashed experiments
+            to_drop = sorted(
+                (stash_revs[rev][0] for rev in to_run if rev in stash_revs),
+                reverse=True,
+            )
+        for index in to_drop:
+            self.scm.repo.git.stash("drop", index)
+
+        result = {}
+        for _, exp_result in exec_results.items():
+            result.update(exp_result)
+        return result
+
+    def _reproduce(self, executors: dict, jobs: Optional[int] = 1) -> dict:
+        """Run dvc repro for the specified ExperimentExecutors in parallel.
+
+        Returns dict containing successfully executed experiments.
+        """
+        result = {}
+
+        with ProcessPoolExecutor(max_workers=jobs) as workers:
+            futures = {}
+            for rev, executor in executors.items():
+                future = workers.submit(
+                    executor.reproduce,
+                    executor.dvc_dir,
+                    cwd=executor.dvc.root_dir,
+                    **executor.repro_kwargs,
+                )
+                futures[future] = (rev, executor)
+            for future in as_completed(futures):
+                rev, executor = futures[future]
+                exc = future.exception()
+                if exc is None:
+                    exp_hash = future.result()
+                    logger.debug(f"ran exp based on {executor.baseline_rev}")
+                    self._scm_checkout(executor.baseline_rev)
+                    self._collect_output(executor.baseline_rev, executor)
+                    remove(self.args_file)
+                    try:
+                        exp_rev = self._commit(exp_hash)
+                    except UnchangedExperimentError:
+                        logger.debug(
+                            "Experiment '%s' identical to baseline '%s'",
+                            rev,
+                            executor.baseline_rev,
+                        )
+                        exp_rev = executor.baseline_rev
+                    logger.info("Reproduced experiment '%s'.", exp_rev[:7])
+                    result[rev] = {exp_rev: exp_hash}
+                else:
+                    logger.exception(
+                        "Failed to reproduce experiment '%s'", rev
+                    )
+                executor.cleanup()
+
+        return result
+
+    def _collect_output(self, rev: str, executor: ExperimentExecutor):
+        logger.debug("copying tmp output from '%s'", executor.tmp_dir)
+        tree = self.scm.get_tree(rev)
+        for fname in tree.walk_files(tree.tree_root):
+            src = executor.path_info / relpath(fname, tree.tree_root)
+            copyfile(src, fname)
 
     def checkout_exp(self, rev, force=False):
         """Checkout an experiment to the user's workspace."""
@@ -162,7 +345,6 @@ class Experiments:
 
         if force:
             self.repo.scm.repo.git.reset(hard=True)
-        logger.debug(f"checkout {rev}")
         self._scm_checkout(rev)
 
         tmp = tempfile.NamedTemporaryFile(delete=False).name

dvc/repo/experiments/executor.py

@@ -0,0 +1,132 @@
+import logging
+import os
+import pickle
+from tempfile import TemporaryDirectory
+
+from funcy import cached_property
+
+from dvc.path_info import PathInfo
+from dvc.stage import PipelineStage
+from dvc.tree.base import BaseTree
+from dvc.utils import relpath
+from dvc.utils.fs import copy_fobj_to_file, makedirs
+
+logger = logging.getLogger(__name__)
+
+
+class ExperimentExecutor:
+    """Base class for executing experiments in parallel.
+
+    Args:
+        src_tree: source tree for this experiment.
+        baseline_rev: baseline revision that this experiment is derived from.
+
+    Optional keyword args:
+        repro_args: Args to be passed into reproduce.
+        repro_kwargs: Keyword args to be passed into reproduce.
+    """
+
+    def __init__(self, src_tree: BaseTree, baseline_rev: str, **kwargs):
+        self.src_tree = src_tree
+        self.baseline_rev = baseline_rev
+        self.repro_args = kwargs.pop("repro_args", [])
+        self.repro_kwargs = kwargs.pop("repro_kwargs", {})
+
+    def run(self):
+        pass
+
+    def cleanup(self):
+        pass
+
+    # TODO: come up with better way to stash repro arguments
+    @staticmethod
+    def pack_repro_args(path, *args, tree=None, **kwargs):
+        open_func = tree.open if tree else open
+        data = {"args": args, "kwargs": kwargs}
+        with open_func(path, "wb") as fobj:
+            pickle.dump(data, fobj)
+
+    @staticmethod
+    def unpack_repro_args(path, tree=None):
+        open_func = tree.open if tree else open
+        with open_func(path, "rb") as fobj:
+            data = pickle.load(fobj)
+        return data["args"], data["kwargs"]
+
+
+class LocalExecutor(ExperimentExecutor):
+    """Local machine exepriment executor."""
+
+    def __init__(self, src_tree: BaseTree, baseline_rev: str, **kwargs):
+        dvc_dir = kwargs.pop("dvc_dir")
+        cache_dir = kwargs.pop("cache_dir")
+        super().__init__(src_tree, baseline_rev, **kwargs)
+        self.tmp_dir = TemporaryDirectory()
+        logger.debug("Init local executor in dir '%s'.", self.tmp_dir)
+        self.dvc_dir = os.path.join(self.tmp_dir.name, dvc_dir)
+        try:
+            for fname in src_tree.walk_files(src_tree.tree_root):
+                dest = self.path_info / relpath(fname, src_tree.tree_root)
+                if not os.path.exists(dest.parent):
+                    makedirs(dest.parent)
+                with src_tree.open(fname, "rb") as fobj:
+                    copy_fobj_to_file(fobj, dest)
+        except Exception:
+            self.tmp_dir.cleanup()
+            raise
+        self._config(cache_dir)
+
+    def _config(self, cache_dir):
+        local_config = os.path.join(self.dvc_dir, "config.local")
+        logger.debug("Writing experiments local config '%s'", local_config)
+        with open(local_config, "w") as fobj:
+            fobj.write("[core]\n    no_scm = true\n")
+            fobj.write(f"[cache]\n    dir = {cache_dir}")
+
+    @cached_property
+    def dvc(self):
+        from dvc.repo import Repo
+
+        return Repo(self.dvc_dir)
+
+    @cached_property
+    def path_info(self):
+        return PathInfo(self.tmp_dir.name)
+
+    @staticmethod
+    def reproduce(dvc_dir, cwd=None, **kwargs):
+        """Run dvc repro and return the result."""
+        from dvc.repo import Repo
+        from dvc.repo.experiments import hash_exp
+
+        unchanged = []
+
+        def filter_pipeline(stage):
+            if isinstance(stage, PipelineStage):
+                unchanged.append(stage)
+
+        if cwd:
+            old_cwd = os.getcwd()
+            os.chdir(cwd)
+        else:
+            old_cwd = None
+            cwd = os.getcwd()
+
+        try:
+            logger.debug("Running repro in '%s'", cwd)
+            dvc = Repo(dvc_dir)
+            dvc.checkout()
+            stages = dvc.reproduce(on_unchanged=filter_pipeline, **kwargs)
+        finally:
+            if old_cwd is not None:
+                os.chdir(old_cwd)
+
+        # ideally we would return stages here like a normal repro() call, but
+        # stages is not currently picklable and cannot be returned across
+        # multiprocessing calls
+        return hash_exp(stages + unchanged)
+
+    def cleanup(self):
+        logger.debug("Removing tmpdir '%s'", self.tmp_dir)
+        self.tmp_dir.cleanup()
+        super().cleanup()

dvc/repo/experiments/show.py

@@ -12,7 +12,7 @@ logger = logging.getLogger(__name__)
 EXP_RE = re.compile(r"(?P<rev_sha>[a-f0-9]{7})-(?P<exp_sha>[a-f0-9]+)")
 
 
-def _collect_experiment(repo, branch):
+def _collect_experiment(repo, branch, stash=False):
     res = defaultdict(dict)
     for rev in repo.brancher(revs=[branch]):
         configs = _collect_configs(repo)
@@ -20,9 +20,10 @@ def _collect_experiment(repo, branch):
         if params:
             res["params"] = params
 
-        metrics = _collect_metrics(repo, None, False)
-        vals = _read_metrics(repo, metrics, rev)
-        if vals:
+        res["queued"] = stash
+        if not stash:
+            metrics = _collect_metrics(repo, None, False)
+            vals = _read_metrics(repo, metrics, rev)
             res["metrics"] = vals
 
     return res
@@ -37,8 +38,9 @@ def show(
     if revs is None:
         revs = [repo.scm.get_rev()]
 
-    revs = set(
-        repo.brancher(
+    revs = OrderedDict(
+        (rev, None)
+        for rev in repo.brancher(
             revs=revs,
             all_branches=all_branches,
             all_tags=all_tags,
@@ -49,6 +51,7 @@ def show(
     for rev in revs:
         res[rev]["baseline"] = _collect_experiment(repo, rev)
 
+    # collect reproduced experiments
     for exp_branch in repo.experiments.scm.list_branches():
         m = re.match(EXP_RE, exp_branch)
         if m:
@@ -61,4 +64,12 @@ def show(
                     )
                 res[rev][exp_rev] = experiment
 
+    # collect queued (not yet reproduced) experiments
+    for stash_rev, (_, baseline_rev) in repo.experiments.stash_revs.items():
+        with repo.experiments.chdir():
+            experiment = _collect_experiment(
+                repo.experiments.exp_dvc, stash_rev, stash=True
+            )
+        res[baseline_rev][stash_rev] = experiment
+
     return res

dvc/repo/reproduce.py

@@ -71,13 +71,20 @@ def reproduce(
         )
 
     experiment = kwargs.pop("experiment", False)
-    if experiment and self.experiments:
+    queue = kwargs.pop("queue", False)
+    run_all = kwargs.pop("run_all", False)
+    jobs = kwargs.pop("jobs", 1)
+    if (experiment or run_all) and self.experiments:
         try:
-            return self.experiments.new(
+            return _reproduce_experiments(
+                self,
                 target=target,
                 recursive=recursive,
                 all_pipelines=all_pipelines,
-                **kwargs
+                queue=queue,
+                run_all=run_all,
+                jobs=jobs,
+                **kwargs,
             )
         except UnchangedExperimentError:
             # If experiment contains no changes, just run regular repro
@@ -109,8 +116,14 @@ def reproduce(
     return _reproduce_stages(active_graph, targets, **kwargs)
 
 
+def _reproduce_experiments(repo, run_all=False, jobs=1, **kwargs):
+    if run_all:
+        return repo.experiments.reproduce_queued(jobs=jobs)
+    return repo.experiments.reproduce_one(**kwargs)
+
+
 def _reproduce_stages(
-    G, stages, downstream=False, single_item=False, **kwargs
+    G, stages, downstream=False, single_item=False, on_unchanged=None, **kwargs
 ):
     r"""Derive the evaluation of the given node for the given graph.
 
@@ -194,7 +207,10 @@ def _reproduce_stages(
                 # dependencies didn't change.
                 kwargs["force"] = True
 
-            result.extend(ret)
+            if ret:
+                result.extend(ret)
+            elif on_unchanged is not None:
+                on_unchanged(stage)
         except Exception as exc:
             raise ReproductionError(stage.relpath) from exc
 

dvc/scheme.py

@@ -9,3 +9,5 @@ class Schemes:
     GDRIVE = "gdrive"
     LOCAL = "local"
     OSS = "oss"
+    WEBDAV = "webdav"
+    WEBDAVS = "webdavs"

dvc/stage/__init__.py

@@ -396,7 +396,7 @@ class Stage(params.StageParams):
 
     @staticmethod
     def _changed_entries(entries):
-        return [str(entry) for entry in entries if entry.changed_checksum()]
+        return [str(entry) for entry in entries if entry.workspace_status()]
 
     def _changed_stage_entry(self):
         return f"'md5' of {self} changed."

dvc/tree/__init__.py

@@ -11,6 +11,8 @@ from .local import LocalTree
 from .oss import OSSTree
 from .s3 import S3Tree
 from .ssh import SSHTree
+from .webdav import WebDAVTree
+from .webdavs import WebDAVSTree
 
 TREES = [
     AzureTree,
@@ -22,6 +24,8 @@ TREES = [
     S3Tree,
     SSHTree,
     OSSTree,
+    WebDAVTree,
+    WebDAVSTree,
     # NOTE: LocalTree is the default
 ]
 

dvc/tree/gdrive.py

@@ -381,6 +381,9 @@ class GDriveTree(BaseTree):
         # it does not create a file on the remote
         gdrive_file = self._drive.CreateFile(param)
 
+        gdrive_file.FetchMetadata(fields="fileSize")
+        size = gdrive_file["fileSize"]
+
         with Tqdm(
             desc=progress_desc,
             disable=no_progress_bar,
@@ -388,7 +391,13 @@ class GDriveTree(BaseTree):
             # explicit `bar_format` as `total` will be set by `update_to`
             bar_format=Tqdm.BAR_FMT_DEFAULT,
         ) as pbar:
-            gdrive_file.GetContentFile(to_file, callback=pbar.update_to)
+            if size:
+                gdrive_file.GetContentFile(to_file, callback=pbar.update_to)
+            else:
+                # PyDrive doesn't like downloading empty files
+                # https://github.com/iterative/dvc/issues/4286
+                with open(to_file, "w"):
+                    pass
 
     @contextmanager
     @_gdrive_retry

dvc/tree/webdav.py

@@ -0,0 +1,253 @@
+import logging
+import os
+import threading
+from collections import deque
+
+from funcy import cached_property, wrap_prop
+
+from dvc.config import ConfigError
+from dvc.exceptions import DvcException
+from dvc.path_info import HTTPURLInfo, WebDAVURLInfo
+from dvc.progress import Tqdm
+from dvc.scheme import Schemes
+
+from .base import BaseTree
+from .http import ask_password
+
+logger = logging.getLogger(__name__)
+
+
+class WebDAVConnectionError(DvcException):
+    def __init__(self, host):
+        super().__init__(f"Unable to connect to WebDAV {host}.")
+
+
+class WebDAVTree(BaseTree):  # pylint:disable=abstract-method
+    # Use webdav scheme
+    scheme = Schemes.WEBDAV
+
+    # URLInfo for Webdav ~ replaces webdav -> http
+    PATH_CLS = WebDAVURLInfo
+
+    # Traversable as walk_files is implemented
+    CAN_TRAVERSE = True
+
+    # Length of walk_files prefix
+    TRAVERSE_PREFIX_LEN = 2
+
+    # Implementation based on webdav3.client
+    REQUIRES = {"webdavclient3": "webdav3.client"}
+
+    # Chunk size for buffered upload/download with progress bar
+    CHUNK_SIZE = 2 ** 16
+
+    # Constructor
+    def __init__(self, repo, config):
+        # Call BaseTree constructor
+        super().__init__(repo, config)
+
+        # Get username from configuration
+        self.user = config.get("user", None)
+
+        # Get password from configuration (might be None ~ not set)
+        self.password = config.get("password", None)
+
+        # Whether to ask for password if it is not set
+        self.ask_password = config.get("ask_password", False)
+
+        # Use token for webdav auth
+        self.token = config.get("token", None)
+
+        # Path to certificate
+        self.cert_path = config.get("cert_path", None)
+
+        # Path to private key
+        self.key_path = config.get("key_path", None)
+
+        # Connection timeout
+        self.timeout = config.get("timeout", 30)
+
+        # Get URL from configuration
+        self.url = config.get("url", None)
+
+        # If URL in config parse path_info
+        if self.url:
+            self.path_info = self.PATH_CLS(self.url)
+
+            # If username not specified try to use from URL
+            if self.user is None and self.path_info.user is not None:
+                self.user = self.path_info.user
+
+            # If username specified add to path_info
+            if self.user is not None:
+                self.path_info.user = self.user
+        else:
+            self.path_info = None
+
+    # Webdav client
+    @wrap_prop(threading.Lock())
+    @cached_property
+    def _client(self):
+        from webdav3.client import Client
+
+        # Construct hostname from path_info by stripping path
+        http_info = HTTPURLInfo(self.path_info.url)
+        hostname = http_info.replace(path="").url
+
+        # Set password or ask for it
+        if self.ask_password and self.password is None and self.token is None:
+            host, user = self.path_info.host, self.path_info.user
+            self.password = ask_password(host, user)
+
+        # Setup webdav client options dictionary
+        options = {
+            "webdav_hostname": hostname,
+            "webdav_login": self.user,
+            "webdav_password": self.password,
+            "webdav_token": self.token,
+            "webdav_cert_path": self.cert_path,
+            "webdav_key_path": self.key_path,
+            "webdav_timeout": self.timeout,
+            "webdav_chunk_size": self.CHUNK_SIZE,
+        }
+
+        client = Client(options)
+
+        # Check whether client options are valid
+        if not client.valid():
+            raise ConfigError(
+                f"Configuration for WebDAV {hostname} is invalid."
+            )
+
+        # Check whether connection is valid (root should always exist)
+        if not client.check(self.path_info.path):
+            raise WebDAVConnectionError(hostname)
+
+        return client
+
+    # Checks whether file/directory exists at remote
+    def exists(self, path_info, use_dvcignore=True):
+        # Use webdav check to test for file existence
+        return self._client.check(path_info.path)
+
+    # Gets file hash 'etag'
+    def get_file_hash(self, path_info):
+        # Use webdav client info method to get etag
+        etag = self._client.info(path_info.path)["etag"].strip('"')
+
+        # From HTTPTree
+        if not etag:
+            raise DvcException(
+                "could not find an ETag or "
+                "Content-MD5 header for '{url}'".format(url=path_info.url)
+            )
+
+        if etag.startswith("W/"):
+            raise DvcException(
+                "Weak ETags are not supported."
+                " (Etag: '{etag}', URL: '{url}')".format(
+                    etag=etag, url=path_info.url
+                )
+            )
+
+        return etag
+
+    # Checks whether path points to directory
+    def isdir(self, path_info):
+        # Use webdav is_dir to test whether path points to a directory
+        return self._client.is_dir(path_info.path)
+
+    # Yields path info to all files
+    def walk_files(self, path_info, **kwargs):
+        # Check whether directory exists
+        if not self.exists(path_info):
+            return
+
+        # Collect directories
+        dirs = deque([path_info.path])
+
+        # Iterate all directories found so far
+        while dirs:
+            # Iterate directory content
+            for entry in self._client.list(dirs.pop(), get_info=True):
+                # Construct path_info to entry
+                info = path_info.replace(path=entry["path"])
+
+                # Check whether entry is a directory
+                if entry["isdir"]:
+                    # Append new found directory to directory list
+                    dirs.append(info.path)
+                else:
+                    # Yield path info to non directory
+                    yield info
+
+    # Removes file/directory
+    def remove(self, path_info):
+        # Use webdav client clean (DELETE) method to remove file/directory
+        self._client.clean(path_info.path)
+
+    # Creates directories
+    def makedirs(self, path_info):
+        # Terminate recursion
+        if path_info.path == self.path_info.path or self.exists(path_info):
+            return
+
+        # Recursively descent to root
+        self.makedirs(path_info.parent)
+
+        # Construct directory at current recursion depth
+        self._client.mkdir(path_info.path)
+
+    # Moves file/directory at remote
+    def move(self, from_info, to_info, mode=None):
+        # Webdav client move
+        self._client.move(from_info.path, to_info.path)
+
+    # Downloads file from remote to file
+    def _download(self, from_info, to_file, name=None, no_progress_bar=False):
+        # Progress from HTTPTree
+        with open(to_file, "wb") as fd:
+            with Tqdm.wrapattr(
+                fd,
+                "write",
+                total=None if no_progress_bar else self._file_size(from_info),
+                leave=False,
+                desc=from_info.url if name is None else name,
+                disable=no_progress_bar,
+            ) as fd_wrapped:
+                # Download from WebDAV via buffer
+                self._client.download_from(
+                    buff=fd_wrapped, remote_path=from_info.path
+                )
+
+    # Uploads file to remote
+    def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
+        # First try to create parent directories
+        self.makedirs(to_info.parent)
+
+        # Progress from HTTPTree
+        def chunks():
+            with open(from_file, "rb") as fd:
+                with Tqdm.wrapattr(
+                    fd,
+                    "read",
+                    total=None
+                    if no_progress_bar
+                    else os.path.getsize(from_file),
+                    leave=False,
+                    desc=to_info.url if name is None else name,
+                    disable=no_progress_bar,
+                ) as fd_wrapped:
+                    while True:
+                        chunk = fd_wrapped.read(self.CHUNK_SIZE)
+                        if not chunk:
+                            break
+                        yield chunk
+
+        # Upload to WebDAV via buffer
+        self._client.upload_to(buff=chunks(), remote_path=to_info.path)
+
+    # Queries size of file at remote
+    def _file_size(self, path_info):
+        # Get file size from info dictionary and convert to int (from str)
+        return int(self._client.info(path_info.path)["size"])

dvc/tree/webdavs.py

@@ -0,0 +1,7 @@
+from dvc.scheme import Schemes
+
+from .webdav import WebDAVTree
+
+
+class WebDAVSTree(WebDAVTree):  # pylint:disable=abstract-method
+    scheme = Schemes.WEBDAVS

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "1.2.2"
+_BASE_VERSION = "1.3.0"
 
 
 def _generate_version(base_version):

setup.py

@@ -93,18 +93,19 @@ azure = ["azure-storage-blob==2.1.0", "knack"]
 oss = ["oss2==2.6.1"]
 ssh = ["paramiko>=2.5.0"]
 hdfs = ["pyarrow>=0.17.0"]
+webdav = ["webdavclient3>=3.14.5"]
 # gssapi should not be included in all_remotes, because it doesn't have wheels
 # for linux and mac, so it will fail to compile if user doesn't have all the
 # requirements, including kerberos itself. Once all the wheels are available,
 # we can start shipping it by default.
 ssh_gssapi = ["paramiko[gssapi]>=2.5.0"]
-all_remotes = gs + s3 + azure + ssh + oss + gdrive + hdfs
+all_remotes = gs + s3 + azure + ssh + oss + gdrive + hdfs + webdav
 
 # Extra dependecies to run tests
 tests_requirements = [
     "wheel>=0.31.1",
     # Test requirements:
-    "pytest>=4.6.0",
+    "pytest>=6.0.1",
     "pytest-docker>=0.7.2",
     "pytest-timeout>=1.3.3",
     "pytest-cov>=2.6.1",
@@ -160,6 +161,7 @@ setup(
         "ssh": ssh,
         "ssh_gssapi": ssh_gssapi,
         "hdfs": hdfs,
+        "webdav": webdav,
         "tests": tests_requirements,
     },
     keywords="data-science data-version-control machine-learning git"

tests/func/experiments/test_show.py

@@ -15,5 +15,6 @@ def test_show_simple(tmp_dir, scm, dvc):
         "baseline": {
             "metrics": {"metrics.yaml": {"foo": 1}},
             "params": {"params.yaml": {"foo": 1}},
+            "queued": False,
         }
     }

tests/func/test_commit.py

@@ -2,7 +2,9 @@ from os import fspath
 
 import pytest
 
+from dvc.dependency.base import DependencyDoesNotExistError
 from dvc.dvcfile import PIPELINE_FILE
+from dvc.output.base import OutputDoesNotExistError
 from dvc.stage.exceptions import StageCommitError
 from dvc.utils.yaml import dump_yaml, load_yaml
 
@@ -86,6 +88,24 @@ def test_commit_no_exec(tmp_dir, dvc):
     assert dvc.status(stage.path) == {}
 
 
+def test_commit_no_exec_missing_dep(tmp_dir, dvc):
+    stage = dvc.run(
+        name="my", cmd="mycmd", deps=["dep"], outs=["out"], no_exec=True
+    )
+    assert dvc.status(stage.path)
+
+    with pytest.raises(DependencyDoesNotExistError):
+        dvc.commit(stage.path, force=True)
+
+
+def test_commit_no_exec_missing_out(tmp_dir, dvc):
+    stage = dvc.run(name="my", cmd="mycmd", outs=["out"], no_exec=True)
+    assert dvc.status(stage.path)
+
+    with pytest.raises(OutputDoesNotExistError):
+        dvc.commit(stage.path, force=True)
+
+
 def test_commit_pipeline_stage(tmp_dir, dvc, run_copy):
     tmp_dir.gen("foo", "foo")
     stage = run_copy("foo", "bar", no_commit=True, name="copy-foo-bar")

tests/func/test_status.py

@@ -1,28 +1,34 @@
 import os
 
-from mock import patch
-
 from dvc.main import main
-from tests.basic_env import TestDvc
 
 
-class TestStatus(TestDvc):
-    def test_quiet(self):
-        self.dvc.add(self.FOO)
+def test_quiet(tmp_dir, dvc, capsys):
+    tmp_dir.dvc_gen("foo", "foo")
 
-        ret = main(["status", "--quiet"])
-        self.assertEqual(ret, 0)
+    # clear
+    capsys.readouterr()
 
-        os.remove(self.FOO)
-        os.rename(self.BAR, self.FOO)
+    assert main(["status", "--quiet"]) == 0
+    out_err = capsys.readouterr()
+    assert not out_err.out
+    assert not out_err.err
 
-        ret = main(["status", "--quiet"])
-        self.assertEqual(ret, 1)
+    tmp_dir.gen("foo", "barr")
 
-    @patch("dvc.repo.status._cloud_status", return_value=True)
-    def test_implied_cloud(self, mock_status):
-        main(["status", "--remote", "something"])
-        mock_status.assert_called()
+    assert main(["status", "--quiet"]) == 1
+    out_err = capsys.readouterr()
+    assert not out_err.out
+    assert not out_err.err
+
+
+def test_implied_cloud(dvc, mocker):
+    mock_status = mocker.patch(
+        "dvc.repo.status._cloud_status", return_value=True
+    )
+
+    main(["status", "--remote", "something"])
+    assert mock_status.called
 
 
 def test_status_non_dvc_repo_import(tmp_dir, dvc, git_dir):

tests/func/test_version.py

@@ -16,15 +16,10 @@ def test_info_in_repo(scm_init, tmp_dir, caplog):
 
     assert main(["version"]) == 0
 
-    assert re.search(r"DVC version: \d+\.\d+\.\d+", caplog.text)
-    assert re.search(r"Python version: \d\.\d\.\d", caplog.text)
-    assert re.search(r"Platform: .*", caplog.text)
-    assert re.search(r"Binary: (True|False)", caplog.text)
-    assert re.search(r"Package: .*", caplog.text)
-    assert re.search(r"Supported remotes: .*", caplog.text)
-    assert re.search(
-        r"(Cache: (.*link - (not )?supported(,\s)?){3})", caplog.text
-    )
+    assert re.search(r"DVC version: \d+\.\d+\.\d+\+.*", caplog.text)
+    assert re.search(r"Platform: Python \d\.\d\.\d on .*", caplog.text)
+    assert re.search(r"Supports: .*", caplog.text)
+    assert re.search(r"Cache types: .*", caplog.text)
 
     if scm_init:
         assert "Repo: dvc, git" in caplog.text
@@ -58,26 +53,20 @@ def test_fs_info_in_repo(tmp_dir, dvc, caplog):
     os.mkdir(dvc.cache.local.cache_dir)
     assert main(["version"]) == 0
 
-    assert "Filesystem type (cache directory): " in caplog.text
-    assert "Filesystem type (workspace): " in caplog.text
+    assert re.search(r"Cache directory: .* on .*", caplog.text)
+    assert re.search(r"Workspace directory: .* on .*", caplog.text)
 
 
 def test_info_outside_of_repo(tmp_dir, caplog):
     assert main(["version"]) == 0
 
-    assert re.search(r"DVC version: \d+\.\d+\.\d+", caplog.text)
-    assert re.search(r"Python version: \d\.\d\.\d", caplog.text)
-    assert re.search(r"Platform: .*", caplog.text)
-    assert re.search(r"Binary: (True|False)", caplog.text)
-    assert re.search(r"Package: .*", caplog.text)
-    assert re.search(r"Supported remotes: .*", caplog.text)
-    assert not re.search(r"(Cache: (.*link - (not )?(,\s)?){3})", caplog.text)
+    assert re.search(r"DVC version: \d+\.\d+\.\d+\+.*", caplog.text)
+    assert re.search(r"Platform: Python \d\.\d\.\d on .*", caplog.text)
+    assert re.search(r"Supports: .*", caplog.text)
+    assert not re.search(r"Cache types: .*", caplog.text)
     assert "Repo:" not in caplog.text
 
 
 @pytest.mark.skipif(psutil is None, reason="No psutil.")
 def test_fs_info_outside_of_repo(tmp_dir, caplog):
     assert main(["version"]) == 0
-
-    assert "Filesystem type (cache directory): " not in caplog.text
-    assert "Filesystem type (workspace): " in caplog.text

tests/unit/command/test_repro.py

@@ -15,6 +15,9 @@ default_arguments = {
     "recursive": False,
     "force_downstream": False,
     "experiment": False,
+    "queue": False,
+    "run_all": False,
+    "jobs": None,
 }
 
 

tests/unit/remote/test_webdav.py

@@ -0,0 +1,42 @@
+from dvc.tree.webdav import WebDAVTree
+
+# Test configuration
+url = "webdavs://example.com/public.php/webdav"
+user = "username"
+userurl = f"webdavs://{user}@example.com/public.php/webdav"
+password = "password"
+
+
+# Test minimum requiered configuration (url)
+def test_init(dvc):
+    config = {"url": url}
+    tree = WebDAVTree(dvc, config)
+
+    assert tree.path_info == url
+
+
+# Test username from configuration
+def test_user(dvc):
+    config = {"url": url, "user": user}
+    tree = WebDAVTree(dvc, config)
+
+    assert tree.user == user
+    assert tree.path_info.user == user
+
+
+# Test username extraction from url
+def test_userurl(dvc):
+    config = {"url": userurl}
+    tree = WebDAVTree(dvc, config)
+
+    assert tree.path_info == userurl
+    assert tree.user == user
+    assert tree.path_info.user == user
+
+
+# test password from config
+def test_password(dvc):
+    config = {"url": url, "user": user, "password": password}
+    tree = WebDAVTree(dvc, config)
+
+    assert tree.password == password