import/get: escape non-pattern LFS path parts (#10416)

* [pre-commit.ci] pre-commit autoupdate

updates:
- [github.com/astral-sh/ruff-pre-commit: v0.4.1 → v0.4.3](https://github.com/astral-sh/ruff-pre-commit/compare/v0.4.1...v0.4.3)

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

.github/mergify.yml

@@ -1,10 +0,0 @@
-pull_request_rules:
-  - name: backport patches to 2.x
-    conditions:
-      - base=main
-      - label=backport
-    actions:
-      backport:
-        branches:
-          - "2.x"
-        title: ({{ destination_branch }}) {{ title }}

.github/workflows/codeql.yml

@@ -2,7 +2,7 @@ name: "CodeQL"
 
 on:
   push:
-    branches: ["main", "2.x"]
+    branches: ["main"]
   pull_request:
     branches: ["main"]
   schedule:

.github/workflows/tests.yaml

@@ -2,7 +2,7 @@ name: Tests
 
 on:
   push:
-    branches: [main, 2.x]
+    branches: [main]
   pull_request:
   schedule:
     - cron: '5 1 * * *'  # every day at 01:05
@@ -66,7 +66,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - name: Slack Notification
-      uses: rtCamp/action-slack-notify@v2.2.1
+      uses: rtCamp/action-slack-notify@v2.3.0
       env:
         SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
         SLACK_COLOR: ${{ job.status }}

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ ci:
 
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.5.0
+    rev: v4.6.0
     hooks:
       - id: check-added-large-files
       - id: check-case-conflict
@@ -22,7 +22,7 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.2.2'
+    rev: 'v0.4.4'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

CITATION.cff

@@ -0,0 +1,25 @@
+cff-version: 1.2.0
+title: 'DVC: Data Version Control - Git for Data & Models"'
+message: >-
+  If you use this software as part of a publication and wish
+  to cite it, please use the metadata from this file.
+type: software
+authors:
+  - name: The DVC team and contributors
+    website: 'https://github.com/iterative'
+repository-code: 'https://github.com/iterative/dvc'
+url: 'https://dvc.org'
+abstract: >-
+  DVC is a tool for data science that takes advantage of existing software engineering toolset.
+  It helps machine learning teams manage large datasets, make projects reproducible, and collaborate better.
+keywords:
+  - ai
+  - collaboration
+  - data-science
+  - data-version-control
+  - developer-tools
+  - git
+  - machine-learning
+  - python
+  - reproducibility
+license: Apache-2.0

dvc/__init__.py

@@ -3,6 +3,7 @@ DVC
 ----
 Make your data science projects reproducible and shareable.
 """
+
 import dvc.logger
 from dvc.build import PKG  # noqa: F401
 from dvc.version import __version__, version_tuple  # noqa: F401

dvc/__main__.py

@@ -1,4 +1,5 @@
 """Main entry point for DVC command line tool."""
+
 import sys
 
 from dvc.cli import main

dvc/_debug.py

@@ -162,7 +162,7 @@ def _get_path_func(tool: str, ext: str):
     fmt = f"{tool}.dvc-{{now:%Y%m%d}}_{{now:%H%M%S}}.{ext}"
 
     def func(now: Optional["datetime"] = None) -> str:
-        return fmt.format(now=now or datetime.now())
+        return fmt.format(now=now or datetime.now())  # noqa: DTZ005
 
     return func
 

dvc/api/artifacts.py

@@ -1,3 +1,4 @@
+import os
 from typing import Any, Optional
 
 from dvc.repo import Repo
@@ -36,12 +37,22 @@ def artifacts_show(
     if version and stage:
         raise ValueError("Artifact version and stage are mutually exclusive.")
 
+    from dvc.repo.artifacts import Artifacts
+    from dvc.utils import as_posix
+
     repo_kwargs: dict[str, Any] = {
         "subrepos": True,
         "uninitialized": True,
     }
+
+    dirname, _ = Artifacts.parse_path(name)
     with Repo.open(repo, **repo_kwargs) as _repo:
         rev = _repo.artifacts.get_rev(name, version=version, stage=stage)
         with _repo.switch(rev):
-            path = _repo.artifacts.get_path(name)
-        return {"rev": rev, "path": path}
+            root = _repo.fs.root_marker
+            _dirname = _repo.fs.join(root, dirname) if dirname else root
+            with Repo(_dirname, fs=_repo.fs, scm=_repo.scm) as r:
+                path = r.artifacts.get_path(name)
+                path = _repo.fs.join(_repo.fs.root_marker, as_posix(path))
+                parts = _repo.fs.relparts(path, _repo.root_dir)
+                return {"rev": rev, "path": os.path.join(*parts)}

dvc/api/data.py

@@ -25,7 +25,14 @@ def _wrap_exceptions(repo, url):
         raise PathMissingError(exc.path, url) from exc
 
 
-def get_url(path, repo=None, rev=None, remote=None):
+def get_url(
+    path: str,
+    repo: Optional[str] = None,
+    rev: Optional[str] = None,
+    remote: Optional[str] = None,
+    config: Optional[dict[str, Any]] = None,
+    remote_config: Optional[dict[str, Any]] = None,
+):
     """
     Returns the URL to the storage location of a data file or directory tracked
     in a DVC repo. For Git repos, HEAD is used unless a rev argument is
@@ -35,15 +42,44 @@ def get_url(path, repo=None, rev=None, remote=None):
 
     NOTE: This function does not check for the actual existence of the file or
     directory in the remote storage.
+
+    Args:
+        path (str): location and file name of the target, relative to the root
+            of `repo`.
+        repo (str, optional): location of the DVC project or Git Repo.
+            Defaults to the current DVC project (found by walking up from the
+            current working directory tree).
+            It can be a URL or a file system path.
+            Both HTTP and SSH protocols are supported for online Git repos
+            (e.g. [user@]server:project.git).
+        rev (str, optional): Any `Git revision`_ such as a branch or tag name,
+            a commit hash or a dvc experiment name.
+            Defaults to HEAD.
+            If `repo` is not a Git repo, this option is ignored.
+        remote (str, optional): Name of the `DVC remote`_ used to form the
+            returned URL string.
+            Defaults to the `default remote`_ of `repo`.
+            For local projects, the cache is tried before the default remote.
+        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:
+        str: URL to the file or directory.
     """
     from dvc.config import NoRemoteError
     from dvc_data.index import StorageKeyError
 
-    repo_kwargs: dict[str, Any] = {}
-    if remote:
-        repo_kwargs["config"] = {"core": {"remote": remote}}
     with Repo.open(
-        repo, rev=rev, subrepos=True, uninitialized=True, **repo_kwargs
+        repo,
+        rev=rev,
+        subrepos=True,
+        uninitialized=True,
+        remote=remote,
+        config=config,
+        remote_config=remote_config,
     ) as _repo:
         index, entry = _repo.get_data_index_entry(path)
         with reraise(

dvc/cli/parser.py

@@ -1,4 +1,5 @@
 """Main parser for the dvc cli."""
+
 import argparse
 import os
 from functools import lru_cache

dvc/commands/add.py

@@ -51,6 +51,7 @@ class CmdAdd(CmdBase):
                 to_remote=self.args.to_remote,
                 remote_jobs=self.args.remote_jobs,
                 force=self.args.force,
+                relink=self.args.relink,
             )
         except FileNotFoundError:
             logger.exception("")
@@ -119,6 +120,13 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Override local file or folder if exists.",
     )
+    parser.add_argument(
+        "--no-relink",
+        dest="relink",
+        action="store_false",
+        help="Don't recreate links from cache to workspace.",
+    )
+    parser.set_defaults(relink=True)
     parser.add_argument(
         "targets", nargs="+", help="Input files/directories to add."
     ).complete = completion.FILE

dvc/commands/commit.py

@@ -20,6 +20,7 @@ class CmdCommit(CmdBase):
                     with_deps=self.args.with_deps,
                     recursive=self.args.recursive,
                     force=self.args.force,
+                    relink=self.args.relink,
                 )
             except DvcException:
                 logger.exception("failed to commit%s", (" " + target) if target else "")
@@ -64,6 +65,13 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Commit cache for subdirectories of the specified directory.",
     )
+    commit_parser.add_argument(
+        "--no-relink",
+        dest="relink",
+        action="store_false",
+        help="Don't recreate links from cache to workspace.",
+    )
+    commit_parser.set_defaults(relink=True)
     commit_parser.add_argument(
         "targets",
         nargs="*",

dvc/commands/gc.py

@@ -60,7 +60,7 @@ class CmdGC(CmdBase):
             msg += " of the current and the following repos:"
 
             for repo_path in self.args.repos:
-                msg += "\n  - %s" % os.path.abspath(repo_path)
+                msg += f"\n  - {os.path.abspath(repo_path)}"
         else:
             msg += " of the current repo."
 

dvc/commands/get.py

@@ -12,7 +12,13 @@ class CmdGet(CmdBaseNoRepo):
         from dvc.api import get_url
         from dvc.ui import ui
 
-        url = get_url(self.args.path, repo=self.args.url, rev=self.args.rev)
+        url = get_url(
+            self.args.path,
+            repo=self.args.url,
+            rev=self.args.rev,
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+        )
         ui.write(url, force=True)
 
         return 0

dvc/commands/studio.py

@@ -13,11 +13,10 @@ logger = logger.getChild(__name__)
 
 class CmdStudioLogin(CmdConfig):
     def run(self):
-        from dvc_studio_client.auth import StudioAuthError, get_access_token
-
         from dvc.env import DVC_STUDIO_URL
         from dvc.ui import ui
         from dvc.utils.studio import STUDIO_URL
+        from dvc_studio_client.auth import StudioAuthError, get_access_token
 
         name = self.args.name
         hostname = self.args.hostname or os.environ.get(DVC_STUDIO_URL) or STUDIO_URL
@@ -46,6 +45,15 @@ class CmdStudioLogin(CmdConfig):
 
         self.save_config(hostname, access_token)
 
+        if not self.config["exp"].get("auto_push", True):
+            from dvc.ui import ui
+
+            ui.warn(
+                "exp.auto_push is disabled. \n"
+                "Enable with 'dvc config exp.auto_push true' "
+                "to automatically push experiments to Studio."
+            )
+
         config_path = self.config.files["global"]
         ui.write(f"Authentication complete. Saved token to {config_path}.")
         return 0
@@ -54,6 +62,8 @@ class CmdStudioLogin(CmdConfig):
         with self.config.edit("global") as conf:
             conf["studio"]["token"] = token
             conf["studio"]["url"] = hostname
+            if "auto_push" not in conf["exp"]:
+                conf["exp"]["auto_push"] = True
 
 
 class CmdStudioLogout(CmdConfig):

dvc/compare.py

@@ -88,12 +88,10 @@ class TabularData(MutableSequence[Sequence["CellT"]]):
         return list(it)
 
     @overload
-    def __setitem__(self, item: int, value: Sequence["CellT"]) -> None:
-        ...
+    def __setitem__(self, item: int, value: Sequence["CellT"]) -> None: ...
 
     @overload
-    def __setitem__(self, item: slice, value: Iterable[Sequence["CellT"]]) -> None:
-        ...
+    def __setitem__(self, item: slice, value: Iterable[Sequence["CellT"]]) -> None: ...
 
     def __setitem__(self, item, value) -> None:
         it = value

dvc/config.py

@@ -1,4 +1,5 @@
 """DVC config objects."""
+
 import os
 import re
 from contextlib import contextmanager

dvc/config_schema.py

@@ -193,6 +193,7 @@ SCHEMA = {
                     "session_token": str,
                     Optional("listobjects", default=False): Bool,  # obsoleted
                     Optional("use_ssl", default=True): Bool,
+                    Optional("allow_anonymous_login", default=False): Bool,
                     "ssl_verify": Any(Bool, str),
                     "sse": str,
                     "sse_kms_key_id": str,
@@ -232,7 +233,12 @@ SCHEMA = {
                     Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },
-                "hdfs": {"user": str, "kerb_ticket": str, **REMOTE_COMMON},
+                "hdfs": {
+                    "user": str,
+                    "kerb_ticket": str,
+                    "replication": int,
+                    **REMOTE_COMMON,
+                },
                 "webhdfs": {
                     "kerberos": Bool,
                     "kerberos_principal": str,
@@ -336,6 +342,8 @@ SCHEMA = {
         "params": str,
         "plots": str,
         "live": str,
+        "auto_push": Bool,
+        "git_remote": str,
     },
     "parsing": {
         "bool": All(Lower, Choices("store_true", "boolean_optional")),

dvc/dependency/db.py

@@ -22,9 +22,12 @@ def download_progress(to: "Output") -> Iterator[Callable[[int], Any]]:
     from dvc.ui import ui
     from dvc.ui._rich_progress import DbDownloadProgress
 
-    with log_durations(logger.debug, f"Saving to {to}"), DbDownloadProgress(
-        console=ui.error_console,
-    ) as progress:
+    with (
+        log_durations(logger.debug, f"Saving to {to}"),
+        DbDownloadProgress(
+            console=ui.error_console,
+        ) as progress,
+    ):
         task = progress.add_task("Saving", total=None, output=to)
         yield lambda n: progress.advance(task, advance=n)
         progress.update(task, description="Saved", total=0)

dvc/dvcfile.py

@@ -68,10 +68,9 @@ def is_git_ignored(repo, path):
 def check_dvcfile_path(repo, path):
     if not is_valid_filename(path):
         raise StageFileBadNameError(
-            "bad DVC file name '{}'. DVC files should be named "
-            "'{}' or have a '.dvc' suffix (e.g. '{}.dvc').".format(
-                relpath(path), PROJECT_FILE, os.path.basename(path)
-            )
+            f"bad DVC file name '{relpath(path)}'. DVC files should be named "
+            f"'{PROJECT_FILE}' or have a '.dvc' suffix "
+            f"(e.g. '{os.path.basename(path)}.dvc')."
         )
 
     if is_git_ignored(repo, path):
@@ -243,16 +242,22 @@ class ProjectFile(FileMixin):
 
     def dump_dataset(self, dataset):
         with modify_yaml(self.path, fs=self.repo.fs) as data:
-            datasets: list[dict] = data.setdefault("datasets", [])
+            parsed = self.datasets if data else []
+            raw = data.setdefault("datasets", [])
             loc = next(
-                (i for i, ds in enumerate(datasets) if ds["name"] == dataset["name"]),
+                (i for i, ds in enumerate(parsed) if ds["name"] == dataset["name"]),
                 None,
             )
             if loc is not None:
-                apply_diff(dataset, datasets[loc])
-                datasets[loc] = dataset
+                if raw[loc] != parsed[loc]:
+                    raise ParametrizedDumpError(
+                        "cannot update a parametrized dataset entry"
+                    )
+
+                apply_diff(dataset, raw[loc])
+                raw[loc] = dataset
             else:
-                datasets.append(dataset)
+                raw.append(dataset)
         self.repo.scm_context.track_file(self.relpath)
 
     def _dump_lockfile(self, stage, **kwargs):
@@ -308,29 +313,29 @@ class ProjectFile(FileMixin):
         return self.LOADER(self, self.contents, self.lockfile_contents)
 
     @property
-    def metrics(self) -> list[str]:
-        return self.contents.get("metrics", [])
+    def artifacts(self) -> dict[str, Optional[dict[str, Any]]]:
+        return self.resolver.resolve_artifacts()
 
     @property
-    def plots(self) -> Any:
-        return self.contents.get("plots", {})
+    def metrics(self) -> list[str]:
+        return self.resolver.resolve_metrics()
 
     @property
     def params(self) -> list[str]:
-        return self.contents.get("params", [])
+        return self.resolver.resolve_params()
+
+    @property
+    def plots(self) -> list[Any]:
+        return self.resolver.resolve_plots()
 
     @property
     def datasets(self) -> list[dict[str, Any]]:
-        return self.contents.get("datasets", [])
+        return self.resolver.resolve_datasets()
 
     @property
     def datasets_lock(self) -> list[dict[str, Any]]:
         return self.lockfile_contents.get("datasets", [])
 
-    @property
-    def artifacts(self) -> dict[str, Optional[dict[str, Any]]]:
-        return self.contents.get("artifacts", {})
-
     def remove(self, force=False):
         if not force:
             logger.warning("Cannot remove pipeline file.")

dvc/env.py

@@ -19,3 +19,4 @@ DVC_STUDIO_URL = "DVC_STUDIO_URL"
 DVC_SQLALCHEMY_ECHO = "DVC_SQLALCHEMY_ECHO"
 DVC_SYSTEM_CONFIG_DIR = "DVC_SYSTEM_CONFIG_DIR"
 DVC_UPDATER_ENDPOINT = "DVC_UPDATER_ENDPOINT"
+DVC_STAGE = "DVC_STAGE"

dvc/exceptions.py

@@ -1,4 +1,5 @@
 """Exceptions raised by the dvc."""
+
 import errno
 from typing import TYPE_CHECKING, Optional
 
@@ -132,11 +133,11 @@ class MoveNotDataSourceError(DvcException):
     def __init__(self, path):
         msg = (
             "move is not permitted for stages that are not data sources. "
-            "You need to either move '{path}' to a new location and edit "
-            "it by hand, or remove '{path}' and create a new one at the "
+            f"You need to either move {path!r} to a new location and edit "
+            f"it by hand, or remove {path!r} and create a new one at the "
             "desired location."
         )
-        super().__init__(msg.format(path=path))
+        super().__init__(msg)
 
 
 class NotDvcRepoError(DvcException):
@@ -270,9 +271,8 @@ class NoOutputInExternalRepoError(DvcException):
         from dvc.utils import relpath
 
         super().__init__(
-            "Output '{}' not found in target repository '{}'".format(
-                relpath(path, external_repo_path), external_repo_url
-            )
+            f"Output {relpath(path, external_repo_path)!r} "
+            f"not found in target repository '{external_repo_url}'"
         )
 
 

dvc/fs/__init__.py

@@ -1,10 +1,10 @@
+import glob
 from typing import Optional
 from urllib.parse import urlparse
 
-from dvc_http import HTTPFileSystem, HTTPSFileSystem  # noqa: F401
-
 from dvc.config import ConfigError as RepoConfigError
 from dvc.config_schema import SCHEMA, Invalid
+from dvc_http import HTTPFileSystem, HTTPSFileSystem  # noqa: F401
 
 # pylint: disable=unused-import
 from dvc_objects.fs import (  # noqa: F401
@@ -70,7 +70,14 @@ def download(
             to_infos = [to]
 
         if isinstance(fs, DVCFileSystem):
-            lfs_prefetch(fs, from_infos)
+            lfs_prefetch(
+                fs,
+                [
+                    f"{fs.normpath(glob.escape(fs_path))}/**"
+                    if fs.isdir(fs_path)
+                    else glob.escape(fs_path)
+                ],
+            )
         cb.set_size(len(from_infos))
         jobs = jobs or fs.jobs
         generic.copy(fs, from_infos, localfs, to_infos, callback=cb, batch_size=jobs)

dvc/fs/dvc.py

@@ -360,7 +360,7 @@ class _DVCFileSystem(AbstractFileSystem):
         except FileNotFoundError:
             return False
 
-    def ls(self, path, detail=True, dvc_only=False, **kwargs):  # noqa: C901
+    def ls(self, path, detail=True, dvc_only=False, **kwargs):  # noqa: C901, PLR0912
         key = self._get_key_from_relative(path)
         repo, dvc_fs, subkey = self._get_subrepo_info(key)
 

dvc/output.py

@@ -1365,7 +1365,7 @@ class Output:
                 dry_run=not self.use_cache,
             )
         except FileNotFoundError as exc:
-            if self.fs_path == path:
+            if not self.exists:
                 raise self.DoesNotExistError(self) from exc
             if not self.is_dir_checksum:
                 raise

dvc/parsing/__init__.py

@@ -21,6 +21,7 @@ from .context import (
     VarsAlreadyLoaded,
 )
 from .interpolate import (
+    check_expression,
     check_recursive_parse_errors,
     is_interpolated_string,
     recurse,
@@ -38,10 +39,16 @@ if TYPE_CHECKING:
 
 logger = logger.getChild(__name__)
 
-STAGES_KWD = "stages"
 VARS_KWD = "vars"
 WDIR_KWD = "wdir"
+
+ARTIFACTS_KWD = "artifacts"
+DATASETS_KWD = "datasets"
+METRICS_KWD = "metrics"
 PARAMS_KWD = "params"
+PLOTS_KWD = "plots"
+STAGES_KWD = "stages"
+
 FOREACH_KWD = "foreach"
 MATRIX_KWD = "matrix"
 DO_KWD = "do"
@@ -163,6 +170,27 @@ class DataResolver:
             for name, definition in stages_data.items()
         }
 
+        self.artifacts = [
+            ArtifactDefinition(self, self.context, name, definition, ARTIFACTS_KWD)
+            for name, definition in d.get(ARTIFACTS_KWD, {}).items()
+        ]
+        self.datasets = [
+            TopDefinition(self, self.context, str(i), definition, DATASETS_KWD)
+            for i, definition in enumerate(d.get(DATASETS_KWD, []))
+        ]
+        self.metrics = [
+            TopDefinition(self, self.context, str(i), definition, METRICS_KWD)
+            for i, definition in enumerate(d.get(METRICS_KWD, []))
+        ]
+        self.params = [
+            TopDefinition(self, self.context, str(i), definition, PARAMS_KWD)
+            for i, definition in enumerate(d.get(PARAMS_KWD, []))
+        ]
+        self.plots = [
+            TopDefinition(self, self.context, str(i), definition, PLOTS_KWD)
+            for i, definition in enumerate(d.get(PLOTS_KWD, []))
+        ]
+
     def resolve_one(self, name: str):
         group, key = split_group_name(name)
 
@@ -186,6 +214,27 @@ class DataResolver:
         logger.trace("Resolved dvc.yaml:\n%s", data)
         return {STAGES_KWD: data}
 
+    # Top-level sections are eagerly evaluated, whereas stages are lazily evaluated,
+    # one-by-one.
+
+    def resolve_artifacts(self) -> dict[str, Optional[dict[str, Any]]]:
+        d: dict[str, Optional[dict[str, Any]]] = {}
+        for item in self.artifacts:
+            d.update(item.resolve())
+        return d
+
+    def resolve_datasets(self) -> list[dict[str, Any]]:
+        return [item.resolve() for item in self.datasets]
+
+    def resolve_metrics(self) -> list[str]:
+        return [item.resolve() for item in self.metrics]
+
+    def resolve_params(self) -> list[str]:
+        return [item.resolve() for item in self.params]
+
+    def resolve_plots(self) -> list[Any]:
+        return [item.resolve() for item in self.plots]
+
     def has_key(self, key: str):
         return self._has_group_and_key(*split_group_name(key))
 
@@ -565,3 +614,43 @@ class MatrixDefinition:
                 return entry.resolve_stage(skip_checks=True)
             except ContextError as exc:
                 format_and_raise(exc, f"stage '{generated}'", self.relpath)
+
+
+class TopDefinition:
+    def __init__(
+        self,
+        resolver: DataResolver,
+        context: Context,
+        name: str,
+        definition: "Any",
+        where: str,
+    ):
+        self.resolver = resolver
+        self.context = context
+        self.name = name
+        self.definition = definition
+        self.where = where
+        self.relpath = self.resolver.relpath
+
+    def resolve(self):
+        try:
+            check_recursive_parse_errors(self.definition)
+            return self.context.resolve(self.definition)
+        except (ParseError, ContextError) as exc:
+            format_and_raise(exc, f"'{self.where}.{self.name}'", self.relpath)
+
+
+class ArtifactDefinition(TopDefinition):
+    def resolve(self) -> dict[str, Optional[dict[str, Any]]]:
+        try:
+            check_expression(self.name)
+            name = self.context.resolve(self.name)
+            if not isinstance(name, str):
+                typ = type(name).__name__
+                raise ResolveError(
+                    f"failed to resolve '{self.where}.{self.name}'"
+                    f" in '{self.relpath}': expected str, got " + typ
+                )
+        except (ParseError, ContextError) as exc:
+            format_and_raise(exc, f"'{self.where}.{self.name}'", self.relpath)
+        return {name: super().resolve()}

dvc/progress.py

@@ -1,4 +1,5 @@
 """Manages progress bars for DVC repo."""
+
 import logging
 import sys
 from threading import RLock

dvc/repo/__init__.py

@@ -551,11 +551,7 @@ class Repo:
 
             if out.protocol == "local" and match(fs_path, out.fs_path):
                 return True
-
-            if recursive and out.fs.isin(out.fs_path, fs_path):
-                return True
-
-            return False
+            return recursive and out.fs.isin(out.fs_path, fs_path)
 
         matched = list(filter(func, outs))
         if not matched:

dvc/repo/add.py

@@ -171,11 +171,16 @@ def _add_transfer(
     stage.dump()
 
 
-def _add(stage: "Stage", source: Optional[str] = None, no_commit: bool = False) -> None:
+def _add(
+    stage: "Stage",
+    source: Optional[str] = None,
+    no_commit: bool = False,
+    relink: bool = True,
+) -> None:
     out = stage.outs[0]
     path = out.fs.abspath(source) if source else None
     try:
-        stage.add_outs(path, no_commit=no_commit)
+        stage.add_outs(path, no_commit=no_commit, relink=relink)
     except CacheLinkError:
         stage.dump()
         raise
@@ -194,6 +199,7 @@ def add(
     to_remote: bool = False,
     remote_jobs: Optional[int] = None,
     force: bool = False,
+    relink: bool = True,
 ) -> list["Stage"]:
     add_targets = find_targets(targets, glob=glob)
     if not add_targets:
@@ -224,7 +230,12 @@ def add(
     with warn_link_failures() as link_failures:
         for source, (stage, output_exists) in progress_iter(stages_with_targets):
             try:
-                _add(stage, source if output_exists else None, no_commit=no_commit)
+                _add(
+                    stage,
+                    source if output_exists else None,
+                    no_commit=no_commit,
+                    relink=relink,
+                )
             except CacheLinkError:
                 link_failures.append(stage.relpath)
     return stages

dvc/repo/artifacts.py

@@ -12,7 +12,7 @@ from dvc.exceptions import (
     InvalidArgumentError,
 )
 from dvc.log import logger
-from dvc.utils import relpath, resolve_output
+from dvc.utils import as_posix, relpath, resolve_output
 from dvc.utils.objects import cached_property
 from dvc.utils.serialize import modify_yaml
 
@@ -99,7 +99,7 @@ class Artifacts:
         """Read artifacts from dvc.yaml."""
         artifacts: dict[str, dict[str, Artifact]] = {}
         for dvcfile, dvcfile_artifacts in self.repo.index._artifacts.items():
-            dvcyaml = relpath(dvcfile, self.repo.root_dir)
+            dvcyaml = self.repo.fs.relpath(dvcfile, self.repo.root_dir)
             artifacts[dvcyaml] = {}
             for name, value in dvcfile_artifacts.items():
                 try:
@@ -147,8 +147,8 @@ class Artifacts:
         gto_tags: list["GTOTag"] = sort_versions(parse_tag(tag) for tag in tags)
         return gto_tags[0].tag.target
 
-    def get_path(self, name: str):
-        """Return repo fspath for the given artifact."""
+    @classmethod
+    def parse_path(cls, name: str) -> tuple[Optional[str], str]:
         from gto.constants import SEPARATOR_IN_NAME, fullname_re
 
         name = _reformat_name(name)
@@ -158,13 +158,37 @@ class Artifacts:
         dirname = m.group("dirname")
         if dirname:
             dirname = dirname.rstrip(SEPARATOR_IN_NAME)
-        dvcyaml = os.path.join(dirname, PROJECT_FILE) if dirname else PROJECT_FILE
-        artifact_name = m.group("name")
+
+        return dirname, m.group("name")
+
+    def get_path(self, name: str):
+        """Return fspath for the given artifact relative to the git root."""
+        from dvc.fs import GitFileSystem
+
+        dirname, artifact_name = self.parse_path(name)
+        # `name`/`dirname` are expected to be a git root relative.
+        # We convert it to dvc-root relative path so that we can read artifacts
+        # from dvc.yaml file.
+        # But we return dirname intact, as we want to return a git-root relative path.
+        # This is useful when reading from `dvcfs` from remote.
+        fs = self.repo.fs
+        assert self.scm
+        if isinstance(fs, GitFileSystem):
+            scm_root = fs.root_marker
+        else:
+            scm_root = self.scm.root_dir
+
+        dirparts = posixpath.normpath(dirname).split(posixpath.sep) if dirname else ()
+        abspath = fs.join(scm_root, *dirparts, PROJECT_FILE)
+        rela = fs.relpath(abspath, self.repo.root_dir)
         try:
-            artifact = self.read()[dvcyaml][artifact_name]
+            artifact = self.read()[rela][artifact_name]
         except KeyError as exc:
             raise ArtifactNotFoundError(name) from exc
-        return os.path.join(dirname, artifact.path) if dirname else artifact.path
+
+        path = posixpath.join(dirname or "", artifact.path)
+        parts = posixpath.normpath(path).split(posixpath.sep)
+        return os.path.join(*parts)
 
     def download(
         self,
@@ -177,15 +201,28 @@ class Artifacts:
     ) -> tuple[int, str]:
         """Download the specified artifact."""
         from dvc.fs import download as fs_download
+        from dvc.repo import Repo
 
         logger.debug("Trying to download artifact '%s' via DVC", name)
         rev = self.get_rev(name, version=version, stage=stage)
+
+        dirname, _ = self.parse_path(name)
         with self.repo.switch(rev):
-            path = self.get_path(name)
+            root = self.repo.fs.root_marker
+            _dirname = self.repo.fs.join(root, dirname) if dirname else root
+            with Repo(_dirname, fs=self.repo.fs, scm=self.repo.scm) as r:
+                path = r.artifacts.get_path(name)
+                path = self.repo.fs.join(root, as_posix(path))
+                path = self.repo.fs.relpath(path, self.repo.root_dir)
+                # when the `repo` is a subrepo, the path `/subrepo/myart.pkl` for dvcfs
+                # should be translated as `/myart.pkl`,
+                # i.e. relative to the root of the subrepo
+                path = self.repo.fs.join(root, path)
+                path = self.repo.fs.normpath(path)
+
             out = resolve_output(path, out, force=force)
             fs = self.repo.dvcfs
-            fs_path = fs.from_os_path(path)
-            count = fs_download(fs, fs_path, os.path.abspath(out), jobs=jobs)
+            count = fs_download(fs, path, os.path.abspath(out), jobs=jobs)
         return count, out
 
     @staticmethod
@@ -200,10 +237,9 @@ class Artifacts:
         dvc_studio_config: Optional[dict[str, Any]] = None,
         **kwargs,
     ) -> tuple[int, str]:
-        from dvc_studio_client.model_registry import get_download_uris
-
         from dvc.fs import HTTPFileSystem, generic, localfs
         from dvc.fs.callbacks import TqdmCallback
+        from dvc_studio_client.model_registry import get_download_uris
 
         logger.debug("Trying to download artifact '%s' via studio", name)
         out = out or os.getcwd()
@@ -231,7 +267,7 @@ class Artifacts:
                 from_infos.append(url)
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(
                 f"Failed to download artifact '{name}' via Studio"
             ) from exc
@@ -333,7 +369,7 @@ class Artifacts:
                 )
             except FileExistsLocallyError:
                 raise
-            except Exception as exc:  # noqa: BLE001
+            except Exception as exc:
                 if saved_exc:
                     logger.exception(str(saved_exc), exc_info=saved_exc.__cause__)
                 raise DvcException(

dvc/repo/datasets.py

@@ -16,8 +16,7 @@ from dvc.types import StrPath
 from dvc_data.hashfile.meta import Meta
 
 if TYPE_CHECKING:
-    from dql.dataset import DatasetRecord  # type: ignore[import]
-    from dvcx.dataset import DatasetVersion  # type: ignore[import]
+    from dvcx.dataset import DatasetRecord, DatasetVersion  # type: ignore[import]
     from typing_extensions import Self
 
     from dvc.repo import Repo

dvc/repo/experiments/collect.py

@@ -78,7 +78,7 @@ def _collect_rev(
             timestamp: Optional[datetime] = None
         else:
             commit = repo.scm.resolve_commit(rev)
-            timestamp = datetime.fromtimestamp(commit.commit_time)
+            timestamp = datetime.fromtimestamp(commit.commit_time)  # noqa: DTZ006
 
         return SerializableExp.from_repo(
             repo,
@@ -338,6 +338,6 @@ def _sorted_ranges(exp_ranges: Iterable["ExpRange"]) -> list["ExpRange"]:
         if head_exp and head_exp.data and head_exp.data.timestamp:
             return head_exp.data.timestamp, head_exp.rev
 
-        return datetime.fromtimestamp(0), ""
+        return datetime.fromtimestamp(0), ""  # noqa: DTZ006
 
     return sorted(exp_ranges, key=_head_timestamp, reverse=True)

dvc/repo/experiments/executor/base.py

@@ -25,6 +25,7 @@ from dvc.utils import dict_sha256, env2bool, relpath
 from dvc.utils.fs import remove
 from dvc.utils.studio import (
     env_to_config,
+    get_repo_url,
     get_subrepo_relpath,
 )
 
@@ -297,13 +298,16 @@ class BaseExecutor(ABC):
             exp_hash = cls.hash_exp(stages)
             if include_untracked:
                 dvc.scm.add(include_untracked, force=True)  # type: ignore[call-arg]
-            cls.commit(
-                dvc.scm,  # type: ignore[arg-type]
-                exp_hash,
-                exp_name=info.name,
-                force=force,
-                message=message,
-            )
+
+            with cls.auto_push(dvc):
+                cls.commit(
+                    dvc.scm,  # type: ignore[arg-type]
+                    exp_hash,
+                    exp_name=info.name,
+                    force=force,
+                    message=message,
+                )
+
             ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
             exp_ref = ExpRefInfo.from_ref(ref) if ref else None
             untracked = dvc.scm.untracked_files()
@@ -460,9 +464,6 @@ class BaseExecutor(ABC):
         from dvc.repo.checkout import checkout as dvc_checkout
         from dvc.ui import ui
 
-        auto_push = env2bool(DVC_EXP_AUTO_PUSH)
-        git_remote = os.getenv(DVC_EXP_GIT_REMOTE, None)
-
         if queue is not None:
             queue.put((rev, os.getpid()))
         if log_errors and log_level is not None:
@@ -483,9 +484,6 @@ class BaseExecutor(ABC):
             message=message,
             **kwargs,
         ) as dvc:
-            if auto_push:
-                cls._validate_remotes(dvc, git_remote)
-
             args, kwargs = cls._repro_args(dvc)
             if args:
                 targets: Optional[Union[list, str]] = args[0]
@@ -519,8 +517,6 @@ class BaseExecutor(ABC):
                     dvc,
                     info,
                     exp_hash,
-                    auto_push,
-                    git_remote,
                     repro_force,
                     message=message,
                 )
@@ -550,20 +546,18 @@ class BaseExecutor(ABC):
         dvc,
         info,
         exp_hash,
-        auto_push,
-        git_remote,
         repro_force,
         message: Optional[str] = None,
     ) -> tuple[Optional[str], Optional["ExpRefInfo"], bool]:
-        cls.commit(
-            dvc.scm,
-            exp_hash,
-            exp_name=info.name,
-            force=repro_force,
-            message=message,
-        )
-        if auto_push:
-            cls._auto_push(dvc, dvc.scm, git_remote)
+        with cls.auto_push(dvc):
+            cls.commit(
+                dvc.scm,
+                exp_hash,
+                exp_name=info.name,
+                force=repro_force,
+                message=message,
+            )
+
         ref: Optional[str] = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
         exp_ref: Optional["ExpRefInfo"] = ExpRefInfo.from_ref(ref) if ref else None
         if cls.WARN_UNTRACKED:
@@ -591,9 +585,8 @@ class BaseExecutor(ABC):
         message: Optional[str] = None,
         **kwargs,
     ) -> Iterator["Repo"]:
-        from dvc_studio_client.post_live_metrics import post_live_metrics
-
         from dvc.repo import Repo
+        from dvc_studio_client.post_live_metrics import post_live_metrics
 
         with Repo(os.path.join(info.root_dir, info.dvc_dir)) as dvc:
             info.status = TaskStatus.RUNNING
@@ -618,6 +611,8 @@ class BaseExecutor(ABC):
             # inferring repo url will fail if not set here
             run_env_config = env_to_config(kwargs.get("run_env", {}))
             dvc_studio_config = run_env_config | dvc_studio_config
+            # override studio repo url if exp git remote set
+            repo_url = get_repo_url(dvc)
             try:
                 post_live_metrics(
                     "start",
@@ -628,6 +623,7 @@ class BaseExecutor(ABC):
                     dvc_studio_config=dvc_studio_config,
                     message=message,
                     subdir=get_subrepo_relpath(dvc),
+                    studio_repo_url=repo_url,
                 )
                 logger.debug("Running repro in '%s'", os.getcwd())
                 yield dvc
@@ -653,6 +649,7 @@ class BaseExecutor(ABC):
                     experiment_rev=dvc.experiments.scm.get_ref(EXEC_BRANCH),
                     metrics=_gather_metrics(dvc, on_error="return"),
                     dvc_studio_config=dvc_studio_config,
+                    studio_repo_url=repo_url,
                 )
 
                 if infofile is not None:
@@ -672,15 +669,46 @@ class BaseExecutor(ABC):
             kwargs = {}
         return args, kwargs
 
+    @classmethod
+    @contextmanager
+    def auto_push(cls, dvc: "Repo") -> Iterator[None]:
+        exp_config = dvc.config.get("exp", {})
+        auto_push = env2bool(DVC_EXP_AUTO_PUSH, exp_config.get("auto_push", False))
+        if not auto_push:
+            yield
+            return
+
+        git_remote = os.getenv(
+            DVC_EXP_GIT_REMOTE, exp_config.get("git_remote", "origin")
+        )
+        try:
+            cls._validate_remotes(dvc, git_remote)
+        except DvcException as exc:
+            logger.warning("Failed to validate remotes. Disabling auto push: %s", exc)
+
+            yield
+            return
+        yield
+        cls._auto_push(dvc, git_remote)
+
     @staticmethod
     def _auto_push(
         dvc: "Repo",
-        scm: "Git",
         git_remote: Optional[str],
         push_cache=True,
         run_cache=True,
     ):
-        branch = scm.get_ref(EXEC_BRANCH, follow=False)
+        from dvc.ui import ui
+        from dvc.utils import format_link
+
+        branch = dvc.scm.get_ref(EXEC_BRANCH, follow=False)
+        link = format_link(
+            "https://dvc.org/doc/user-guide/experiment-management/sharing-experiments"
+        )
+        ui.write(
+            f"Pushing experiment to '{git_remote}'. Cancel with CTRL+C. "
+            f"See {link} for more info."
+        )
         try:
             dvc.experiments.push(
                 git_remote,
@@ -688,7 +716,7 @@ class BaseExecutor(ABC):
                 push_cache=push_cache,
                 run_cache=run_cache,
             )
-        except BaseException as exc:  # noqa: BLE001
+        except DvcException as exc:
             logger.warning(
                 (
                     "Something went wrong while auto pushing experiment "
@@ -708,6 +736,7 @@ class BaseExecutor(ABC):
         message: Optional[str] = None,
     ):
         """Commit stages as an experiment and return the commit SHA."""
+
         rev = scm.get_rev()
         if not scm.is_dirty(untracked_files=False):
             logger.debug("No changes to commit")

dvc/repo/experiments/executor/local.py

@@ -96,9 +96,10 @@ class TempDirExecutor(BaseLocalExecutor):
             temp_merge: stash_rev,
             temp_baseline: entry.baseline_rev,
         }
-        with get_exp_rwlock(
-            repo, writes=[temp_head, temp_merge, temp_baseline]
-        ), self.set_temp_refs(scm, temp_ref_dict):
+        with (
+            get_exp_rwlock(repo, writes=[temp_head, temp_merge, temp_baseline]),
+            self.set_temp_refs(scm, temp_ref_dict),
+        ):
             # Executor will be initialized with an empty git repo that
             # we populate by pushing:
             #   EXEC_HEAD - the base commit for this experiment
@@ -130,6 +131,8 @@ class TempDirExecutor(BaseLocalExecutor):
 
         self.scm.stash.apply(merge_rev)
         self._update_config(repo.config.read("local"))
+        local_git_config = os.path.join(repo.scm.root_dir, ".git", "config")
+        self._update_git_config(ConfigObj(local_git_config, list_values=False))
 
     def _update_config(self, update):
         local_config = os.path.join(self.root_dir, self.dvc_dir, "config.local")
@@ -143,6 +146,18 @@ class TempDirExecutor(BaseLocalExecutor):
             with open(local_config, "wb") as fobj:
                 conf_obj.write(fobj)
 
+    def _update_git_config(self, update):
+        local_config = os.path.join(self.scm.root_dir, ".git", "config")
+        logger.debug("Writing experiments local Git config '%s'", local_config)
+        if os.path.exists(local_config):
+            conf_obj = ConfigObj(local_config, list_values=False)
+            conf_obj.merge(update)
+        else:
+            conf_obj = ConfigObj(update, list_values=False)
+        if conf_obj:
+            with open(local_config, "wb") as fobj:
+                conf_obj.write(fobj)
+
     def init_cache(
         self,
         repo: "Repo",

dvc/repo/experiments/queue/base.py

@@ -4,7 +4,6 @@ from collections.abc import Collection, Generator, Iterable, Mapping
 from dataclasses import asdict, dataclass
 from typing import TYPE_CHECKING, Any, NamedTuple, Optional, Union
 
-from dvc_studio_client.post_live_metrics import get_studio_config
 from funcy import retry
 
 from dvc.dependency import ParamsDependency
@@ -24,6 +23,7 @@ from dvc.repo.experiments.utils import (
 )
 from dvc.utils.objects import cached_property
 from dvc.utils.studio import config_to_env
+from dvc_studio_client.post_live_metrics import get_studio_config
 
 from .utils import get_remote_executor_refs
 
@@ -180,7 +180,7 @@ class BaseStashQueue(ABC):
 
         def _get_timestamp(rev: str) -> datetime:
             commit = self.scm.resolve_commit(rev)
-            return datetime.fromtimestamp(commit.commit_time)
+            return datetime.fromtimestamp(commit.commit_time)  # noqa: DTZ006
 
         def _format_entry(
             entry: QueueEntry,

dvc/repo/experiments/queue/celery.py

@@ -474,10 +474,10 @@ class LocalCeleryQueue(BaseStashQueue):
             self.scm, exp_names, git_remote
         )
         if not git_remote:
-            queue_entry_match: dict[
-                str, Optional["QueueEntry"]
-            ] = self.match_queue_entry_by_name(
-                exp_names, self.iter_queued(), self.iter_done()
+            queue_entry_match: dict[str, Optional["QueueEntry"]] = (
+                self.match_queue_entry_by_name(
+                    exp_names, self.iter_queued(), self.iter_done()
+                )
             )
 
         for exp_name in exp_names:

dvc/repo/experiments/queue/tempdir.py

@@ -125,7 +125,7 @@ class TempDirQueue(WorkspaceQueue):
                 )
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'") from exc
         finally:
             executor.cleanup(infofile)

dvc/repo/experiments/queue/workspace.py

@@ -131,7 +131,7 @@ class WorkspaceQueue(BaseStashQueue):
                 )
         except DvcException:
             raise
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             raise DvcException(f"Failed to reproduce experiment '{rev[:7]}'") from exc
         finally:
             executor.cleanup(infofile)

dvc/repo/experiments/remove.py

@@ -44,9 +44,9 @@ def remove(  # noqa: C901, PLR0912
     exp_ref_list: list["ExpRefInfo"] = []
     queue_entry_list: list["QueueEntry"] = []
     if exp_names:
-        results: dict[
-            str, "ExpRefAndQueueEntry"
-        ] = celery_queue.get_ref_and_entry_by_names(exp_names, git_remote)
+        results: dict[str, "ExpRefAndQueueEntry"] = (
+            celery_queue.get_ref_and_entry_by_names(exp_names, git_remote)
+        )
         remained: list[str] = []
         for name, result in results.items():
             if not result.exp_ref_info and not result.queue_entry:

dvc/repo/experiments/show.py

@@ -306,7 +306,7 @@ def format_time(
         return fill_value
     if iso:
         return timestamp.isoformat()
-    if timestamp.date() == date.today():
+    if timestamp.date() == date.today():  # noqa: DTZ011
         fmt = "%I:%M %p"
     else:
         fmt = "%b %d, %Y"

dvc/repo/experiments/utils.py

@@ -287,7 +287,7 @@ def gen_random_name():
     # fmt: on
     # Use custom random generator to make sure that names are random even if
     # global random seed is set (common for ML pipelines).
-    random_generator = random.Random()
+    random_generator = random.Random()  # noqa: S311
     adjective = random_generator.choice(ADJECTIVES)
     noun = random_generator.choice(NOUNS)
     return f"{adjective}-{noun}"

dvc/repo/fetch.py

@@ -49,14 +49,12 @@ def _collect_indexes(  # noqa: PLR0913
         config["core"] = core
 
     def stage_filter(stage: "Stage") -> bool:
-        if push and stage.is_repo_import:
-            return False
-        return True
+        return not (push and stage.is_repo_import)
 
     def outs_filter(out: "Output") -> bool:
         if push and not out.can_push:
             return False
-        return True
+        return not (remote and out.remote and remote != out.remote)
 
     for rev in repo.brancher(
         revs=revs,

dvc/repo/gc.py

@@ -50,7 +50,7 @@ def _used_obj_ids_not_in_remote(
 
 
 @locked
-def gc(  # noqa: PLR0913, C901
+def gc(  # noqa: C901, PLR0912, PLR0913
     self: "Repo",
     all_branches: bool = False,
     cloud: bool = False,

dvc/repo/metrics/show.py

@@ -1,3 +1,4 @@
+import logging
 import os
 from collections.abc import Iterable, Iterator
 from itertools import chain
@@ -8,7 +9,7 @@ from scmrepo.exceptions import SCMError
 
 from dvc.log import logger
 from dvc.scm import NoSCMError
-from dvc.utils import as_posix, expand_paths
+from dvc.utils import as_posix
 from dvc.utils.collections import ensure_list
 from dvc.utils.serialize import load_path
 
@@ -103,7 +104,7 @@ def _collect_metrics(
     paths = (fs.from_os_path(metric) for metric in metrics)
     # make paths absolute for DVCFileSystem
     repo_paths = (f"{fs.root_marker}{path}" for path in paths)
-    return ldistinct(expand_paths(fs, repo_paths))
+    return ldistinct(try_expand_paths(fs, repo_paths))
 
 
 class FileResult(TypedDict, total=False):
@@ -116,6 +117,22 @@ class Result(TypedDict, total=False):
     error: Exception
 
 
+def try_expand_paths(fs: "FileSystem", paths: Iterable[str]) -> Iterator[str]:
+    for path in paths:
+        try:
+            if fs.isdir(path):
+                yield from fs.find(path)
+                continue
+        except Exception as e:  # noqa: BLE001
+            logger.debug(
+                "failed to expand %r: %s",
+                path,
+                e,
+                exc_info=logger.isEnabledFor(logging.TRACE),  # type: ignore[attr-defined]
+            )
+        yield path
+
+
 def to_relpath(fs: "FileSystem", root_dir: str, d: Result) -> Result:
     relpath = fs.relpath
     cwd = fs.getcwd()
@@ -207,7 +224,7 @@ def show(
                 on_error=on_error,
             )
             res[rev] = Result(data=result)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             if on_error == "raise":
                 raise
 

dvc/repo/params/show.py

@@ -6,9 +6,9 @@ from typing import TYPE_CHECKING, Any, Optional, Union
 
 from dvc.dependency.param import ParamsDependency, read_param_file
 from dvc.log import logger
-from dvc.repo.metrics.show import FileResult, Result
+from dvc.repo.metrics.show import FileResult, Result, try_expand_paths
 from dvc.stage import PipelineStage
-from dvc.utils import as_posix, expand_paths
+from dvc.utils import as_posix
 from dvc.utils.collections import ensure_list
 
 if TYPE_CHECKING:
@@ -77,7 +77,7 @@ def _collect_params(
         path = fs.from_os_path(param)
         # make paths absolute for DVCFileSystem
         repo_path = f"{fs.root_marker}{path}"
-        ret.update(dict.fromkeys(expand_paths(fs, [repo_path]), _params))
+        ret.update(dict.fromkeys(try_expand_paths(fs, [repo_path]), _params))
     return ret
 
 
@@ -191,7 +191,7 @@ def show(
                 on_error=on_error,
             )
             res[rev] = Result(data=params)
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             if on_error == "raise":
                 raise
             logger.warning("failed to load params in revision %r, %s", rev, str(exc))

dvc/repo/plots/__init__.py

@@ -369,11 +369,9 @@ def _matches(targets, config_file, plot_id):
         return True
 
     full_id = get_plot_id(plot_id, config_file)
-    if any(
+    return any(
         (re.match(target, plot_id) or re.match(target, full_id)) for target in targets
-    ):
-        return True
-    return False
+    )
 
 
 def _normpath(path):
@@ -462,11 +460,12 @@ def _resolve_definitions(
                     fs, data_path, props=plot_props | props, onerror=onerror
                 )
                 # use config for parent directory with most specific definition
-                unpacked["data"] = {
-                    k: v
-                    for k, v in unpacked["data"].items()
-                    if _closest_parent(fs, k, plot_ids_parents) == data_path
-                }
+                if unpacked.get("data"):
+                    unpacked["data"] = {
+                        k: v
+                        for k, v in unpacked["data"].items()
+                        if _closest_parent(fs, k, plot_ids_parents) == data_path
+                    }
                 dpath.merge(result, unpacked)
         elif _matches(targets, config_path, plot_id):
             adjusted_props = _adjust_sources(fs, plot_props, config_dir)

dvc/repo/trie.py

@@ -25,15 +25,11 @@ def build_outs_trie(stages):
                 overlapping = out
             if parent and overlapping:
                 msg = (
-                    "The output paths:\n'{}'('{}')\n'{}'('{}')\n"
+                    f"The output paths:\n'{parent!s}'('{parent.stage.addressing}')\n"
+                    f"'{overlapping!s}'('{overlapping.stage.addressing}')\n"
                     "overlap and are thus in the same tracked directory.\n"
                     "To keep reproducibility, outputs should be in separate "
                     "tracked directories or tracked individually."
-                ).format(
-                    str(parent),
-                    parent.stage.addressing,
-                    str(overlapping),
-                    overlapping.stage.addressing,
                 )
                 raise OverlappingOutputPathsError(parent, overlapping, msg)
 

dvc/repo/worktree.py

@@ -76,14 +76,10 @@ def worktree_view(
     """
 
     def stage_filter(stage: "Stage") -> bool:
-        if push and stage.is_repo_import:
-            return False
-        return True
+        return not (push and stage.is_repo_import)
 
     def outs_filter(out: "Output") -> bool:
-        if not out.is_in_repo or not out.use_cache or (push and not out.can_push):
-            return False
-        return True
+        return out.is_in_repo and out.use_cache and (not push or out.can_push)
 
     return index.targets_view(
         targets,

dvc/scm.py

@@ -1,4 +1,5 @@
 """Manages source control systems (e.g. Git)."""
+
 import os
 from collections.abc import Iterator, Mapping
 from contextlib import contextmanager
@@ -70,8 +71,7 @@ def SCM(
     *,
     search_parent_directories: bool = ...,
     no_scm: Literal[False] = ...,
-) -> "Git":
-    ...
+) -> "Git": ...
 
 
 @overload
@@ -80,8 +80,7 @@ def SCM(
     *,
     search_parent_directories: bool = ...,
     no_scm: Literal[True],
-) -> "NoSCM":
-    ...
+) -> "NoSCM": ...
 
 
 @overload
@@ -90,8 +89,7 @@ def SCM(
     *,
     search_parent_directories: bool = ...,
     no_scm: bool = ...,
-) -> Union["Git", "NoSCM"]:
-    ...
+) -> Union["Git", "NoSCM"]: ...
 
 
 def SCM(root_dir, *, search_parent_directories=True, no_scm=False):
