fetch: merge config instead of updating

Update will replace sections instead of merging them together, which
is not desirable here.

Related https://github.com/iterative/studio/pull/6541

.flake8

@@ -1,19 +1,20 @@
 [flake8]
-ignore=
+max_line_length = 88
+extend-ignore=
     # Whitespace before ':'
     E203
-    # Too many leading '#' for block comment
-    E266
     # Line break occurred before a binary operator
     W503
     # unindexed parameters in the str.format, see:
     # https://pypi.org/project/flake8-string-format/
     P1
-    # use `!r` instead of surrounding by quotes
-    B028
-max_line_length = 88
 max-complexity = 10
-select = B,C,E,F,W,T4,B902,T,P,TYP
 show_source = true
 count = true
 min_python_version = 3.8.0
+per-file-ignores =
+    dvc/testing/benchmarks/conftest.py:F401
+    dvc/testing/conftest.py:F401
+    dvc/testing/plugin.py:F401
+    tests/conftest.py:F401
+    tests/func/experiments/executor/test_ssh.py:F401

.pre-commit-config.yaml

@@ -23,17 +23,6 @@ repos:
       - id: mixed-line-ending
       - id: sort-simple-yaml
       - id: trailing-whitespace
-  - repo: https://github.com/lovesegfault/beautysh
-    rev: v6.2.1
-    hooks:
-      - id: beautysh
-        args: ["-i", "2"]
-  - repo: https://github.com/asottile/pyupgrade
-    rev: v3.9.0
-    hooks:
-      - id: pyupgrade
-        args:
-          - --py38-plus
   - repo: https://github.com/psf/black
     rev: 23.7.0
     hooks:
@@ -43,22 +32,20 @@ repos:
     hooks:
       - id: isort
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.0.278'
+    rev: 'v0.0.282'
     hooks:
       - id: ruff
+        args: [--fix, --exit-non-zero-on-fix]
   - repo: https://github.com/codespell-project/codespell
     rev: v2.2.5
     hooks:
       - id: codespell
         additional_dependencies: ["tomli"]
   - repo: https://github.com/PyCQA/flake8
-    rev: 6.0.0
+    rev: 6.1.0
     hooks:
       - id: flake8
         additional_dependencies:
-          - flake8-bugbear
-          - flake8-comprehensions
-          - flake8-debugger
           - flake8-string-format
           - flake8-typing-imports
   - repo: https://github.com/PyCQA/bandit

dvc/api/data.py

@@ -67,7 +67,7 @@ class _OpenContextManager(GCM):
         raise AttributeError("dvc.api.open() should be used in a with statement.")
 
 
