build: bump s3/gcsfs from 2021.10.1 to 2021.11.0

Bumps [gcsfs](https://github.com/fsspec/gcsfs) from 2021.10.1 to 2021.11.0.
- [Release notes](https://github.com/fsspec/gcsfs/releases)
Bumps [s3fs](https://github.com/fsspec/s3fs) from 2021.10.1 to 2021.11.0.
- [Release notes](https://github.com/fsspec/s3fs/releases)
- [Changelog](https://github.com/fsspec/s3fs/blob/main/release-procedure.md)

---
updated-dependencies:
- dependency-name: s3fs
  dependency-type: direct:development
  update-type: version-update:semver-minor
- dependency-name: gcsfs
  dependency-type: direct:development
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>

.github/workflows/packages.yaml

@@ -10,7 +10,7 @@ jobs:
   deb:
     runs-on: ubuntu-18.04
     steps:
-    - uses: actions/checkout@v2.3.5
+    - uses: actions/checkout@v2.4.0
       with:
           fetch-depth: 0
     - name: Set up Python 3.8
@@ -41,7 +41,7 @@ jobs:
   rpm:
     runs-on: ubuntu-18.04
     steps:
-    - uses: actions/checkout@v2.3.5
+    - uses: actions/checkout@v2.4.0
       with:
         fetch-depth: 0
     - name: Set up Python 3.8
@@ -72,7 +72,7 @@ jobs:
   osxpkg:
     runs-on: macos-10.15
     steps:
-    - uses: actions/checkout@v2.3.5
+    - uses: actions/checkout@v2.4.0
       with:
         fetch-depth: 0
     - name: Set up Python 3.8
@@ -106,7 +106,7 @@ jobs:
   exe:
     runs-on: windows-2019
     steps:
-    - uses: actions/checkout@v2.3.5
+    - uses: actions/checkout@v2.4.0
       with:
         fetch-depth: 0
     - name: Set up Python 3.8
@@ -136,7 +136,7 @@ jobs:
   pip:
     runs-on: ubuntu-18.04
     steps:
-    - uses: actions/checkout@v2.3.5
+    - uses: actions/checkout@v2.4.0
       with:
         fetch-depth: 0
     - name: Set up Python 3.7

.github/workflows/tests.yaml

@@ -20,7 +20,7 @@ jobs:
       uses: styfle/[email protected]
       with:
         access_token: ${{ github.token }}
-    - uses: actions/checkout@v2.3.5
+    - uses: actions/checkout@v2.4.0
       with:
         fetch-depth: 0
     - name: Set up Python 3.8
@@ -51,7 +51,7 @@ jobs:
           - os: windows-latest
             pyv: "3.10"
     steps:
-    - uses: actions/checkout@v2.3.5
+    - uses: actions/checkout@v2.4.0
       with:
         fetch-depth: 0
     - name: Set up Python
@@ -107,7 +107,7 @@ jobs:
         OSS_ACCESS_KEY_SECRET: ${{ secrets.OSS_ACCESS_KEY_SECRET}}
         OSS_ENDPOINT: ${{ secrets.OSS_ENDPOINT}}
       run: >-
-        python -m tests -n=4
+        python -m tests -n=auto
         --cov-report=xml --cov-report=term
         --enable-ssh ${{ env.extra_test_args }}
     - name: upload coverage report

.pre-commit-config.yaml

@@ -1,12 +1,12 @@
 ci:
-  skip: [pylint, dvc-pre-commit]
+  skip: [mypy, pylint, dvc-pre-commit]
 
 repos:
   - hooks:
       - id: black
         language_version: python3
     repo: https://github.com/ambv/black
-    rev: 21.9b0
+    rev: 21.10b0
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.0.1
     hooks:
@@ -34,19 +34,14 @@ repos:
           - flake8-string-format
     repo: https://gitlab.com/pycqa/flake8
     rev: 3.9.2
-  - hooks:
-      -  id: mypy
-         language_version: python3
-         files: ^dvc/
-         additional_dependencies:
-           - types-requests
-           - types-paramiko
-           - types-tabulate
-           - types-toml
-    repo: https://github.com/pre-commit/mirrors-mypy
-    rev: 'v0.910-1'
   - repo: local
     hooks:
+    - id: mypy
+      name: mypy
+      entry: mypy
+      files: ^dvc/
+      language: system
+      types: [python]
     - id: pylint
       name: pylint
       entry: pylint

dvc/command/experiments.py

@@ -806,6 +806,10 @@ class CmdExperimentsInit(CmdBase):
         "plots": PLOTS,
         "live": DVCLIVE,
     }
+    EXP_LINK = (
+        "https://dvc.org/doc"
+        "/user-guide/experiment-management/running-experiments"
+    )
 
     def run(self):
         from dvc.command.stage import parse_cmd
@@ -843,23 +847,37 @@ class CmdExperimentsInit(CmdBase):
             interactive=self.args.interactive,
             force=self.args.force,
         )
+
+        name = self.args.name or self.args.type
+
+        text = ui.rich_text.assemble(
+            "\n" if self.args.interactive else "",
+            "Created ",
+            (name, "bright_blue"),
+            " stage in ",
+            ("dvc.yaml", "green"),
+            ".",
+        )
+        if not self.args.run:
+            text.append_text(
+                ui.rich_text.assemble(
+                    " To run, use ",
+                    ('"dvc exp run"', "green"),
+                    ".\nSee ",
+                    (self.EXP_LINK, "repr.url"),
+                    ".",
+                )
+            )
+
+        ui.write(text, styled=True)
         if self.args.run:
             return self.repo.experiments.run(
                 targets=[initialized_stage.addressing]
             )
+
         return 0
 
 
-class RawDefaultsHelpFormatter(
-    argparse.RawDescriptionHelpFormatter,
-    argparse.ArgumentDefaultsHelpFormatter,
-):
-    def _get_help_string(self, action: argparse.Action) -> Optional[str]:
-        if action.default:
-            return super()._get_help_string(action)
-        return action.help
-
-
 def add_parser(subparsers, parent_parser):
     EXPERIMENTS_HELP = "Commands to run and compare experiments."
 
@@ -1395,7 +1413,7 @@ def add_parser(subparsers, parent_parser):
         "init",
         parents=[parent_parser],
         description=append_doc_link(EXPERIMENTS_INIT_HELP, "exp/init"),
-        formatter_class=RawDefaultsHelpFormatter,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     experiments_init_parser.add_argument(
         "cmd",
@@ -1421,7 +1439,6 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Overwrite existing stage",
     )
-
     experiments_init_parser.add_argument(
         "--explicit",
         action="store_true",
@@ -1433,46 +1450,46 @@ def add_parser(subparsers, parent_parser):
     )
     experiments_init_parser.add_argument(
         "--code",
-        default=CmdExperimentsInit.CODE,
         help="Path to the source file or directory "
-        "which your experiments depend",
+        "which your experiments depend"
+        f" (default: {CmdExperimentsInit.CODE})",
     )
     experiments_init_parser.add_argument(
         "--data",
-        default=CmdExperimentsInit.DATA,
         help="Path to the data file or directory "
-        "which your experiments depend",
+        "which your experiments depend"
+        f" (default: {CmdExperimentsInit.DATA})",
     )
     experiments_init_parser.add_argument(
         "--models",
-        default=CmdExperimentsInit.MODELS,
-        help="Path to the model file or directory for your experiments",
+        help="Path to the model file or directory for your experiments"
+        f" (default: {CmdExperimentsInit.MODELS})",
     )
     experiments_init_parser.add_argument(
         "--params",
-        default=CmdExperimentsInit.DEFAULT_PARAMS,
-        help="Path to the parameters file for your experiments",
+        help="Path to the parameters file for your experiments"
+        f" (default: {CmdExperimentsInit.DEFAULT_PARAMS})",
     )
     experiments_init_parser.add_argument(
         "--metrics",
-        default=CmdExperimentsInit.DEFAULT_METRICS,
-        help="Path to the metrics file for your experiments",
+        help="Path to the metrics file for your experiments"
+        f" (default: {CmdExperimentsInit.DEFAULT_METRICS})",
     )
     experiments_init_parser.add_argument(
         "--plots",
-        default=CmdExperimentsInit.PLOTS,
-        help="Path to the plots file or directory for your experiments",
+        help="Path to the plots file or directory for your experiments"
+        f" (default: {CmdExperimentsInit.PLOTS})",
     )
     experiments_init_parser.add_argument(
         "--live",
-        help="Path to log dvclive outputs for your experiments "
-        f"(default: {CmdExperimentsInit.DVCLIVE})",
+        help="Path to log dvclive outputs for your experiments"
+        f" (default: {CmdExperimentsInit.DVCLIVE})",
     )
     experiments_init_parser.add_argument(
         "--type",
         choices=["default", "live"],
         default="default",
-        help="Select type of stage to create",
+        help="Select type of stage to create (default: %(default)s)",
     )
     experiments_init_parser.set_defaults(func=CmdExperimentsInit)
 

dvc/command/machine.py

@@ -193,7 +193,14 @@ class CmdMachineCreate(CmdBase):
 
 
 class CmdMachineStatus(CmdBase):
-    SHOWN_FIELD = ["name", "cloud", "instance_hdd_size", "instance_ip"]
+    SHOWN_FIELD = [
+        "name",
+        "cloud",
+        "instance_ip",
+        "instance_type",
+        "instance_hdd_size",
+        "instance_gpu",
+    ]
 
     def run(self):
         if self.repo.machine is None:

dvc/command/plots.py

@@ -68,8 +68,12 @@ class CmdPlots(CmdBase):
 
             if self.args.open:
                 import webbrowser
+                from platform import uname
 
-                opened = webbrowser.open(index_path)
+                if "Microsoft" in uname().release:
+                    url = Path(rel) / "index.html"
+
+                opened = webbrowser.open(url)
                 if not opened:
                     ui.error_write(
                         "Failed to open. Please try opening it manually."

dvc/compare.py

@@ -12,6 +12,7 @@ from typing import (
     Mapping,
     MutableSequence,
     Sequence,
+    Set,
     Tuple,
     Union,
     overload,
@@ -181,6 +182,35 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             {k: self._columns[k][i] for k in keys} for i in range(len(self))
         ]
 
+    def dropna(self, axis: str = "rows"):
+        if axis not in ["rows", "cols"]:
+            raise ValueError(
+                f"Invalid 'axis' value {axis}."
+                "Choose one of ['rows', 'cols']"
+            )
+
+        to_drop: Set = set()
+        for n_row, row in enumerate(self):
+            for n_col, col in enumerate(row):
+                if col == self._fill_value:
+                    if axis == "rows":
+                        to_drop.add(n_row)
+                        break
+                    else:
+                        to_drop.add(self.keys()[n_col])
+
+        if axis == "rows":
+            for name in self.keys():
+                self._columns[name] = Column(
+                    [
+                        x
+                        for n, x in enumerate(self._columns[name])
+                        if n not in to_drop
+                    ]
+                )
+        else:
+            self.drop(*to_drop)
+
 
 def _normalize_float(val: float, precision: int):
     return f"{val:.{precision}g}"

dvc/config_schema.py

@@ -179,10 +179,12 @@ SCHEMA = {
                 },
                 "hdfs": {"user": str, "kerb_ticket": str, **REMOTE_COMMON},
                 "webhdfs": {
-                    "hdfscli_config": str,
-                    "webhdfs_token": str,
-                    "user": str,
-                    "webhdfs_alias": str,
+                    "kerberos": Bool,
+                    "kerberos_principal": str,
+                    "proxy_to": str,
+                    "ssl_verify": Any(Bool, str),
+                    "token": str,
+                    "use_https": Bool,
                     **REMOTE_COMMON,
                 },
                 "azure": {

dvc/fs/local.py

@@ -39,7 +39,11 @@ class LocalFileSystem(BaseFileSystem):
         return os.path.lexists(path_info)
 
     def checksum(self, path_info) -> str:
-        return self.fs.checksum(path_info)
+        from fsspec.utils import tokenize
+
+        st = os.stat(path_info)
+
+        return str(int(tokenize([st.st_ino, st.st_mtime, st.st_size]), 16))
 
     def isfile(self, path_info) -> bool:
         return os.path.isfile(path_info)

dvc/fs/utils.py

@@ -1,7 +1,9 @@
+import errno
 import logging
 import os
 from typing import TYPE_CHECKING
 
+from .base import RemoteActionNotImplemented
 from .local import LocalFileSystem
 
 if TYPE_CHECKING:
@@ -12,17 +14,48 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _link(
+    from_fs: "BaseFileSystem",
+    from_info: "AnyPath",
+    to_fs: "BaseFileSystem",
+    to_info: "DvcPath",
+    hardlink: bool = False,
+) -> None:
+    if not isinstance(from_fs, type(to_fs)):
+        raise OSError(errno.EXDEV, "can't link across filesystems")
+
+    links = ["reflink"] + ["hardlink"] if hardlink else []
+    while links:
+        link = links.pop(0)
+        try:
+            func = getattr(to_fs, link)
+        except AttributeError:
+            continue
+
+        try:
+            return func(from_info, to_info)
+        except RemoteActionNotImplemented:
+            continue
+        except OSError as exc:
+            if exc.errno not in [errno.EXDEV, errno.ENOTSUP]:
+                raise
+
+    raise OSError(errno.ENOTSUP, "reflink and hardlink are not supported")
+
+
 def transfer(
     from_fs: "BaseFileSystem",
     from_info: "AnyPath",
     to_fs: "BaseFileSystem",
     to_info: "DvcPath",
-    move: bool = False,
+    hardlink: bool = False,
 ) -> None:
-    use_move = isinstance(from_fs, type(to_fs)) and move
     try:
-        if use_move:
-            return to_fs.move(from_info, to_info)
+        try:
+            return _link(from_fs, from_info, to_fs, to_info, hardlink=hardlink)
+        except OSError as exc:
+            if exc.errno not in [errno.EXDEV, errno.ENOTSUP]:
+                raise
 
         if isinstance(from_fs, LocalFileSystem):
             if not isinstance(from_info, from_fs.PATH_CLS):
@@ -49,8 +82,6 @@ def transfer(
             and isinstance(exc.__context__, FileExistsError)
         ):
             logger.debug("'%s' file already exists, skipping", to_info)
-            if use_move:
-                from_fs.remove(from_info)
             return None
 
         raise

dvc/fs/webhdfs.py

@@ -31,9 +31,10 @@ class WebHDFSFileSystem(CallbackMixin, FSSpecWrapper):
         )
 
     def _prepare_credentials(self, **config):
-        if "webhdfs_token" in config:
-            config["token"] = config.pop("webhdfs_token")
-
+        self._ssl_verify = config.pop("ssl_verify", True)
+        principal = config.pop("kerberos_principal", None)
+        if principal:
+            config["kerb_kwargs"] = {"principal": principal}
         return config
 
     @wrap_prop(threading.Lock())
@@ -41,7 +42,9 @@ class WebHDFSFileSystem(CallbackMixin, FSSpecWrapper):
     def fs(self):
         from fsspec.implementations.webhdfs import WebHDFS
 
-        return WebHDFS(**self.fs_args)
+        fs = WebHDFS(**self.fs_args)
+        fs.session.verify = self._ssl_verify
+        return fs
 
     def checksum(self, path_info):
         path = self._with_bucket(path_info)

dvc/info.py

@@ -1,3 +1,4 @@
+import errno
 import itertools
 import os
 import pathlib
@@ -7,7 +8,7 @@ import uuid
 import psutil
 
 from dvc import __version__
-from dvc.exceptions import DvcException, NotDvcRepoError
+from dvc.exceptions import NotDvcRepoError
 from dvc.fs import FS_MAP, get_fs_cls, get_fs_config
 from dvc.repo import Repo
 from dvc.scm.base import SCMError
@@ -103,7 +104,9 @@ def _get_linktype_support_info(repo):
             if is_link and not is_link(dst):
                 status = "broken"
             os.unlink(dst)
-        except DvcException:
+        except OSError as exc:
+            if exc.errno != errno.ENOTSUP:
+                raise
             status = "not supported"
 
         if status == "supported":

dvc/machine/__init__.py

@@ -211,3 +211,7 @@ class MachineManager:
         """move configuration to a new location if the machine is running."""
         config, backend = self.get_config_and_backend(name)
         return backend.rename(new=new, **config)
+
+    def get_executor_kwargs(self, name: Optional[str]):
+        config, backend = self.get_config_and_backend(name)
+        return backend.get_executor_kwargs(**config)

dvc/machine/backend/base.py

@@ -4,7 +4,6 @@ from typing import TYPE_CHECKING, Iterator, Optional
 
 if TYPE_CHECKING:
     from dvc.fs.ssh import SSHFileSystem
-    from dvc.repo.experiments.executor.base import BaseExecutor
     from dvc.types import StrPath
 
 
@@ -34,10 +33,8 @@ class BaseMachineBackend(ABC):
         """Spawn an interactive SSH shell for the specified machine."""
 
     @abstractmethod
-    def get_executor(
-        self, name: Optional[str] = None, **config
-    ) -> "BaseExecutor":
-        """Return an executor instance which can be used for DVC
+    def get_executor_kwargs(self, name: str, **config) -> dict:
+        """Return SSHExecutor kwargs which can be used for DVC
         experiment/pipeline execution on the specified machine.
         """
 

dvc/machine/backend/terraform.py

@@ -1,6 +1,6 @@
 import os
 from contextlib import contextmanager
-from functools import partialmethod
+from functools import partial, partialmethod
 from typing import TYPE_CHECKING, Iterator, Optional
 
 from dvc.exceptions import DvcException
@@ -10,10 +10,22 @@ from dvc.utils.fs import makedirs
 from .base import BaseMachineBackend
 
 if TYPE_CHECKING:
-    from dvc.repo.experiments.executor.base import BaseExecutor
     from dvc.types import StrPath
 
 
+@contextmanager
+def _sshfs(resource: dict):
+    from tpi import TerraformProviderIterative
+
+    with TerraformProviderIterative.pemfile(resource) as pem:
+        fs = SSHFileSystem(
+            host=resource["instance_ip"],
+            user="ubuntu",
+            keyfile=pem,
+        )
+        yield fs
+
+
 class TerraformBackend(BaseMachineBackend):
     def __init__(self, tmp_dir: "StrPath", **kwargs):
         super().__init__(tmp_dir, **kwargs)
@@ -51,26 +63,24 @@ class TerraformBackend(BaseMachineBackend):
         _tpi_func, "run_shell"
     )  # type: ignore[assignment]
 
-    def get_executor(
-        self, name: Optional[str] = None, **config
-    ) -> "BaseExecutor":
-        raise NotImplementedError
+    def get_executor_kwargs(self, name: str, **config) -> dict:
+        with self.make_tpi(name) as tpi:
+            resource = tpi.default_resource(name)
+        return {
+            "host": resource["instance_ip"],
+            "port": SSHFileSystem.DEFAULT_PORT,
+            "username": "ubuntu",
+            "fs_factory": partial(_sshfs, dict(resource)),
+        }
 
     @contextmanager
     def get_sshfs(  # pylint: disable=unused-argument
         self, name: Optional[str] = None, **config
     ) -> Iterator["SSHFileSystem"]:
-        from tpi import TerraformProviderIterative
-
         assert name
         with self.make_tpi(name) as tpi:
             resource = tpi.default_resource(name)
-        with TerraformProviderIterative.pemfile(resource) as pem:
-            fs = SSHFileSystem(
-                host=resource["instance_ip"],
-                user="ubuntu",
-                keyfile=pem,
-            )
+        with _sshfs(resource) as fs:
             yield fs
 
     def rename(self, name: str, new: str, **config):

dvc/objects/checkout.py

@@ -1,3 +1,4 @@
+import errno
 import logging
 from itertools import chain
 
@@ -77,7 +78,9 @@ def _try_links(cache, from_info, to_info, link_types):
             _verify_link(cache, to_info, link_types[0])
             return
 
-        except DvcException as exc:
+        except OSError as exc:
+            if exc.errno not in [errno.EXDEV, errno.ENOTSUP]:
+                raise
             logger.debug(
                 "Cache type '%s' is not supported: %s", link_types[0], exc
             )
@@ -94,13 +97,13 @@ def _link(cache, from_info, to_info):
         raise CheckoutError([str(to_info)]) from exc
 
 
-def _cache_is_copy(cache, path_info):
+def _confirm_cache_type(cache, path_info):
     """Checks whether cache uses copies."""
     if cache.cache_type_confirmed:
-        return cache.cache_types[0] == "copy"
+        return
 
     if set(cache.cache_types) <= {"copy"}:
-        return True
+        return
 
     workspace_file = path_info.with_name("." + uuid())
     test_cache_file = cache.path_info / ".cache_type_test_file"
@@ -115,7 +118,16 @@ def _cache_is_copy(cache, path_info):
         cache.fs.remove(test_cache_file)
 
     cache.cache_type_confirmed = True
-    return cache.cache_types[0] == "copy"
+
+
+def _relink(cache, cache_info, fs, path_info, in_cache, force):
+    _remove(path_info, fs, in_cache, force=force)
+    _link(cache, cache_info, path_info)
+    # NOTE: Depending on a file system (e.g. on NTFS), `_remove` might reset
+    # read-only permissions in order to delete a hardlink to protected object,
+    # which will also reset it for the object itself, making it unprotected,
+    # so we need to protect it back.
+    cache.protect(cache_info)
 
 
 def _checkout_file(
@@ -130,18 +142,33 @@ def _checkout_file(
 ):
     """The file is changed we need to checkout a new copy"""
     modified = False
+
+    _confirm_cache_type(cache, path_info)
+
     cache_info = cache.hash_to_path_info(change.new.oid.value)
     if change.old.oid:
         if relink:
-            if fs.iscopy(path_info) and _cache_is_copy(cache, path_info):
+            if fs.iscopy(path_info) and cache.cache_types[0] == "copy":
                 cache.unprotect(path_info)
             else:
-                _remove(path_info, fs, change.old.in_cache, force=force)
-                _link(cache, cache_info, path_info)
+                _relink(
+                    cache,
+                    cache_info,
+                    fs,
+                    path_info,
+                    change.old.in_cache,
+                    force=force,
+                )
         else:
             modified = True
-            _remove(path_info, fs, change.old.in_cache, force=force)
-            _link(cache, cache_info, path_info)
+            _relink(
+                cache,
+                cache_info,
+                fs,
+                path_info,
+                change.old.in_cache,
+                force=force,
+            )
     else:
         _link(cache, cache_info, path_info)
         modified = True

dvc/objects/db/base.py

@@ -81,13 +81,17 @@ class ObjectDB:
         from_info: "AnyPath",
         to_info: "DvcPath",
         _hash_info: "HashInfo",
-        move: bool = False,
+        hardlink: bool = False,
     ):
         from dvc import fs
 
         self.makedirs(to_info.parent)
         return fs.utils.transfer(
-            from_fs, from_info, self.fs, to_info, move=move
+            from_fs,
+            from_info,
+            self.fs,
+            to_info,
+            hardlink=hardlink,
         )
 
     def add(
@@ -95,7 +99,7 @@ class ObjectDB:
         path_info: "AnyPath",
         fs: "BaseFileSystem",
         hash_info: "HashInfo",
-        move: bool = True,
+        hardlink: bool = False,
         verify: Optional[bool] = None,
     ):
         if self.read_only:
@@ -110,7 +114,7 @@ class ObjectDB:
             pass
 
         cache_info = self.hash_to_path_info(hash_info.value)
-        self._add_file(fs, path_info, cache_info, hash_info, move=move)
+        self._add_file(fs, path_info, cache_info, hash_info, hardlink=hardlink)
 
         try:
             if verify:

dvc/objects/db/reference.py

@@ -58,12 +58,16 @@ class ReferenceObjectDB(ObjectDB):
         from_info: "AnyPath",
         to_info: "DvcPath",
         hash_info: "HashInfo",
-        move: bool = False,
+        hardlink: bool = False,
     ):
         self.makedirs(to_info.parent)
         if hash_info.isdir:
             return super()._add_file(
-                from_fs, from_info, to_info, hash_info, move
+                from_fs,
+                from_info,
+                to_info,
+                hash_info,
+                hardlink=hardlink,
             )
         ref_file = ReferenceHashFile(from_info, from_fs, hash_info)
         self._obj_cache[hash_info] = ref_file

dvc/objects/stage.py

@@ -88,7 +88,7 @@ def _stage_file(path_info, fs, name, odb=None, upload_odb=None, dry_run=False):
     if dry_run:
         obj = HashFile(path_info, fs, hash_info)
     else:
-        odb.add(path_info, fs, hash_info, move=False)
+        odb.add(path_info, fs, hash_info, hardlink=False)
         obj = odb.get(hash_info)
 
     return path_info, meta, obj
@@ -175,7 +175,7 @@ def _stage_tree(path_info, fs, fs_info, name, odb=None, **kwargs):
             path_info, fs
         )
     tree.digest(hash_info=hash_info)
-    odb.add(tree.path_info, tree.fs, tree.hash_info, move=True)
+    odb.add(tree.path_info, tree.fs, tree.hash_info, hardlink=False)
     raw = odb.get(tree.hash_info)
     # cleanup unneeded memfs tmpfile and return tree based on the
     # ODB fs/path

dvc/objects/transfer.py

@@ -133,7 +133,7 @@ def transfer(
     obj_ids: Iterable["HashInfo"],
     jobs: Optional[int] = None,
     verify: bool = False,
-    move: bool = False,
+    hardlink: bool = False,
     **kwargs,
 ) -> int:
     """Transfer (copy) the specified objects from one ODB to another.
@@ -157,7 +157,11 @@ def transfer(
     def func(hash_info: "HashInfo") -> None:
         obj = src.get(hash_info)
         return dest.add(
-            obj.path_info, obj.fs, obj.hash_info, verify=verify, move=move
+            obj.path_info,
+            obj.fs,
+            obj.hash_info,
+            verify=verify,
+            hardlink=hardlink,
         )
 
     total = len(status.new)

dvc/output.py

@@ -555,10 +555,6 @@ class Output:
 
         assert not self.IS_DEPENDENCY
 
-        if not self.changed():
-            logger.debug("Output '%s' didn't change. Skipping saving.", self)
-            return
-
         _, self.meta, self.obj = ostage(
             self.odb,
             self.path_info,
@@ -600,7 +596,7 @@ class Output:
                     self.odb,
                     {obj.hash_info},
                     shallow=False,
-                    move=True,
+                    hardlink=True,
                 )
             checkout(
                 filter_info or self.path_info,
@@ -629,7 +625,7 @@ class Output:
             self.odb,
             {save_obj.hash_info} | {oid for _, _, oid in save_obj},
             shallow=True,
-            move=True,
+            hardlink=True,
         )
         return checkout_obj
 
@@ -826,7 +822,7 @@ class Output:
             odb,
             {obj.hash_info},
             jobs=jobs,
-            move=upload,
+            hardlink=False,
             shallow=False,
         )
 

dvc/repo/experiments/__init__.py

@@ -473,6 +473,17 @@ class Experiments:
             "\tdvc exp branch <exp> <branch>\n"
         )
 
+    def _validate_new_ref(self, exp_ref: ExpRefInfo):
+        from .utils import check_ref_format
+
+        if not exp_ref.name:
+            return
+
+        check_ref_format(self.scm, exp_ref)
+
+        if self.scm.get_ref(str(exp_ref)):
+            raise ExperimentExistsError(exp_ref.name)
+
     @scm_locked
     def new(self, *args, checkpoint_resume: Optional[str] = None, **kwargs):
         """Create a new experiment.
@@ -485,6 +496,16 @@ class Experiments:
                 *args, resume_rev=checkpoint_resume, **kwargs
             )
 
+        name = kwargs.get("name", None)
+        baseline_sha = kwargs.get("baseline_rev") or self.repo.scm.get_rev()
+        exp_ref = ExpRefInfo(baseline_sha=baseline_sha, name=name)
+
+        try:
+            self._validate_new_ref(exp_ref)
+        except ExperimentExistsError as err:
+            if not (kwargs.get("force", False) or kwargs.get("reset", False)):
+                raise err
+
         return self._stash_exp(*args, **kwargs)
 
     def _resume_checkpoint(
@@ -653,8 +674,8 @@ class Experiments:
                 name=item.name,
                 branch=item.branch,
                 tmp_dir=base_tmp_dir,
-                cache_dir=self.repo.odb.local.cache_dir,
             )
+            executor.init_cache(self.repo, stash_rev)
             executors[stash_rev] = executor
 
         for ref in (EXEC_HEAD, EXEC_MERGE, EXEC_BASELINE):

dvc/repo/experiments/base.py

@@ -36,7 +36,7 @@ class BaselineMismatchError(DvcException):
 class ExperimentExistsError(DvcException):
     def __init__(self, name: str):
         msg = (
-            "Reproduced experiment conflicts with existing experiment "
+            "Experiment conflicts with existing experiment "
             f"'{name}'. To overwrite the existing experiment run:\n\n"
             "\tdvc exp run -f ...\n\n"
             "To run this experiment with a different name run:\n\n"

dvc/repo/experiments/executor/base.py

@@ -14,8 +14,6 @@ from typing import (
     Union,
 )
 
-from funcy import cached_property
-
 from dvc.env import DVC_EXP_AUTO_PUSH, DVC_EXP_GIT_REMOTE
 from dvc.exceptions import DvcException
 from dvc.path_info import PathInfo
@@ -24,8 +22,6 @@ from dvc.repo.experiments.base import (
     EXEC_BASELINE,
     EXEC_BRANCH,
     EXEC_CHECKPOINT,
-    EXEC_HEAD,
-    EXEC_MERGE,
     EXEC_NAMESPACE,
     EXPS_NAMESPACE,
     EXPS_STASH,
@@ -34,7 +30,6 @@ from dvc.repo.experiments.base import (
     ExpRefInfo,
     UnchangedExperimentError,
 )
-from dvc.scm import SCM
 from dvc.stage import PipelineStage
 from dvc.stage.monitor import CheckpointKilledError
 from dvc.stage.serialize import to_lockfile
@@ -117,44 +112,21 @@ class BaseExecutor(ABC):
         self._init_git(src, branch)
         self.name = name
 
-    def _init_git(self, scm: "Git", branch: Optional[str] = None):
-        """Init git repo and collect executor refs from the specified SCM."""
-        from dulwich.repo import Repo as DulwichRepo
-
-        DulwichRepo.init(os.fspath(self.root_dir))
-
-        cwd = os.getcwd()
-        os.chdir(self.root_dir)
-        try:
-            refspec = f"{EXEC_NAMESPACE}/"
-            scm.push_refspec(self.git_url, refspec, refspec)
-            if branch:
-                scm.push_refspec(self.git_url, branch, branch)
-                self.scm.set_ref(EXEC_BRANCH, branch, symbolic=True)
-            elif self.scm.get_ref(EXEC_BRANCH):
-                self.scm.remove_ref(EXEC_BRANCH)
-
-            if self.scm.get_ref(EXEC_CHECKPOINT):
-                self.scm.remove_ref(EXEC_CHECKPOINT)
-
-            # checkout EXEC_HEAD and apply EXEC_MERGE on top of it without
-            # committing
-            head = EXEC_BRANCH if branch else EXEC_HEAD
-            self.scm.checkout(head, detach=True)
-            merge_rev = self.scm.get_ref(EXEC_MERGE)
-            self.scm.merge(merge_rev, squash=True, commit=False)
-        finally:
-            os.chdir(cwd)
-
-    @cached_property
-    def scm(self):
-        return SCM(self.root_dir)
+    @abstractmethod
+    def _init_git(self, scm: "Git", branch: Optional[str] = None, **kwargs):
+        """Init git repo and populate it using exp refs from the specified
+        SCM instance.
+        """
 
     @property
     @abstractmethod
     def git_url(self) -> str:
         pass
 
+    @abstractmethod
+    def init_cache(self, dvc: "Repo", rev: str, run_cache: bool = True):
+        """Initialize DVC (cache)."""
+
     @property
     def dvc_dir(self) -> str:
         return os.path.join(self.root_dir, self._dvc_dir)
@@ -168,8 +140,7 @@ class BaseExecutor(ABC):
         return dict_sha256(exp_data)
 
     def cleanup(self):
-        self.scm.close()
-        del self.scm
+        pass
 
     # TODO: come up with better way to stash repro arguments
     @staticmethod

dvc/repo/experiments/executor/local.py

@@ -1,10 +1,25 @@
 import logging
 import os
 from tempfile import TemporaryDirectory
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
+
+from funcy import cached_property
+
+from dvc.repo.experiments.base import (
+    EXEC_BRANCH,
+    EXEC_CHECKPOINT,
+    EXEC_HEAD,
+    EXEC_MERGE,
+    EXEC_NAMESPACE,
+)
+from dvc.scm import SCM
 
 from .base import BaseExecutor
 
+if TYPE_CHECKING:
+    from dvc.repo import Repo
+    from dvc.scm.git import Git
+
 logger = logging.getLogger(__name__)
 
 
@@ -36,6 +51,15 @@ class BaseLocalExecutor(BaseExecutor):
             root_dir = root_dir.replace(os.sep, "/")
         return f"file://{root_dir}"
 
+    @cached_property
+    def scm(self):
+        return SCM(self.root_dir)
+
+    def cleanup(self):
+        super().cleanup()
+        self.scm.close()
+        del self.scm
+
 
 class TempDirExecutor(BaseLocalExecutor):
     """Temp directory experiment executor."""
@@ -51,22 +75,46 @@ class TempDirExecutor(BaseLocalExecutor):
         self,
         *args,
         tmp_dir: Optional[str] = None,
-        cache_dir: Optional[str] = None,
         **kwargs,
     ):
         self._tmp_dir = ExpTemporaryDirectory(dir=tmp_dir)
         kwargs["root_dir"] = self._tmp_dir.name
         super().__init__(*args, **kwargs)
-        if cache_dir:
-            self._config(cache_dir)
         logger.debug("Init temp dir executor in dir '%s'", self._tmp_dir)
 
+    def _init_git(self, scm: "Git", branch: Optional[str] = None, **kwargs):
+        from dulwich.repo import Repo as DulwichRepo
+
+        DulwichRepo.init(os.fspath(self.root_dir))
+
+        refspec = f"{EXEC_NAMESPACE}/"
+        scm.push_refspec(self.git_url, refspec, refspec, **kwargs)
+        if branch:
+            scm.push_refspec(self.git_url, branch, branch, **kwargs)
+            self.scm.set_ref(EXEC_BRANCH, branch, symbolic=True)
+        elif self.scm.get_ref(EXEC_BRANCH):
+            self.scm.remove_ref(EXEC_BRANCH)
+
+        if self.scm.get_ref(EXEC_CHECKPOINT):
+            self.scm.remove_ref(EXEC_CHECKPOINT)
+
+        # checkout EXEC_HEAD and apply EXEC_MERGE on top of it without
+        # committing
+        head = EXEC_BRANCH if branch else EXEC_HEAD
+        self.scm.checkout(head, detach=True)
+        merge_rev = self.scm.get_ref(EXEC_MERGE)
+        self.scm.merge(merge_rev, squash=True, commit=False)
+
     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", encoding="utf-8") as fobj:
             fobj.write(f"[cache]\n    dir = {cache_dir}")
 
+    def init_cache(self, dvc: "Repo", rev: str, run_cache: bool = True):
+        """Initialize DVC (cache)."""
+        self._config(dvc.odb.local.cache_dir)
+
     def cleanup(self):
         super().cleanup()
         logger.debug("Removing tmpdir '%s'", self._tmp_dir)

dvc/repo/experiments/executor/ssh.py

@@ -0,0 +1,159 @@
+import logging
+import posixpath
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Callable, Optional
+
+from funcy import first
+
+from dvc.fs.ssh import SSHFileSystem
+from dvc.repo.experiments.base import (
+    EXEC_BRANCH,
+    EXEC_CHECKPOINT,
+    EXEC_HEAD,
+    EXEC_MERGE,
+    EXEC_NAMESPACE,
+)
+
+from .base import BaseExecutor
+
+if TYPE_CHECKING:
+    from dvc.machine import MachineManager
+    from dvc.repo import Repo
+    from dvc.scm.git import Git
+
+logger = logging.getLogger(__name__)
+
+
+@contextmanager
+def _sshfs(fs_factory, **kwargs):
+    if fs_factory:
+        with fs_factory() as fs:
+            yield fs
+        return
+    yield SSHFileSystem(**kwargs)
+
+
+class SSHExecutor(BaseExecutor):
+    """SSH experiment executor."""
+
+    WARN_UNTRACKED = True
+    QUIET = True
+
+    def __init__(
+        self,
+        *args,
+        host: Optional[str] = None,
+        port: Optional[int] = None,
+        username: Optional[str] = None,
+        fs_factory: Optional[Callable] = None,
+        **kwargs,
+    ):
+        assert host
+
+        self.host: str = host
+        self.port = port
+        self.username = username
+        self._fs_factory = fs_factory
+
+        super().__init__(*args, **kwargs)
+        logger.debug("Init SSH executor for host '%s'", self.host)
+
+    @classmethod
+    def gen_dirname(cls, name: Optional[str] = None):
+        from shortuuid import uuid
+
+        return "-".join([name or "dvc-exp", "executor", uuid()])
+
+    @classmethod
+    def from_machine(
+        cls,
+        manager: "MachineManager",
+        machine_name: Optional[str],
+        *args,
+        **kwargs,
+    ):
+        kwargs["root_dir"] = cls.gen_dirname(kwargs.get("name"))
+        kwargs.update(manager.get_executor_kwargs(machine_name))
+        return cls(*args, **kwargs)
+
+    def sshfs(self):
+        return _sshfs(self._fs_factory, host=self.host, port=self.port)
+
+    @property
+    def git_url(self) -> str:
+        user = f"{self.username}@" if self.username else ""
+        port = f":{self.port}" if self.port is not None else ""
+        path = f"{self.root_dir}" if self.root_dir else ""
+        if path and not posixpath.isabs(path):
+            path = f"/~/{path}"
+        return f"ssh://{user}{self.host}{port}{path}"
+
+    @property
+    def abs_url(self) -> str:
+        assert self._repo_abspath
+        user = f"{self.username}@" if self.username else ""
+        port = f":{self.port}" if self.port is not None else ""
+        return f"ssh://{user}{self.host}{port}{self._repo_abspath}"
+
+    @staticmethod
+    def _git_client_args(fs):
+        kwargs = {
+            "password": fs.fs_args.get("password"),
+            "key_filename": first(fs.fs_args.get("client_keys", [])),
+        }
+        return kwargs
+
+    def _init_git(self, scm: "Git", branch: Optional[str] = None, **kwargs):
+        with self.sshfs() as fs:
+            fs.makedirs(self.root_dir)
+            self._ssh_cmd(fs, "git init .")
+            self._ssh_cmd(fs, "git config user.name dvc-exp")
+            self._ssh_cmd(
+                fs, "git config user.email [email protected]"
+            )
+
+            result = self._ssh_cmd(fs, "pwd")
+            path = result.stdout.strip()
+            self._repo_abspath = path
+
+            # TODO: support multiple client key retries in git backends
+            # (see https://github.com/iterative/dvc/issues/6508)
+            kwargs.update(self._git_client_args(fs))
+            refspec = f"{EXEC_NAMESPACE}/"
+            scm.push_refspec(self.git_url, refspec, refspec, **kwargs)
+            if branch:
+                scm.push_refspec(self.git_url, branch, branch, **kwargs)
+                self._ssh_cmd(fs, f"git symbolic-ref {EXEC_BRANCH} {branch}")
+            else:
+                self._ssh_cmd(
+                    fs, f"git symbolic-ref -d {EXEC_BRANCH}", check=False
+                )
+            self._ssh_cmd(
+                fs, f"git update-ref -d {EXEC_CHECKPOINT}", check=False
+            )
+
+            # checkout EXEC_HEAD and apply EXEC_MERGE on top of it without
+            # committing
+            head = EXEC_BRANCH if branch else EXEC_HEAD
+            self._ssh_cmd(fs, f"git checkout {head}")
+            merge_rev = scm.get_ref(EXEC_MERGE)
+            self._ssh_cmd(fs, f"git merge --squash --no-commit {merge_rev}")
+
+    def _ssh_cmd(self, sshfs, cmd, chdir=None, **kwargs):
+        working_dir = chdir or self.root_dir
+        return sshfs.fs.execute(f"cd {working_dir};{cmd}", **kwargs)
+
+    def init_cache(self, dvc: "Repo", rev: str, run_cache: bool = True):
+        from dvc.objects.db import ODBManager, get_odb
+        from dvc.repo import Repo
+        from dvc.repo.push import push
+
+        cache_url = posixpath.join(
+            self.abs_url,
+            self._dvc_dir,
+            Repo.DVC_DIR,
+            ODBManager.CACHE_DIR,
+        )
+        with self.sshfs() as fs:
+            odb = get_odb(fs, fs.PATH_CLS(cache_url), **fs.config)
+            push(dvc, revs=[rev], run_cache=run_cache, odb=odb)

dvc/repo/experiments/init.py

@@ -1,10 +1,18 @@
+import logging
 import os
-from typing import TYPE_CHECKING, Dict, Iterable
+from contextlib import contextmanager
+from typing import TYPE_CHECKING, Dict, Iterable, cast
 
 from funcy import compact, lremove, post_processing
-from rich.prompt import Prompt
+from rich.prompt import Confirm, Prompt
+from rich.rule import Rule
+from rich.syntax import Syntax
 
+from dvc.exceptions import DvcException
+from dvc.stage import PipelineStage
+from dvc.stage.serialize import to_pipeline_file
 from dvc.types import OptStr
+from dvc.utils.serialize import dumps_yaml
 
 if TYPE_CHECKING:
     from dvc.repo import Repo
@@ -25,7 +33,37 @@ PROMPTS = {
 }
 
 
-class RequiredPrompt(Prompt):
+class RichInputMixin:
+    """Prevents exc message from printing in the same line on Ctrl + D/C."""
+
+    @classmethod
+    def get_input(cls, *args, **kwargs) -> str:
+        try:
+            return super().get_input(*args, **kwargs)  # type: ignore[misc]
+        except (KeyboardInterrupt, EOFError):
+            ui.error_write()
+            raise
+
+
+class ConfirmationPrompt(RichInputMixin, Confirm):
+    def make_prompt(self, default):
+        prompt = self.prompt.copy()
+        prompt.end = ""
+
+        prompt.append(" [")
+        yes, no = self.choices
+        for idx, (val, choice) in enumerate(((True, yes), (False, no))):
+            if idx:
+                prompt.append("/")
+            if val == default:
+                prompt.append(choice.upper(), "green")
+            else:
+                prompt.append(choice)
+        prompt.append("] ")
+        return prompt
+
+
+class RequiredPrompt(RichInputMixin, Prompt):
     def process_response(self, value: str):
         from rich.prompt import InvalidResponse
 
@@ -36,14 +74,6 @@ class RequiredPrompt(Prompt):
             )
         return ret
 
-    @classmethod
-    def get_input(cls, *args, **kwargs) -> str:
-        try:
-            return super().get_input(*args, **kwargs)
-        except (KeyboardInterrupt, EOFError):
-            ui.error_write()
-            raise
-
     def render_default(self, default):
         from rich.text import Text
 
@@ -71,7 +101,7 @@ class SkippablePrompt(RequiredPrompt):
             prompt.append(_default)
             prompt.append(", ")
 
-        prompt.append(f"{self.skip_value} to skip", style="italic")
+        prompt.append(f"{self.skip_value} to omit", style="italic")
         prompt.append("]")
         prompt.append(self.prompt_suffix)
         return prompt
@@ -85,36 +115,84 @@ def _prompts(keys: Iterable[str], defaults: Dict[str, OptStr]):
         else:
             prompt_cls = SkippablePrompt
         kwargs = {"default": defaults[key]} if key in defaults else {}
-        prompt = PROMPTS.get(key)
-        value = prompt_cls.ask(prompt, console=ui.error_console, **kwargs)
+        prompt = PROMPTS[key]
+        value = prompt_cls.ask(  # type: ignore[call-overload]
+            prompt, console=ui.error_console, **kwargs
+        )
         yield key, value
 
 
+@contextmanager
+def _disable_logging(highest_level=logging.CRITICAL):
+    previous_level = logging.root.manager.disable
+
+    logging.disable(highest_level)
+
+    try:
+        yield
+    finally:
+        logging.disable(previous_level)
+
+
+PIPELINE_FILE_LINK = (
+    "https://dvc.org/doc/user-guide/project-structure/pipelines-files"
+)
+
+
 def init_interactive(
+    name: str,
     defaults: Dict[str, str],
-    provided: Iterable[str],
-    show_heading: bool = False,
+    provided: Dict[str, str],
+    show_tree: bool = False,
     live: bool = False,
 ) -> Dict[str, str]:
-    primary = lremove(provided, ["cmd", "code", "data", "models", "params"])
-    secondary = lremove(provided, ["live"] if live else ["metrics", "plots"])
-
-    if not (primary or secondary):
-        return {}
-
-    message = (
-        "This command will guide you to set up your first stage in "
-        "[green]dvc.yaml[/green].\n"
+    command = provided.pop("cmd", None)
+    primary = lremove(provided.keys(), ["code", "data", "models", "params"])
+    secondary = lremove(
+        provided.keys(), ["live"] if live else ["metrics", "plots"]
     )
-    if show_heading:
-        ui.error_write(message, styled=True)
 
-    return compact(
-        {
-            **_prompts(primary, defaults),
-            **_prompts(secondary, defaults),
-        }
+    ret: Dict[str, str] = {}
+    if not (command or primary or secondary):
+        return ret
+
+    message = ui.rich_text.assemble(
+        "This command will guide you to set up a ",
+        (name, "bright_blue"),
+        " stage in ",
+        ("dvc.yaml", "green"),
+        ".",
     )
+    doc_link = ui.rich_text.assemble(
+        "See ", (PIPELINE_FILE_LINK, "repr.url"), "."
+    )
+    ui.error_write(message, doc_link, "", sep="\n", styled=True)
+
+    if not command:
+        ret.update(_prompts(["cmd"], defaults))
+        ui.write()
+    else:
+        ret.update({"cmd": command})
+
+    ui.write("Enter the paths for dependencies and outputs of the command.")
+    workspace = {**defaults, **provided}
+    if show_tree and workspace:
+        from rich.tree import Tree
+
+        tree = Tree(
+            "DVC assumes the following workspace structure:",
+            highlight=True,
+        )
+        if not live and "live" not in provided:
+            workspace.pop("live", None)
+        for value in sorted(workspace.values()):
+            tree.add(f"[green]{value}[/green]")
+        ui.error_write(tree, styled=True)
+
+    ui.error_write()
+    ret.update(_prompts(primary, defaults))
+    ret.update(_prompts(secondary, defaults))
+    return compact(ret)
 
 
 def _check_stage_exists(
@@ -151,36 +229,37 @@ def init(
     with_live = type == "live"
     if interactive:
         defaults = init_interactive(
-            defaults=defaults or {},
-            show_heading=not dvcfile.exists(),
+            name,
+            defaults=defaults,
             live=with_live,
-            provided=overrides.keys(),
+            provided=overrides,
+            show_tree=True,
         )
     else:
         if with_live:
             # suppress `metrics`/`params` if live is selected, unless
             # it is also provided via overrides/cli.
             # This makes output to be a checkpoint as well.
-            defaults.pop("metrics")
-            defaults.pop("params")
+            defaults.pop("metrics", None)
+            defaults.pop("params", None)
         else:
-            defaults.pop("live")  # suppress live otherwise
+            defaults.pop("live", None)  # suppress live otherwise
 
     context: Dict[str, str] = {**defaults, **overrides}
     assert "cmd" in context
 
     params_kv = []
-    if context.get("params"):
+    params = context.get("params")
+    if params:
         from dvc.utils.serialize import LOADERS
 
-        path = context["params"]
-        assert isinstance(path, str)
-        _, ext = os.path.splitext(path)
-        params_kv = [{path: list(LOADERS[ext](path))}]
+        assert isinstance(params, str)
+        _, ext = os.path.splitext(params)
+        params_kv = [{params: list(LOADERS[ext](params))}]
 
     checkpoint_out = bool(context.get("live"))
     models = context.get("models")
-    return repo.stage.add(
+    stage = repo.stage.create(
         name=name,
         cmd=context["cmd"],
         deps=compact([context.get("code"), context.get("data")]),
@@ -191,3 +270,24 @@ def init(
         force=force,
         **{"checkpoints" if checkpoint_out else "outs": compact([models])},
     )
+
+    if interactive:
+        ui.write(Rule(style="green"), styled=True)
+        _yaml = dumps_yaml(to_pipeline_file(cast(PipelineStage, stage)))
+        syn = Syntax(_yaml, "yaml", theme="ansi_dark")
+        ui.error_write(syn, styled=True)
+
+    if not interactive or ConfirmationPrompt.ask(
+        "Do you want to add the above contents to dvc.yaml?",
+        console=ui.error_console,
+        default=True,
+    ):
+        scm = repo.scm
+        with _disable_logging(), scm.track_file_changes(autostage=True):
+            stage.dump(update_lock=False)
+            stage.ignore_outs()
+            if params:
+                scm.track_file(params)
+    else:
+        raise DvcException("Aborting ...")
+    return stage

dvc/repo/experiments/utils.py

@@ -157,3 +157,12 @@ def resolve_exp_ref(
         msg.extend([f"\t{info}" for info in exp_ref_list])
         raise InvalidArgumentError("\n".join(msg))
     return exp_ref_list[0]
+
+
+def check_ref_format(scm: "Git", ref: ExpRefInfo):
+    # "/" forbidden, only in dvc exp as we didn't support it for now.
+    if not scm.check_ref_format(str(ref)) or "/" in ref.name:
+        raise InvalidArgumentError(
+            f"Invalid exp name {ref.name}, the exp name must follow rules in "
+            "https://git-scm.com/docs/git-check-ref-format"
+        )

dvc/repo/install.py

@@ -1,2 +1,69 @@
-def install(self, use_pre_commit_tool):
-    self.scm.install(use_pre_commit_tool)
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from dvc.repo import Repo
+    from dvc.scm.git import Git
+
+
+def pre_commit_install(scm: "Git") -> None:
+    import os
+
+    from dvc.utils.serialize import modify_yaml
+
+    config_path = os.path.join(scm.root_dir, ".pre-commit-config.yaml")
+    with modify_yaml(config_path) as config:
+        entry = {
+            "repo": "https://github.com/iterative/dvc",
+            "rev": "master",
+            "hooks": [
+                {
+                    "id": "dvc-pre-commit",
+                    "language_version": "python3",
+                    "stages": ["commit"],
+                },
+                {
+                    "id": "dvc-pre-push",
+                    "language_version": "python3",
+                    "stages": ["push"],
+                },
+                {
+                    "id": "dvc-post-checkout",
+                    "language_version": "python3",
+                    "stages": ["post-checkout"],
+                    "always_run": True,
+                },
+            ],
+        }
+
+        config["repos"] = config.get("repos", [])
+        if entry not in config["repos"]:
+            config["repos"].append(entry)
+
+
+def install_hooks(scm: "Git") -> None:
+    hooks = ["post-checkout", "pre-commit", "pre-push"]
+    for hook in hooks:
+        scm.verify_hook(hook)
+    for hook in hooks:
+        scm.install_hook(hook, f"exec dvc git-hook {hook} $@")
+
+
+def install(self: "Repo", use_pre_commit_tool: bool = False) -> None:
+    """Adds dvc commands to SCM hooks for the repo.
+
+    If use_pre_commit_tool is set and pre-commit is installed it will be used
+    to install the hooks.
+    """
+    from dvc.scm.git import Git
+
+    scm = self.scm
+    if not isinstance(scm, Git):
+        return
+
+    driver = "dvc git-hook merge-driver --ancestor %O --our %A --their %B "
+    scm.install_merge_driver("dvc", "DVC merge driver", driver)
+
+    if use_pre_commit_tool:
+        return pre_commit_install(scm)
+
+    return install_hooks(scm)

dvc/repo/live.py

@@ -1,5 +1,4 @@
 import logging
-import os
 from typing import TYPE_CHECKING, List, Optional
 
 from dvc.render.utils import render
@@ -19,10 +18,7 @@ def create_summary(out):
 
     html_path = out.path_info.fspath + "_dvc_plots"
 
-    index_path = render(
-        out.repo, plots, metrics=metrics, path=html_path, refresh_seconds=5
-    )
-    logger.info(f"\nfile://{os.path.abspath(index_path)}")
+    render(out.repo, plots, metrics=metrics, path=html_path, refresh_seconds=5)
 
 
 def summary_path_info(out: "Output") -> Optional["PathInfo"]:

dvc/repo/plots/template.py

@@ -109,16 +109,16 @@ class PlotTemplates:
     @staticmethod
     def _find(templates, template_name):
         for template in templates:
-            if (
-                template_name == template
-                or template_name + ".json" == template
+            if template.endswith(template_name) or template.endswith(
+                template_name + ".json"
             ):
                 return template
         return None
 
     def _find_in_project(self, name: str) -> Optional["StrPath"]:
-        if os.path.exists(name):
-            return name
+        full_path = os.path.abspath(name)
+        if os.path.exists(full_path):
+            return full_path
 
         if os.path.exists(self.templates_dir):
             templates = [

dvc/repo/push.py

@@ -1,8 +1,13 @@
+from typing import TYPE_CHECKING, Optional
+
 from dvc.exceptions import FileTransferError, UploadError
 
 from ..utils import glob_targets
 from . import locked
 
+if TYPE_CHECKING:
+    from dvc.objects.db.base import ObjectDB
+
 
 @locked
 def push(
@@ -18,8 +23,11 @@ def push(
     run_cache=False,
     revs=None,
     glob=False,
+    odb: Optional["ObjectDB"] = None,
 ):
-    used_run_cache = self.stage_cache.push(remote) if run_cache else []
+    used_run_cache = (
+        self.stage_cache.push(remote, odb=odb) if run_cache else []
+    )
 
     if isinstance(targets, str):
         targets = [targets]
@@ -41,11 +49,13 @@ def push(
     )
 
     pushed = len(used_run_cache)
-    for odb, obj_ids in used.items():
-        if odb and odb.read_only:
+    for dest_odb, obj_ids in used.items():
+        if dest_odb and dest_odb.read_only:
             continue
         try:
-            pushed += self.cloud.push(obj_ids, jobs, remote=remote, odb=odb)
+            pushed += self.cloud.push(
+                obj_ids, jobs, remote=remote, odb=odb or dest_odb
+            )
         except FileTransferError as exc:
             raise UploadError(exc.amount)
     return pushed

dvc/scm/base.py

@@ -149,14 +149,6 @@ class Base:
         """Returns a list of commits in the repo."""
         return []
 
-    def install(self, use_pre_commit_tool=False):
-        """
-        Adds dvc commands to SCM hooks for the repo.
-
-        If use_pre_commit_tool is set and pre-commit is
-        installed it will be used to install the hooks.
-        """
-
     def cleanup_ignores(self):
         """
         This method should clean up ignores (eg. entries in .gitignore),
@@ -191,8 +183,10 @@ class Base:
         """
 
     @contextmanager
-    def track_file_changes(self, config=None):
-        autostage = get_in(config or {}, ["core", "autostage"])
+    def track_file_changes(self, config=None, autostage=False):
+        autostage = get_in(
+            config or {}, ["core", "autostage"], default=autostage
+        )
 
         try:
             yield

dvc/scm/git/__init__.py

@@ -94,6 +94,12 @@ class Git(Base):
     def dir(self):
         return first(self.backends.values()).dir
 
+    @cached_property
+    def hooks_dir(self):
+        from pathlib import Path
+
+        return Path(self.root_dir) / self.GIT_DIR / "hooks"
+
     @property
     def gitpython(self):
         return self.backends["gitpython"]
@@ -209,67 +215,27 @@ class Git(Base):
 
         self.track_file(relpath(gitignore))
 
-    def _install_hook(self, name):
-        hook = self._hook_path(name)
-        with open(hook, "w+", encoding="utf-8") as fobj:
-            fobj.write(f"#!/bin/sh\nexec dvc git-hook {name} $@\n")
+    def verify_hook(self, name):
+        if (self.hooks_dir / name).exists():
+            raise GitHookAlreadyExistsError(name)
 
-        os.chmod(hook, 0o777)
+    def install_hook(
+        self, name: str, script: str, interpreter: str = "sh"
+    ) -> None:
+        import shutil
 
-    def _install_merge_driver(self):
-        self.gitpython.repo.git.config("merge.dvc.name", "DVC merge driver")
-        self.gitpython.repo.git.config(
-            "merge.dvc.driver",
-            (
-                "dvc git-hook merge-driver "
-                "--ancestor %O "
-                "--our %A "
-                "--their %B "
-            ),
-        )
+        self.hooks_dir.mkdir(exist_ok=True)
+        hook = self.hooks_dir / name
 
-    def install(self, use_pre_commit_tool=False):
-        self._install_merge_driver()
+        directive = f"#!{shutil.which(interpreter)}"
+        hook.write_text(f"{directive}\n{script}\n", encoding="utf-8")
+        hook.chmod(0o777)
 
-        if not use_pre_commit_tool:
-            self._verify_dvc_hooks()
-            self._install_hook("post-checkout")
-            self._install_hook("pre-commit")
-            self._install_hook("pre-push")
-            return
-
-        from dvc.utils.serialize import modify_yaml
-
-        config_path = os.path.join(self.root_dir, ".pre-commit-config.yaml")
-        with modify_yaml(config_path) as config:
-            entry = {
-                "repo": "https://github.com/iterative/dvc",
-                "rev": "master",
-                "hooks": [
-                    {
-                        "id": "dvc-pre-commit",
-                        "language_version": "python3",
-                        "stages": ["commit"],
-                    },
-                    {
-                        "id": "dvc-pre-push",
-                        "language_version": "python3",
-                        "stages": ["push"],
-                    },
-                    {
-                        "id": "dvc-post-checkout",
-                        "language_version": "python3",
-                        "stages": ["post-checkout"],
-                        "always_run": True,
-                    },
-                ],
-            }
-
-            if "repos" not in config:
-                config["repos"] = []
-
-            if entry not in config["repos"]:
-                config["repos"].append(entry)
+    def install_merge_driver(
+        self, name: str, description: str, driver: str
+    ) -> None:
+        self.gitpython.repo.git.config(f"merge.{name}.name", description)
+        self.gitpython.repo.git.config(f"merge.{name}.driver", driver)
 
     def cleanup_ignores(self):
         for path in self.ignored_paths:
@@ -319,22 +285,6 @@ class Git(Base):
     def close(self):
         self.backends.close_initialized()
 
-    @cached_property
-    def _hooks_home(self):
-        return os.path.join(self.root_dir, self.GIT_DIR, "hooks")
-
-    def _hook_path(self, name):
-        return os.path.join(self._hooks_home, name)
-
-    def _verify_hook(self, name):
-        if os.path.exists(self._hook_path(name)):
-            raise GitHookAlreadyExistsError(name)
-
-    def _verify_dvc_hooks(self):
-        self._verify_hook("post-checkout")
-        self._verify_hook("pre-commit")
-        self._verify_hook("pre-push")
-
     @property
     def no_commits(self):
         return not bool(self.get_ref("HEAD"))
@@ -396,6 +346,7 @@ class Git(Base):
     status = partialmethod(_backend_func, "status")
     merge = partialmethod(_backend_func, "merge")
     validate_git_remote = partialmethod(_backend_func, "validate_git_remote")
+    check_ref_format = partialmethod(_backend_func, "check_ref_format")
 
     def resolve_rev(self, rev: str) -> str:
         from dvc.repo.experiments.utils import exp_refs_by_name

dvc/scm/git/backend/dulwich/__init__.py

@@ -85,7 +85,7 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
     from .asyncssh_vendor import AsyncSSHVendor
 
     # monkeypatch dulwich client's default SSH vendor to use asyncssh
-    client.get_ssh_vendor = AsyncSSHVendor
+    client.get_ssh_vendor = AsyncSSHVendor  # type: ignore[assignment]
 
     # Dulwich progress will return messages equivalent to git CLI,
     # our pbars should just display the messages as formatted by dulwich
@@ -681,3 +681,8 @@ class DulwichBackend(BaseGitBackend):  # pylint:disable=abstract-method
             os.path.join("", path)
         ):
             raise InvalidRemoteSCMRepo(url)
+
+    def check_ref_format(self, refname: str):
+        from dulwich.refs import check_ref_format
+
+        return check_ref_format(refname.encode())

dvc/scm/git/backend/gitpython.py

@@ -148,7 +148,8 @@ class GitPythonBackend(BaseGitBackend):  # pylint:disable=abstract-method
         rev: Optional[str] = None,
         shallow_branch: Optional[str] = None,
     ):
-        import git
+        from git import Repo
+        from git.exc import GitCommandError
 
         ld_key = "LD_LIBRARY_PATH"
 
@@ -170,7 +171,7 @@ class GitPythonBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 url = f"file://{url}"
             with TqdmGit(desc="Cloning", unit="obj") as pbar:
                 clone_from = partial(
-                    git.Repo.clone_from,
+                    Repo.clone_from,
                     url,
                     to_path,
                     env=env,  # needed before we can fix it in __init__
@@ -182,7 +183,7 @@ class GitPythonBackend(BaseGitBackend):  # pylint:disable=abstract-method
                 else:
                     tmp_repo = clone_from(branch=shallow_branch, depth=1)
             tmp_repo.close()
-        except git.exc.GitCommandError as exc:  # pylint: disable=no-member
+        except GitCommandError as exc:  # pylint: disable=no-member
             raise CloneError(url, to_path) from exc
 
         # NOTE: using our wrapper to make sure that env is fixed in __init__
@@ -191,7 +192,7 @@ class GitPythonBackend(BaseGitBackend):  # pylint:disable=abstract-method
         if rev:
             try:
                 repo.checkout(rev)
-            except git.exc.GitCommandError as exc:  # pylint: disable=no-member
+            except GitCommandError as exc:  # pylint: disable=no-member
                 raise RevError(
                     "failed to access revision '{}' for repo '{}'".format(
                         rev, url

dvc/stage/cache.py

@@ -2,7 +2,7 @@ import logging
 import os
 import tempfile
 from contextlib import contextmanager
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
 
 from funcy import cached_property, first
 
@@ -10,6 +10,9 @@ from dvc.exceptions import DvcException
 from dvc.path_info import PathInfo
 from dvc.utils import dict_sha256, relpath
 
+if TYPE_CHECKING:
+    from dvc.objects.db.base import ObjectDB
+
 logger = logging.getLogger(__name__)
 
 
@@ -229,10 +232,11 @@ class StageCache:
 
         return ret
 
-    def push(self, remote: Optional[str]):
+    def push(self, remote: Optional[str], odb: Optional["ObjectDB"] = None):
         from dvc.objects.transfer import _log_exceptions
 
-        odb = self.repo.cloud.get_remote_odb(remote)
+        if odb is None:
+            odb = self.repo.cloud.get_remote_odb(remote)
         return self._transfer(
             _log_exceptions(odb.fs.upload),
             self.repo.odb.local,

dvc/stage/monitor.py

@@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Callable, List
 from dvc.repo.live import create_summary
 from dvc.stage.decorators import relock_repo
 from dvc.stage.exceptions import StageCmdFailedError
+from dvc.ui import ui
 
 if TYPE_CHECKING:
     from dvc.output import Output
@@ -86,11 +87,16 @@ class LiveTask(MonitorTask):
     error_cls = LiveKilledError
 
     def __init__(self, stage: "Stage", out: "Output", proc: subprocess.Popen):
+        self.output_path = os.path.join(
+            os.getcwd(), out.path_info.fspath + "_dvc_plots", "index.html"
+        )
+        ui.write(f"Live summary will be created at:\n{self.output_path}")
         super().__init__(stage, functools.partial(create_summary, out), proc)
 
     def after_run(self):
         # make sure summary is prepared for all the data
         self.execute()
+        ui.write(f"Live summary has been created at:\n{self.output_path}")
 
 
 class Monitor:

dvc/system.py

@@ -2,35 +2,58 @@ import errno
 import logging
 import os
 import platform
-
-from dvc.exceptions import DvcException
+import stat
+import sys
 
 logger = logging.getLogger(__name__)
 
 
+if os.name == "nt" and sys.version_info < (3, 8):
+    # NOTE: using backports for `os.path.realpath`
+    # See https://bugs.python.org/issue9949 for more info.
+    # pylint: disable=import-error, no-name-in-module
+    from jaraco.windows.filesystem.backports import realpath as _realpath
+
+    def realpath(path):
+        return _realpath(os.fspath(path))
+
+
+else:
+    realpath = os.path.realpath
+
+
 class System:
     @staticmethod
     def hardlink(source, link_name):
-        try:
-            os.link(source, link_name)
-        except OSError as exc:
-            raise DvcException("failed to link") from exc
+        # NOTE: we should really be using `os.link()` here with
+        # `follow_symlinks=True`, but unfortunately the implementation is
+        # buggy across platforms, so until it is fixed, we just dereference
+        # the symlink ourselves here.
+        #
+        # See https://bugs.python.org/issue41355 for more info.
+        st = os.lstat(source)
+        if stat.S_ISLNK(st.st_mode):
+            src = realpath(source)
+        else:
+            src = source
+
+        os.link(src, link_name)
 
     @staticmethod
     def symlink(source, link_name):
-        try:
-            os.symlink(source, link_name)
-        except OSError as exc:
-            raise DvcException("failed to symlink") from exc
+        os.symlink(source, link_name)
 
     @staticmethod
     def _reflink_darwin(src, dst):
         import ctypes
 
+        def _cdll(name):
+            return ctypes.CDLL(name, use_errno=True)
+
         LIBC = "libc.dylib"
         LIBC_FALLBACK = "/usr/lib/libSystem.dylib"
         try:
-            clib = ctypes.CDLL(LIBC)
+            clib = _cdll(LIBC)
         except OSError as exc:
             logger.debug(
                 "unable to access '{}' (errno '{}'). "
@@ -39,40 +62,43 @@ class System:
             if exc.errno != errno.ENOENT:
                 raise
             # NOTE: trying to bypass System Integrity Protection (SIP)
-            clib = ctypes.CDLL(LIBC_FALLBACK)
+            clib = _cdll(LIBC_FALLBACK)
 
         if not hasattr(clib, "clonefile"):
-            return -1
+            raise OSError(
+                errno.ENOTSUP,
+                "'clonefile' not supported by the standard library",
+            )
 
         clonefile = clib.clonefile
         clonefile.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_int]
         clonefile.restype = ctypes.c_int
 
-        return clonefile(
+        ret = clonefile(
             ctypes.c_char_p(os.fsencode(src)),
             ctypes.c_char_p(os.fsencode(dst)),
             ctypes.c_int(0),
         )
-
-    @staticmethod
-    def _reflink_windows(_src, _dst):
-        return -1
+        if ret:
+            err = ctypes.get_errno()
+            msg = os.strerror(err)
+            raise OSError(err, msg)
 
     @staticmethod
     def _reflink_linux(src, dst):
         import fcntl  # pylint: disable=import-error
 
+        from funcy import suppress
+
         FICLONE = 0x40049409
 
         try:
-            ret = 255
             with open(src, "rb") as s, open(dst, "wb+") as d:
-                ret = fcntl.ioctl(d.fileno(), FICLONE, s.fileno())
-        finally:
-            if ret != 0:
+                fcntl.ioctl(d.fileno(), FICLONE, s.fileno())
+        except OSError:
+            with suppress(OSError):
                 os.unlink(dst)
-
-        return ret
+            raise
 
     @staticmethod
     def reflink(source, link_name):
@@ -81,20 +107,15 @@ class System:
         source, link_name = os.fspath(source), os.fspath(link_name)
 
         system = platform.system()
-        try:
-            if system == "Windows":
-                ret = System._reflink_windows(source, link_name)
-            elif system == "Darwin":
-                ret = System._reflink_darwin(source, link_name)
-            elif system == "Linux":
-                ret = System._reflink_linux(source, link_name)
-            else:
-                ret = -1
-        except OSError:
-            ret = -1
-
-        if ret != 0:
-            raise DvcException("reflink is not supported")
+        if system == "Darwin":
+            System._reflink_darwin(source, link_name)
+        elif system == "Linux":
+            System._reflink_linux(source, link_name)
+        else:
+            raise OSError(
+                errno.ENOTSUP,
+                f"reflink is not supported on '{system}'",
+            )
 
         # NOTE: reflink has a new inode, but has the same mode as the src,
         # so we need to chmod it to look like a normal copy.

dvc/ui/__init__.py

@@ -84,21 +84,35 @@ class Console:
         )
 
     def write_json(
-        self, data: Any, indent: int = 2, default: Callable = None
+        self,
+        data: Any,
+        indent: int = None,
+        highlight: bool = None,
+        stderr: bool = False,
+        skip_keys: bool = False,
+        ensure_ascii: bool = True,
+        check_circular: bool = True,
+        allow_nan: bool = True,
+        default: Optional[Callable[[Any], Any]] = None,
+        sort_keys: bool = False,
     ) -> None:
-        import json
+        if highlight is None:
+            highlight = self.isatty()
+        if indent is None and self.isatty():
+            indent = 2
 
-        if self.isatty():
-            from rich.highlighter import JSONHighlighter
-
-            j = json.dumps(data, indent=indent, default=default)
-            highlighter = JSONHighlighter()
-            text = highlighter(j)
-            text.no_wrap = True
-            text.overflow = None
-            return self.write(text, styled=True)
-
-        return self.write(json.dumps(data, default=default))
+        console = self.error_console if stderr else self.rich_console
+        return console.print_json(
+            data=data,
+            indent=indent,
+            highlight=bool(highlight),
+            skip_keys=skip_keys,
+            ensure_ascii=ensure_ascii,
+            check_circular=check_circular,
+            allow_nan=allow_nan,
+            default=default,
+            sort_keys=sort_keys,
+        )
 
     def write(
         self,

dvc/utils/fs.py

@@ -206,7 +206,7 @@ def copyfile(src, dest, callback=None, no_progress_bar=False, name=None):
 
     try:
         System.reflink(src, dest)
-    except DvcException:
+    except OSError:
         from dvc.progress import tdqm_or_callback_wrapped
 
         with open(src, "rb") as fsrc, open(dest, "wb+") as fdest:

scripts/build-requirements.txt

@@ -1,2 +1 @@
-PyInstaller==4.5.1
-numpy==1.19.3
+PyInstaller==4.6

setup.cfg

@@ -67,14 +67,15 @@ install_requires =
     pygtrie>=2.3.2
     dpath>=2.0.2,<3
     shtab>=1.3.4,<2
-    rich>=10.9.0
+    rich>=10.13.0
     dictdiffer>=0.8.1
     python-benedict>=0.24.2
-    pyparsing==2.4.7
+    pyparsing>=2.4.7
     typing_extensions>=3.7.4
     fsspec[http]>=2021.10.1
     aiohttp-retry>=2.4.5
     diskcache>=5.2.1
+    jaraco.windows>=5.7.0; python_version < '3.8' and sys_platform == 'win32'
 
 [options.extras_require]
 # gssapi should not be included in all_remotes, because it doesn't have wheels
@@ -99,18 +100,19 @@ azure =
     azure-identity>=1.4.0
     knack
 gdrive = pydrive2[fsspec]>=1.9.4
-gs = gcsfs==2021.10.1
+gs = gcsfs==2021.11.0
 hdfs =
-    fsspec[arrow]; python_version < '3.10'
+    fsspec[arrow]
 oss = ossfs==2021.8.0
 s3 =
-    s3fs==2021.10.1
+    s3fs==2021.11.0
     aiobotocore[boto3]>1.0.1
 ssh = sshfs[bcrypt]>=2021.8.1
 ssh_gssapi = sshfs[gssapi]>=2021.8.1
 webdav = webdav4>=0.9.3
 # not to break `dvc[webhdfs]`
 webhdfs =
+    requests-kerberos==0.13.0
 terraform = tpi[ssh]>=2.1.0
 tests =
     %(terraform)s
@@ -127,23 +129,22 @@ tests =
     rangehttpserver==1.2.0
     mock-ssh-server==0.9.1
     wget==3.2
-    filelock==3.3.1
+    filelock==3.3.2
     wsgidav==3.1.1
     crc32c==2.2.post0
     xmltodict==0.12.0
     google-compute-engine==2.8.13
     google-cloud-storage==1.42.3
-    dvclive==0.4.1
+    dvclive==0.4.2
     hdfs==2.6.0
     # required by collective.checkdocs
     Pygments==2.10.0
     collective.checkdocs==0.2
     pydocstyle==6.1.1
-    jaraco.windows==5.7.0
     # pylint requirements
     pylint==2.11.1
     # we use this to suppress pytest-related false positives in our tests.
-    pylint-pytest==1.0.3
+    pylint-pytest==1.1.2
     # we use this to suppress some messages in tests, eg: foo/bar naming,
     # and, protected method calls in our tests
     pylint-plugin-utils==0.6

tests/func/experiments/executor/test_ssh.py

@@ -0,0 +1,87 @@
+import posixpath
+from contextlib import contextmanager
+from functools import partial
+
+import pytest
+
+from dvc.fs.ssh import SSHFileSystem
+from dvc.path_info import URLInfo
+from dvc.repo.experiments.base import EXEC_HEAD, EXEC_MERGE
+from dvc.repo.experiments.executor.ssh import SSHExecutor
+from tests.func.machine.conftest import *  # noqa, pylint: disable=wildcard-import
+from tests.remotes.ssh import TEST_SSH_KEY_PATH, TEST_SSH_USER
+
+
+@contextmanager
+def _ssh_factory(cloud):
+    yield SSHFileSystem(
+        host=cloud.host,
+        port=cloud.port,
+        user=TEST_SSH_USER,
+        keyfile=TEST_SSH_KEY_PATH,
+    )
+
+
+def test_from_machine(tmp_dir, scm, dvc, machine_instance, mocker):
+    mocker.patch.object(SSHExecutor, "_init_git")
+    executor = SSHExecutor.from_machine(dvc.machine, "foo", scm, "")
+    assert executor.host == machine_instance["instance_ip"]
+
+
+@pytest.mark.needs_internet
+@pytest.mark.parametrize("cloud", [pytest.lazy_fixture("git_ssh")])
+def test_init_git(tmp_dir, scm, cloud):
+    tmp_dir.scm_gen({"foo": "foo", "dir": {"bar": "bar"}}, commit="init")
+    rev = scm.get_rev()
+    scm.set_ref(EXEC_HEAD, rev)
+    tmp_dir.gen("foo", "stashed")
+    scm.gitpython.git.stash()
+    rev = scm.resolve_rev("stash@{0}")
+    scm.set_ref(EXEC_MERGE, rev)
+
+    root_url = URLInfo(str(cloud)) / SSHExecutor.gen_dirname()
+
+    executor = SSHExecutor(
+        scm,
+        ".",
+        root_dir=root_url.path,
+        host=root_url.host,
+        port=root_url.port,
+        username=TEST_SSH_USER,
+        fs_factory=partial(_ssh_factory, cloud),
+    )
+    assert root_url.path == executor._repo_abspath
+
+    fs = cloud._ssh
+    assert fs.exists(posixpath.join(executor._repo_abspath, "foo"))
+    assert fs.exists(posixpath.join(executor._repo_abspath, "dir"))
+    assert fs.exists(posixpath.join(executor._repo_abspath, "dir", "bar"))
+
+
+@pytest.mark.needs_internet
+@pytest.mark.parametrize("cloud", [pytest.lazy_fixture("git_ssh")])
+def test_init_cache(tmp_dir, dvc, scm, cloud):
+    foo = tmp_dir.dvc_gen("foo", "foo", commit="init")[0].outs[0]
+    rev = scm.get_rev()
+    scm.set_ref(EXEC_HEAD, rev)
+    scm.set_ref(EXEC_MERGE, rev)
+    root_url = URLInfo(str(cloud)) / SSHExecutor.gen_dirname()
+
+    executor = SSHExecutor(
+        scm,
+        ".",
+        root_dir=root_url.path,
+        host=root_url.host,
+        port=root_url.port,
+        username=TEST_SSH_USER,
+        fs_factory=partial(_ssh_factory, cloud),
+    )
+    executor.init_cache(dvc, rev)
+
+    fs = cloud._ssh
+    foo_hash = foo.hash_info.value
+    assert fs.exists(
+        posixpath.join(
+            executor._repo_abspath, ".dvc", "cache", foo_hash[:2], foo_hash[2:]
+        )
+    )

tests/func/experiments/test_experiments.py

@@ -53,6 +53,7 @@ def test_experiment_exists(tmp_dir, scm, dvc, exp_stage, mocker, workspace):
         tmp_dir=not workspace,
     )
 
+    new_mock = mocker.spy(dvc.experiments, "_stash_exp")
     with pytest.raises(ExperimentExistsError):
         dvc.experiments.run(
             exp_stage.addressing,
@@ -60,6 +61,7 @@ def test_experiment_exists(tmp_dir, scm, dvc, exp_stage, mocker, workspace):
             params=["foo=3"],
             tmp_dir=not workspace,
         )
+    new_mock.assert_not_called()
 
     results = dvc.experiments.run(
         exp_stage.addressing,
@@ -685,3 +687,16 @@ def test_exp_run_recursive(tmp_dir, scm, dvc, run_copy_metrics):
     )
     assert dvc.experiments.run(".", recursive=True)
     assert (tmp_dir / "metric.json").parse() == {"foo": 1}
+
+
+def test_experiment_name_invalid(tmp_dir, scm, dvc, exp_stage, mocker):
+    from dvc.exceptions import InvalidArgumentError
+
+    new_mock = mocker.spy(dvc.experiments, "_stash_exp")
+    with pytest.raises(InvalidArgumentError):
+        dvc.experiments.run(
+            exp_stage.addressing,
+            name="fo^o",
+            params=["foo=3"],
+        )
+    new_mock.assert_not_called()

tests/func/machine/__init__.py

@@ -1,10 +0,0 @@
-import textwrap
-
-CONFIG_TEXT = textwrap.dedent(
-    """\
-        [feature]
-            machine = true
-        ['machine "foo"']
-            cloud = aws
-    """
-)

tests/func/machine/conftest.py

@@ -0,0 +1,53 @@
+import textwrap
+
+import pytest
+
+BASIC_CONFIG = textwrap.dedent(
+    """\
+        [feature]
+            machine = true
+        ['machine "foo"']
+            cloud = aws
+    """
+)
+
+TEST_INSTANCE = {
+    "aws_security_group": None,
+    "cloud": "aws",
+    "id": "iterative-2jyhw8j9ieov6",
+    "image": "ubuntu-bionic-18.04-amd64-server-20210818",
+    "instance_gpu": None,
+    "instance_hdd_size": 35,
+    "instance_ip": "123.123.123.123",
+    "instance_launch_time": "2021-08-25T07:13:03Z",
+    "instance_type": "m",
+    "name": "test-resource",
+    "region": "us-west",
+    "spot": False,
+    "spot_price": -1,
+    "ssh_name": None,
+    "ssh_private": "-----BEGIN RSA PRIVATE KEY-----\\n",
+    "startup_script": "IyEvYmluL2Jhc2g=",
+    "timeouts": None,
+}
+
+
+@pytest.fixture
+def machine_config(tmp_dir):
+    (tmp_dir / ".dvc" / "config").write_text(BASIC_CONFIG)
+    yield BASIC_CONFIG
+
+
+@pytest.fixture
+def machine_instance(tmp_dir, dvc, mocker):
+    from tpi.terraform import TerraformBackend
+
+    with dvc.config.edit() as conf:
+        conf["machine"]["foo"] = {"cloud": "aws"}
+    mocker.patch.object(
+        TerraformBackend,
+        "instances",
+        autospec=True,
+        return_value=[TEST_INSTANCE],
+    )
+    yield TEST_INSTANCE

tests/func/machine/test_machine_config.py

@@ -6,7 +6,7 @@ import tpi
 
 from dvc.main import main
 
-from . import CONFIG_TEXT
+from .conftest import BASIC_CONFIG
 
 
 @pytest.mark.parametrize(
@@ -24,25 +24,23 @@ from . import CONFIG_TEXT
         ("ssh_private", "secret"),
     ],
 )
-def test_machine_modify_susccess(tmp_dir, dvc, slot, value):
-    (tmp_dir / ".dvc" / "config").write_text(CONFIG_TEXT)
+def test_machine_modify_susccess(tmp_dir, dvc, machine_config, slot, value):
     assert main(["machine", "modify", "foo", slot, value]) == 0
     assert (
         tmp_dir / ".dvc" / "config"
-    ).read_text() == CONFIG_TEXT + f"    {slot} = {value}\n"
+    ).read_text() == machine_config + f"    {slot} = {value}\n"
     assert main(["machine", "modify", "--unset", "foo", slot]) == 0
-    assert (tmp_dir / ".dvc" / "config").read_text() == CONFIG_TEXT
+    assert (tmp_dir / ".dvc" / "config").read_text() == machine_config
 
 
-def test_machine_modify_startup_script(tmp_dir, dvc):
+def test_machine_modify_startup_script(tmp_dir, dvc, machine_config):
     slot, value = "startup_script", "start.sh"
-    (tmp_dir / ".dvc" / "config").write_text(CONFIG_TEXT)
     assert main(["machine", "modify", "foo", slot, value]) == 0
     assert (
         tmp_dir / ".dvc" / "config"
-    ).read_text() == CONFIG_TEXT + f"    {slot} = ../{value}\n"
+    ).read_text() == machine_config + f"    {slot} = ../{value}\n"
     assert main(["machine", "modify", "--unset", "foo", slot]) == 0
-    assert (tmp_dir / ".dvc" / "config").read_text() == CONFIG_TEXT
+    assert (tmp_dir / ".dvc" / "config").read_text() == machine_config
 
 
 @pytest.mark.parametrize(
@@ -57,11 +55,11 @@ def test_machine_modify_startup_script(tmp_dir, dvc):
         ("instance_hdd_size", "BIG", "expected int"),
     ],
 )
-def test_machine_modify_fail(tmp_dir, dvc, caplog, slot, value, msg):
-    (tmp_dir / ".dvc" / "config").write_text(CONFIG_TEXT)
-
+def test_machine_modify_fail(
+    tmp_dir, dvc, machine_config, caplog, slot, value, msg
+):
     assert main(["machine", "modify", "foo", slot, value]) == 251
-    assert (tmp_dir / ".dvc" / "config").read_text() == CONFIG_TEXT
+    assert (tmp_dir / ".dvc" / "config").read_text() == machine_config
     assert msg in caplog.text
 
 
@@ -117,9 +115,10 @@ def test_machine_list(tmp_dir, dvc, capsys):
     )
 
 
-def test_machine_rename_success(tmp_dir, scm, dvc, capsys, mocker):
+def test_machine_rename_success(
+    tmp_dir, scm, dvc, machine_config, capsys, mocker
+):
     config_file = tmp_dir / ".dvc" / "config"
-    config_file.write_text(CONFIG_TEXT)
 
     mocker.patch.object(
         tpi.terraform.TerraformBackend,
@@ -133,7 +132,7 @@ def test_machine_rename_success(tmp_dir, scm, dvc, capsys, mocker):
     assert main(["machine", "rename", "foo", "bar"]) == 0
     cap = capsys.readouterr()
     assert "Rename machine 'foo' to 'bar'." in cap.out
-    assert config_file.read_text() == CONFIG_TEXT.replace("foo", "bar")
+    assert config_file.read_text() == machine_config.replace("foo", "bar")
     assert not (
         tmp_dir / ".dvc" / "tmp" / "machine" / "terraform" / "foo"
     ).exists()
@@ -143,7 +142,7 @@ def test_machine_rename_success(tmp_dir, scm, dvc, capsys, mocker):
 
 
 def test_machine_rename_none_exist(tmp_dir, scm, dvc, caplog):
-    config_alice = CONFIG_TEXT.replace("foo", "alice")
+    config_alice = BASIC_CONFIG.replace("foo", "alice")
     config_file = tmp_dir / ".dvc" / "config"
     config_file.write_text(config_alice)
     assert main(["machine", "rename", "foo", "bar"]) == 251
@@ -152,7 +151,7 @@ def test_machine_rename_none_exist(tmp_dir, scm, dvc, caplog):
 
 
 def test_machine_rename_exist(tmp_dir, scm, dvc, caplog):
-    config_bar = CONFIG_TEXT + "['machine \"bar\"']\n    cloud = aws"
+    config_bar = BASIC_CONFIG + "['machine \"bar\"']\n    cloud = aws"
     config_file = tmp_dir / ".dvc" / "config"
     config_file.write_text(config_bar)
     assert main(["machine", "rename", "foo", "bar"]) == 251
@@ -160,10 +159,10 @@ def test_machine_rename_exist(tmp_dir, scm, dvc, caplog):
     assert "Machine 'bar' already exists." in caplog.text
 
 
-def test_machine_rename_error(tmp_dir, scm, dvc, caplog, mocker):
+def test_machine_rename_error(
+    tmp_dir, scm, dvc, machine_config, caplog, mocker
+):
     config_file = tmp_dir / ".dvc" / "config"
-    config_file.write_text(CONFIG_TEXT)
-
     os.makedirs((tmp_dir / ".dvc" / "tmp" / "machine" / "terraform" / "foo"))
 
     def cmd_error(self, source, destination, **kwargs):
@@ -172,5 +171,5 @@ def test_machine_rename_error(tmp_dir, scm, dvc, caplog, mocker):
     mocker.patch.object(tpi.terraform.TerraformBackend, "state_mv", cmd_error)
 
     assert main(["machine", "rename", "foo", "bar"]) == 251
-    assert config_file.read_text() == CONFIG_TEXT
+    assert config_file.read_text() == machine_config
     assert "rename failed" in caplog.text

tests/func/machine/test_machine_status.py

@@ -1,41 +1,11 @@
-import json
-
-from tpi.terraform import TerraformBackend
-
 from dvc.command.machine import CmdMachineStatus
 from dvc.main import main
 
-from . import CONFIG_TEXT
-
-STATUS = """{
-    "aws_security_group": null,
-    "cloud": "aws",
-    "id": "iterative-2jyhw8j9ieov6",
-    "image": "ubuntu-bionic-18.04-amd64-server-20210818",
-    "instance_gpu": null,
-    "instance_hdd_size": 35,
-    "instance_ip": "123.123.123.123",
-    "instance_launch_time": "2021-08-25T07:13:03Z",
-    "instance_type": "m",
-    "name": "test-resource",
-    "region": "us-west",
-    "spot": false,
-    "spot_price": -1,
-    "ssh_name": null,
-    "ssh_private": "-----BEGIN RSA PRIVATE KEY-----\\n",
-    "startup_script": "IyEvYmluL2Jhc2g=",
-    "timeouts": null
-}"""
-
-
-def test_status(tmp_dir, scm, dvc, mocker, capsys):
-    (tmp_dir / ".dvc" / "config").write_text(CONFIG_TEXT)
-    status = json.loads(STATUS)
-
-    mocker.patch.object(
-        TerraformBackend, "instances", autospec=True, return_value=[status]
-    )
 
+def test_status(
+    tmp_dir, scm, dvc, machine_config, machine_instance, mocker, capsys
+):
+    status = machine_instance
     assert main(["machine", "status", "foo"]) == 0
     cap = capsys.readouterr()
     assert "instance_num_1:" in cap.out

tests/func/objects/db/test_index.py

@@ -66,7 +66,7 @@ def test_clear_on_download_err(tmp_dir, dvc, index, mocker):
 
     assert list(index.hashes())
 
-    mocker.patch("dvc.fs.local.LocalFileSystem.upload", side_effect=Exception)
+    mocker.patch("dvc.fs.utils.transfer", side_effect=Exception)
     with pytest.raises(DownloadError):
         dvc.pull()
     assert not list(index.hashes())
@@ -83,7 +83,7 @@ def test_partial_upload(tmp_dir, dvc, index, mocker):
             raise Exception("stop baz")
         return original(self, from_file, to_info, name, **kwargs)
 
-    mocker.patch.object(LocalFileSystem, "upload", unreliable_upload)
+    mocker.patch("dvc.fs.utils.transfer", unreliable_upload)
     with pytest.raises(UploadError):
         dvc.push()
     assert not list(index.hashes())

tests/func/test_add.py

@@ -496,27 +496,27 @@ def test_should_update_state_entry_for_directory_after_add(
 
     ret = main(["add", "data"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 4
+    assert file_md5_counter.mock.call_count == 3
 
     ret = main(["status"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 4
+    assert file_md5_counter.mock.call_count == 3
 
     ls = "dir" if os.name == "nt" else "ls"
     ret = main(
         ["run", "--single-stage", "-d", "data", "{} {}".format(ls, "data")]
     )
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 4
+    assert file_md5_counter.mock.call_count == 3
 
     os.rename("data", "data" + ".back")
     ret = main(["checkout"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 4
+    assert file_md5_counter.mock.call_count == 3
 
     ret = main(["status"])
     assert ret == 0
-    assert file_md5_counter.mock.call_count == 4
+    assert file_md5_counter.mock.call_count == 3
 
 
 class TestAddCommit(TestDvc):
@@ -537,15 +537,15 @@ def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc):
     counter = mocker.spy(dvc_module.objects.stage, "_stage_tree")
     ret = main(["add", "data"])
     assert ret == 0
-    assert counter.mock.call_count == 2
+    assert counter.mock.call_count == 1
 
     ret = main(["status"])
     assert ret == 0
-    assert counter.mock.call_count == 2
+    assert counter.mock.call_count == 1
 
     ret = main(["status"])
     assert ret == 0
-    assert counter.mock.call_count == 2
+    assert counter.mock.call_count == 1
 
 
 class TestShouldPlaceStageInDataDirIfRepositoryBelowSymlink(TestDvc):
@@ -958,13 +958,13 @@ def test_add_with_cache_link_error(tmp_dir, dvc, mocker, capsys):
 
     mocker.patch(
         "dvc.objects.checkout._do_link",
-        side_effect=DvcException("link failed"),
+        side_effect=OSError(errno.ENOTSUP, "link failed"),
     )
     dvc.add("foo")
     err = capsys.readouterr()[1]
     assert "reconfigure cache types" in err
 
-    assert not (tmp_dir / "foo").exists()
+    assert (tmp_dir / "foo").exists()
     assert (tmp_dir / "foo.dvc").exists()
     assert (tmp_dir / ".dvc" / "cache").read_text() == {
         "ac": {"bd18db4cc2f85cedef654fccc4a4d8": "foo"}

tests/func/test_external_repo.py

@@ -2,7 +2,6 @@ import os
 from unittest.mock import ANY, patch
 
 from dvc.external_repo import CLONES, external_repo
-from dvc.fs.local import LocalFileSystem
 from dvc.objects.stage import stage
 from dvc.objects.transfer import transfer
 from dvc.path_info import PathInfo
@@ -48,12 +47,14 @@ def test_source_change(erepo_dir):
 
 
 def test_cache_reused(erepo_dir, mocker, local_cloud):
+    import dvc.fs.utils
+
     erepo_dir.add_remote(config=local_cloud.config)
     with erepo_dir.chdir():
         erepo_dir.dvc_gen("file", "text", commit="add file")
     erepo_dir.dvc.push()
 
-    download_spy = mocker.spy(LocalFileSystem, "upload")
+    download_spy = mocker.spy(dvc.fs.utils, "transfer")
 
     # Use URL to prevent any fishy optimizations
     url = f"file://{erepo_dir}"
@@ -215,7 +216,7 @@ def test_subrepos_are_ignored(tmp_dir, erepo_dir):
             repo.odb.local,
             {obj.hash_info},
             shallow=False,
-            move=True,
+            hardlink=True,
         )
         assert set(cache_dir.glob("??/*")) == {
             cache_dir / "e1" / "d9e8eae5374860ae025ec84cfd85c7.dir",

tests/func/test_import.py

@@ -308,9 +308,9 @@ def test_download_error_pulling_imported_stage(tmp_dir, dvc, erepo_dir):
     remove("foo_imported")
     remove(dst_cache)
 
-    with patch(
-        "dvc.fs.local.LocalFileSystem.upload", side_effect=Exception
-    ), pytest.raises(DownloadError):
+    with patch("dvc.fs.utils.transfer", side_effect=Exception), pytest.raises(
+        DownloadError
+    ):
         dvc.pull(["foo_imported.dvc"])
 
 

tests/func/test_install.py

@@ -17,7 +17,7 @@ class TestInstall:
         return pathlib.Path(".git") / "hooks" / name
 
     def test_create_hooks(self, scm, dvc):
-        scm.install()
+        dvc.install()
 
         hooks_with_commands = [
             ("post-checkout", "exec dvc git-hook post-checkout"),
@@ -30,22 +30,22 @@ class TestInstall:
             assert hook_path.is_file()
             assert command in hook_path.read_text()
 
-    def test_install_pre_commit_tool(self, scm):
-        scm.install(use_pre_commit_tool=True)
+    def test_install_pre_commit_tool(self, scm, dvc):
+        dvc.install(use_pre_commit_tool=True)
 
         precommit_path = pathlib.Path(".") / ".pre-commit-config.yaml"
         assert precommit_path.is_file()
 
-    def test_fail_if_hook_exists(self, scm):
+    def test_fail_if_hook_exists(self, scm, dvc):
         self._hook("post-checkout").write_text("hook content")
 
         with pytest.raises(GitHookAlreadyExistsError):
-            scm.install()
+            dvc.install()
 
     def test_pre_commit_hook(self, tmp_dir, scm, dvc, caplog):
         tmp_dir.dvc_gen("file", "file content", commit="create foo")
         tmp_dir.gen("file", "file modified")
-        scm.install()
+        dvc.install()
 
         # scm.commit bypasses hooks
         with pytest.raises(GitCommandError, match=r"modified:\s*file"):
@@ -54,7 +54,7 @@ class TestInstall:
     def test_post_checkout(self, tmp_dir, scm, dvc):
         tmp_dir.dvc_gen({"file": "file content"}, commit="add")
         os.unlink("file")
-        scm.install()
+        dvc.install()
 
         scm.gitpython.git.checkout("-b", "new_branch")
 
@@ -78,7 +78,7 @@ class TestInstall:
         scm.gitpython.repo.clone(os.fspath(git_remote))
         scm.gitpython.repo.create_remote("origin", os.fspath(git_remote))
 
-        scm.install()
+        dvc.install()
 
         assert not expected_storage_path.is_file()
         scm.gitpython.repo.git.push("origin", "master")
@@ -100,7 +100,7 @@ def test_merge_driver_no_ancestor(tmp_dir, scm, dvc):
 
     # installing hook only before merge, as it runs `dvc` commands which makes
     # `checkouts` and `commits` above slower
-    scm.install()
+    dvc.install()
     (tmp_dir / ".gitattributes").write_text("*.dvc merge=dvc")
 
     scm.gitpython.repo.git.merge(
@@ -137,7 +137,7 @@ def test_merge_driver(tmp_dir, scm, dvc):
 
     # installing hook only before merge, as it runs `dvc` commands on
     # `checkouts` and `commits` which slows tests down
-    scm.install()
+    dvc.install()
     (tmp_dir / ".gitattributes").write_text("*.dvc merge=dvc")
 
     scm.gitpython.repo.git.merge(

tests/func/test_remote.py

@@ -8,7 +8,6 @@ import pytest
 
 from dvc.config import Config
 from dvc.exceptions import DownloadError, RemoteCacheRequiredError, UploadError
-from dvc.fs.local import LocalFileSystem
 from dvc.main import main
 from dvc.path_info import PathInfo
 from dvc.utils.fs import remove
@@ -172,22 +171,24 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc):
 
 
 def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
+    import dvc.fs.utils as fs_utils
+
     foo = tmp_dir.dvc_gen({"foo": "foo content"})[0].outs[0]
     bar = tmp_dir.dvc_gen({"bar": "bar content"})[0].outs[0]
     baz = tmp_dir.dvc_gen({"baz": {"foo": "foo content"}})[0].outs[0]
 
     # Faulty upload version, failing on foo
-    original = LocalFileSystem.upload
+    original = fs_utils.transfer
     odb = dvc.cloud.get_remote_odb("upstream")
 
-    def unreliable_upload(self, fobj, to_info, **kwargs):
+    def unreliable_upload(from_fs, from_info, to_fs, to_info, **kwargs):
         if os.path.abspath(to_info) == os.path.abspath(
             odb.get(foo.hash_info).path_info
         ):
             raise Exception("stop foo")
-        return original(self, fobj, to_info, **kwargs)
+        return original(from_fs, from_info, to_fs, to_info, **kwargs)
 
-    with patch.object(LocalFileSystem, "upload", unreliable_upload):
+    with patch.object(fs_utils, "transfer", unreliable_upload):
         with pytest.raises(UploadError) as upload_error_info:
             dvc.push()
         assert upload_error_info.value.amount == 2
@@ -201,7 +202,7 @@ def test_partial_push_n_pull(tmp_dir, dvc, tmp_path_factory, local_remote):
     remove(dvc.odb.local.cache_dir)
 
     baz._collect_used_dir_cache()
-    with patch.object(LocalFileSystem, "upload", side_effect=Exception):
+    with patch.object(fs_utils, "transfer", side_effect=Exception):
         with pytest.raises(DownloadError) as download_error_info:
             dvc.pull()
         # error count should be len(.dir + standalone file checksums)
@@ -214,9 +215,8 @@ def test_raise_on_too_many_open_files(
 ):
     tmp_dir.dvc_gen({"file": "file content"})
 
-    mocker.patch.object(
-        LocalFileSystem,
-        "upload",
+    mocker.patch(
+        "dvc.fs.utils.transfer",
         side_effect=OSError(errno.EMFILE, "Too many open files"),
     )
 
@@ -263,19 +263,19 @@ def test_external_dir_resource_on_no_cache(tmp_dir, dvc, tmp_path_factory):
 
 
 def test_push_order(tmp_dir, dvc, tmp_path_factory, mocker, local_remote):
+    import dvc.fs.utils as fs_utils
+
     foo = tmp_dir.dvc_gen({"foo": {"bar": "bar content"}})[0].outs[0]
     tmp_dir.dvc_gen({"baz": "baz content"})
 
-    mocked_upload = mocker.patch.object(
-        LocalFileSystem, "upload", return_value=0
-    )
+    mocked_upload = mocker.spy(fs_utils, "transfer")
     dvc.push()
 
     # foo .dir file should be uploaded after bar
     odb = dvc.cloud.get_remote_odb("upstream")
     foo_path = odb.hash_to_path_info(foo.hash_info.value)
     bar_path = odb.hash_to_path_info(foo.obj.trie[("bar",)][1].value)
-    paths = [args[1] for args, _ in mocked_upload.call_args_list]
+    paths = [args[3] for args, _ in mocked_upload.call_args_list]
     assert paths.index(foo_path) > paths.index(bar_path)
 
 

tests/remotes/hdfs.py

@@ -2,7 +2,6 @@ import locale
 import os
 import platform
 import subprocess
-import sys
 import uuid
 from contextlib import contextmanager
 from pathlib import Path
@@ -74,10 +73,7 @@ class HDFS(Base, URLInfo):  # pylint: disable=abstract-method
 @pytest.fixture(scope="session")
 def hadoop(test_config):
     test_config.requires("real_hdfs")
-
-    if sys.version_info >= (3, 10):
-        pytest.skip("pyarrow is not available for 3.10 yet.")
-
+    pytest.importorskip("pyarrow.fs")
     if platform.system() != "Linux":
         pytest.skip("only supported on Linux")
 
@@ -280,9 +276,7 @@ def hdfs(test_config, mocker):
     # "The pyarrow installation is not built with support for
     # 'HadoopFileSystem'"
     test_config.requires("hdfs")
-
-    if sys.version_info >= (3, 10):
-        pytest.skip("pyarrow is not available for 3.10 yet.")
+    pytest.importorskip("pyarrow.fs")
 
     mocker.patch("pyarrow.fs._not_imported", [])
     mocker.patch(

tests/unit/command/test_plots.py

@@ -1,5 +1,6 @@
 import os
 import posixpath
+from pathlib import Path
 
 import pytest
 
@@ -149,12 +150,31 @@ def test_plots_diff_open(tmp_dir, dvc, mocker, capsys, plots_data):
     mocker.patch("dvc.command.plots.render", return_value=index_path)
 
     assert cmd.run() == 0
-    mocked_open.assert_called_once_with(index_path)
+    mocked_open.assert_called_once_with(index_path.as_uri())
 
     out, _ = capsys.readouterr()
     assert index_path.as_uri() in out
 
 
+def test_plots_diff_open_WSL(tmp_dir, dvc, mocker, plots_data):
+    mocked_open = mocker.patch("webbrowser.open", return_value=True)
+    mocked_uname_result = mocker.MagicMock()
+    mocked_uname_result.release = "Microsoft"
+    mocker.patch("platform.uname", return_value=mocked_uname_result)
+
+    cli_args = parse_args(
+        ["plots", "diff", "--targets", "plots.csv", "--open"]
+    )
+    cmd = cli_args.func(cli_args)
+    mocker.patch("dvc.repo.plots.diff.diff", return_value=plots_data)
+
+    index_path = tmp_dir / "dvc_plots" / "index.html"
+    mocker.patch("dvc.command.plots.render", return_value=index_path)
+
+    assert cmd.run() == 0
+    mocked_open.assert_called_once_with(Path("dvc_plots") / "index.html")
+
+
 def test_plots_diff_open_failed(tmp_dir, dvc, mocker, capsys, plots_data):
     mocked_open = mocker.patch("webbrowser.open", return_value=False)
     cli_args = parse_args(
@@ -167,7 +187,7 @@ def test_plots_diff_open_failed(tmp_dir, dvc, mocker, capsys, plots_data):
 
     assert cmd.run() == 1
     expected_url = tmp_dir / "dvc_plots" / "index.html"
-    mocked_open.assert_called_once_with(expected_url)
+    mocked_open.assert_called_once_with(expected_url.as_uri())
 
     error_message = "Failed to open. Please try opening it manually."
 

tests/unit/objects/db/test_local.py

@@ -103,10 +103,9 @@ def test_staging_file(tmp_dir, dvc):
         check(local_odb, obj)
     check(staging_odb, obj)
 
-    transfer(staging_odb, local_odb, {obj.hash_info}, move=True)
+    transfer(staging_odb, local_odb, {obj.hash_info}, hardlink=True)
     check(local_odb, obj)
-    with pytest.raises(FileNotFoundError):
-        check(staging_odb, obj)
+    check(staging_odb, obj)
 
     path_info = local_odb.hash_to_path_info(obj.hash_info.value)
     assert fs.exists(path_info)
@@ -130,10 +129,11 @@ def test_staging_dir(tmp_dir, dvc):
         check(local_odb, obj)
     check(staging_odb, obj)
 
-    transfer(staging_odb, local_odb, {obj.hash_info}, shallow=False, move=True)
+    transfer(
+        staging_odb, local_odb, {obj.hash_info}, shallow=False, hardlink=True
+    )
     check(local_odb, obj)
-    with pytest.raises(FileNotFoundError):
-        check(staging_odb, obj)
+    check(staging_odb, obj)
 
     path_info = local_odb.hash_to_path_info(obj.hash_info.value)
     assert fs.exists(path_info)

tests/unit/remote/test_webhdfs.py

@@ -1,21 +1,54 @@
+from unittest.mock import Mock, create_autospec
+
+import pytest
+import requests
+
 from dvc.fs.webhdfs import WebHDFSFileSystem
 
+host = "host"
+kerberos = False
+kerberos_principal = "principal"
+port = 12345
+proxy_to = "proxy"
+ssl_verify = False
+token = "token"
+use_https = True
 user = "test"
-webhdfs_token = "token"
-webhdfs_alias = "alias-name"
-hdfscli_config = "path/to/cli/config"
 
 
-def test_init(dvc):
-    url = "webhdfs://[email protected]:50070"
-    config = {
-        "host": url,
-        "webhdfs_token": webhdfs_token,
-        "webhdfs_alias": webhdfs_alias,
-        "hdfscli_config": hdfscli_config,
-        "user": user,
+@pytest.fixture()
+def webhdfs_config():
+    url = f"webhdfs://{user}@{host}:{port}"
+    url_config = WebHDFSFileSystem._get_kwargs_from_urls(url)
+    return {
+        "kerberos": kerberos,
+        "kerberos_principal": kerberos_principal,
+        "proxy_to": proxy_to,
+        "ssl_verify": ssl_verify,
+        "token": token,
+        "use_https": use_https,
+        **url_config,
     }
 
-    fs = WebHDFSFileSystem(**config)
-    assert fs.fs_args["token"] == webhdfs_token
+
+def test_init(dvc, webhdfs_config):
+    fs = WebHDFSFileSystem(**webhdfs_config)
+    assert fs.fs_args["host"] == host
+    assert fs.fs_args["token"] == token
     assert fs.fs_args["user"] == user
+    assert fs.fs_args["port"] == port
+    assert fs.fs_args["kerberos"] == kerberos
+    assert fs.fs_args["kerb_kwargs"] == {"principal": kerberos_principal}
+    assert fs.fs_args["proxy_to"] == proxy_to
+    assert fs.fs_args["use_https"] == use_https
+
+
+def test_verify_ssl(dvc, webhdfs_config, monkeypatch):
+    mock_session = create_autospec(requests.Session)
+    monkeypatch.setattr(requests, "Session", Mock(return_value=mock_session))
+    # can't have token at the same time as user or proxy_to
+    del webhdfs_config["token"]
+    fs = WebHDFSFileSystem(**webhdfs_config)
+    # ssl verify can't be set until after the file system is instantiated
+    fs.fs  # pylint: disable=pointless-statement
+    assert mock_session.verify == ssl_verify

tests/unit/render/test_vega.py

@@ -451,3 +451,29 @@ def test_find_vega(tmp_dir, dvc):
         first(plot_content["layer"])["encoding"]["x"]["field"] == INDEX_FIELD
     )
     assert first(plot_content["layer"])["encoding"]["y"]["field"] == "y"
+
+
+@pytest.mark.parametrize(
+    "template_path, target_name",
+    [
+        (os.path.join(".dvc", "plots", "template.json"), "template"),
+        (os.path.join(".dvc", "plots", "template.json"), "template.json"),
+        (
+            os.path.join(".dvc", "plots", "subdir", "template.json"),
+            os.path.join("subdir", "template.json"),
+        ),
+        (
+            os.path.join(".dvc", "plots", "subdir", "template.json"),
+            os.path.join("subdir", "template"),
+        ),
+        ("template.json", "template.json"),
+    ],
+)
+def test_should_resolve_template(tmp_dir, dvc, template_path, target_name):
+    os.makedirs(os.path.abspath(os.path.dirname(template_path)), exist_ok=True)
+    with open(template_path, "w", encoding="utf-8") as fd:
+        fd.write("template_content")
+
+    assert dvc.plots.templates._find_in_project(
+        target_name
+    ) == os.path.abspath(template_path)

tests/unit/repo/experiments/test_utils.py

@@ -1,7 +1,8 @@
 import pytest
 
+from dvc.exceptions import InvalidArgumentError
 from dvc.repo.experiments.base import EXPS_NAMESPACE, ExpRefInfo
-from dvc.repo.experiments.utils import resolve_exp_ref
+from dvc.repo.experiments.utils import check_ref_format, resolve_exp_ref
 
 
 def commit_exp_ref(tmp_dir, scm, file="foo", contents="foo", name="foo"):
@@ -25,3 +26,16 @@ def test_resolve_exp_ref(tmp_dir, scm, git_upstream, name_only, use_url):
     remote_ref_info = resolve_exp_ref(scm, "foo" if name_only else ref, remote)
     assert isinstance(remote_ref_info, ExpRefInfo)
     assert str(remote_ref_info) == ref
+
+
+@pytest.mark.parametrize(
+    "name,result", [("name", True), ("group/name", False), ("na me", False)]
+)
+def test_run_check_ref_format(scm, name, result):
+
+    ref = ExpRefInfo("abc123", name)
+    if result:
+        check_ref_format(scm, ref)
+    else:
+        with pytest.raises(InvalidArgumentError):
+            check_ref_format(scm, ref)

tests/unit/test_tabular_data.py

@@ -177,3 +177,36 @@ def test_row_from_dict():
         ["", "", "value3", "value4", "", ""],
         ["", "", "value3", "", "value5", "value6"],
     ]
+
+
+@pytest.mark.parametrize(
+    "axis,expected",
+    [
+        (
+            "rows",
+            [
+                ["foo", "bar", "foobar"],
+            ],
+        ),
+        ("cols", [["foo"], ["foo"], ["foo"]]),
+    ],
+)
+def test_dropna(axis, expected):
+    td = TabularData(["col-1", "col-2", "col-3"])
+    td.extend([["foo"], ["foo", "bar"], ["foo", "bar", "foobar"]])
+    assert list(td) == [
+        ["foo", "", ""],
+        ["foo", "bar", ""],
+        ["foo", "bar", "foobar"],
+    ]
+
+    td.dropna(axis)
+
+    assert list(td) == expected
+
+
+def test_dropna_invalid_axis():
+    td = TabularData(["col-1", "col-2", "col-3"])
+
+    with pytest.raises(ValueError, match="Invalid 'axis' value foo."):
+        td.dropna("foo")