@@ -241,7 +239,9 @@ def iter_revs(
         if commit_date:
             from datetime import datetime
 
-            commit_datestamp = datetime.strptime(commit_date, "%Y-%m-%d").timestamp()
+            commit_datestamp = (
+                datetime.strptime(commit_date, "%Y-%m-%d").timestamp()  # noqa: DTZ007
+            )
 
             def _time_filter(rev):
                 try:

dvc/stage/run.py

@@ -39,7 +39,7 @@ def _enforce_cmd_list(cmd):
 
 
 def prepare_kwargs(stage, run_env=None):
-    from dvc.env import DVC_ROOT
+    from dvc.env import DVC_ROOT, DVC_STAGE
 
     kwargs = {"cwd": stage.wdir, "env": fix_env(None), "close_fds": True}
 
@@ -48,6 +48,9 @@ def prepare_kwargs(stage, run_env=None):
     if DVC_ROOT not in kwargs["env"]:
         kwargs["env"][DVC_ROOT] = stage.repo.root_dir
 
+    # Create DVC_STAGE env variable for every command
+    kwargs["env"][DVC_STAGE] = stage.addressing
+
     # NOTE: when you specify `shell=True`, `Popen` [1] will default to
     # `/bin/sh` on *nix and will add ["/bin/sh", "-c"] to your command.
     # But we actually want to run the same shell that we are running

dvc/stage/utils.py

@@ -92,9 +92,7 @@ def check_no_externals(stage):
     from dvc.utils import format_link
 
     def _is_cached_external(out):
-        if out.is_in_repo or not out.use_cache:
-            return False
-        return True
+        return not out.is_in_repo and out.use_cache
 
     outs = [str(out) for out in stage.outs if _is_cached_external(out)]
     if not outs:

dvc/testing/api_tests.py

@@ -33,9 +33,9 @@ class TestAPI:
         [
             {},
             {"url": "{path}"},
-            {"url": "{path}", "rev": "master"},
+            {"url": "{path}", "rev": "{default_branch}"},
             {"url": "file://{posixpath}"},
-            {"url": "file://{posixpath}", "rev": "master"},
+            {"url": "file://{posixpath}", "rev": "{default_branch}"},
         ],
         ids=["current", "local", "local_rev", "git", "git_rev"],
     )