-def open(  # noqa, pylint: disable=redefined-builtin
+def open(  # noqa: A001, pylint: disable=redefined-builtin
     path: str,
     repo: Optional[str] = None,
     rev: Optional[str] = None,
@@ -75,6 +75,7 @@ def open(  # noqa, pylint: disable=redefined-builtin
     mode: str = "r",
     encoding: Optional[str] = None,
     config: Optional[Dict[str, Any]] = None,
+    remote_config: Optional[Dict[str, Any]] = None,
 ):
     """
     Opens a file tracked in a DVC project.
@@ -118,6 +119,9 @@ def open(  # noqa, pylint: disable=redefined-builtin
             Mirrors the namesake parameter in builtin `open()`_.
         config(dict, optional): config to be passed to the DVC repository.
             Defaults to None.
+        remote_config(dict, optional): remote config to be passed to the DVC
+            repository.
+            Defaults to None.
 
     Returns:
         _OpenContextManager: A context manager that generatse a corresponding
@@ -214,22 +218,27 @@ def open(  # noqa, pylint: disable=redefined-builtin
         "mode": mode,
         "encoding": encoding,
         "config": config,
+        "remote_config": remote_config,
     }
     return _OpenContextManager(_open, args, kwargs)
 
 
-def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
-    if remote:
-        if config is not None:
-            raise ValueError(
-                "can't specify both `remote` and `config` at the same time"
-            )
-        config = {"core": {"remote": remote}}
-
+def _open(
+    path,
+    repo=None,
+    rev=None,
+    remote=None,
+    mode="r",
+    encoding=None,
+    config=None,
+    remote_config=None,
+):
     repo_kwargs: Dict[str, Any] = {
         "subrepos": True,
         "uninitialized": True,
+        "remote": remote,
         "config": config,
+        "remote_config": remote_config,
     }
 
     with Repo.open(repo, rev=rev, **repo_kwargs) as _repo:
@@ -265,7 +274,16 @@ def _open(path, repo=None, rev=None, remote=None, mode="r", encoding=None, confi
                 raise DvcIsADirectoryError(f"'{path}' is a directory") from exc
 
 
-def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config=None):
+def read(
+    path,
+    repo=None,
+    rev=None,
+    remote=None,
+    mode="r",
+    encoding=None,
+    config=None,
+    remote_config=None,
+):
     """
     Returns the contents of a tracked file (by DVC or Git). For Git repos, HEAD
     is used unless a rev argument is supplied. The default remote is tried
@@ -279,5 +297,6 @@ def read(path, repo=None, rev=None, remote=None, mode="r", encoding=None, config
         mode=mode,
         encoding=encoding,
         config=config,
+        remote_config=remote_config,
     ) as fd:
         return fd.read()

dvc/cli/__init__.py

@@ -230,7 +230,7 @@ def main(argv=None):  # noqa: C901, PLR0912, PLR0915
         logger.exception("")
     except DvcParserError:
         ret = 254
-    except Exception as exc:  # noqa, pylint: disable=broad-except
+    except Exception as exc:  # noqa: BLE001, pylint: disable=broad-except
         # pylint: disable=no-member
         ret = _log_exceptions(exc) or 255
 

dvc/cli/utils.py

@@ -1,3 +1,27 @@
+import argparse
+
+
+class DictAction(argparse.Action):
+    def __call__(self, parser, args, values, option_string=None):  # noqa: ARG002
+        d = getattr(args, self.dest) or {}
+
+        if isinstance(values, list):
+            kvs = values
+        else:
+            kvs = [values]
+
+        for kv in kvs:
+            key, value = kv.split("=")
+            if not value:
+                raise argparse.ArgumentError(
+                    self,
+                    f'Could not parse argument "{values}" as k1=v1 k2=v2 ... format',
+                )
+            d[key] = value
+
+        setattr(args, self.dest, d)
+
+
 def fix_subparsers(subparsers):
     """Workaround for bug in Python 3. See more info at:
     https://bugs.python.org/issue16308

dvc/commands/commit.py

@@ -47,7 +47,10 @@ def add_parser(subparsers, parent_parser):
         "--force",
         action="store_true",
         default=False,
-        help="Commit even if hash value for dependencies/outputs changed.",
+        help=(
+            "Commit data even if hash values for dependencies or "
+            "outputs did not change."
+        ),
     )
     commit_parser.add_argument(
         "-d",

dvc/commands/completion.py

@@ -9,6 +9,9 @@ from dvc.ui import ui
 logger = logging.getLogger(__name__)
 
 
+SUPPORTED_SHELLS = ["bash", "zsh"]
+
+
 class CmdCompletion(CmdBaseNoRepo):
     def run(self):
         import shtab
@@ -35,6 +38,6 @@ def add_parser(subparsers, parent_parser):
         "--shell",
         help="Shell syntax for completions.",
         default="bash",
-        choices=["bash", "zsh"],
+        choices=SUPPORTED_SHELLS,
     )
     completion_parser.set_defaults(func=CmdCompletion)

dvc/commands/get.py

@@ -3,7 +3,7 @@ import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
-from dvc.cli.utils import append_doc_link
+from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)
@@ -38,6 +38,8 @@ class CmdGet(CmdBaseNoRepo):
                 jobs=self.args.jobs,
                 force=self.args.force,
                 config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
             )
             return 0
         except CloneError:
@@ -111,4 +113,19 @@ def add_parser(subparsers, parent_parser):
             "in the target repository."
         ),
     )
+    get_parser.add_argument(
+        "--remote",
+        type=str,
+        help="Remote name to set as a default in the target repository.",
+    )
+    get_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository."
+        ),
+    )
     get_parser.set_defaults(func=CmdGet)

dvc/commands/imp.py

@@ -3,7 +3,7 @@ import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBase
-from dvc.cli.utils import append_doc_link
+from dvc.cli.utils import DictAction, append_doc_link
 from dvc.exceptions import DvcException
 
 logger = logging.getLogger(__name__)
@@ -23,6 +23,8 @@ class CmdImport(CmdBase):
                 no_download=self.args.no_download,
                 jobs=self.args.jobs,
                 config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
             )
         except CloneError:
             logger.exception("failed to import '%s'", self.args.path)
@@ -103,4 +105,19 @@ def add_parser(subparsers, parent_parser):
             "in the target repository."
         ),
     )
+    import_parser.add_argument(
+        "--remote",
+        type=str,
+        help="Remote name to set as a default in the target repository.",
+    )
+    import_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository."
+        ),
+    )
     import_parser.set_defaults(func=CmdImport)

dvc/commands/ls/__init__.py

@@ -3,7 +3,7 @@ import logging
 
 from dvc.cli import completion
 from dvc.cli.command import CmdBaseNoRepo
-from dvc.cli.utils import append_doc_link
+from dvc.cli.utils import DictAction, append_doc_link
 from dvc.commands.ls.ls_colors import LsColors
 from dvc.exceptions import DvcException
 from dvc.ui import ui
@@ -35,6 +35,8 @@ class CmdList(CmdBaseNoRepo):
                 recursive=self.args.recursive,
                 dvc_only=self.args.dvc_only,
                 config=self.args.config,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
             )
             if self.args.json:
                 ui.write_json(entries)
@@ -89,6 +91,21 @@ def add_parser(subparsers, parent_parser):
             "in the target repository."
         ),
     )
+    list_parser.add_argument(
+        "--remote",
+        type=str,
+        help="Remote name to set as a default in the target repository.",
+    )
+    list_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository."
+        ),
+    )
     list_parser.add_argument(
         "path",
         nargs="?",

dvc/compare.py

@@ -57,7 +57,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         return self._columns[name]
 
     def items(self) -> ItemsView[str, Column]:
-        projection = {k: self.column(k) for k in self.keys()}  # noqa: SIM118
+        projection = {k: self.column(k) for k in self.keys()}
         return projection.items()
 
     def keys(self) -> List[str]:
@@ -171,8 +171,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
                 self.add_column(key)
 
         row: List["CellT"] = [
-            with_value(d.get(key), self._fill_value)
-            for key in self.keys()  # noqa: SIM118
+            with_value(d.get(key), self._fill_value) for key in self.keys()
         ]
         self.append(row)
 
@@ -227,7 +226,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
             to_drop -= match_line
 
         if axis == "rows":
-            for name in self.keys():  # noqa: SIM118
+            for name in self.keys():
                 self._columns[name] = Column(
                     [x for n, x in enumerate(self._columns[name]) if n not in to_drop]
                 )
@@ -275,7 +274,7 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
                 else:
                     unique_rows.append(tuple_row)
 
-            for name in self.keys():  # noqa: SIM118
+            for name in self.keys():
                 self._columns[name] = Column(
                     [
                         x

dvc/config.py

@@ -90,9 +90,12 @@ class Config(dict):
     def __init__(
         self,
         dvc_dir: Optional["StrPath"] = None,
+        local_dvc_dir: Optional["StrPath"] = None,
         validate: bool = True,
         fs: Optional["FileSystem"] = None,
         config: Optional["DictStrAny"] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
     ):  # pylint: disable=super-init-not-called
         from dvc.fs import LocalFileSystem
 
@@ -101,9 +104,15 @@ class Config(dict):
         self.fs = fs or self.wfs
 
         if dvc_dir:
-            self.dvc_dir = self.fs.path.abspath(self.fs.path.realpath(dvc_dir))
+            self.dvc_dir = self.fs.path.abspath(dvc_dir)
 
-        self.load(validate=validate, config=config)
+        self.local_dvc_dir = local_dvc_dir
+        if not fs and not local_dvc_dir:
+            self.local_dvc_dir = dvc_dir
+
+        self.load(
+            validate=validate, config=config, remote=remote, remote_config=remote_config
+        )
 
     @classmethod
     def from_cwd(cls, fs: Optional["FileSystem"] = None, **kwargs):
@@ -136,7 +145,9 @@ class Config(dict):
 
         if self.dvc_dir is not None:
             files["repo"] = self.fs.path.join(self.dvc_dir, self.CONFIG)
-            files["local"] = self.fs.path.join(self.dvc_dir, self.CONFIG_LOCAL)
+
+        if self.local_dvc_dir is not None:
+            files["local"] = self.wfs.path.join(self.local_dvc_dir, self.CONFIG_LOCAL)
 
         return files
 
@@ -154,7 +165,16 @@ class Config(dict):
         with open(config_file, "w+", encoding="utf-8"):
             return Config(dvc_dir)
 
-    def load(self, validate: bool = True, config: Optional["DictStrAny"] = None):
+    def merge(self, config):
+        merge(self, config)
+
+    def load(
+        self,
+        validate: bool = True,
+        config: Optional["DictStrAny"] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
+    ):
         """Loads config from all the config files.
 
         Raises:
@@ -169,6 +189,17 @@ class Config(dict):
             conf = self.validate(conf)
 
         self.clear()
+
+        if remote:
+            conf["core"]["remote"] = remote
+
+        if remote_config:
+            remote = remote or conf["core"].get("remote")
+            if not remote:
+                raise ValueError("Missing remote name")
+
+            merge(conf, {"remote": {remote: remote_config}})
+
         self.update(conf)
 
     def _get_fs(self, level):
@@ -229,21 +260,30 @@ class Config(dict):
         return conf
 
     @staticmethod
-    def _load_paths(conf, filename):
-        abs_conf_dir = os.path.abspath(os.path.dirname(filename))
+    def _resolve(conf_dir, path):
+        from .config_schema import ExpPath, RelPath
 
-        def resolve(path):
-            from .config_schema import RelPath
+        if re.match(r"\w+://", path):
+            return path
 
-            if os.path.isabs(path) or re.match(r"\w+://", path):
-                return path
+        if os.path.isabs(path):
+            return path
 
-            # on windows convert slashes to backslashes
-            # to have path compatible with abs_conf_dir
-            if os.path.sep == "\\" and "/" in path:
-                path = path.replace("/", "\\")
+        # on windows convert slashes to backslashes
+        # to have path compatible with abs_conf_dir
+        if os.path.sep == "\\" and "/" in path:
+            path = path.replace("/", "\\")
 
-            return RelPath(os.path.join(abs_conf_dir, path))
+        expanded = os.path.expanduser(path)
+        if os.path.isabs(expanded):
+            return ExpPath(expanded, path)
+
+        return RelPath(os.path.abspath(os.path.join(conf_dir, path)))
+
+    @classmethod
+    def _load_paths(cls, conf, filename):
+        conf_dir = os.path.abspath(os.path.dirname(filename))
+        resolve = partial(cls._resolve, conf_dir)
 
         return Config._map_dirs(conf, resolve)
 
@@ -252,11 +292,17 @@ class Config(dict):
         from dvc.fs import localfs
         from dvc.utils import relpath
 
-        from .config_schema import RelPath
+        from .config_schema import ExpPath, RelPath
 
         if re.match(r"\w+://", path):
             return path
 
+        if isinstance(path, ExpPath):
+            return path.def_path
+
+        if os.path.expanduser(path) != path:
+            return localfs.path.as_posix(path)
+
         if isinstance(path, RelPath) or not os.path.isabs(path):
             path = relpath(path, conf_dir)
             return localfs.path.as_posix(path)

dvc/config_schema.py

@@ -8,6 +8,7 @@ from voluptuous import (
     All,
     Any,
     Coerce,
+    Exclusive,
     Invalid,
     Lower,
     Optional,
@@ -74,6 +75,17 @@ def ByUrl(mapping):  # noqa: N802
     return validate
 
 
+class ExpPath(str):
+    __slots__ = ("def_path",)
+
+    def_path: str
+
+    def __new__(cls, string, def_path):
+        ret = super().__new__(cls, string)
+        ret.def_path = def_path
+        return ret
+
+
 class RelPath(str):
     __slots__ = ()
 
@@ -324,7 +336,8 @@ SCHEMA = {
     },
     "hydra": {
         Optional("enabled", default=False): Bool,
-        "config_dir": str,
+        Exclusive("config_dir", "config_source"): str,
+        Exclusive("config_module", "config_source"): str,
         "config_name": str,
     },
     "studio": {

dvc/data_cloud.py

@@ -4,6 +4,7 @@ import logging
 from typing import TYPE_CHECKING, Iterable, Optional, Set, Tuple
 
 from dvc.config import NoRemoteError, RemoteConfigError
+from dvc.fs.callbacks import Callback
 from dvc.utils.objects import cached_property
 from dvc_data.hashfile.db import get_index
 from dvc_data.hashfile.transfer import TransferResult
@@ -209,15 +210,20 @@ class DataCloud:
             cache = self.repo.cache.legacy
         else:
             cache = self.repo.cache.local
-        return self.transfer(
-            cache,
-            odb,
-            objs,
-            jobs=jobs,
-            dest_index=get_index(odb),
-            cache_odb=cache,
-            validate_status=self._log_missing,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Pushing to {odb.fs.unstrip_protocol(odb.path)}",
+            unit="file",
+        ) as cb:
+            return self.transfer(
+                cache,
+                odb,
+                objs,
+                jobs=jobs,
+                dest_index=get_index(odb),
+                cache_odb=cache,
+                validate_status=self._log_missing,
+                callback=cb,
+            )
 
     def pull(
         self,
@@ -263,16 +269,21 @@ class DataCloud:
             cache = self.repo.cache.legacy
         else:
             cache = self.repo.cache.local
-        return self.transfer(
-            odb,
-            cache,
-            objs,
-            jobs=jobs,
-            src_index=get_index(odb),
-            cache_odb=cache,
-            verify=odb.verify,
-            validate_status=self._log_missing,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Fetching from {odb.fs.unstrip_protocol(odb.path)}",
+            unit="file",
+        ) as cb:
+            return self.transfer(
+                odb,
+                cache,
+                objs,
+                jobs=jobs,
+                src_index=get_index(odb),
+                cache_odb=cache,
+                verify=odb.verify,
+                validate_status=self._log_missing,
+                callback=cb,
+            )
 
     def status(
         self,

dvc/dependency/repo.py

@@ -1,6 +1,6 @@
 from typing import TYPE_CHECKING, Dict, Optional, Union
 
-from voluptuous import Required
+from voluptuous import Any, Required
 
 from dvc.utils import as_posix
 
@@ -17,6 +17,7 @@ class RepoDependency(Dependency):
     PARAM_REV = "rev"
     PARAM_REV_LOCK = "rev_lock"
     PARAM_CONFIG = "config"
+    PARAM_REMOTE = "remote"
 
     REPO_SCHEMA = {
         PARAM_REPO: {
@@ -24,6 +25,7 @@ class RepoDependency(Dependency):
             PARAM_REV: str,
             PARAM_REV_LOCK: str,
             PARAM_CONFIG: str,
+            PARAM_REMOTE: Any(str, dict),
         }
     }
 
@@ -76,6 +78,10 @@ class RepoDependency(Dependency):
         if config:
             repo[self.PARAM_CONFIG] = config
 
+        remote = self.def_repo.get(self.PARAM_REMOTE)
+        if remote:
+            repo[self.PARAM_REMOTE] = remote
+
         return {
             self.PARAM_PATH: self.def_path,
             self.PARAM_REPO: repo,
@@ -99,6 +105,14 @@ class RepoDependency(Dependency):
         from dvc.config import Config
         from dvc.fs import DVCFileSystem
 
+        rem = self.def_repo.get("remote")
+        if isinstance(rem, dict):
+            remote = None  # type: ignore[unreachable]
+            remote_config = rem
+        else:
+            remote = rem
+            remote_config = None
+
         conf = self.def_repo.get("config")
         if conf:
             config = Config.load_file(conf)
@@ -113,6 +127,8 @@ class RepoDependency(Dependency):
             rev=rev or self._get_rev(locked=locked),
             subrepos=True,
             config=config,
+            remote=remote,
+            remote_config=remote_config,
         )
 
     def _get_rev(self, locked: bool = True):

dvc/exceptions.py

@@ -71,6 +71,10 @@ class OutputNotFoundError(DvcException):
         )
 
 
+class StageNotFoundError(DvcException):
+    pass
+
+
 class StagePathAsOutputError(DvcException):
     """Thrown if directory that stage is going to be saved in is specified as
     an output of another stage.
@@ -293,7 +297,7 @@ class URLMissingError(DvcException):
         super().__init__(f"The path '{url}' does not exist")
 
 
-class IsADirectoryError(DvcException):  # noqa,pylint:disable=redefined-builtin
+class IsADirectoryError(DvcException):  # noqa: A001,pylint:disable=redefined-builtin
     """Raised when a file operation is requested on a directory."""
 
 

dvc/fs/dvc.py

@@ -352,7 +352,7 @@ class _DVCFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
                 dvc_info = dvc_fs.fs.index.info(subkey)
                 dvc_path = _get_dvc_path(dvc_fs, subkey)
                 dvc_info["name"] = dvc_path
-            except FileNotFoundError:
+            except KeyError:
                 pass
 
         fs_info = None

dvc/logger.py

@@ -160,7 +160,7 @@ class LoggerHandler(logging.StreamHandler):
             self.flush()
         except (BrokenPipeError, RecursionError):
             raise
-        except Exception:  # noqa, pylint: disable=broad-except
+        except Exception:  # noqa: BLE001, pylint: disable=broad-except
             self.handleError(record)
 
 

dvc/output.py

@@ -488,6 +488,9 @@ class Output:
         ):
             return str(self.def_path)
 
+        if not self.fs.path.isin(self.fs_path, self.repo.root_dir):
+            return self.fs_path
+
         cur_dir = self.fs.path.getcwd()
         if self.fs.path.isin(cur_dir, self.repo.root_dir):
             return self.fs.path.relpath(self.fs_path, cur_dir)
@@ -784,13 +787,18 @@ class Output:
                     self.hash_name,
                     ignore=self.dvcignore,
                 )
-                otransfer(
-                    staging,
-                    self.cache,
-                    {obj.hash_info},
-                    shallow=False,
-                    hardlink=hardlink,
-                )
+                with Callback.as_tqdm_callback(
+                    desc=f"Committing {self} to cache",
+                    unit="file",
+                ) as cb:
+                    otransfer(
+                        staging,
+                        self.cache,
+                        {obj.hash_info},
+                        shallow=False,
+                        hardlink=hardlink,
+                        callback=cb,
+                    )
             if relink:
                 rel = self.fs.path.relpath(filter_info or self.fs_path)
                 with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as cb:
@@ -819,13 +827,18 @@ class Output:
         save_obj = obj.filter(prefix)
         assert isinstance(save_obj, Tree)
         checkout_obj = save_obj.get_obj(self.cache, prefix)
-        otransfer(
-            staging,
-            self.cache,
-            {save_obj.hash_info} | {oid for _, _, oid in save_obj},
-            shallow=True,
-            hardlink=hardlink,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Committing {self} to cache",
+            unit="file",
+        ) as cb:
+            otransfer(
+                staging,
+                self.cache,
+                {save_obj.hash_info} | {oid for _, _, oid in save_obj},
+                shallow=True,
+                hardlink=hardlink,
+                callback=cb,
+            )
         return checkout_obj
 
     def dumpd(self, **kwargs):  # noqa: C901, PLR0912
@@ -1045,14 +1058,19 @@ class Output:
             upload=upload,
             no_progress_bar=no_progress_bar,
         )
-        otransfer(
-            staging,
-            odb,
-            {obj.hash_info},
-            jobs=jobs,
-            hardlink=False,
-            shallow=False,
-        )
+        with Callback.as_tqdm_callback(
+            desc=f"Transferring to {odb.fs.unstrip_protocol(odb.path)}",
+            unit="file",
+        ) as cb:
+            otransfer(
+                staging,
+                odb,
+                {obj.hash_info},
+                jobs=jobs,
+                hardlink=False,
+                shallow=False,
+                callback=cb,
+            )
 
         self.hash_info = obj.hash_info
         self.files = None
@@ -1415,7 +1433,18 @@ class Output:
 
         assert staging
         assert obj.hash_info
-        otransfer(staging, self.cache, {obj.hash_info}, hardlink=relink, shallow=False)
+        with Callback.as_tqdm_callback(
+            desc=f"Adding {self} to cache",
+            unit="file",
+        ) as cb:
+            otransfer(
+                staging,
+                self.cache,
+                {obj.hash_info},
+                hardlink=relink,
+                shallow=False,
+                callback=cb,
+            )
 
         if relink:
             with CheckoutCallback(

dvc/parsing/__init__.py

@@ -2,6 +2,7 @@ import logging
 import os
 from collections.abc import Mapping, Sequence
 from copy import deepcopy
+from itertools import product
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -51,6 +52,7 @@ VARS_KWD = "vars"
 WDIR_KWD = "wdir"
 PARAMS_KWD = "params"
 FOREACH_KWD = "foreach"
+MATRIX_KWD = "matrix"
 DO_KWD = "do"
 
 DEFAULT_PARAMS_FILE = "params.yaml"
@@ -105,7 +107,7 @@ def is_map_or_seq(data: Any) -> bool:
     return not isinstance(data, str) and _is_map_or_seq(data)
 
 
-def split_foreach_name(name: str) -> Tuple[str, Optional[str]]:
+def split_group_name(name: str) -> Tuple[str, Optional[str]]:
     group, *keys = name.rsplit(JOIN, maxsplit=1)
     return group, first(keys)
 
@@ -120,13 +122,15 @@ def check_interpolations(data: "DictStrAny", where: str, path: str):
     return recurse(func)(data)
 
 
-Definition = Union["ForeachDefinition", "EntryDefinition"]
+Definition = Union["ForeachDefinition", "EntryDefinition", "MatrixDefinition"]
 
 
 def make_definition(
     resolver: "DataResolver", name: str, definition: "DictStrAny", **kwargs
 ) -> Definition:
     args = resolver, resolver.context, name, definition
+    if MATRIX_KWD in definition:
+        return MatrixDefinition(*args, **kwargs)
     if FOREACH_KWD in definition:
         return ForeachDefinition(*args, **kwargs)
     return EntryDefinition(*args, **kwargs)
@@ -159,7 +163,8 @@ class DataResolver:
         self.tracked_vars: Dict[str, Mapping] = {}
 
         stages_data = d.get(STAGES_KWD, {})
-        # we wrap the definitions into ForeachDefinition and EntryDefinition,
+        # we wrap the definitions into:
+        # ForeachDefinition, MatrixDefinition, and EntryDefinition
         # that helps us to optimize, cache and selectively load each one of
         # them as we need, and simplify all of this DSL/parsing logic.
         self.definitions: Dict[str, Definition] = {
@@ -168,12 +173,13 @@ class DataResolver:
         }
 
     def resolve_one(self, name: str):
-        group, key = split_foreach_name(name)
+        group, key = split_group_name(name)
 
         if not self._has_group_and_key(group, key):
             raise EntryNotFound(f"Could not find '{name}'")
 
-        # all of the checks for `key` not being None for `ForeachDefinition`
+        # all of the checks for `key` not being None for
+        # `ForeachDefinition`/`MatrixDefinition`
         # and/or `group` not existing in the `interim`, etc. should be
         # handled by the `self.has_key()` above.
         definition = self.definitions[group]
@@ -190,7 +196,7 @@ class DataResolver:
         return {STAGES_KWD: data}
 
     def has_key(self, key: str):
-        return self._has_group_and_key(*split_foreach_name(key))
+        return self._has_group_and_key(*split_group_name(key))
 
     def _has_group_and_key(self, group: str, key: Optional[str] = None):
         try:
@@ -198,16 +204,14 @@ class DataResolver:
         except KeyError:
             return False
 
-        if key:
-            return isinstance(definition, ForeachDefinition) and definition.has_member(
-                key
-            )
-        return not isinstance(definition, ForeachDefinition)
+        if not isinstance(definition, (ForeachDefinition, MatrixDefinition)):
+            return key is None
+        return key is not None and definition.has_member(key)
 
     @collecting
     def get_keys(self):
         for name, definition in self.definitions.items():
-            if isinstance(definition, ForeachDefinition):
+            if isinstance(definition, (ForeachDefinition, MatrixDefinition)):
                 yield from definition.get_generated_names()
                 continue
             yield name
@@ -256,8 +260,9 @@ class EntryDefinition:
         name = self.name
         if not skip_checks:
             # we can check for syntax errors as we go for interpolated entries,
-            # but for foreach-generated ones, once is enough, which it does
-            # that itself. See `ForeachDefinition.do_definition`.
+            # but for foreach and matrix generated ones, once is enough, which it does
+            # that itself. See `ForeachDefinition.template`
+            # and `MatrixDefinition.template`.
             check_syntax_errors(self.definition, name, self.relpath)
 
         # we need to pop vars from generated/evaluated data
@@ -331,17 +336,18 @@ class ForeachDefinition:
         self.name = name
 
         assert DO_KWD in definition
+        assert MATRIX_KWD not in definition
         self.foreach_data = definition[FOREACH_KWD]
-        self._do_definition = definition[DO_KWD]
+        self._template = definition[DO_KWD]
 
         self.pair = IterationPair()
         self.where = where
 
     @cached_property
-    def do_definition(self):
+    def template(self):
         # optimization: check for syntax errors only once for `foreach` stages
-        check_syntax_errors(self._do_definition, self.name, self.relpath)
-        return self._do_definition
+        check_syntax_errors(self._template, self.name, self.relpath)
+        return self._template
 
     @cached_property
     def resolved_iterable(self):
@@ -444,16 +450,111 @@ class ForeachDefinition:
             # i.e. quadratic complexity).
             generated = self._generate_name(key)
             entry = EntryDefinition(
-                self.resolver, self.context, generated, self.do_definition
+                self.resolver, self.context, generated, self.template
             )
             try:
                 # optimization: skip checking for syntax errors on each foreach
-                # generated stages. We do it once when accessing do_definition.
+                # generated stages. We do it once when accessing template.
                 return entry.resolve_stage(skip_checks=True)
             except ContextError as exc:
                 format_and_raise(exc, f"stage '{generated}'", self.relpath)
 
-            # let mypy know that this state is unreachable as format_and_raise
-            # does not return at all (it's not able to understand it for some
-            # reason)
-            raise AssertionError("unreachable")
+
+class MatrixDefinition:
+    def __init__(
+        self,
+        resolver: DataResolver,
+        context: Context,
+        name: str,
+        definition: "DictStrAny",
+        where: str = STAGES_KWD,
+    ):
+        self.resolver = resolver
+        self.relpath = self.resolver.relpath
+        self.context = context
+        self.name = name
+
+        assert MATRIX_KWD in definition
+        assert DO_KWD not in definition
+        assert FOREACH_KWD not in definition
+
+        self._template = definition.copy()
+        self.matrix_data = self._template.pop(MATRIX_KWD)
+
+        self.pair = IterationPair()
+        self.where = where
+
+    @cached_property
+    def template(self) -> "DictStrAny":
+        # optimization: check for syntax errors only once for `matrix` stages
+        check_syntax_errors(self._template, self.name, self.relpath)
+        return self._template
+
+    @cached_property
+    def resolved_iterable(self) -> Dict[str, List]:
+        return self._resolve_matrix_data()
+
+    def _resolve_matrix_data(self) -> Dict[str, List]:
+        try:
+            iterable = self.context.resolve(self.matrix_data, unwrap=False)
+        except (ContextError, ParseError) as exc:
+            format_and_raise(exc, f"'{self.where}.{self.name}.matrix'", self.relpath)
+        return iterable
+
+    @cached_property
+    def normalized_iterable(self) -> Dict[str, "DictStrAny"]:
+        """Convert sequence to Mapping with keys normalized."""
+        iterable = self.resolved_iterable
+        assert isinstance(iterable, Mapping)
+
+        ret: Dict[str, "DictStrAny"] = {}
+        matrix = {key: enumerate(v) for key, v in iterable.items()}
+        for combination in product(*matrix.values()):
+            d: "DictStrAny" = {}
+            fragments: List[str] = []
+            for k, (i, v) in zip(matrix.keys(), combination):
+                d[k] = v
+                fragments.append(f"{k}{i}" if is_map_or_seq(v) else to_str(v))
+
+            key = "-".join(fragments)
+            ret[key] = d
+        return ret
+
+    def has_member(self, key: str) -> bool:
+        return key in self.normalized_iterable
+
+    def get_generated_names(self) -> List[str]:
+        return list(map(self._generate_name, self.normalized_iterable))
+
+    def _generate_name(self, key: str) -> str:
+        return f"{self.name}{JOIN}{key}"
+
+    def resolve_all(self) -> "DictStrAny":
+        return join(map(self.resolve_one, self.normalized_iterable))
+
+    def resolve_one(self, key: str) -> "DictStrAny":
+        return self._each_iter(key)
+
+    def _each_iter(self, key: str) -> "DictStrAny":
+        err_message = f"Could not find '{key}' in matrix group '{self.name}'"
+        with reraise(KeyError, EntryNotFound(err_message)):
+            value = self.normalized_iterable[key]
+
+        temp_dict = {self.pair.value: value}
+        with self.context.set_temporarily(temp_dict, reserve=True):
+            # optimization: item and key can be removed on __exit__() as they
+            # are top-level values, and are not merged recursively.
+            # This helps us avoid cloning context, which is slower
+            # (increasing the size of the context might increase
+            # the no. of items to be generated which means more cloning,
+            # i.e. quadratic complexity).
+            generated = self._generate_name(key)
+            entry = EntryDefinition(
+                self.resolver, self.context, generated, self.template
+            )
+            try:
+                # optimization: skip checking for syntax errors on each matrix
+                # generated stages. We do it once when accessing template.
+                return entry.resolve_stage(skip_checks=True)
+            except ContextError as exc:
+                format_and_raise(exc, f"stage '{generated}'", self.relpath)

dvc/repo/__init__.py

@@ -93,6 +93,7 @@ class Repo:
     from dvc.repo.status import status  # type: ignore[misc]
     from dvc.repo.update import update  # type: ignore[misc]
 
+    from .cache import check_missing as cache_check_missing  # type: ignore[misc]
     from .data import status as data_status  # type: ignore[misc]
 
     ls = staticmethod(_ls)
@@ -133,7 +134,7 @@ class Repo:
         assert root_dir
         return root_dir, dvc_dir
 
-    def __init__(  # noqa: PLR0915
+    def __init__(  # noqa: PLR0915, PLR0913
         self,
         root_dir: Optional[str] = None,
         fs: Optional["FileSystem"] = None,
@@ -144,6 +145,8 @@ class Repo:
         url: Optional[str] = None,
         repo_factory: Optional[Callable] = None,
         scm: Optional[Union["Git", "NoSCM"]] = None,
+        remote: Optional[str] = None,
+        remote_config: Optional["DictStrAny"] = None,
     ):
         from dvc.cachemgr import CacheManager
         from dvc.data_cloud import DataCloud
@@ -163,6 +166,8 @@ class Repo:
         self._fs = fs or localfs
         self._scm = scm
         self._config = config
+        self._remote = remote
+        self._remote_config = remote_config
         self._data_index = None
 
         if rev and not fs:
@@ -237,7 +242,14 @@ class Repo:
     def config(self):
         from dvc.config import Config
 
-        return Config(self.dvc_dir, fs=self.fs, config=self._config)
+        return Config(
+            self.dvc_dir,
+            local_dvc_dir=self.local_dvc_dir,
+            fs=self.fs,
+            config=self._config,
+            remote=self._remote,
+            remote_config=self._remote_config,
+        )
 
     @cached_property
     def local_dvc_dir(self):
@@ -411,7 +423,7 @@ class Repo:
 
         fs = fs or localfs
         root = root or os.curdir
-        root_dir = fs.path.realpath(root)
+        root_dir = fs.path.abspath(root)
 
         if not fs.isdir(root_dir):
             raise NotDvcRepoError(f"directory '{root}' does not exist")
@@ -487,7 +499,7 @@ class Repo:
         used_run_cache=None,
         revs=None,
         num=1,
-        push: bool = False,  # noqa: F811
+        push: bool = False,
     ):
         """Get the stages related to the given target and collect
         the `info` of its outputs.

dvc/repo/cache.py

@@ -0,0 +1,31 @@
+import os
+
+
+def check_missing(repo, rev=None, max_size=None, types=None):
+    from dvc_data.index import StorageKeyError
+
+    with repo.switch(rev or "workspace"):
+        idx = repo.index.targets_view(
+            None,
+            max_size=max_size,
+            types=types,
+        )
+
+        index = idx.data["repo"]
+
+    def onerror(_entry, _exc):
+        pass
+
+    index.onerror = onerror
+
+    ret = []
+    for _, entry in index.iteritems():
+        try:
+            fs, path = index.storage_map.get_cache(entry)
+        except (StorageKeyError, ValueError):
+            continue
+
+        if not fs.exists(path):
+            ret.append(os.path.join(*entry.key))
+
+    return ret

dvc/repo/experiments/executor/base.py

@@ -607,7 +607,7 @@ class BaseExecutor(ABC):
             old_cwd = os.getcwd()
 
             for path in copy_paths or []:
-                cls._copy_path(os.path.realpath(path), os.path.join(dvc.root_dir, path))
+                cls._copy_path(os.path.abspath(path), os.path.join(dvc.root_dir, path))
 
             if info.wdir:
                 os.chdir(os.path.join(dvc.scm.root_dir, info.wdir))

dvc/repo/experiments/queue/base.py

@@ -476,13 +476,17 @@ class BaseStashQueue(ABC):
         hydra_output_file = ParamsDependency.DEFAULT_PARAMS_FILE
         for path, overrides in params.items():
             if hydra_enabled and path == hydra_output_file:
-                config_dir = os.path.join(
-                    self.repo.root_dir, hydra_config.get("config_dir", "conf")
-                )
+                if (config_module := hydra_config.get("config_module")) is None:
+                    config_dir = os.path.join(
+                        self.repo.root_dir, hydra_config.get("config_dir", "conf")
+                    )
+                else:
+                    config_dir = None
                 config_name = hydra_config.get("config_name", "config")
                 compose_and_dump(
                     path,
                     config_dir,
+                    config_module,
                     config_name,
                     overrides,
                 )

dvc/repo/fetch.py

@@ -9,6 +9,14 @@ from . import locked
 logger = logging.getLogger(__name__)
 
 
+def _make_index_onerror(onerror, rev):
+    def _onerror(entry, exc):
+        if onerror:
+            return onerror(rev, entry, exc)
+
+    return _onerror
+
+
 def _collect_indexes(  # noqa: PLR0913
     repo,
     targets=None,
@@ -22,6 +30,7 @@ def _collect_indexes(  # noqa: PLR0913
     max_size=None,
     types=None,
     config=None,
+    onerror=None,
 ):
     indexes = {}
     collection_exc = None
@@ -39,7 +48,7 @@ def _collect_indexes(  # noqa: PLR0913
         all_commits=all_commits,
     ):
         try:
-            repo.config.update(config)
+            repo.config.merge(config)
 
             idx = repo.index.targets_view(
                 targets,
@@ -48,8 +57,14 @@ def _collect_indexes(  # noqa: PLR0913
                 max_size=max_size,
                 types=types,
             )
-            indexes[idx.data_tree.hash_info.value] = idx.data["repo"]
+
+            data = idx.data["repo"]
+            data.onerror = _make_index_onerror(onerror, rev)
+
+            indexes[idx.data_tree.hash_info.value] = data
         except Exception as exc:  # pylint: disable=broad-except
+            if onerror:
+                onerror(rev, None, exc)
             collection_exc = exc
             logger.exception("failed to collect '%s'", rev or "workspace")
 
@@ -75,6 +90,7 @@ def fetch(  # noqa: C901, PLR0913
     max_size=None,
     types=None,
     config=None,
+    onerror=None,
 ) -> int:
     """Download data items from a cloud and imported repositories
 
@@ -119,6 +135,7 @@ def fetch(  # noqa: C901, PLR0913
         max_size=max_size,
         types=types,
         config=config,
+        onerror=onerror,
     )
 
     cache_key = ("fetch", tokenize(sorted(indexes.keys())))

dvc/repo/get.py

@@ -19,7 +19,17 @@ class GetDVCFileError(DvcException):
         )
 
 
-def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
+def get(
+    url,
+    path,
+    out=None,
+    rev=None,
+    jobs=None,
+    force=False,
+    config=None,
+    remote=None,
+    remote_config=None,
+):
     from dvc.config import Config
     from dvc.dvcfile import is_valid_filename
     from dvc.repo import Repo
@@ -38,6 +48,8 @@ def get(url, path, out=None, rev=None, jobs=None, force=False, config=None):
         subrepos=True,
         uninitialized=True,
         config=config,
+        remote=remote,
+        remote_config=remote_config,
     ) as repo:
         from dvc.fs import download
         from dvc.fs.data import DataFileSystem

dvc/repo/graph.py

@@ -31,6 +31,9 @@ def check_acyclic(graph: "DiGraph") -> None:
 
 def get_pipeline(pipelines, node):
     found = [i for i in pipelines if i.has_node(node)]
+    if not found:
+        return None
+
     assert len(found) == 1
     return found[0]
 
@@ -60,6 +63,9 @@ def collect_pipeline(stage: "Stage", graph: "DiGraph") -> Iterator["Stage"]:
     import networkx as nx
 
     pipeline = get_pipeline(get_pipelines(graph), stage)
+    if not pipeline:
+        return iter([])
+
     return nx.dfs_postorder_nodes(pipeline, stage)
 
 

dvc/repo/imp.py

@@ -1,4 +1,14 @@
-def imp(self, url, path, out=None, rev=None, config=None, **kwargs):
+def imp(
+    self,
+    url,
+    path,
+    out=None,
+    rev=None,
+    config=None,
+    remote=None,
+    remote_config=None,
+    **kwargs,
+):
     erepo = {"url": url}
     if rev is not None:
         erepo["rev"] = rev
@@ -6,4 +16,17 @@ def imp(self, url, path, out=None, rev=None, config=None, **kwargs):
     if config is not None:
         erepo["config"] = config
 
+    if remote is not None and remote_config is not None:
+        conf = erepo.get("config") or {}
+        remotes = conf.get("remote") or {}
+        remote_conf = remotes.get(remote) or {}
+        remote_conf.update(remote_config)
+        remotes[remote] = remote_conf
+        conf["remote"] = remotes
+        erepo["config"] = conf
+    elif remote is not None:
+        erepo["remote"] = remote
+    elif remote_config is not None:
+        erepo["remote"] = remote_config
+
     return self.imp_url(path, out=out, erepo=erepo, frozen=True, **kwargs)

dvc/repo/index.py

@@ -413,7 +413,7 @@ class Index:
             by_workspace[workspace].add(key)
 
         for path in self._plot_sources:
-            key = self.repo.fs.path.relparts(path, self.repo.root_dir)
+            key = self.repo.fs.path.parts(path)
             by_workspace["repo"].add(key)
 
         return dict(by_workspace)

dvc/repo/init.py

@@ -37,7 +37,7 @@ def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False):  # noqa:
             "Cannot initialize repo with `--no-scm` and `--subdir`"
         )
 
-    root_dir = os.path.realpath(root_dir)
+    root_dir = os.path.abspath(root_dir)
     dvc_dir = os.path.join(root_dir, Repo.DVC_DIR)
 
     try:

dvc/repo/ls.py

@@ -14,6 +14,8 @@ def ls(
     recursive: Optional[bool] = None,
     dvc_only: bool = False,
     config: Optional[str] = None,
+    remote: Optional[str] = None,
+    remote_config: Optional[dict] = None,
 ):
     """Methods for getting files and outputs for the repo.
 
@@ -23,7 +25,9 @@ def ls(
         rev (str, optional): SHA commit, branch or tag name
         recursive (bool, optional): recursively walk the repo
         dvc_only (bool, optional): show only DVC-artifacts
-        config (bool, optional): path to config file
+        config (str, optional): path to config file
+        remote (str, optional): remote name to set as a default remote in the repo
+        remote_config (str, dict): remote config to merge with a remote in the repo
 
     Returns:
         list of `entry`
@@ -47,7 +51,13 @@ def ls(
         config_dict = None
 
     with Repo.open(
-        url, rev=rev, subrepos=True, uninitialized=True, config=config_dict
+        url,
+        rev=rev,
+        subrepos=True,
+        uninitialized=True,
+        config=config_dict,
+        remote=remote,
+        remote_config=remote_config,
     ) as repo:
         path = path or ""
 

dvc/repo/plots/__init__.py

@@ -483,7 +483,7 @@ def _collect_pipeline_files(repo, targets: List[str], props, onerror=None):
             if isinstance(elem, str):
                 dvcfile_defs_dict[elem] = None
             else:
-                k, v = list(elem.items())[0]
+                k, v = next(iter(elem.items()))
                 dvcfile_defs_dict[k] = v
 
         resolved = _resolve_definitions(

dvc/repo/stage.py

@@ -37,11 +37,19 @@ StageSet = Set["Stage"]
 
 
 def _collect_with_deps(stages: StageList, graph: "DiGraph") -> StageSet:
+    from dvc.exceptions import StageNotFoundError
     from dvc.repo.graph import collect_pipeline
 
     res: StageSet = set()
     for stage in stages:
-        res.update(collect_pipeline(stage, graph=graph))
+        pl = list(collect_pipeline(stage, graph=graph))
+        if not pl:
+            raise StageNotFoundError(
+                f"Stage {stage} is not found in the project. "
+                "Check that there are no symlinks in the parents "
+                "leading up to it within the project."
+            )
+        res.update(pl)
     return res
 
 
@@ -211,7 +219,7 @@ class StageLoad:
         self, path: Optional[str] = None, name: Optional[str] = None
     ) -> str:
         if path:
-            return self.repo.fs.path.realpath(path)
+            return self.repo.fs.path.abspath(path)
 
         path = PROJECT_FILE
         logger.debug("Assuming '%s' to be a stage inside '%s'", name, path)
@@ -235,7 +243,7 @@ class StageLoad:
     ) -> Iterable[str]:
         if not name:
             return stages.keys()
-        if accept_group and stages.is_foreach_generated(name):
+        if accept_group and stages.is_foreach_or_matrix_generated(name):
             return self._get_group_keys(stages, name)
         if glob:
             return fnmatch.filter(stages.keys(), name)

dvc/repo/worktree.py

@@ -427,7 +427,7 @@ def _get_update_diff_index(
         meta_cmp_key=partial(_meta_checksum, remote.fs),
         with_unchanged=True,
     ):
-        if change.typ == ADD or change.typ == MODIFY:
+        if change.typ in (ADD, MODIFY):
             entry = change.new
             # preserve md5's which were calculated in out.save() after
             # downloading

dvc/schema.py

@@ -11,7 +11,7 @@ from dvc.output import (
     META_SCHEMA,
     Output,
 )
-from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
+from dvc.parsing import DO_KWD, FOREACH_KWD, MATRIX_KWD, VARS_KWD
 from dvc.stage.params import StageParams
 
 STAGES = "stages"
@@ -77,6 +77,7 @@ PARAM_PSTAGE_NON_DEFAULT_SCHEMA = {str: [str]}
 VARS_SCHEMA = [str, dict]
 
 STAGE_DEFINITION = {
+    MATRIX_KWD: {str: Any(str, list)},
     Required(StageParams.PARAM_CMD): Any(str, list),
     Optional(StageParams.PARAM_WDIR): str,
     Optional(StageParams.PARAM_DEPS): [str],

dvc/scm.py

@@ -67,7 +67,7 @@ def map_scm_exception(with_cause: bool = False) -> Iterator[None]:
         into = SCMError(str(exc))
         if with_cause:
             raise into from exc
-        raise into
+        raise into  # noqa: B904
 
 
 @overload

dvc/stage/loader.py

@@ -7,7 +7,7 @@ from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple
 from funcy import get_in, lcat, once, project
 
 from dvc import dependency, output
-from dvc.parsing import FOREACH_KWD, JOIN, EntryNotFound
+from dvc.parsing import FOREACH_KWD, JOIN, MATRIX_KWD, EntryNotFound
 from dvc.utils.objects import cached_property
 from dvc_data.hashfile.meta import Meta
 
@@ -167,8 +167,11 @@ class StageLoader(Mapping):
     def __contains__(self, name):
         return self.resolver.has_key(name)  # noqa: W601
 
-    def is_foreach_generated(self, name: str):
-        return name in self.stages_data and FOREACH_KWD in self.stages_data[name]
+    def is_foreach_or_matrix_generated(self, name: str) -> bool:
+        return (
+            name in self.stages_data
+            and {FOREACH_KWD, MATRIX_KWD} & self.stages_data[name].keys()
+        )
 
 
 class SingleStageLoader(Mapping):

dvc/stage/utils.py

@@ -32,14 +32,14 @@ def check_stage_path(repo, path, is_wdir=False):
         wdir_or_path="stage working dir" if is_wdir else "file path", path=path
     )
 
-    real_path = os.path.realpath(path)
+    real_path = os.path.abspath(path)
     if not os.path.exists(real_path):
         raise StagePathNotFoundError(error_msg.format("does not exist"))
 
     if not os.path.isdir(real_path):
         raise StagePathNotDirectoryError(error_msg.format("is not directory"))
 
-    proj_dir = os.path.realpath(repo.root_dir)
+    proj_dir = os.path.abspath(repo.root_dir)
     if real_path != proj_dir and not path_isin(real_path, proj_dir):
         raise StagePathOutsideError(error_msg.format("is outside of DVC repo"))
 

dvc/testing/benchmarks/conftest.py

@@ -1 +1 @@
-from .fixtures import *  # noqa, pylint: disable=wildcard-import
+from .fixtures import *  # noqa: F403 # pylint: disable=wildcard-import

dvc/testing/conftest.py

@@ -1 +1 @@
-from .fixtures import *  # noqa, pylint: disable=wildcard-import
+from .fixtures import *  # noqa: F403, pylint: disable=wildcard-import

dvc/testing/plugin.py

@@ -1,5 +1,5 @@
-from .benchmarks.fixtures import *  # noqa, pylint: disable=wildcard-import
-from .fixtures import *  # noqa, pylint: disable=wildcard-import
+from .benchmarks.fixtures import *  # noqa: F403, pylint: disable=wildcard-import
+from .fixtures import *  # noqa: F403, pylint: disable=wildcard-import
 
 
 def pytest_generate_tests(metafunc):

dvc/utils/__init__.py

@@ -218,23 +218,13 @@ def _visual_center(line, width):
 
 
 def relpath(path, start=os.curdir):
-    path = os.fspath(path)
+    path = os.path.abspath(os.fspath(path))
     start = os.path.abspath(os.fspath(start))
 
     # Windows path on different drive than curdir doesn't have relpath
-    if os.name == "nt":
-        # Since python 3.8 os.realpath resolves network shares to their UNC
-        # path. So, to be certain that relative paths correctly captured,
-        # we need to resolve to UNC path first. We resolve only the drive
-        # name so that we don't follow any 'real' symlinks on the path
-        def resolve_network_drive_windows(path_to_resolve):
-            drive, tail = os.path.splitdrive(path_to_resolve)
-            return os.path.join(os.path.realpath(drive), tail)
+    if os.name == "nt" and not os.path.commonprefix([start, path]):
+        return path
 
-        path = resolve_network_drive_windows(os.path.abspath(path))
-        start = resolve_network_drive_windows(start)
-        if not os.path.commonprefix([start, path]):
-            return path
     return os.path.relpath(path, start)
 
 

dvc/utils/hydra.py

@@ -1,6 +1,6 @@
 import logging
 from pathlib import Path
-from typing import TYPE_CHECKING, List
+from typing import TYPE_CHECKING, List, Optional
 
 from dvc.exceptions import InvalidArgumentError
 
@@ -15,7 +15,8 @@ logger = logging.getLogger(__name__)
 
 def compose_and_dump(
     output_file: "StrPath",
-    config_dir: str,
+    config_dir: Optional[str],
+    config_module: Optional[str],
     config_name: str,
     overrides: List[str],
 ) -> None:
@@ -25,6 +26,8 @@ def compose_and_dump(
         output_file: File where the composed config will be dumped.
         config_dir: Folder containing the Hydra config files.
             Must be absolute file system path.
+        config_module: Module containing the Hydra config files.
+            Ignored if `config_dir` is not `None`.
         config_name: Name of the config file containing defaults,
             without the .yaml extension.
         overrides: List of `Hydra Override`_ patterns.
@@ -32,12 +35,21 @@ def compose_and_dump(
     .. _Hydra Override:
         https://hydra.cc/docs/advanced/override_grammar/basic/
     """
-    from hydra import compose, initialize_config_dir
+    from hydra import compose, initialize_config_dir, initialize_config_module
     from omegaconf import OmegaConf
 
     from .serialize import DUMPERS
 
-    with initialize_config_dir(config_dir, version_base=None):
+    config_source = config_dir or config_module
+    if not config_source:
+        raise ValueError("Either `config_dir` or `config_module` should be provided.")
+    initialize_config = (
+        initialize_config_dir if config_dir else initialize_config_module
+    )
+
+    with initialize_config(  # type: ignore[attr-defined]
+        config_source, version_base=None
+    ):
         cfg = compose(config_name=config_name, overrides=overrides)
 
     OmegaConf.resolve(cfg)

dvc/utils/serialize/__init__.py

@@ -1,11 +1,11 @@
 from collections import defaultdict
 from typing import DefaultDict
 
-from ._common import *  # noqa, pylint: disable=wildcard-import
-from ._json import *  # noqa, pylint: disable=wildcard-import
-from ._py import *  # noqa, pylint: disable=wildcard-import
-from ._toml import *  # noqa, pylint: disable=wildcard-import
-from ._yaml import *  # noqa, pylint: disable=wildcard-import
+from ._common import *  # noqa: F403, pylint: disable=wildcard-import
+from ._json import *  # noqa: F403, pylint: disable=wildcard-import
+from ._py import *  # noqa: F403, pylint: disable=wildcard-import
+from ._toml import *  # noqa: F403, pylint: disable=wildcard-import
+from ._yaml import *  # noqa: F403, pylint: disable=wildcard-import
 
 LOADERS: DefaultDict[str, LoaderFn] = defaultdict(  # noqa: F405
     lambda: load_yaml  # noqa: F405

pyproject.toml

@@ -34,7 +34,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.6.0,<2.7.0",
+    "dvc-data>=2.12.0,<2.13.0",
     "dvc-http>=2.29.0",
     "dvc-render>=0.3.1,<1",
     "dvc-studio-client>=0.9.2,<1",
@@ -76,7 +76,7 @@ gs = ["dvc-gs==2.22.1"]
 hdfs = ["dvc-hdfs==2.19"]
 lint = [
     "mypy==1.4.1",
-    "pylint==2.17.4",
+    "pylint==2.17.5",
     "types-colorama",
     "types-psutil",
     "types-pyinstaller",
@@ -300,7 +300,7 @@ ignore-variadic-names = true
 known-first-party = ["dvc", "dvc_data", "dvc_objects", "dvc_render", "dvc_task", "tests"]
 
 [tool.ruff.pep8-naming]
-ignore-names = ["M", "SCM"]
+extend-ignore-names = ["M", "SCM"]
 
 [tool.ruff.per-file-ignores]
 "dvc/commands/**" = ["N806"]

tests/conftest.py

@@ -7,11 +7,11 @@ import pytest
 
 from dvc import env
 from dvc.stage import PipelineStage
-from dvc.testing.fixtures import *  # noqa, pylint: disable=wildcard-import
+from dvc.testing.fixtures import *  # noqa: F403, pylint: disable=wildcard-import
 
-from .dir_helpers import *  # noqa, pylint: disable=wildcard-import
-from .remotes import *  # noqa, pylint: disable=wildcard-import
-from .scripts import *  # noqa, pylint: disable=wildcard-import
+from .dir_helpers import *  # noqa: F403 # pylint: disable=wildcard-import
+from .remotes import *  # noqa: F403 # pylint: disable=wildcard-import
+from .scripts import *  # noqa: F403, pylint: disable=wildcard-import
 
 # Prevent updater and analytics from running their processes
 os.environ["DVC_TEST"] = "true"

tests/func/api/test_data.py

@@ -5,7 +5,7 @@ from funcy import first, get_in
 
 from dvc import api
 from dvc.exceptions import OutputNotFoundError, PathMissingError
-from dvc.testing.api_tests import TestAPI  # noqa, pylint: disable=unused-import
+from dvc.testing.api_tests import TestAPI  # noqa: F401, pylint: disable=unused-import
 from dvc.testing.tmp_dir import make_subrepo
 from dvc.utils.fs import remove
 

tests/func/experiments/conftest.py

@@ -1,6 +1,7 @@
 import pytest
 
-from tests.unit.repo.experiments.conftest import (  # noqa, pylint disable=unused-argument
+# pylint: disable-next=unused-argument
+from tests.unit.repo.experiments.conftest import (  # noqa: F401
     exp_stage,
     failed_exp_stage,
     session_app,

tests/func/experiments/executor/test_ssh.py

@@ -10,7 +10,9 @@ from dvc_ssh.tests.cloud import TEST_SSH_KEY_PATH, TEST_SSH_USER
 from dvc.repo.experiments.executor.base import ExecutorInfo, ExecutorResult
 from dvc.repo.experiments.executor.ssh import SSHExecutor
 from dvc.repo.experiments.refs import EXEC_HEAD, EXEC_MERGE
-from tests.func.machine.conftest import *  # noqa, pylint: disable=wildcard-import
+
+# pylint: disable-next=wildcard-import
+from tests.func.machine.conftest import *  # noqa: F403
 
 
 @contextmanager

tests/func/parsing/test_matrix.py

@@ -0,0 +1,52 @@
+import pytest
+
+from dvc.parsing import DataResolver, MatrixDefinition
+
+MATRIX_DATA = {
+    "os": ["win", "linux"],
+    "pyv": [3.7, 3.8],
+    "dict": [{"arg1": 1}, {"arg2": 2}],
+    "list": [["out1", "out11"], ["out2", "out22"]],
+}
+
+
+@pytest.mark.parametrize(
+    "matrix",
+    [
+        MATRIX_DATA,
+        {
+            "os": "${os}",
+            "pyv": "${pyv}",
+            "dict": "${dict}",
+            "list": "${list}",
+        },
+    ],
+)
+def test_matrix_interpolated(tmp_dir, dvc, matrix):
+    (tmp_dir / "params.yaml").dump(MATRIX_DATA)
+    resolver = DataResolver(dvc, tmp_dir.fs_path, {})
+    data = {
+        "matrix": matrix,
+        "cmd": "echo ${item.os} ${item.pyv} ${item.dict}"
+        " -- ${item.list.0} ${item.list.1}",
+    }
+    definition = MatrixDefinition(resolver, resolver.context, "build", data)
+
+    assert definition.resolve_all() == {
+        "[email protected]": {"cmd": "echo win 3.7 --arg1 1 -- out1 out11"},
+        "[email protected]": {"cmd": "echo win 3.7 --arg1 1 -- out2 out22"},
+        "[email protected]": {"cmd": "echo win 3.7 --arg2 2 -- out1 out11"},
+        "[email protected]": {"cmd": "echo win 3.7 --arg2 2 -- out2 out22"},
+        "[email protected]": {"cmd": "echo win 3.8 --arg1 1 -- out1 out11"},
+        "[email protected]": {"cmd": "echo win 3.8 --arg1 1 -- out2 out22"},
+        "[email protected]": {"cmd": "echo win 3.8 --arg2 2 -- out1 out11"},
+        "[email protected]": {"cmd": "echo win 3.8 --arg2 2 -- out2 out22"},
+        "[email protected]": {"cmd": "echo linux 3.7 --arg1 1 -- out1 out11"},
+        "[email protected]": {"cmd": "echo linux 3.7 --arg1 1 -- out2 out22"},
+        "[email protected]": {"cmd": "echo linux 3.7 --arg2 2 -- out1 out11"},
+        "[email protected]": {"cmd": "echo linux 3.7 --arg2 2 -- out2 out22"},
+        "[email protected]": {"cmd": "echo linux 3.8 --arg1 1 -- out1 out11"},
+        "[email protected]": {"cmd": "echo linux 3.8 --arg1 1 -- out2 out22"},
+        "[email protected]": {"cmd": "echo linux 3.8 --arg2 2 -- out1 out11"},
+        "[email protected]": {"cmd": "echo linux 3.8 --arg2 2 -- out2 out22"},
+    }

tests/func/repro/test_repro.py

@@ -1326,14 +1326,18 @@ def test_repro_keep_going(mocker, tmp_dir, dvc, copy_script):
     with pytest.raises(ReproductionError):
         dvc.reproduce(on_error="keep-going", repro_fn=spy)
 
-    assert spy.call_args_list == [
-        mocker.call(bar_stage, upstream=[], force=False, interactive=False),
-        mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
-        mocker.call(foo_stage, upstream=[], force=False, interactive=False),
-    ]
+    bar_call = mocker.call(bar_stage, upstream=[], force=False, interactive=False)
+    stage1_call = mocker.call(
+        stage1, upstream=[bar_stage], force=False, interactive=False
+    )
+    foo_call = mocker.call(foo_stage, upstream=[], force=False, interactive=False)
+    assert spy.call_args_list in (
+        [bar_call, stage1_call, foo_call],
+        [foo_call, bar_call, stage1_call],
+    )
 
 
-def test_repro_ignore_errors(M, mocker, tmp_dir, dvc, copy_script):
+def test_repro_ignore_errors(mocker, tmp_dir, dvc, copy_script):
     from dvc.repo import reproduce
 
     (bar_stage, foo_stage) = tmp_dir.dvc_gen({"bar": "bar", "foo": "foo"})
@@ -1347,14 +1351,18 @@ def test_repro_ignore_errors(M, mocker, tmp_dir, dvc, copy_script):
     spy = mocker.spy(reproduce, "_reproduce_stage")
     dvc.reproduce(on_error="ignore", repro_fn=spy)
 
-    assert spy.call_args_list == [
-        mocker.call(bar_stage, upstream=[], force=False, interactive=False),
-        mocker.call(stage1, upstream=[bar_stage], force=False, interactive=False),
-        mocker.call(foo_stage, upstream=[], force=False, interactive=False),
-        mocker.call(
-            stage2,
-            upstream=[foo_stage, stage1],
-            force=False,
-            interactive=False,
-        ),
-    ]
+    bar_call = mocker.call(bar_stage, upstream=[], force=False, interactive=False)
+    foo_call = mocker.call(foo_stage, upstream=[], force=False, interactive=False)
+    stage1_call = mocker.call(
+        stage1, upstream=[bar_stage], force=False, interactive=False
+    )
+    stage2_call = mocker.call(
+        stage2,
+        upstream=[foo_stage, stage1],
+        force=False,
+        interactive=False,
+    )
+    assert spy.call_args_list in (
+        [bar_call, stage1_call, foo_call, stage2_call],
+        [foo_call, bar_call, stage1_call, stage2_call],
+    )

tests/func/test_add.py

@@ -452,9 +452,9 @@ def temporary_windows_drive(tmp_path_factory):
         if len(s) > 0
     ]
 
-    new_drive_name = [
+    new_drive_name = next(
         letter for letter in string.ascii_uppercase if letter not in drives
-    ][0]
+    )
     new_drive = f"{new_drive_name}:"
 
     target_path = tmp_path_factory.mktemp("tmp_windows_drive")

tests/func/test_config.py

@@ -218,7 +218,7 @@ def test_list_bad_args(tmp_dir, dvc, caplog, args):
 
 
 def test_set_invalid_key(dvc):
-    with pytest.raises(ConfigError, match=r"extra keys not allowed"):  # noqa: PT012
+    with pytest.raises(ConfigError, match=r"extra keys not allowed"):
         with dvc.config.edit() as conf:
             conf["core"]["invalid_key"] = "value"
 
@@ -227,9 +227,7 @@ def test_merging_two_levels(dvc):
     with dvc.config.edit() as conf:
         conf["remote"]["test"] = {"url": "ssh://example.com"}
 
-    with pytest.raises(  # noqa: PT012
-        ConfigError, match=r"expected 'url' for dictionary value"
-    ):
+    with pytest.raises(ConfigError, match=r"expected 'url' for dictionary value"):
         with dvc.config.edit("global") as conf:
             conf["remote"]["test"] = {"password": "1"}
 
@@ -267,14 +265,16 @@ def test_load_relative_paths(dvc, field, remote_url):
 
     # check if written paths are correct
     dvc_dir = dvc.config.dvc_dir
-    assert dvc.config["remote"]["test"][field] == os.path.join(
-        dvc_dir, "..", "file.txt"
+    assert dvc.config["remote"]["test"][field] == os.path.abspath(
+        os.path.join(dvc_dir, "..", "file.txt")
     )
 
     # load config and check that it contains what we expect
     # (relative paths are evaluated correctly)
     cfg = Config(dvc_dir)
-    assert cfg["remote"]["test"][field] == os.path.join(dvc_dir, "..", "file.txt")
+    assert cfg["remote"]["test"][field] == os.path.abspath(
+        os.path.join(dvc_dir, "..", "file.txt")
+    )
 
 
 def test_config_gdrive_fields(tmp_dir, dvc):

tests/func/test_data_cloud.py

@@ -10,7 +10,9 @@ from dvc.cli import main
 from dvc.exceptions import CheckoutError
 from dvc.repo.open_repo import clean_repos
 from dvc.stage.exceptions import StageNotFound
-from dvc.testing.remote_tests import TestRemote  # noqa, pylint: disable=unused-import
+from dvc.testing.remote_tests import (  # noqa: F401, pylint: disable=unused-import
+    TestRemote,
+)
 from dvc.utils.fs import remove
 from dvc_data.hashfile.db import HashFileDB
 from dvc_data.hashfile.db.local import LocalHashFileDB

tests/func/test_get.py

@@ -64,6 +64,22 @@ def test_get_repo_dir(tmp_dir, erepo_dir):
     assert (tmp_dir / "dir_imported").read_text() == {"file": "contents"}
 
 
+def test_get_repo_broken_dir(tmp_dir, erepo_dir):
+    import shutil
+
+    from dvc_data.index import DataIndexDirError
+
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen({"broken": {"file": "contents"}})
+        erepo_dir.dvc.cache.local.clear()
+        shutil.rmtree(erepo_dir / "broken")
+
+    with pytest.raises(DataIndexDirError):
+        Repo.get(os.fspath(erepo_dir), "broken", "out")
+
+    assert not (tmp_dir / "out").exists()
+
+
 @pytest.mark.parametrize(
     "erepo", [pytest.lazy_fixture("git_dir"), pytest.lazy_fixture("erepo_dir")]
 )

tests/func/test_import.py

@@ -12,6 +12,7 @@ from dvc.scm import Git
 from dvc.stage.exceptions import StagePathNotFoundError
 from dvc.testing.tmp_dir import make_subrepo
 from dvc.utils.fs import remove
+from dvc_data.index.index import DataIndexDirError
 
 
 def test_import(tmp_dir, scm, dvc, erepo_dir):
@@ -96,6 +97,19 @@ def test_import_dir(tmp_dir, scm, dvc, erepo_dir):
     }
 
 
+def test_import_broken_dir(tmp_dir, scm, dvc, erepo_dir):
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen({"dir": {"foo": "foo content"}}, commit="create dir")
+        erepo_dir.dvc.cache.local.clear()
+        remove(erepo_dir / "dir")
+
+    with pytest.raises(DataIndexDirError):
+        dvc.imp(os.fspath(erepo_dir), "dir", "dir_imported")
+
+    assert not (tmp_dir / "dir_imported").exists()
+    assert not (tmp_dir / "dir_imported.dvc").exists()
+
+
 def test_import_file_from_dir(tmp_dir, scm, dvc, erepo_dir):
     with erepo_dir.chdir():
         erepo_dir.dvc_gen(
@@ -346,7 +360,8 @@ def test_push_wildcard_from_bare_git_repo(
     with dvc_repo.chdir():
         dvc_repo.dvc.imp(os.fspath(tmp_dir), "dirextra")
 
-        dvc_repo.dvc.imp(os.fspath(tmp_dir), "dir123")
+        with pytest.raises(DataIndexDirError):
+            dvc_repo.dvc.imp(os.fspath(tmp_dir), "dir123")
 
 
 @pytest.mark.parametrize("dname", [".", "dir", "dir/subdir"])

tests/func/test_ls.py

@@ -583,3 +583,40 @@ def test_broken_symlink(tmp_dir, dvc):
             "path": "link",
         },
     ]
+
+
+def test_ls_broken_dir(tmp_dir, dvc):
+    from dvc_data.index import DataIndexDirError
+
+    tmp_dir.dvc_gen(
+        {
+            "broken": {"baz": "baz"},
+        }
+    )
+
+    shutil.rmtree(tmp_dir / "broken")
+    dvc.cache.local.clear()
+
+    tmp_dir.dvc_gen(
+        {
+            "foo": "foo",
+            "dir": {"bar": "bar"},
+        }
+    )
+
+    entries = Repo.ls(os.fspath(tmp_dir))
+    assert entries == [
+        {"isdir": False, "isexec": False, "isout": False, "path": ".dvcignore"},
+        {"isdir": True, "isexec": False, "isout": True, "path": "broken"},
+        {"isdir": False, "isexec": False, "isout": False, "path": "broken.dvc"},
+        {"isdir": True, "isexec": False, "isout": True, "path": "dir"},
+        {"isdir": False, "isexec": False, "isout": False, "path": "dir.dvc"},
+        {"isdir": False, "isexec": False, "isout": True, "path": "foo"},
+        {"isdir": False, "isexec": False, "isout": False, "path": "foo.dvc"},
+    ]
+
+    with pytest.raises(DataIndexDirError):
+        Repo.ls(os.fspath(tmp_dir), "broken")
+
+    with pytest.raises(DataIndexDirError):
+        Repo.ls(os.fspath(tmp_dir), recursive=True)

tests/func/test_stage.py

@@ -84,7 +84,7 @@ def test_default_wdir_ignored_in_checksum(tmp_dir, dvc):
     stage = dvc.run(cmd="cp bar foo", deps=["bar"], outs=["foo"], name="copy-foo-bar")
 
     d = stage.dumpd()
-    assert Stage.PARAM_WDIR not in d.keys()
+    assert Stage.PARAM_WDIR not in d
 
     d = load_yaml("dvc.yaml")
     assert Stage.PARAM_WDIR not in d["stages"]["copy-foo-bar"]
@@ -230,17 +230,35 @@ def test_parent_repo_collect_stages(tmp_dir, scm, dvc):
 
 @pytest.mark.parametrize("with_deps", (False, True))
 def test_collect_symlink(tmp_dir, dvc, with_deps):
+    from dvc.exceptions import StageNotFoundError
+
     tmp_dir.gen({"data": {"foo": "foo contents"}})
     foo_path = os.path.join("data", "foo")
     dvc.add(foo_path)
 
     data_link = tmp_dir / "data_link"
     data_link.symlink_to("data")
-    stage = list(
-        dvc.stage.collect(target=str(data_link / "foo.dvc"), with_deps=with_deps)
-    )[0]
 
-    assert stage.addressing == f"{foo_path}.dvc"
+    if with_deps:
+        # NOTE: with_deps means that we'll need to collect and use dvcfiles in the repo
+        # and we currently don't follow symlinks when collecting those, so it will not
+        # be able to find the target stage.
+        with pytest.raises(StageNotFoundError):
+            dvc.stage.collect(target=str(data_link / "foo.dvc"), with_deps=with_deps)
+    else:
+        stage = next(
+            iter(
+                dvc.stage.collect(
+                    target=str(data_link / "foo.dvc"), with_deps=with_deps
+                )
+            )
+        )
+
+        assert stage.addressing == os.path.join("data_link", "foo.dvc")
+
+    stage = next(iter(dvc.stage.collect(target=f"{foo_path}.dvc", with_deps=with_deps)))
+
+    assert stage.addressing == os.path.join("data", "foo.dvc")
 
 
 def test_stage_strings_representation(tmp_dir, dvc, run_copy):

tests/func/utils/test_hydra.py

@@ -1,3 +1,5 @@
+from contextlib import nullcontext as does_not_raise
+
 import pytest
 
 from dvc.exceptions import InvalidArgumentError
@@ -167,16 +169,70 @@ def hydra_setup(tmp_dir, config_dir, config_name):
         ),
     ],
 )
-def test_compose_and_dump(tmp_dir, suffix, overrides, expected):
+def test_compose_and_dump_overrides(tmp_dir, suffix, overrides, expected):
     from dvc.utils.hydra import compose_and_dump
 
     config_name = "config"
     output_file = tmp_dir / f"params.{suffix}"
     config_dir = hydra_setup(tmp_dir, "conf", "config")
-    compose_and_dump(output_file, config_dir, config_name, overrides)
+    config_module = None
+    compose_and_dump(output_file, config_dir, config_module, config_name, overrides)
     assert output_file.parse() == expected
 
 
+def hydra_setup_dir_basic(tmp_dir, config_subdir, config_name, config_content):
+    if config_subdir is None:
+        return None
+
+    config_dir = tmp_dir / config_subdir
+    config_dir.mkdir()
+    (config_dir / f"{config_name}.yaml").dump(config_content)
+    return str(config_dir)
+
+
+@pytest.mark.parametrize(
+    "config_subdir,config_module,config_content,error_context",
+    [
+        ("conf", None, {"normal_yaml_config": False}, does_not_raise()),
+        (
+            None,
+            "hydra.test_utils.configs",
+            {"normal_yaml_config": True},
+            does_not_raise(),
+        ),
+        (
+            "conf",
+            "hydra.test_utils.configs",
+            {"normal_yaml_config": False},
+            does_not_raise(),
+        ),
+        (
+            None,
+            None,
+            None,
+            pytest.raises(
+                ValueError,
+                match="Either `config_dir` or `config_module` should be provided.",
+            ),
+        ),
+    ],
+)
+def test_compose_and_dump_dir_module(
+    tmp_dir, config_subdir, config_module, config_content, error_context
+):
+    from dvc.utils.hydra import compose_and_dump
+
+    output_file = tmp_dir / "params.yaml"
+    config_name = "config"
+    config_dir = hydra_setup_dir_basic(
+        tmp_dir, config_subdir, config_name, config_content
+    )
+
+    with error_context:
+        compose_and_dump(output_file, config_dir, config_module, config_name, [])
+        assert output_file.parse() == config_content
+
+
 def test_compose_and_dump_yaml_handles_string(tmp_dir):
     """Regression test for https://github.com/iterative/dvc/issues/8583"""
     from dvc.utils.hydra import compose_and_dump
@@ -185,7 +241,7 @@ def test_compose_and_dump_yaml_handles_string(tmp_dir):
     config.parent.mkdir()
     config.write_text("foo: 'no'\n")
     output_file = tmp_dir / "params.yaml"
-    compose_and_dump(output_file, str(config.parent), "config", [])
+    compose_and_dump(output_file, str(config.parent), None, "config", [])
     assert output_file.read_text() == "foo: 'no'\n"
 
 
@@ -197,7 +253,7 @@ def test_compose_and_dump_resolves_interpolation(tmp_dir):
     config.parent.mkdir()
     config.dump({"data": {"root": "path/to/root", "raw": "${.root}/raw"}})
     output_file = tmp_dir / "params.yaml"
-    compose_and_dump(output_file, str(config.parent), "config", [])
+    compose_and_dump(output_file, str(config.parent), None, "config", [])
     assert output_file.parse() == {
         "data": {"root": "path/to/root", "raw": "path/to/root/raw"}
     }

tests/remotes/__init__.py

@@ -1,4 +1,4 @@
-from dvc_ssh.tests.fixtures import (  # noqa, pylint: disable=unused-import
+from dvc_ssh.tests.fixtures import (  # noqa: F401, pylint: disable=unused-import
     make_ssh,
     ssh,
     ssh_server,

tests/unit/command/ls/test_ls.py

@@ -19,7 +19,14 @@ def test_list(mocker):
     url = "local_dir"
     m = _test_cli(mocker, url)
     m.assert_called_once_with(
-        url, None, recursive=False, rev=None, dvc_only=False, config=None
+        url,
+        None,
+        recursive=False,
+        rev=None,
+        dvc_only=False,
+        config=None,
+        remote=None,
+        remote_config=None,
     )
 
 
@@ -27,7 +34,14 @@ def test_list_recursive(mocker):
     url = "local_dir"
     m = _test_cli(mocker, url, "-R")
     m.assert_called_once_with(
-        url, None, recursive=True, rev=None, dvc_only=False, config=None
+        url,
+        None,
+        recursive=True,
+        rev=None,
+        dvc_only=False,
+        config=None,
+        remote=None,
+        remote_config=None,
     )
 
 
@@ -35,7 +49,14 @@ def test_list_git_ssh_rev(mocker):
     url = "[email protected]:repo"
     m = _test_cli(mocker, url, "--rev", "123")
     m.assert_called_once_with(
-        url, None, recursive=False, rev="123", dvc_only=False, config=None
+        url,
+        None,
+        recursive=False,
+        rev="123",
+        dvc_only=False,
+        config=None,
+        remote=None,
+        remote_config=None,
     )
 
 
@@ -44,7 +65,14 @@ def test_list_targets(mocker):
     target = "subdir"
     m = _test_cli(mocker, url, target)
     m.assert_called_once_with(
-        url, target, recursive=False, rev=None, dvc_only=False, config=None
+        url,
+        target,
+        recursive=False,
+        rev=None,
+        dvc_only=False,
+        config=None,
+        remote=None,
+        remote_config=None,
     )
 
 
@@ -52,15 +80,40 @@ def test_list_outputs_only(mocker):
     url = "local_dir"
     m = _test_cli(mocker, url, None, "--dvc-only")
     m.assert_called_once_with(
-        url, None, recursive=False, rev=None, dvc_only=True, config=None
+        url,
+        None,
+        recursive=False,
+        rev=None,
+        dvc_only=True,
+        config=None,
+        remote=None,
+        remote_config=None,
     )
 
 
 def test_list_config(mocker):
     url = "local_dir"
-    m = _test_cli(mocker, url, None, "--config", "myconfig")
+    m = _test_cli(
+        mocker,
+        url,
+        None,
+        "--config",
+        "myconfig",
+        "--remote",
+        "myremote",
+        "--remote-config",
+        "k1=v1",
+        "k2=v2",
+    )
     m.assert_called_once_with(
-        url, None, recursive=False, rev=None, dvc_only=False, config="myconfig"
+        url,
+        None,
+        recursive=False,
+        rev=None,
+        dvc_only=False,
+        config="myconfig",
+        remote="myremote",
+        remote_config={"k1": "v1", "k2": "v2"},
     )
 
 

tests/unit/command/test_completion.py

@@ -0,0 +1,17 @@
+import logging
+
+import pytest
+
+from dvc.cli import main
+from dvc.commands.completion import SUPPORTED_SHELLS
+
+
+@pytest.mark.parametrize("shell", SUPPORTED_SHELLS)
+def test_completion(caplog, capsys, shell):
+    with caplog.at_level(logging.INFO):
+        assert main(["completion", "-s", shell]) == 0
+    assert not caplog.text
+
+    out, err = capsys.readouterr()
+    assert not err
+    assert out

tests/unit/command/test_get.py

@@ -16,6 +16,11 @@ def test_get(mocker):
             "4",
             "--config",
             "myconfig",
+            "--remote",
+            "myremote",
+            "--remote-config",
+            "k1=v1",
+            "k2=v2",
         ]
     )
     assert cli_args.func == CmdGet
@@ -33,6 +38,8 @@ def test_get(mocker):
         jobs=4,
         config="myconfig",
         force=False,
+        remote="myremote",
+        remote_config={"k1": "v1", "k2": "v2"},
     )
 
 

tests/unit/command/test_help.py

@@ -0,0 +1,48 @@
+import logging
+from argparse import SUPPRESS, ArgumentParser
+from typing import Tuple
+
+import pytest
+import shtab
+
+from dvc.cli import main
+from dvc.cli.parser import get_main_parser
+
+
+def command_tuples():
+    root: Tuple[str, ...] = ()
+    commands = [root]
+
+    def recurse_parser(parser: ArgumentParser, parents: Tuple[str, ...] = root) -> None:
+        for positional in parser._get_positional_actions():
+            if positional.help != SUPPRESS and isinstance(positional.choices, dict):
+                public_cmds = shtab.get_public_subcommands(positional)
+                for subcmd, subparser in positional.choices.items():
+                    cmd = (*parents, subcmd)
+                    if subcmd in public_cmds:
+                        commands.append(cmd)
+                    recurse_parser(subparser, cmd)
+
+    main_parser = get_main_parser()
+    recurse_parser(main_parser)
+    return sorted(commands)
+
+
+def pytest_generate_tests(metafunc):
+    def ids(values):
+        if values:
+            return "-".join(values)
+        return "dvc"
+
+    metafunc.parametrize("command_tuples", command_tuples(), ids=ids)
+
+
+def test_help(caplog, capsys, command_tuples):
+    with caplog.at_level(logging.INFO), pytest.raises(SystemExit) as e:
+        main([*command_tuples, "--help"])
+    assert e.value.code == 0
+    assert not caplog.text
+
+    out, err = capsys.readouterr()
+    assert not err
+    assert out

tests/unit/command/test_imp.py

@@ -16,6 +16,11 @@ def test_import(mocker, dvc):
             "3",
             "--config",
             "myconfig",
+            "--remote",
+            "myremote",
+            "--remote-config",
+            "k1=v1",
+            "k2=v2",
         ]
     )
     assert cli_args.func == CmdImport
@@ -34,6 +39,8 @@ def test_import(mocker, dvc):
         no_download=False,
         jobs=3,
         config="myconfig",
+        remote="myremote",
+        remote_config={"k1": "v1", "k2": "v2"},
     )
 
 
@@ -65,6 +72,8 @@ def test_import_no_exec(mocker, dvc):
         no_download=False,
         jobs=None,
         config=None,
+        remote=None,
+        remote_config=None,
     )
 
 
@@ -96,4 +105,6 @@ def test_import_no_download(mocker, dvc):
         no_download=True,
         jobs=None,
         config=None,
+        remote=None,
+        remote_config=None,
     )

tests/unit/output/test_local.py

@@ -2,7 +2,6 @@ import os
 
 from dvc.output import Output
 from dvc.stage import Stage
-from dvc.utils import relpath
 from dvc_data.hashfile.hash_info import HashInfo
 from dvc_data.hashfile.meta import Meta
 
@@ -11,7 +10,7 @@ def test_str_workdir_outside_repo(tmp_dir, erepo_dir):
     stage = Stage(erepo_dir.dvc)
     output = Output(stage, "path", cache=False)
 
-    assert relpath("path", erepo_dir.dvc.root_dir) == str(output)
+    assert os.path.abspath("path") == str(output)
 
 
 def test_str_workdir_inside_repo(dvc):

tests/unit/repo/experiments/test_utils.py

@@ -1,8 +1,15 @@
+import re
+
 import pytest
 
 from dvc.exceptions import InvalidArgumentError
 from dvc.repo.experiments.refs import EXPS_NAMESPACE, ExpRefInfo
-from dvc.repo.experiments.utils import check_ref_format, resolve_name, to_studio_params
+from dvc.repo.experiments.utils import (
+    check_ref_format,
+    gen_random_name,
+    resolve_name,
+    to_studio_params,
+)
 
 
 def commit_exp_ref(tmp_dir, scm, file="foo", contents="foo", name="foo"):
@@ -74,3 +81,7 @@ def test_run_check_ref_format(scm, name, result):
 )
 def test_to_studio_params(params, expected):
     assert to_studio_params(params) == expected
+
+
+def test_gen_random_name():
+    assert re.match("[a-zA-Z]+-[a-zA-Z]+", gen_random_name())

tests/unit/test_config.py

@@ -21,6 +21,29 @@ def test_to_relpath(path, expected):
     assert Config._to_relpath(os.path.join(".", "config"), path) == expected
 
 
+@pytest.mark.parametrize(
+    "path, expected",
+    [
+        ("cache", os.path.abspath(os.path.join("conf_dir", "cache"))),
+        ("dir/cache", os.path.abspath(os.path.join("conf_dir", "dir", "cache"))),
+        ("../cache", os.path.abspath("cache")),
+        (os.getcwd(), os.getcwd()),
+        ("ssh://some/path", "ssh://some/path"),
+    ],
+)
+def test_resolve(path, expected):
+    conf_dir = os.path.abspath("conf_dir")
+    assert Config._resolve(conf_dir, path) == expected
+
+
+def test_resolve_homedir():
+    # NOTE: our test suit patches $HOME, but that only works within the
+    # test itself, so we can't use expanduser in @parametrize here.
+    conf_dir = os.path.abspath("conf_dir")
+    expected = os.path.expanduser(os.path.join("~", "cache"))
+    assert Config._resolve(conf_dir, "~/cache") == expected
+
+
 def test_get_fs(tmp_dir, scm):
     tmp_dir.scm_gen("foo", "foo", commit="add foo")
 

tests/unit/utils/test_utils.py

@@ -40,19 +40,8 @@ def test_fix_env_pyenv(path, orig):
 
 
 @pytest.mark.skipif(os.name != "nt", reason="Windows specific")
-def test_relpath_windows(monkeypatch):
-    """test that relpath correctly generated when run on a
-    windows network share. The drive mapped path is mapped
-    to a UNC path by os.path.realpath"""
-
-    def dummy_realpath(path):
-        return path.replace("x:", "\\\\server\\share")
-
-    monkeypatch.setattr(os.path, "realpath", dummy_realpath)
-    assert (
-        relpath("x:\\dir1\\dir2\\file.txt", "\\\\server\\share\\dir1")
-        == "dir2\\file.txt"
-    )
+def test_relpath_windows():
+    assert relpath("x:\\dir1\\dir2\\file.txt", "x:\\dir1") == "dir2\\file.txt"
 
     assert (
         relpath("y:\\dir1\\dir2\\file.txt", "\\\\server\\share\\dir1")