@@ -61,6 +61,8 @@ class TestAPI:
 
         if url := fs_kwargs.get("url"):
             fs_kwargs["url"] = url.format(path=tmp_dir, posixpath=tmp_dir.as_posix())
+        if rev := fs_kwargs.get("rev"):
+            fs_kwargs["rev"] = rev.format(default_branch=scm.active_branch())
 
         fs = DVCFileSystem(**fs_kwargs)
 

dvc/testing/benchmarks/cli/commands/test_get.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-@pytest.mark.flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 def test_get(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
     dataset = make_dataset(
         cache=False, files=False, dvcfile=True, commit=True, remote=True

dvc/testing/benchmarks/cli/commands/test_import.py

@@ -1,7 +1,7 @@
 import pytest
 
 
-@pytest.mark.flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 def test_import(bench_dvc, tmp_dir, scm, dvc, make_dataset, remote):
     dataset = make_dataset(
         cache=False, files=False, dvcfile=True, commit=True, remote=True

dvc/testing/benchmarks/cli/stories/test_modify_data.py

@@ -1,6 +1,7 @@
 """
 Tests for modifications to an existing dataset.
 """
+
 import glob
 import os
 import random

dvc/testing/fixtures.py

@@ -221,15 +221,8 @@ def docker_compose_project_name():
 
 
 @pytest.fixture(scope="session")
-def docker_services(
-    tmp_path_factory,
-    docker_compose_command,
-    docker_compose_file,
-    docker_compose_project_name,
-    docker_setup,
-):
+def docker_services(tmp_path_factory, request):
     from filelock import FileLock
-    from pytest_docker.plugin import DockerComposeExecutor, Services
 
     if os.environ.get("CI") and os.name == "nt":
         pytest.skip("disabled for Windows on CI")
@@ -251,18 +244,12 @@ def docker_services(
         out = (err.output or b"").decode("utf-8")
         pytest.skip(f"docker-compose is not installed: {out}")
 
-    executor = DockerComposeExecutor(
-        docker_compose_command, docker_compose_file, docker_compose_project_name
-    )
-
     # making sure we don't accidentally launch docker-compose in parallel,
     # as it might result in network conflicts. Inspired by:
     # https://github.com/pytest-dev/pytest-xdist#making-session-scoped-fixtures-execute-only-once
     lockfile = tmp_path_factory.getbasetemp().parent / "docker-compose.lock"
     with FileLock(os.fspath(lockfile)):
-        executor.execute(docker_setup)
-        # note: we are not tearing down the containers here
-        return Services(executor)
+        return request.getfixturevalue("docker_services")
 
 
 @pytest.fixture

dvc/utils/__init__.py

@@ -5,7 +5,6 @@ import json
 import os
 import re
 import sys
-from collections.abc import Iterable, Iterator
 from typing import TYPE_CHECKING, Optional
 
 import colorama
@@ -13,8 +12,6 @@ import colorama
 if TYPE_CHECKING:
     from typing import TextIO
 
-    from dvc.fs import FileSystem
-
 
 LARGE_DIR_SIZE = 100
 TARGET_REGEX = re.compile(r"(?P<path>.*?)(:(?P<name>[^\\/:]*))??$")
@@ -410,14 +407,6 @@ def errored_revisions(rev_data: dict) -> list:
     return result
 
 
-def expand_paths(fs: "FileSystem", paths: Iterable[str]) -> Iterator[str]:
-    for path in paths:
-        if fs.isdir(path):
-            yield from fs.find(path)
-        else:
-            yield path
-
-
 def isatty(stream: "Optional[TextIO]") -> bool:
     if stream is None:
         return False

dvc/utils/collections.py

@@ -39,9 +39,7 @@ def apply_diff(src, dest):  # noqa: C901
                     dest[i] = value
     else:
         raise AssertionError(  # noqa: TRY004
-            "Can't apply diff from {} to {}".format(
-                src.__class__.__name__, dest.__class__.__name__
-            )
+            f"Can't apply diff from {type(src).__name__} to {type(dest).__name__}"
         )
 
 

dvc/utils/serialize/_common.py

@@ -1,4 +1,5 @@
 """Common utilities for serialize."""
+
 import os
 from contextlib import AbstractContextManager, contextmanager
 from typing import TYPE_CHECKING, Any, Callable, Optional, Protocol, TextIO, Union
@@ -15,25 +16,21 @@ if TYPE_CHECKING:
 class DumperFn(Protocol):
     def __call__(
         self, path: "StrPath", data: Any, fs: Optional["FileSystem"] = None
-    ) -> Any:
-        ...
+    ) -> Any: ...
 
 
 class DumpersFn(Protocol):
-    def __call__(self, data: Any, stream: TextIO) -> Any:
-        ...
+    def __call__(self, data: Any, stream: TextIO) -> Any: ...
 
 
 class ModifierFn(Protocol):
     def __call__(
         self, path: "StrPath", fs: Optional["FileSystem"] = None
-    ) -> AbstractContextManager[dict]:
-        ...
+    ) -> AbstractContextManager[dict]: ...
 
 
 class LoaderFn(Protocol):
-    def __call__(self, path: "StrPath", fs: Optional["FileSystem"] = None) -> Any:
-        ...
+    def __call__(self, path: "StrPath", fs: Optional["FileSystem"] = None) -> Any: ...
 
 
 ReadType = Union[bytes, None, str]

dvc/utils/strictyaml.py

@@ -6,6 +6,7 @@ Used for parsing dvc.yaml, dvc.lock and .dvc files.
 
 Not to be confused with strictyaml, a python library with similar motivations.
 """
+
 import re
 import typing
 from contextlib import suppress

dvc/utils/studio.py

@@ -1,3 +1,4 @@
+import os
 from typing import TYPE_CHECKING, Any, Optional
 from urllib.parse import urljoin
 
@@ -125,3 +126,19 @@ def get_subrepo_relpath(repo: "Repo") -> str:
     relpath = as_posix(repo.fs.relpath(repo.root_dir, scm_root_dir))
 
     return "" if relpath == "." else relpath
+
+
+def get_repo_url(repo: "Repo") -> str:
+    from dulwich.porcelain import get_remote_repo
+
+    from dvc.env import DVC_EXP_GIT_REMOTE
+
+    repo_url = os.getenv(
+        DVC_EXP_GIT_REMOTE, repo.config.get("exp", {}).get("git_remote")
+    )
+    if repo_url:
+        try:
+            _, repo_url = get_remote_repo(repo.scm.dulwich.repo, repo_url)
+        except IndexError:
+            pass
+    return repo_url

pyproject.toml

@@ -31,24 +31,30 @@ classifiers = [
 dynamic = ["version"]
 
 dependencies = [
+    "attrs>=22.2.0",
+    "celery",
     "colorama>=0.3.9",
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=3.13,<3.15",
+    "dulwich",
+    "dvc-data>=3.15,<3.16",
     "dvc-http>=2.29.0",
+    "dvc-objects",
     "dvc-render>=1.0.1,<2",
     "dvc-studio-client>=0.20,<1",
     "dvc-task>=0.3.0,<1",
     "flatten_dict<1,>=0.4.1",
-    # https://github.com/iterative/dvc/issues/9654
-    "flufl.lock>=5,<8",
+    "flufl.lock>=5,<8", # https://github.com/iterative/dvc/issues/9654
+    "fsspec",
     "funcy>=1.14",
     "grandalf<1,>=0.7",
     "gto>=1.6.0,<2",
     "hydra-core>=1.1",
     "iterative-telemetry>=0.0.7",
+    "kombu",
     "networkx>=2.5",
+    "omegaconf",
     "packaging>=19",
     "pathspec>=0.10.3",
     "platformdirs<4,>=3.1.1",
@@ -59,7 +65,7 @@ dependencies = [
     "requests>=2.22",
     "rich>=12",
     "ruamel.yaml>=0.17.11",
-    "scmrepo>=3,<4",
+    "scmrepo>=3.3.2,<4",
     "shortuuid>=0.5",
     "shtab<2,>=1.3.4",
     "tabulate>=0.8.7",
@@ -77,7 +83,7 @@ gdrive = ["dvc-gdrive>=3,<4"]
 gs = ["dvc-gs>=3,<4"]
 hdfs = ["dvc-hdfs>=3,<4"]
 lint = [
-    "mypy==1.8.0",
+    "mypy==1.10.0",
     "pandas-stubs",
     "types-colorama",
     "types-psutil",
@@ -89,7 +95,7 @@ lint = [
     "typing-extensions",
 ]
 oss = ["dvc-oss>=3,<4"]
-s3 = ["dvc-s3>=3,<4"]
+s3 = ["dvc-s3>=3.2.0,<4"]
 ssh = ["dvc-ssh>=4,<5"]
 ssh_gssapi = ["dvc-ssh[gssapi]>=4,<5"]
 testing = [
@@ -102,8 +108,8 @@ tests = [
     "beautifulsoup4>=4.4",
     "dvc-ssh",
     "filelock",
-    "flaky",
     "pytest>=7,<9",
+    "pytest-rerunfailures",
     "pytest-cov>=4.1.0",
     "pytest-docker>=1,<4",
     "pytest-mock",
@@ -261,10 +267,10 @@ ignore = [
     "TRY003", "TRY300", "PLR2004", "PLW2901", "LOG007",
 ]
 select = [
-    "F", "E", "W", "C90", "I", "N", "UP", "YTT", "ASYNC", "S", "BLE", "B", "A", "C4", "T10",
+    "F", "E", "W", "C90", "I", "N", "UP", "YTT", "ASYNC", "S", "BLE", "B", "A", "C4", "DTZ", "T10",
     "EXE", "ISC", "ICN", "G", "INP", "PIE", "T20", "PYI", "PT", "Q", "RSE", "RET",
     "SLOT", "SIM", "TID", "TCH", "ARG", "PGH", "PLC", "PLE", "PLR", "PLW", "TRY",
-    "FLY", "PERF101", "LOG", "RUF", "RUF022", "RUF023", "RUF024", "RUF025", "RUF026",
+    "FLY", "PERF101", "LOG", "RUF", "RUF022", "RUF023", "RUF024", "RUF025", "RUF026", "RUF027"
 ]
 preview = true
 explicit-preview-rules = true
@@ -286,7 +292,7 @@ strict = true
 ignore-variadic-names = true
 
 [tool.ruff.lint.isort]
-known-first-party = ["dvc", "dvc_data", "dvc_objects", "dvc_render", "dvc_task", "tests"]
+known-first-party = ["dvc", "dvc_*", "tests"]
 
 [tool.ruff.lint.pep8-naming]
 extend-ignore-names = ["M", "SCM"]

tests/conftest.py

@@ -123,7 +123,7 @@ def pytest_runtest_setup(item):
     if "CI" in os.environ and item.get_closest_marker("needs_internet") is not None:
         # remotes that need internet connection might be flaky,
         # so we rerun them in case it fails.
-        item.add_marker(pytest.mark.flaky(max_runs=5, min_passes=1))
+        item.add_marker(pytest.mark.flaky(reruns=5))
 
 
 @pytest.fixture(scope="session")

tests/dir_helpers.py

@@ -43,7 +43,6 @@ from global repo template to creating everything inplace, which:
     - does not create unnecessary files
 """
 
-
 import os
 
 import pytest

tests/func/api/test_artifacts.py

@@ -0,0 +1,74 @@
+from os.path import join, normpath
+
+import pytest
+
+from dvc.api import artifacts_show
+from dvc.testing.tmp_dir import make_subrepo
+from dvc.utils import as_posix
+from tests.func.artifacts.test_artifacts import get_tag_and_name, make_artifact
+
+
+@pytest.mark.parametrize("sub", ["sub", ""])
+def test_artifacts_show(tmp_dir, dvc, scm, sub):
+    subdir = tmp_dir / sub
+
+    dirname = str(subdir.relative_to(tmp_dir))
+    tag, name = get_tag_and_name(as_posix(dirname), "myart", "v2.0.0")
+    make_artifact(tmp_dir, "myart", tag, subdir / "myart.pkl")
+
+    assert artifacts_show(name) == {
+        "path": normpath(join(dirname, "myart.pkl")),
+        "rev": scm.get_rev(),
+    }
+    assert artifacts_show(name, repo=tmp_dir.fs_path) == {
+        "path": normpath(join(dirname, "myart.pkl")),
+        "rev": scm.get_rev(),
+    }
+    assert artifacts_show(name, repo=f"file://{tmp_dir.as_posix()}") == {
+        "path": normpath(join(dirname, "myart.pkl")),
+        "rev": scm.get_rev(),
+    }
+
+    assert artifacts_show(name, repo=subdir.fs_path) == {
+        "path": normpath(join(dirname, "myart.pkl")),
+        "rev": scm.get_rev(),
+    }
+    with subdir.chdir():
+        assert artifacts_show(name) == {
+            "path": normpath(join(dirname, "myart.pkl")),
+            "rev": scm.get_rev(),
+        }
+
+
+@pytest.mark.parametrize("sub", ["sub", ""])
+def test_artifacts_show_subrepo(tmp_dir, scm, sub):
+    subrepo = tmp_dir / "subrepo"
+    make_subrepo(subrepo, scm)
+    subdir = subrepo / sub
+
+    dirname = str(subdir.relative_to(tmp_dir))
+    tag, name = get_tag_and_name(as_posix(dirname), "myart", "v2.0.0")
+    make_artifact(subrepo, "myart", tag, subdir / "myart.pkl")
+
+    assert artifacts_show(name) == {
+        "path": join(dirname, "myart.pkl"),
+        "rev": scm.get_rev(),
+    }
+    assert artifacts_show(name, repo=tmp_dir.fs_path) == {
+        "path": join(dirname, "myart.pkl"),
+        "rev": scm.get_rev(),
+    }
+    assert artifacts_show(name, repo=f"file://{tmp_dir.as_posix()}") == {
+        "path": join(dirname, "myart.pkl"),
+        "rev": scm.get_rev(),
+    }
+
+    assert artifacts_show(name, repo=subdir.fs_path) == {
+        "path": str((subdir / "myart.pkl").relative_to(subrepo)),
+        "rev": scm.get_rev(),
+    }
+    with subdir.chdir():
+        assert artifacts_show(name) == {
+            "path": str((subdir / "myart.pkl").relative_to(subrepo)),
+            "rev": scm.get_rev(),
+        }

tests/func/api/test_data.py

@@ -31,6 +31,40 @@ def test_get_url_requires_dvc(tmp_dir, scm):
         api.get_url("foo", repo=f"file://{tmp_dir.as_posix()}")
 
 
+def test_get_url_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):
+    erepo_dir.add_remote(config=cloud.config, name="other")
+    erepo_dir.add_remote(config=local_cloud.config, default=True)
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen("foo", "foo", commit="add foo")
+
+    # Using file url to force clone to tmp repo
+    repo_url = f"file://{erepo_dir.as_posix()}"
+    expected_rel_path = os.path.join(
+        "files", "md5", "ac/bd18db4cc2f85cedef654fccc4a4d8"
+    )
+
+    # Test default remote
+    assert api.get_url("foo", repo=repo_url) == (local_cloud / expected_rel_path).url
+
+    # Test remote arg
+    assert (
+        api.get_url("foo", repo=repo_url, remote="other")
+        == (cloud / expected_rel_path).url
+    )
+
+    # Test config arg
+    assert (
+        api.get_url("foo", repo=repo_url, config={"core": {"remote": "other"}})
+        == (cloud / expected_rel_path).url
+    )
+
+    # Test remote_config arg
+    assert (
+        api.get_url("foo", repo=repo_url, remote_config={"url": cloud.url})
+        == (cloud / expected_rel_path).url
+    )
+
+
 def test_open_external(tmp_dir, erepo_dir, cloud):
     erepo_dir.add_remote(config=cloud.config)
 
@@ -253,3 +287,12 @@ def test_read_from_remote(tmp_dir, erepo_dir, cloud, local_cloud):
         )
         == "foo content"
     )
+
+    assert (
+        api.read(
+            os.path.join("dir", "foo"),
+            repo=f"file://{erepo_dir.as_posix()}",
+            remote_config={"url": cloud.url},
+        )
+        == "foo content"
+    )

tests/func/artifacts/test_artifacts.py

@@ -6,7 +6,9 @@ import pytest
 
 from dvc.annotations import Artifact
 from dvc.exceptions import ArtifactNotFoundError, InvalidArgumentError
-from dvc.repo.artifacts import check_name_format
+from dvc.repo.artifacts import Artifacts, check_name_format
+from dvc.testing.tmp_dir import make_subrepo
+from dvc.utils import as_posix
 from dvc.utils.strictyaml import YAMLSyntaxError, YAMLValidationError
 
 dvcyaml = {
@@ -185,7 +187,7 @@ def test_get_rev(tmp_dir, dvc, scm):
         dvc.artifacts.get_rev("myart", stage="prod")
 
 
-def test_get_path(tmp_dir, dvc):
+def test_get_path(tmp_dir, dvc, scm):
     (tmp_dir / "dvc.yaml").dump(dvcyaml)
     subdir = tmp_dir / "subdir"
     subdir.mkdir()
@@ -196,3 +198,86 @@ def test_get_path(tmp_dir, dvc):
     assert dvc.artifacts.get_path("subdir/dvc.yaml:myart") == os.path.join(
         "subdir", "myart.pkl"
     )
+
+
+def test_parametrized(tmp_dir, dvc):
+    (tmp_dir / "params.yaml").dump({"path": "myart.pkl"})
+    (tmp_dir / "dvc.yaml").dump(
+        {"artifacts": {"myart": {"type": "model", "path": "${path}"}}}
+    )
+    assert tmp_dir.dvc.artifacts.read() == {
+        "dvc.yaml": {"myart": Artifact(path="myart.pkl", type="model")}
+    }
+
+
+def test_get_path_subrepo(tmp_dir, scm, dvc):
+    subrepo = tmp_dir / "subrepo"
+    make_subrepo(subrepo, scm)
+    (subrepo / "dvc.yaml").dump(dvcyaml)
+
+    assert dvc.artifacts.get_path("subrepo:myart") == os.path.join(
+        "subrepo", "myart.pkl"
+    )
+    assert dvc.artifacts.get_path("subrepo/dvc.yaml:myart") == os.path.join(
+        "subrepo", "myart.pkl"
+    )
+
+    assert subrepo.dvc.artifacts.get_path("subrepo:myart") == os.path.join(
+        "subrepo", "myart.pkl"
+    )
+    assert subrepo.dvc.artifacts.get_path("subrepo/dvc.yaml:myart") == os.path.join(
+        "subrepo", "myart.pkl"
+    )
+
+
+def get_tag_and_name(dirname, name, version):
+    tagname = f"{name}@{version}"
+    if dirname in (os.curdir, ""):
+        return tagname, name
+    return f"{dirname}={tagname}", f"{dirname}:{name}"
+
+
+def make_artifact(tmp_dir, name, tag, path) -> Artifact:
+    artifact = Artifact(path=path.name, type="model")
+    dvcfile = path.with_name("dvc.yaml")
+
+    tmp_dir.scm_gen(path, "hello_world", commit="add myart.pkl")
+    tmp_dir.dvc.artifacts.add(name, artifact, dvcfile=os.fspath(dvcfile))
+    tmp_dir.scm.add_commit([dvcfile], message="add dvc.yaml")
+    tmp_dir.scm.tag(tag, annotated=True, message="foo")
+    return artifact
+
+
+@pytest.mark.parametrize("sub", ["sub", ""])
+def test_artifacts_download(tmp_dir, dvc, scm, sub):
+    subdir = tmp_dir / sub
+    dirname = str(subdir.relative_to(tmp_dir))
+    tag, name = get_tag_and_name(as_posix(dirname), "myart", "v2.0.0")
+    make_artifact(tmp_dir, "myart", tag, subdir / "myart.pkl")
+
+    result = (1, "myart.pkl")
+    assert Artifacts.get(".", name, force=True) == result
+    assert Artifacts.get(tmp_dir.fs_path, name, force=True) == result
+    assert Artifacts.get(f"file://{tmp_dir.as_posix()}", name, force=True) == result
+    assert Artifacts.get(subdir.fs_path, name, force=True) == result
+    with subdir.chdir():
+        assert Artifacts.get(".", name, force=True) == result
+
+
+@pytest.mark.parametrize("sub", ["sub", ""])
+def test_artifacts_download_subrepo(tmp_dir, scm, sub):
+    subrepo = tmp_dir / "subrepo"
+    make_subrepo(subrepo, scm)
+    subdir = subrepo / sub
+
+    dirname = str(subdir.relative_to(tmp_dir))
+    tag, name = get_tag_and_name(as_posix(dirname), "myart", "v2.0.0")
+    make_artifact(subrepo, "myart", tag, subdir / "myart.pkl")
+
+    result = (1, "myart.pkl")
+    assert Artifacts.get(".", name) == result
+    assert Artifacts.get(tmp_dir.fs_path, name, force=True) == result
+    assert Artifacts.get(f"file://{tmp_dir.as_posix()}", name, force=True) == result
+    assert Artifacts.get(subdir.fs_path, name, force=True) == result
+    with subdir.chdir():
+        assert Artifacts.get(".", name, force=True) == result

tests/func/experiments/test_remote.py

@@ -372,3 +372,68 @@ def test_push_pull_invalid_workspace(
         dvc.experiments.push(git_upstream.remote, push_cache=True)
         dvc.experiments.pull(git_upstream.remote, pull_cache=True)
         assert "failed to collect" not in caplog.text
+
+
+@pytest.mark.parametrize(
+    "auto_push, expected_key", [(True, "up_to_date"), (False, "success")]
+)
+def test_auto_push_on_run(
+    tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage, auto_push, expected_key
+):
+    remote = git_upstream.remote
+
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = auto_push
+        conf["exp"]["git_remote"] = remote
+
+    exp_name = "foo"
+    dvc.experiments.run(exp_stage.addressing, params=["foo=2"], name=exp_name)
+
+    assert first(dvc.experiments.push(name=exp_name, git_remote=remote)) == expected_key
+
+
+@pytest.mark.parametrize(
+    "auto_push, expected_key", [(True, "up_to_date"), (False, "success")]
+)
+def test_auto_push_on_save(
+    tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage, auto_push, expected_key
+):
+    remote = git_upstream.remote
+    exp_name = "foo"
+    dvc.experiments.run(exp_stage.addressing, params=["foo=2"], name=exp_name)
+
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = auto_push
+        conf["exp"]["git_remote"] = remote
+
+    dvc.experiments.save(name=exp_name, force=True)
+
+    assert first(dvc.experiments.push(name=exp_name, git_remote=remote)) == expected_key
+
+
+def test_auto_push_misconfigured(
+    tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage, caplog
+):
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = True
+        conf["exp"]["git_remote"] = "notfound"
+
+    exp_name = "foo"
+    with caplog.at_level(logging.WARNING, logger="dvc"):
+        dvc.experiments.run(exp_stage.addressing, params=["foo=2"], name=exp_name)
+        assert "Failed to validate remotes" in caplog.text
+
+
+def test_auto_push_tmp_dir(tmp_dir, scm, dvc, git_upstream, local_remote, exp_stage):
+    remote = git_upstream.remote
+
+    with dvc.config.edit() as conf:
+        conf["exp"]["auto_push"] = True
+        conf["exp"]["git_remote"] = remote
+
+    exp_name = "foo"
+    dvc.experiments.run(
+        exp_stage.addressing, params=["foo=2"], name=exp_name, tmp_dir=True
+    )
+
+    assert first(dvc.experiments.push(name=exp_name, git_remote=remote)) == "up_to_date"

tests/func/experiments/test_show.py

@@ -101,7 +101,7 @@ def test_show_simple(tmp_dir, scm, dvc, exp_stage):
 @pytest.mark.parametrize("workspace", [True, False])
 def test_show_experiment(tmp_dir, scm, dvc, exp_stage, workspace):
     baseline_rev = scm.get_rev()
-    timestamp = datetime.fromtimestamp(
+    timestamp = datetime.fromtimestamp(  # noqa: DTZ006
         scm.gitpython.repo.rev_parse(baseline_rev).committed_date
     )
 
@@ -423,7 +423,7 @@ def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
     baseline_rev = scm.get_rev()
 
     def _get_rev_isotimestamp(rev):
-        return datetime.fromtimestamp(
+        return datetime.fromtimestamp(  # noqa: DTZ006
             scm.gitpython.repo.rev_parse(rev).committed_date
         ).isoformat()
 
@@ -453,15 +453,11 @@ def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
         in cap.out
     )
     assert (
-        "{},{},branch_base,{},,2,2,{}".format(
-            ref_info1.name, rev1[:7], _get_rev_isotimestamp(rev1), data_hash
-        )
+        f"{ref_info1.name},{rev1[:7]},branch_base,{_get_rev_isotimestamp(rev1)},,2,2,{data_hash}"
         in cap.out
     )
     assert (
-        "{},{},branch_commit,{},,3,3,{}".format(
-            ref_info2.name, rev2[:7], _get_rev_isotimestamp(rev2), data_hash
-        )
+        f"{ref_info2.name},{rev2[:7]},branch_commit,{_get_rev_isotimestamp(rev2)},,3,3,{data_hash}"
         in cap.out
     )
 
@@ -656,7 +652,7 @@ def test_metrics_renaming(tmp_dir, dvc, scm, capsys, copy_script):
     cap = capsys.readouterr()
 
     def _get_rev_isotimestamp(rev):
-        return datetime.fromtimestamp(
+        return datetime.fromtimestamp(  # noqa: DTZ006
             scm.gitpython.repo.rev_parse(rev).committed_date
         ).isoformat()
 

tests/func/metrics/test_show.py

@@ -1,5 +1,6 @@
 import json
 import os
+import shutil
 from os.path import join
 
 import pytest
@@ -9,8 +10,10 @@ from dvc.cli import main
 from dvc.dvcfile import PROJECT_FILE
 from dvc.exceptions import OverlappingOutputPathsError
 from dvc.repo import Repo
+from dvc.repo.metrics.show import FileResult, Result
 from dvc.utils.fs import remove
 from dvc.utils.serialize import JSONFileCorruptedError
+from dvc_data.index import DataIndexDirError
 
 
 def test_show_simple(tmp_dir, dvc, run_copy_metrics):
@@ -342,3 +345,27 @@ def test_cached_metrics(tmp_dir, dvc, scm, remote):
             }
         }
     }
+
+
+def test_top_level_parametrized(tmp_dir, dvc):
+    tmp_dir.dvc_gen("metrics.yaml", "foo: 3\nbar: 10")
+    (tmp_dir / "params.yaml").dump({"metric_file": "metrics.yaml"})
+    (tmp_dir / "dvc.yaml").dump({"metrics": ["${metric_file}"]})
+    assert dvc.metrics.show() == {
+        "": {"data": {"metrics.yaml": {"data": {"foo": 3, "bar": 10}}}}
+    }
+
+
+def test_metric_in_a_tracked_directory_with_missing_dir_file(M, tmp_dir, dvc):
+    tmp_dir.dvc_gen({"dir": {"file": "2"}})
+    (tmp_dir / "dvc.yaml").dump({"metrics": [join("dir", "file")]})
+    shutil.rmtree(tmp_dir / "dir")  # remove from workspace
+    dvc.cache.local.clear()  # remove .dir file
+
+    assert dvc.metrics.show() == {
+        "": Result(
+            data={
+                join("dir", "file"): FileResult(error=M.instance_of(DataIndexDirError)),
+            }
+        )
+    }

tests/func/params/test_show.py

@@ -1,9 +1,12 @@
+import shutil
 from os.path import join
 
 import pytest
 
+from dvc.dvcfile import PROJECT_FILE
 from dvc.repo import Repo
-from dvc.repo.stage import PROJECT_FILE
+from dvc.repo.metrics.show import FileResult, Result
+from dvc_data.index import DataIndexDirError
 
 
 def test_show_empty(dvc):
@@ -178,3 +181,32 @@ def test_cached_params(tmp_dir, dvc, scm, remote):
             }
         }
     }
+
+
+def test_top_level_parametrized(tmp_dir, dvc):
+    (tmp_dir / "param.json").dump({"foo": 3, "bar": 10})
+    (tmp_dir / "params.yaml").dump({"param_file": "param.json"})
+    (tmp_dir / "dvc.yaml").dump({"params": ["${param_file}"]})
+    assert dvc.params.show() == {
+        "": {
+            "data": {
+                "param.json": {"data": {"foo": 3, "bar": 10}},
+                "params.yaml": {"data": {"param_file": "param.json"}},
+            }
+        }
+    }
+
+
+def test_param_in_a_tracked_directory_with_missing_dir_file(M, tmp_dir, dvc):
+    tmp_dir.dvc_gen({"dir": {"file": "2"}})
+    (tmp_dir / "dvc.yaml").dump({"params": [join("dir", "file")]})
+    shutil.rmtree(tmp_dir / "dir")  # remove from workspace
+    dvc.cache.local.clear()  # remove .dir file
+
+    assert dvc.params.show() == {
+        "": Result(
+            data={
+                join("dir", "file"): FileResult(error=M.instance_of(DataIndexDirError)),
+            }
+        )
+    }

tests/func/parsing/test_errors.py

@@ -1,6 +1,5 @@
 """Negative tests for the parametrization."""
 
-
 import logging
 import re
 

tests/func/parsing/test_foreach.py

@@ -1,4 +1,5 @@
 """Testing happy paths for the foreach."""
+
 import os
 
 import pytest

tests/func/parsing/test_top_level.py

@@ -0,0 +1,168 @@
+from dvc.parsing import DataResolver
+
+
+def test_params(tmp_dir, dvc):
+    (tmp_dir / "params.yaml").dump(
+        {"params": {"param1": "params.json", "param2": "params.toml"}}
+    )
+
+    template = {"params": ["${params.param1}", "param11", "${params.param2}"]}
+    resolver = DataResolver(dvc, tmp_dir, template)
+    assert resolver.resolve_params() == ["params.json", "param11", "params.toml"]
+
+
+def test_metrics(tmp_dir, dvc):
+    (tmp_dir / "params.yaml").dump(
+        {"metrics": {"metric1": "metrics.json", "metric2": "metrics.toml"}}
+    )
+
+    template = {"metrics": ["${metrics.metric1}", "metric11", "${metrics.metric2}"]}
+    resolver = DataResolver(dvc, tmp_dir, template)
+    assert resolver.resolve_metrics() == ["metrics.json", "metric11", "metrics.toml"]
+
+
+def test_plots(tmp_dir, dvc):
+    template = {
+        "plots": [
+            {
+                "${plots.plot1_name}": {
+                    "x": "${plots.x_cls}",
+                    "y": {
+                        "train_classes.csv": "${plots.y_train_cls}",
+                        "test_classes.csv": [
+                            "${plots.y_train_cls}",
+                            "${plots.y_test_cls}",
+                        ],
+                    },
+                    "title": "Compare test vs train confusion matrix",
+                    "template": "confusion",
+                    "x_label": "Actual class",
+                    "y_label": "Predicted class",
+                }
+            },
+            {"eval/importance2.png": None},
+            {"${plots.plot3_name}": None},
+            "eval/importance4.png",
+            "${plots.plot5_name}",
+        ],
+    }
+
+    (tmp_dir / "params.yaml").dump(
+        {
+            "plots": {
+                "x_cls": "actual_class",
+                "y_train_cls": "predicted_class",
+                "y_test_cls": "predicted_class2",
+                "plot1_name": "eval/importance1.png",
+                "plot3_name": "eval/importance3.png",
+                "plot5_name": "eval/importance5.png",
+            }
+        }
+    )
+    resolver = DataResolver(dvc, tmp_dir, template)
+    assert resolver.resolve_plots() == [
+        {
+            "eval/importance1.png": {
+                "x": "actual_class",
+                "y": {
+                    "train_classes.csv": "predicted_class",
+                    "test_classes.csv": ["predicted_class", "predicted_class2"],
+                },
+                "title": "Compare test vs train confusion matrix",
+                "template": "confusion",
+                "x_label": "Actual class",
+                "y_label": "Predicted class",
+            }
+        },
+        {"eval/importance2.png": None},
+        {"eval/importance3.png": None},
+        "eval/importance4.png",
+        "eval/importance5.png",
+    ]
+
+
+def test_artifacts(tmp_dir, dvc):
+    template = {
+        "artifacts": {
+            "${artifacts.name}": {
+                "path": "${artifacts.path}",
+                "type": "model",
+                "desc": "CV classification model, ResNet50",
+                "labels": ["${artifacts.label1}", "${artifacts.label2}"],
+                "meta": {"framework": "${artifacts.framework}"},
+            }
+        }
+    }
+
+    (tmp_dir / "params.yaml").dump(
+        {
+            "artifacts": {
+                "name": "cv-classification",
+                "path": "models/resnet.pt",
+                "label1": "resnet50",
+                "label2": "classification",
+                "framework": "pytorch",
+            }
+        }
+    )
+
+    resolver = DataResolver(dvc, tmp_dir, template)
+    assert resolver.resolve_artifacts() == {
+        "cv-classification": {
+            "path": "models/resnet.pt",
+            "type": "model",
+            "desc": "CV classification model, ResNet50",
+            "labels": ["resnet50", "classification"],
+            "meta": {"framework": "pytorch"},
+        }
+    }
+
+
+def test_datasets(tmp_dir, dvc):
+    template = {
+        "datasets": [
+            {"name": "${ds1.name}", "url": "${ds1.url}", "type": "dvcx"},
+            {
+                "name": "${ds2.name}",
+                "url": "${ds2.url}",
+                "type": "dvc",
+                "path": "${ds2.path}",
+            },
+            {
+                "name": "${ds3.name}",
+                "url": "${ds3.url}",
+                "type": "url",
+            },
+        ]
+    }
+
+    (tmp_dir / "params.yaml").dump(
+        {
+            "ds1": {"name": "dogs", "url": "dvcx://dogs"},
+            "ds2": {
+                "name": "example-get-started",
+                "url": "[email protected]:iterative/example-get-started.git",
+                "path": "path",
+            },
+            "ds3": {
+                "name": "cloud-versioning-demo",
+                "url": "s3://cloud-versioning-demo",
+            },
+        }
+    )
+
+    resolver = DataResolver(dvc, tmp_dir, template)
+    assert resolver.resolve_datasets() == [
+        {"name": "dogs", "url": "dvcx://dogs", "type": "dvcx"},
+        {
+            "name": "example-get-started",
+            "url": "[email protected]:iterative/example-get-started.git",
+            "type": "dvc",
+            "path": "path",
+        },
+        {
+            "name": "cloud-versioning-demo",
+            "url": "s3://cloud-versioning-demo",
+            "type": "url",
+        },
+    ]

tests/func/plots/test_show.py

@@ -500,3 +500,73 @@ def test_show_plots_defined_with_native_os_path(tmp_dir, dvc, scm, capsys):
     json_data = json_out["data"]
     assert json_data[f"{top_level_plot}"]
     assert json_data[stage_plot]
+
+
+@pytest.mark.parametrize(
+    "plot_config,expanded_config,expected_datafiles",
+    [
+        (
+            {
+                "comparison": {
+                    "x": {"${data1}": "${a}"},
+                    "y": {"sub/dir/data2.json": "${b}"},
+                }
+            },
+            {
+                "comparison": {
+                    "x": {"data1.json": "a"},
+                    "y": {"sub/dir/data2.json": "b"},
+                }
+            },
+            ["data1.json", os.path.join("sub", "dir", "data2.json")],
+        ),
+        (
+            {"${data1}": None},
+            {"data1.json": {}},
+            ["data1.json"],
+        ),
+        (
+            "${data1}",
+            {"data1.json": {}},
+            ["data1.json"],
+        ),
+    ],
+)
+def test_top_level_parametrized(
+    tmp_dir, dvc, plot_config, expanded_config, expected_datafiles
+):
+    (tmp_dir / "params.yaml").dump(
+        {"data1": "data1.json", "a": "a", "b": "b", "c": "c"}
+    )
+    data = {
+        "data1.json": [
+            {"a": 1, "b": 0.1, "c": 0.01},
+            {"a": 2, "b": 0.2, "c": 0.02},
+        ],
+        os.path.join("sub", "dir", "data.json"): [
+            {"a": 6, "b": 0.6, "c": 0.06},
+            {"a": 7, "b": 0.7, "c": 0.07},
+        ],
+    }
+
+    for filename, content in data.items():
+        dirname = os.path.dirname(filename)
+        if dirname:
+            os.makedirs(dirname)
+        (tmp_dir / filename).dump_json(content, sort_keys=True)
+
+    config_file = "dvc.yaml"
+    with modify_yaml(config_file) as dvcfile_content:
+        dvcfile_content["plots"] = [plot_config]
+
+    result = dvc.plots.show()
+
+    assert expanded_config == get_plot(
+        result, "workspace", typ="definitions", file=config_file
+    )
+
+    for filename, content in data.items():
+        if filename in expected_datafiles:
+            assert content == get_plot(result, "workspace", file=filename)
+        else:
+            assert filename not in get_plot(result, "workspace")

tests/func/test_data_cloud.py

@@ -1,14 +1,15 @@
 import logging
 import os
 import shutil
+from os.path import join
 
 import pytest
-from flaky.flaky_decorator import flaky
 
 import dvc_data
 from dvc.cli import main
 from dvc.exceptions import CheckoutError
 from dvc.repo.open_repo import clean_repos
+from dvc.scm import CloneError
 from dvc.stage.exceptions import StageNotFound
 from dvc.testing.remote_tests import TestRemote  # noqa: F401
 from dvc.utils.fs import remove
@@ -207,7 +208,7 @@ def test_verify_hashes(tmp_dir, scm, dvc, mocker, tmp_path_factory, local_remote
     assert hash_spy.call_count == 10
 
 
-@flaky(max_runs=3, min_passes=1)
+@pytest.mark.flaky(reruns=3)
 @pytest.mark.parametrize("erepo_type", ["git_dir", "erepo_dir"])
 def test_pull_git_imports(request, tmp_dir, dvc, scm, erepo_type):
     erepo = request.getfixturevalue(erepo_type)
@@ -581,6 +582,58 @@ def test_target_remote(tmp_dir, dvc, make_remote):
     }
 
 
+def test_output_target_remote(tmp_dir, dvc, make_remote):
+    make_remote("default", default=True)
+    make_remote("for_foo", default=False)
+    make_remote("for_bar", default=False)
+
+    tmp_dir.dvc_gen("foo", "foo")
+    tmp_dir.dvc_gen("bar", "bar")
+    tmp_dir.dvc_gen("data", {"one": "one", "two": "two"})
+
+    with (tmp_dir / "foo.dvc").modify() as d:
+        d["outs"][0]["remote"] = "for_foo"
+
+    with (tmp_dir / "bar.dvc").modify() as d:
+        d["outs"][0]["remote"] = "for_bar"
+
+    # push foo and data to for_foo remote
+    dvc.push(remote="for_foo")
+
+    default = dvc.cloud.get_remote_odb("default")
+    for_foo = dvc.cloud.get_remote_odb("for_foo")
+    for_bar = dvc.cloud.get_remote_odb("for_bar")
+
+    # hashes for foo and data, but not bar
+    expected = {
+        "acbd18db4cc2f85cedef654fccc4a4d8",
+        "f97c5d29941bfb1b2fdab0874906ab82",
+        "6b18131dc289fd37006705affe961ef8.dir",
+        "b8a9f715dbb64fd5c56e7783c6820a61",
+    }
+
+    assert set(default.all()) == set()
+    assert set(for_foo.all()) == expected
+    assert set(for_bar.all()) == set()
+
+    # push everything without specifying remote
+    dvc.push()
+    assert set(default.all()) == {
+        "f97c5d29941bfb1b2fdab0874906ab82",
+        "6b18131dc289fd37006705affe961ef8.dir",
+        "b8a9f715dbb64fd5c56e7783c6820a61",
+    }
+    assert set(for_foo.all()) == expected
+    assert set(for_bar.all()) == {"37b51d194a7513e45b56f6524f2d51f2"}
+
+    clean(["foo", "bar", "data"], dvc)
+
+    # pull foo and data from for_foo remote
+    dvc.pull(remote="for_foo", allow_missing=True)
+
+    assert set(dvc.cache.local.all()) == expected
+
+
 def test_pull_allow_missing(tmp_dir, dvc, local_remote):
     dvc.stage.add(name="bar", outs=["bar"], cmd="echo bar > bar")
 
@@ -593,3 +646,35 @@ def test_pull_allow_missing(tmp_dir, dvc, local_remote):
 
     stats = dvc.pull(allow_missing=True)
     assert stats["fetched"] == 1
+
+
+def test_pull_granular_excluding_import_that_cannot_be_pulled(
+    tmp_dir, dvc, local_remote, mocker
+):
+    """Regression test for https://github.com/iterative/dvc/issues/10309."""
+
+    mocker.patch("dvc.fs.dvc._DVCFileSystem", side_effect=CloneError("SCM error"))
+    (stage,) = tmp_dir.dvc_gen({"dir": {"foo": "foo", "bar": "bar"}})
+    imp_stage = dvc.imp(
+        "https://user:[email protected]/iterative/dvc.git",
+        "dir",
+        out="new_dir",
+        rev="HEAD",
+        no_exec=True,
+    )
+    dvc.push()
+
+    shutil.rmtree("dir")
+    dvc.cache.local.clear()
+
+    assert dvc.pull(stage.addressing) == {
+        "added": [join("dir", "")],
+        "deleted": [],
+        "modified": [],
+        "fetched": 3,
+    }
+
+    with pytest.raises(CloneError, match="SCM error"):
+        dvc.pull()
+    with pytest.raises(CloneError, match="SCM error"):
+        dvc.pull(imp_stage.addressing)

tests/func/test_dataset.py

@@ -412,3 +412,63 @@ def test_collect(tmp_dir, dvc):
 
     with pytest.raises(DatasetNotFoundError, match=r"^dataset not found$"):
         dvc.datasets["not-existing"]
+
+
+def test_parametrized(tmp_dir, dvc):
+    (tmp_dir / "dvc.yaml").dump(
+        {
+            "datasets": [
+                {"name": "${ds1.name}", "url": "${ds1.url}", "type": "dvcx"},
+                {
+                    "name": "${ds2.name}",
+                    "url": "${ds2.url}",
+                    "type": "dvc",
+                    "path": "${ds2.path}",
+                },
+                {
+                    "name": "${ds3.name}",
+                    "url": "${ds3.url}",
+                    "type": "url",
+                },
+            ]
+        }
+    )
+    (tmp_dir / "params.yaml").dump(
+        {
+            "ds1": {"name": "dogs", "url": "dvcx://dogs"},
+            "ds2": {
+                "name": "example-get-started",
+                "url": "[email protected]:iterative/example-get-started.git",
+                "path": "path",
+            },
+            "ds3": {
+                "name": "cloud-versioning-demo",
+                "url": "s3://cloud-versioning-demo",
+            },
+        }
+    )
+
+    path = (tmp_dir / "dvc.yaml").fs_path
+    assert dict(dvc.datasets.items()) == {
+        "dogs": DVCXDataset(
+            manifest_path=path,
+            spec=DatasetSpec(name="dogs", url="dvcx://dogs", type="dvcx"),
+        ),
+        "example-get-started": DVCDataset(
+            manifest_path=path,
+            spec=DVCDatasetSpec(
+                name="example-get-started",
+                url="[email protected]:iterative/example-get-started.git",
+                path="path",
+                type="dvc",
+            ),
+        ),
+        "cloud-versioning-demo": URLDataset(
+            manifest_path=path,
+            spec=DatasetSpec(
+                name="cloud-versioning-demo",
+                url="s3://cloud-versioning-demo",
+                type="url",
+            ),
+        ),
+    }

tests/func/test_download.py

@@ -0,0 +1,73 @@
+import os
+from unittest.mock import ANY
+
+import pytest
+
+from dvc.fs import download
+
+lfs_prefetch_params = [
+    pytest.param("abc", "abc", id="plain"),
+    pytest.param(
+        "*",
+        "[*]",
+        marks=pytest.mark.skipif(
+            os.name == "nt",
+            reason="forbidden character `*` on Windows filesystem",
+        ),
+        id="escape-*",
+    ),
+    pytest.param(
+        "**",
+        "[*][*]",
+        marks=pytest.mark.skipif(
+            os.name == "nt", reason="forbidden character `*` on Windows filesystem"
+        ),
+        id="escape-**",
+    ),
+    pytest.param(
+        "?",
+        "[?]",
+        marks=pytest.mark.skipif(
+            os.name == "nt", reason="forbidden character `?` on Windows filesystem"
+        ),
+        id="escape-?",
+    ),
+    pytest.param("[abc]", "[[]abc]", id="escape-[seq]"),
+    pytest.param("[!abc]", "[[]!abc]", id="escape-[!seq]"),
+]
+
+
+@pytest.mark.parametrize("dirname, include_name", lfs_prefetch_params)
+def test_lfs_prefetch_directory(tmp_dir, dvc, scm, mocker, dirname, include_name):
+    mock_fetch = mocker.patch("scmrepo.git.lfs.fetch")
+    tmp_dir.scm_gen(
+        {
+            ".gitattributes": "data/**/* filter=lfs diff=lfs merge=lfs -text",
+            f"data/{dirname}/test.txt": "test data",
+        },
+        commit="init lfs",
+    )
+    rev = scm.get_rev()
+    with dvc.switch(rev):
+        download(dvc.dvcfs, f"data/{dirname}", "data")
+        mock_fetch.assert_called_once_with(
+            scm, [rev], include=[f"/data/{include_name}/**"], progress=ANY
+        )
+
+
+@pytest.mark.parametrize("basename, include_name", lfs_prefetch_params)
+def test_lfs_prefetch_file(tmp_dir, dvc, scm, mocker, basename, include_name):
+    mock_fetch = mocker.patch("scmrepo.git.lfs.fetch")
+    tmp_dir.scm_gen(
+        {
+            ".gitattributes": "data/**/* filter=lfs diff=lfs merge=lfs -text",
+            f"data/{basename}.txt": "test data",
+        },
+        commit="init lfs",
+    )
+    rev = scm.get_rev()
+    with dvc.switch(rev):
+        download(dvc.dvcfs, f"data/{basename}.txt", "data")
+        mock_fetch.assert_called_once_with(
+            scm, [rev], include=[f"/data/{include_name}.txt"], progress=ANY
+        )

tests/func/test_gc.py

@@ -297,7 +297,7 @@ def test_gc_rev_num(tmp_dir, scm, dvc):
 def test_date(tmp_dir, scm, dvc):
     tmp_dir.dvc_gen("testfile", "content", commit="add testfile")
 
-    now = datetime.datetime.now()
+    now = datetime.datetime.now(tz=datetime.timezone.utc)
     datestamp = (now.date() + datetime.timedelta(days=1)).isoformat()
 
     tmp_dir.dvc_gen("testfile", "modified", commit="modified")

tests/integration/plots/test_plots.py

@@ -163,7 +163,7 @@ def verify_vega_props(plot_id, json_result, title, x, y, **kwargs):
     assert len(data) == 1
     data = first(data)
 
-    assert dpath.get(data, ["content", "title"]) == title
+    assert dpath.get(data, ["content", "title", "text"]) == title
 
     try:
         # TODO confusion_matrix_plot - need to find better way of asserting
@@ -634,3 +634,15 @@ def test_nested_x_defn_collection(tmp_dir, dvc, scm, capsys):
         {"step": 1, "Max_Leaf_Nodes": "50", "rev": "workspace"},
         {"step": 2, "Max_Leaf_Nodes": "500", "rev": "workspace"},
     ]
+
+
+def test_plots_empty_directory(tmp_dir, dvc, scm, capsys):
+    (tmp_dir / "empty").mkdir()
+    (tmp_dir / "dvc.yaml").dump({"plots": [{"empty": {}}]})
+
+    scm.add(["dvc.yaml"])
+    scm.commit("commit dvc files")
+
+    html_path, _, split_json_result = call(capsys)
+    assert split_json_result == {}
+    assert html_path == ""

tests/integration/test_studio_live_experiments.py

@@ -1,8 +1,8 @@
 import pytest
-from dvc_studio_client import env, post_live_metrics
 from funcy import first
 
 from dvc.env import (
+    DVC_EXP_GIT_REMOTE,
     DVC_STUDIO_OFFLINE,
     DVC_STUDIO_REPO_URL,
     DVC_STUDIO_TOKEN,
@@ -11,13 +11,15 @@ from dvc.env import (
 from dvc.repo import Repo
 from dvc.testing.scripts import COPY_SCRIPT
 from dvc.utils.studio import get_subrepo_relpath
+from dvc_studio_client import env, post_live_metrics
 
 
 @pytest.mark.studio
 @pytest.mark.parametrize("tmp", [True, False])
 @pytest.mark.parametrize("offline", [True, False])
+@pytest.mark.parametrize("dvc_exp_git_remote", [None, "DVC_EXP_GIT_REMOTE"])
 def test_post_to_studio(
-    tmp_dir, dvc, scm, exp_stage, mocker, monkeypatch, tmp, offline
+    tmp_dir, dvc, scm, exp_stage, mocker, monkeypatch, tmp, offline, dvc_exp_git_remote
 ):
     valid_response = mocker.MagicMock()
     valid_response.status_code = 200
@@ -28,6 +30,8 @@ def test_post_to_studio(
     monkeypatch.setenv(DVC_STUDIO_TOKEN, "STUDIO_TOKEN")
     monkeypatch.setenv(DVC_STUDIO_URL, "https://0.0.0.0")
     monkeypatch.setenv(DVC_STUDIO_OFFLINE, offline)
+    if dvc_exp_git_remote:
+        monkeypatch.setenv(DVC_EXP_GIT_REMOTE, dvc_exp_git_remote)
 
     baseline_sha = scm.get_rev()
     exp_rev = first(
@@ -52,7 +56,7 @@ def test_post_to_studio(
 
         assert start_call.kwargs["json"] == {
             "type": "start",
-            "repo_url": "STUDIO_REPO_URL",
+            "repo_url": dvc_exp_git_remote or "STUDIO_REPO_URL",
             "baseline_sha": baseline_sha,
             "name": name,
             "params": {"params.yaml": {"foo": 1}},
@@ -61,7 +65,7 @@ def test_post_to_studio(
 
         assert done_call.kwargs["json"] == {
             "type": "done",
-            "repo_url": "STUDIO_REPO_URL",
+            "repo_url": dvc_exp_git_remote or "STUDIO_REPO_URL",
             "baseline_sha": baseline_sha,
             "name": name,
             "client": "dvc",

tests/remotes/git_server.py

@@ -1,4 +1,5 @@
 import pytest
+
 from dvc_ssh.tests.cloud import SSH, TEST_SSH_KEY_PATH, TEST_SSH_USER
 
 

tests/unit/command/test_add.py

@@ -22,6 +22,7 @@ def test_add(mocker, dvc):
         to_remote=False,
         remote_jobs=None,
         force=False,
+        relink=True,
     )
 
 
@@ -45,6 +46,7 @@ def test_add_to_remote(mocker, dvc):
         to_remote=True,
         remote_jobs=None,
         force=False,
+        relink=True,
     )
 
 

tests/unit/command/test_dataset.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 
 import pytest
 
@@ -15,7 +15,7 @@ from dvc.cli import main
         ),
         (
             {"name": "mydataset", "url": "dvcx://dataset", "type": "dvcx"},
-            {"rev_lock": "0" * 40, "version": 1, "created_at": datetime.now()},
+            {"version": 1, "created_at": datetime.now(tz=timezone.utc)},
             "Adding mydataset (dvcx://dataset @ v1)\n",
         ),
         (
@@ -65,7 +65,7 @@ def test_add_already_exists(dvc, caplog, mocker):
         ),
         (
             {"name": "mydataset", "url": "dvcx://dataset", "type": "dvcx"},
-            {"rev_lock": "0" * 40, "version": 1, "created_at": datetime.now()},
+            {"version": 1, "created_at": datetime.now(tz=timezone.utc)},
             {"version": 2},
             {
                 "missing": "Updating mydataset (dvcx://dataset @ v2)\n",
@@ -75,7 +75,7 @@ def test_add_already_exists(dvc, caplog, mocker):
         ),
         (
             {"name": "mydataset", "url": "dvcx://dataset", "type": "dvcx"},
-            {"rev_lock": "0" * 40, "version": 2, "created_at": datetime.now()},
+            {"version": 2, "created_at": datetime.now(tz=timezone.utc)},
             {"version": 1},
             {
                 "missing": "Updating mydataset (dvcx://dataset @ v1)\n",

tests/unit/command/test_get.py

@@ -44,7 +44,21 @@ def test_get(mocker):
 
 
 def test_get_url(mocker, capsys):
-    cli_args = parse_args(["get", "repo_url", "src", "--rev", "version", "--show-url"])
+    cli_args = parse_args(
+        [
+            "get",
+            "repo_url",
+            "src",
+            "--rev",
+            "version",
+            "--remote",
+            "myremote",
+            "--show-url",
+            "--remote-config",
+            "k1=v1",
+            "k2=v2",
+        ]
+    )
     assert cli_args.func == CmdGet
 
     cmd = cli_args.func(cli_args)
@@ -54,4 +68,10 @@ def test_get_url(mocker, capsys):
     out, _ = capsys.readouterr()
     assert "resource_url" in out
 
-    m.assert_called_once_with("src", repo="repo_url", rev="version")
+    m.assert_called_once_with(
+        "src",
+        repo="repo_url",
+        rev="version",
+        remote="myremote",
+        remote_config={"k1": "v1", "k2": "v2"},
+    )

tests/unit/command/test_queue.py

@@ -166,13 +166,13 @@ def test_experiments_status(dvc, scm, mocker, capsys):
         {
             "rev": "c61a525a4ff39007301b4516fb6e54b323a0587b",
             "name": "I40",
-            "timestamp": datetime(2022, 6, 9, 20, 49, 48),
+            "timestamp": datetime(2022, 6, 9, 20, 49, 48),  # noqa: DTZ001
             "status": "Queued",
         },
         {
             "rev": "8da9c339da30636261a3491a90aafdb760a4168f",
             "name": "I60",
-            "timestamp": datetime(2022, 6, 9, 20, 49, 43),
+            "timestamp": datetime(2022, 6, 9, 20, 49, 43),  # noqa: DTZ001
             "status": "Running",
         },
     ]

tests/unit/command/test_studio.py

@@ -1,7 +1,16 @@
-from dvc_studio_client.auth import AuthorizationExpiredError
+import pytest
 
+from dvc import env
 from dvc.cli import main
 from dvc.utils.studio import STUDIO_URL
+from dvc_studio_client.auth import AuthorizationExpiredError
+
+
+@pytest.fixture(autouse=True)
+def global_config_dir(monkeypatch, tmp_dir_factory):
+    monkeypatch.setenv(
+        env.DVC_GLOBAL_CONFIG_DIR, str(tmp_dir_factory.mktemp("studio-login"))
+    )
 
 
 def test_studio_login_token_check_failed(mocker):

tests/unit/fs/test_fs.py

@@ -1,10 +1,10 @@
 import pytest
-from dvc_http import HTTPFileSystem, HTTPSFileSystem
-from dvc_s3 import S3FileSystem
-from dvc_ssh import SSHFileSystem
 
 from dvc.config import RemoteNotFoundError
 from dvc.fs import LocalFileSystem, get_cloud_fs, get_fs_cls, get_fs_config
+from dvc_http import HTTPFileSystem, HTTPSFileSystem
+from dvc_s3 import S3FileSystem
+from dvc_ssh import SSHFileSystem
 
 url_cls_pairs = [
     ("s3://bucket/path", S3FileSystem),

tests/unit/output/test_load.py

@@ -1,10 +1,10 @@
 import pytest
-from dvc_s3 import S3FileSystem
 
 from dvc import output
 from dvc.fs import LocalFileSystem
 from dvc.output import Output
 from dvc.stage import Stage
+from dvc_s3 import S3FileSystem
 
 
 @pytest.mark.parametrize(

tests/unit/remote/test_remote.py

@@ -1,8 +1,8 @@
 import pytest
-from dvc_gs import GSFileSystem
-from dvc_s3 import S3FileSystem
 
 from dvc.fs import get_cloud_fs
+from dvc_gs import GSFileSystem
+from dvc_s3 import S3FileSystem
 
 
 def test_remote_with_hash_jobs(dvc):

tests/unit/remote/test_webdav.py

@@ -1,7 +1,7 @@
 import pytest
-from dvc_webdav import WebDAVFileSystem, WebDAVSFileSystem
 
 from dvc.fs import get_cloud_fs
+from dvc_webdav import WebDAVFileSystem, WebDAVSFileSystem
 from tests.utils.asserts import issubset
 
 url_fmt = "{scheme}://{user}@example.com/public.php/webdav"

tests/unit/remote/test_webhdfs.py

@@ -1,5 +1,6 @@
 import pytest
 import requests
+
 from dvc_webhdfs import WebHDFSFileSystem
 
 host = "host"

tests/unit/repo/experiments/queue/test_celery.py

@@ -189,13 +189,13 @@ def test_queue_status(test_queue, scm, mocker):
 
     def resolve_commit(rev):
         if rev == "active":
-            commit_time = datetime(2022, 8, 7).timestamp()
+            commit_time = datetime(2022, 8, 7).timestamp()  # noqa: DTZ001
         elif rev == "queued":
-            commit_time = datetime(2022, 8, 6).timestamp()
+            commit_time = datetime(2022, 8, 6).timestamp()  # noqa: DTZ001
         elif rev == "failed":
-            commit_time = datetime(2022, 8, 5).timestamp()
+            commit_time = datetime(2022, 8, 5).timestamp()  # noqa: DTZ001
         elif rev == "success":
-            commit_time = datetime(2022, 8, 4).timestamp()
+            commit_time = datetime(2022, 8, 4).timestamp()  # noqa: DTZ001
         return mocker.Mock(commit_time=commit_time)
 
     mocker.patch.object(
@@ -230,24 +230,24 @@ def test_queue_status(test_queue, scm, mocker):
             "name": "foo",
             "rev": "active",
             "status": "Running",
-            "timestamp": datetime(2022, 8, 7, 0, 0, 0),
+            "timestamp": datetime(2022, 8, 7, 0, 0, 0),  # noqa: DTZ001
         },
         {
             "name": None,
             "rev": "queued",
             "status": "Queued",
-            "timestamp": datetime(2022, 8, 6, 0, 0, 0),
+            "timestamp": datetime(2022, 8, 6, 0, 0, 0),  # noqa: DTZ001
         },
         {
             "name": "bar",
             "rev": "failed",
             "status": "Failed",
-            "timestamp": datetime(2022, 8, 5, 0, 0, 0),
+            "timestamp": datetime(2022, 8, 5, 0, 0, 0),  # noqa: DTZ001
         },
         {
             "name": "foobar",
             "rev": "success",
             "status": "Success",
-            "timestamp": datetime(2022, 8, 4, 0, 0, 0),
+            "timestamp": datetime(2022, 8, 4, 0, 0, 0),  # noqa: DTZ001
         },
     ]

tests/unit/repo/experiments/test_collect.py

@@ -22,7 +22,7 @@ def test_collect_stable_sorting(dvc, scm, mocker):
     ]
 
     def collect_queued_patched(_, baseline_revs) -> dict[str, list["ExpRange"]]:
-        single_timestamp = datetime.datetime(2023, 6, 20, 0, 0, 0)
+        single_timestamp = datetime.datetime(2023, 6, 20, 0, 0, 0)  # noqa: DTZ001
 
         exp_ranges = [
             ExpRange(

tests/unit/scm/test_scm.py

@@ -68,7 +68,7 @@ def test_iter_revs(tmp_dir, scm, mocker):
         if rev == rev_root:
             return GitCommit(
                 "dummy",
-                commit_time=datetime(2022, 6, 28).timestamp(),
+                commit_time=datetime(2022, 6, 28).timestamp(),  # noqa: DTZ001
                 commit_time_offset=0,
                 message="dummy",
                 parents=["dummy"],
@@ -76,14 +76,14 @@ def test_iter_revs(tmp_dir, scm, mocker):
                 committer_email="dummy",
                 author_name="dummy",
                 author_email="dummy",
-                author_time=datetime(2022, 6, 28).timestamp(),
+                author_time=datetime(2022, 6, 28).timestamp(),  # noqa: DTZ001
                 author_time_offset=0,
             )
         if rev == rev_old:
             raise SCMError
         return GitCommit(
             "dummy",
-            commit_time=datetime(2022, 6, 30).timestamp(),
+            commit_time=datetime(2022, 6, 30).timestamp(),  # noqa: DTZ001
             commit_time_offset=0,
             message="dummy",
             parents=["dummy"],
@@ -91,7 +91,7 @@ def test_iter_revs(tmp_dir, scm, mocker):
             committer_email="dummy",
             author_name="dummy",
             author_email="dummy",
-            author_time=datetime(2022, 6, 28).timestamp(),
+            author_time=datetime(2022, 6, 28).timestamp(),  # noqa: DTZ001
             author_time_offset=0,
         )
 

tests/unit/test_logger.py

@@ -23,7 +23,8 @@ colors = {
 @pytest.fixture
 def dt(mocker):
     mocker.patch(
-        "time.time", return_value=time.mktime(datetime(2020, 2, 2).timetuple())
+        "time.time",
+        return_value=time.mktime(datetime(2020, 2, 2).timetuple()),  # noqa: DTZ001
     )
     return "2020-02-02 00:00:00,000"
 

tests/unit/ui/test_console.py

@@ -47,7 +47,7 @@ def test_write_json(capsys, mocker, isatty, expected_output):
 
     console = Console(enable=True)
     mocker.patch.object(console, "isatty", return_value=isatty)
-    message = {"hello": "world", "date": datetime.datetime(1970, 1, 1)}
+    message = {"hello": "world", "date": datetime.datetime(1970, 1, 1)}  # noqa: DTZ001
     console.write_json(message, default=str)
     captured = capsys.readouterr()
     assert captured.out == expected_output

tests/unit/utils/test_fs.py

@@ -43,9 +43,7 @@ def test_should_return_false_when_base_path_is_symlink(mocker):
     target_path = os.path.join(base_path, "bar")
 
     def base_path_is_symlink(path):
-        if path == base_path:
-            return True
-        return False
+        return path == base_path
 
     mocker.patch.object(
         system,

tests/unit/utils/test_studio.py

@@ -4,12 +4,19 @@ import pytest
 from requests import Response
 
 from dvc.env import (
+    DVC_EXP_GIT_REMOTE,
     DVC_STUDIO_OFFLINE,
     DVC_STUDIO_REPO_URL,
     DVC_STUDIO_TOKEN,
     DVC_STUDIO_URL,
 )
-from dvc.utils.studio import STUDIO_URL, config_to_env, env_to_config, notify_refs
+from dvc.utils.studio import (
+    STUDIO_URL,
+    config_to_env,
+    env_to_config,
+    get_repo_url,
+    notify_refs,
+)
 
 CONFIG = {"offline": True, "repo_url": "repo_url", "token": "token", "url": "url"}
 
@@ -67,3 +74,18 @@ def test_config_to_env():
 
 def test_env_to_config():
     assert env_to_config(ENV) == CONFIG
+
+
+@pytest.mark.parametrize(
+    "exp_git_remote, repo_url",
+    [
+        (None, None),
+        ("origin", "git@url"),
+        ("http://url", "http://url"),
+    ],
+)
+def test_get_repo_url(dvc, scm, monkeypatch, exp_git_remote, repo_url):
+    scm.gitpython.repo.create_remote("origin", "git@url")
+    if exp_git_remote:
+        monkeypatch.setenv(DVC_EXP_GIT_REMOTE, exp_git_remote)
+    assert get_repo_url(dvc) == repo_url