fix(pull): normalize dependency path (#10446)

Co-authored-by: Saugat Pachhai (सौगात) <[email protected]>

dvc/commands/artifacts.py

@@ -13,6 +13,9 @@ class CmdArtifactsGet(CmdBaseNoRepo):
         from dvc.scm import CloneError
         from dvc.ui import ui
 
+        if self.args.show_url:
+            return self._show_url()
+
         try:
             count, out = Artifacts.get(
                 self.args.url,
@@ -36,6 +39,28 @@ class CmdArtifactsGet(CmdBaseNoRepo):
             )
             return 1
 
+    def _show_url(self):
+        from dvc.api import artifacts_show, get_url
+        from dvc.ui import ui
+
+        artifact = artifacts_show(
+            self.args.name,
+            version=self.args.rev,
+            stage=self.args.stage,
+            repo=self.args.url,
+        )
+
+        url = get_url(
+            artifact["path"],
+            repo=self.args.url,
+            rev=artifact["rev"],
+            remote=self.args.remote,
+            remote_config=self.args.remote_config,
+        )
+        ui.write(url, force=True)
+
+        return 0
+
 
 def add_parser(subparsers, parent_parser):
     ARTIFACTS_HELP = "DVC model registry artifact commands."
@@ -84,6 +109,14 @@ def add_parser(subparsers, parent_parser):
         help="Destination path to download artifact to",
         metavar="<path>",
     ).complete = completion.DIR
+    get_parser.add_argument(
+        "--show-url",
+        action="store_true",
+        help=(
+            "Print the storage location (URL) the target data would be "
+            "downloaded from, and exit."
+        ),
+    )
     get_parser.add_argument(
         "-j",
         "--jobs",

dvc/commands/data_sync.py

@@ -194,8 +194,8 @@ def add_parser(subparsers, _parent_parser):
     )
     pull_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Fetch run history for all stages.",
     )
     pull_parser.add_argument(
@@ -262,8 +262,8 @@ def add_parser(subparsers, _parent_parser):
     )
     push_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Push run history for all stages.",
     )
     push_parser.add_argument(
@@ -324,8 +324,8 @@ def add_parser(subparsers, _parent_parser):
     )
     fetch_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Fetch run history for all stages.",
     )
     fetch_parser.add_argument(

dvc/commands/experiments/pull.py

@@ -1,3 +1,5 @@
+import argparse
+
 from dvc.cli import formatter
 from dvc.cli.command import CmdBase
 from dvc.cli.utils import append_doc_link
@@ -81,8 +83,8 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     experiments_pull_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Pull run history for all stages.",
     )
     experiments_pull_parser.add_argument(

dvc/commands/experiments/push.py

@@ -1,3 +1,4 @@
+import argparse
 from typing import Any
 
 from dvc.cli import completion, formatter
@@ -120,8 +121,8 @@ def add_parser(experiments_subparsers, parent_parser):
     )
     experiments_push_parser.add_argument(
         "--run-cache",
-        action="store_true",
-        default=False,
+        action=argparse.BooleanOptionalAction,
+        default=True,
         help="Push run history for all stages.",
     )
     experiments_push_parser.add_argument(

dvc/commands/repro.py

@@ -117,10 +117,7 @@ and then the stage name name.
         "--pull",
         action="store_true",
         default=False,
-        help=(
-            "Try automatically pulling missing cache for outputs restored "
-            "from the run-cache."
-        ),
+        help="Try automatically pulling missing data.",
     )
     repro_parser.add_argument(
         "--allow-missing",

dvc/dependency/repo.py

@@ -35,7 +35,7 @@ class RepoDependency(Dependency):
         super().__init__(stage, *args, **kwargs)
 
         self.fs = self._make_fs()
-        self.fs_path = as_posix(self.def_path)
+        self.fs_path = as_posix(self.fs.normpath(self.def_path))
 
     def _parse_path(self, fs, fs_path):  # noqa: ARG002
         return None

dvc/repo/artifacts.py

@@ -314,6 +314,13 @@ class Artifacts:
 
         name = _reformat_name(name)
         saved_exc: Optional[Exception] = None
+
+        local_dvc_studio_config = Config().get("studio", {})
+        args_dvc_studio_config = {}
+        if config and not isinstance(config, dict):
+            config = Config.load_file(config)
+            args_dvc_studio_config = config.get("studio", {})
+
         try:
             logger.trace("Trying studio-only config")
             return cls._download_studio(
@@ -324,14 +331,13 @@ class Artifacts:
                 out=out,
                 force=force,
                 jobs=jobs,
+                dvc_studio_config=local_dvc_studio_config | args_dvc_studio_config,
             )
         except FileExistsLocallyError:
             raise
         except Exception as exc:  # noqa: BLE001
             saved_exc = exc
 
-        if config and not isinstance(config, dict):
-            config = Config.load_file(config)
         with Repo.open(
             url=url,
             subrepos=True,
@@ -341,7 +347,7 @@ class Artifacts:
             remote_config=remote_config,
         ) as repo:
             logger.trace("Trying repo [studio] config")
-            dvc_studio_config = dict(repo.config.get("studio"))
+            repo_dvc_studio_config = repo.config.get("studio", {})
             try:
                 return cls._download_studio(
                     url,
@@ -351,7 +357,9 @@ class Artifacts:
                     out=out,
                     force=force,
                     jobs=jobs,
-                    dvc_studio_config=dvc_studio_config,
+                    dvc_studio_config=local_dvc_studio_config
+                    | repo_dvc_studio_config
+                    | args_dvc_studio_config,
                 )
             except FileExistsLocallyError:
                 raise

dvc/repo/fetch.py

@@ -2,6 +2,7 @@ from typing import TYPE_CHECKING
 
 from dvc.exceptions import DownloadError
 from dvc.log import logger
+from dvc.stage.cache import RunCacheNotSupported
 from dvc.ui import ui
 from dvc_data.index import DataIndex, FileStorage
 
@@ -136,6 +137,8 @@ def fetch(  # noqa: PLR0913
     try:
         if run_cache:
             self.stage_cache.pull(remote)
+    except RunCacheNotSupported as e:
+        logger.debug("failed to pull run cache: %s", e)
     except DownloadError as exc:
         failed_count += exc.amount
 

dvc/repo/push.py

@@ -1,10 +1,14 @@
 from contextlib import suppress
 
 from dvc.exceptions import InvalidArgumentError, UploadError
+from dvc.log import logger
+from dvc.stage.cache import RunCacheNotSupported
 from dvc.ui import ui
 
 from . import locked
 
+logger = logger.getChild(__name__)
+
 
 def _rebuild(idx, path, fs, cb):
     from dvc_data.index import DataIndex, DataIndexEntry, Meta
@@ -95,8 +99,11 @@ def push(  # noqa: PLR0913
                 "Multiple rev push is unsupported for cloud versioned remotes"
             )
 
-    used_run_cache = self.stage_cache.push(remote) if run_cache else []
-    transferred_count += len(used_run_cache)
+    try:
+        used_run_cache = self.stage_cache.push(remote) if run_cache else []
+        transferred_count += len(used_run_cache)
+    except RunCacheNotSupported as e:
+        logger.debug("failed to push run cache: %s", e)
 
     if isinstance(targets, str):
         targets = [targets]

dvc/stage/__init__.py

@@ -339,7 +339,7 @@ class Stage(params.StageParams):
         for out in self.outs:
             status = out.status()
             if status:
-                if allow_missing and status[str(out)] == "not in cache":
+                if allow_missing and status[str(out)] in ["not in cache", "deleted"]:
                     continue
                 logger.debug(
                     "Output '%s' of %s changed because it is '%s'.",
@@ -418,17 +418,28 @@ class Stage(params.StageParams):
 
     @rwlocked(read=["deps"], write=["outs"])
     def reproduce(self, interactive=False, **kwargs) -> Optional["Stage"]:
-        if not (
-            kwargs.get("force", False)
-            or self.changed(
-                kwargs.get("allow_missing", False), kwargs.pop("upstream", None)
-            )
-        ):
+        force = kwargs.get("force", False)
+        allow_missing = kwargs.get("allow_missing", False)
+        pull = kwargs.get("pull", False)
+        upstream = kwargs.pop("upstream", None)
+        if force:
+            pass
+        # Skip stages with missing data if otherwise unchanged
+        elif not self.changed(allow_missing, upstream):
             if not isinstance(self, PipelineStage) and self.is_data_source:
                 logger.info("'%s' didn't change, skipping", self.addressing)
             else:
                 logger.info("Stage '%s' didn't change, skipping", self.addressing)
             return None
+        # Pull stages with missing data if otherwise unchanged
+        elif not self.changed(True, upstream) and pull:
+            try:
+                logger.info("Pulling data for %s", self)
+                self.repo.pull(self.addressing, jobs=kwargs.get("jobs", None))
+                self.checkout()
+                return None
+            except CheckoutError:
+                logger.info("Unable to pull data for %s", self)
 
         msg = f"Going to reproduce {self}. Are you sure you want to continue?"
         if interactive and not prompt.confirm(msg):
@@ -597,16 +608,10 @@ class Stage(params.StageParams):
         if (self.cmd or self.is_import) and not self.frozen and not dry:
             self.remove_outs(ignore_remove=False, force=False)
 
-        if (
-            self.is_import and (not self.frozen or kwargs.get("pull"))
-        ) or self.is_partial_import:
+        if (self.is_import and not self.frozen) or self.is_partial_import:
             self._sync_import(dry, force, kwargs.get("jobs", None), no_download)
         elif not self.frozen and self.cmd:
             self._run_stage(dry, force, **kwargs)
-        elif kwargs.get("pull"):
-            logger.info("Pulling data for %s", self)
-            self.repo.pull(self.addressing, jobs=kwargs.get("jobs", None))
-            self.checkout()
         elif not dry:
             args = ("outputs", "frozen ") if self.frozen else ("data sources", "")
             logger.info("Verifying %s in %s%s", *args, self)

dvc/stage/run.py

@@ -2,6 +2,9 @@ import os
 import signal
 import subprocess
 import threading
+from functools import cache
+
+from packaging.version import InvalidVersion, Version
 
 from dvc.log import logger
 from dvc.utils import fix_env
@@ -12,25 +15,59 @@ from .exceptions import StageCmdFailedError
 logger = logger.getChild(__name__)
 
 
+@cache
+def _fish_supports_no_config(executable) -> bool:
+    """
+    Check if the fish shell supports the --no-config option.
+
+    Parameters:
+    executable (str): The path to the fish shell executable.
+
+    Returns:
+    bool: True if the fish version is greater than 3.3.0, False otherwise.
+    """
+    try:
+        output = subprocess.check_output(
+            [executable, "--version"],  # noqa: S603
+            text=True,
+        )
+        version = Version(output.split(" ")[-1].strip())
+        version_to_check = Version("3.3.0")
+        return version >= version_to_check
+    except (subprocess.CalledProcessError, IndexError, InvalidVersion):
+        logger.trace("could not check fish version, defaulting to False")
+        return False
+
+
+def _warn_if_fish(executable):
+    if (
+        executable is None
+        or os.path.basename(executable) != "fish"
+        or _fish_supports_no_config(executable)
+    ):
+        return
+    logger.warning(
+        "DVC detected that you are using a version of fish shell below 3.3.0 "
+        "Be aware that it might cause problems by overwriting "
+        "your current environment variables with values defined "
+        "in 'config.fish', which might affect your command. See "
+        "https://github.com/iterative/dvc/issues/1307. "
+    )
+
+
 def _make_cmd(executable, cmd):
     if executable is None:
         return cmd
-    opts = {"zsh": ["--no-rcs"], "bash": ["--noprofile", "--norc"]}
+    opts = {
+        "zsh": ["--no-rcs"],
+        "bash": ["--noprofile", "--norc"],
+        "fish": [],
+    }
     name = os.path.basename(executable).lower()
-    return [executable, *opts.get(name, []), "-c", cmd]
-
-
-def warn_if_fish(executable):
-    if executable is None or os.path.basename(os.path.realpath(executable)) != "fish":
-        return
-
-    logger.warning(
-        "DVC detected that you are using fish as your default "
-        "shell. Be aware that it might cause problems by overwriting "
-        "your current environment variables with values defined "
-        "in '.fishrc', which might affect your command. See "
-        "https://github.com/iterative/dvc/issues/1307. "
-    )
+    opt = opts.get(name, [])
+    if name == "fish" and _fish_supports_no_config(executable):
+        opt.append("--no-config")
+    return [executable, *opt, "-c", cmd]
 
 
 def _enforce_cmd_list(cmd):
@@ -106,7 +143,7 @@ def cmd_run(stage, dry=False, run_env=None):
     executable = get_executable()
 
     if not dry:
-        warn_if_fish(executable)
+        _warn_if_fish(executable)
 
     for cmd in commands:
         display_command(cmd)

dvc/testing/benchmarks/fixtures.py

@@ -104,6 +104,11 @@ def make_dvc_bin(
             venv.run("pip", "install", f"git+file://{dvc_git_repo}@{dvc_rev}#egg={egg}")
             if dvc_rev in ["2.18.1", "2.11.0", "2.6.3"]:
                 venv.run("pip", "install", "fsspec==2022.11.0")
+            try:
+                if version.Version(dvc_rev) < version.Version("3.50.3"):
+                    venv.run("pip", "install", "pygit2==1.14.1")
+            except version.InvalidVersion:
+                pass
             dvc_venvs[dvc_rev] = venv
         dvc_bin = venv.which("dvc")
     else:

pyproject.toml

@@ -65,7 +65,7 @@ dependencies = [
     "requests>=2.22",
     "rich>=12",
     "ruamel.yaml>=0.17.11",
-    "scmrepo>=3.3.2,<4",
+    "scmrepo>=3.3.4,<4",
     "shortuuid>=0.5",
     "shtab<2,>=1.3.4",
     "tabulate>=0.8.7",

tests/func/artifacts/test_artifacts.py

@@ -281,3 +281,12 @@ def test_artifacts_download_subrepo(tmp_dir, scm, sub):
     assert Artifacts.get(subdir.fs_path, name, force=True) == result
     with subdir.chdir():
         assert Artifacts.get(".", name, force=True) == result
+
+
+def test_artifacts_download_studio(tmp_dir, dvc, mocker):
+    with dvc.config.edit("global") as conf:
+        conf["studio"]["token"] = "mytoken"
+
+    download_studio = mocker.patch("dvc.repo.artifacts.Artifacts._download_studio")
+    Artifacts.get("myart.pkl", "myart.pkl")
+    assert download_studio.call_args.kwargs["dvc_studio_config"]["token"] == "mytoken"

tests/func/repro/test_repro_allow_missing.py

@@ -55,3 +55,16 @@ def test_repro_allow_missing_upstream_stage_modified(
     ret = dvc.reproduce(pull=True, allow_missing=True)
     # create-foo is skipped ; copy-foo pulls modified dep
     assert len(ret) == 1
+
+
+def test_repro_allow_missing_cached(tmp_dir, dvc):
+    tmp_dir.gen("fixed", "fixed")
+    dvc.stage.add(name="create-foo", cmd="echo foo > foo", deps=["fixed"], outs=["foo"])
+    dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
+    dvc.reproduce()
+
+    remove("foo")
+
+    ret = dvc.reproduce(allow_missing=True)
+    # both stages are skipped
+    assert not ret

tests/func/repro/test_repro_pull.py

@@ -1,5 +1,7 @@
 import os
 
+import pytest
+
 from dvc.stage.cache import RunCacheNotSupported
 from dvc.utils.fs import remove
 
@@ -49,11 +51,14 @@ def test_repro_skip_pull_if_no_run_cache_is_passed(tmp_dir, dvc, mocker, local_r
     (foo,) = tmp_dir.dvc_gen("foo", "foo")
 
     dvc.push()
-    spy_pull = mocker.spy(dvc.stage_cache, "pull")
     dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
+
+    assert dvc.reproduce(pull=True)
+
     remove("foo")
     remove(foo.outs[0].cache_path)
-
+    remove("dvc.lock")
+    spy_pull = mocker.spy(dvc.stage_cache, "pull")
     assert dvc.reproduce(pull=True, run_cache=False)
     assert not spy_pull.called
 
@@ -68,3 +73,46 @@ def test_repro_skip_pull_if_single_item_is_passed(tmp_dir, dvc, mocker, local_re
     remove(foo.outs[0].cache_path)
 
     assert dvc.reproduce(pull=True, single_item=True)
+
+
+def test_repro_pulls_persisted_output(tmp_dir, dvc, mocker, local_remote):
+    tmp_dir.dvc_gen("foo", "foo")
+
+    dvc.push()
+
+    dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs_persist=["bar"])
+    dvc.reproduce()
+    remove("bar")
+
+    # stage is skipped
+    assert not dvc.reproduce(pull=True)
+
+
+@pytest.mark.parametrize("allow_missing", [True, False])
+def test_repro_pulls_allow_missing(tmp_dir, dvc, mocker, local_remote, allow_missing):
+    tmp_dir.dvc_gen("foo", "foo")
+
+    dvc.push()
+
+    dvc.stage.add(name="copy-foo", cmd="cp foo bar", deps=["foo"], outs=["bar"])
+    dvc.reproduce()
+    remove("foo")
+
+    # stage is skipped
+    assert not dvc.reproduce(pull=True, allow_missing=allow_missing)
+    # data only pulled if allow_missing is false
+    assert (tmp_dir / "foo").exists() != allow_missing
+
+
+def test_repro_pull_fails(tmp_dir, dvc, mocker, local_remote):
+    tmp_dir.dvc_gen("foo", "foo")
+
+    dvc.stage.add(
+        name="concat-foo", cmd="cat foo foo > bar", deps=["foo"], outs=["bar"]
+    )
+    stages = dvc.reproduce()
+    remove("bar")
+    remove(stages[0].outs[0].cache_path)
+
+    # stage is run
+    assert dvc.reproduce(pull=True)

tests/func/test_import.py

@@ -305,11 +305,12 @@ def test_cache_type_is_properly_overridden(tmp_dir, scm, dvc, erepo_dir):
     assert scm.is_ignored("foo_imported")
 
 
-def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir):
+@pytest.mark.parametrize("dirpath", ["dir", "dir/"])
+def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir, dirpath):
     with erepo_dir.chdir():
         erepo_dir.dvc_gen({"dir": {"foo": "foo content"}}, commit="create dir")
 
-    dvc.imp(os.fspath(erepo_dir), "dir", "dir_imported")
+    stage = dvc.imp(os.fspath(erepo_dir), dirpath, "dir_imported")
 
     remove("dir_imported")
     dvc.cache.local.clear()
@@ -317,6 +318,7 @@ def test_pull_imported_directory_stage(tmp_dir, dvc, erepo_dir):
     dvc.pull(["dir_imported.dvc"])
 
     assert (tmp_dir / "dir_imported").read_text() == {"foo": "foo content"}
+    assert stage.deps[0].fs_path == "dir"
 
 
 def test_pull_wildcard_imported_directory_stage(tmp_dir, dvc, erepo_dir):

tests/func/test_run_cache.py

@@ -1,3 +1,4 @@
+import logging
 import os
 
 import pytest
@@ -231,13 +232,15 @@ def test_restore_pull(tmp_dir, dvc, run_copy, mocker, local_remote):
     assert (tmp_dir / LOCK_FILE).exists()
 
 
-def test_push_pull_unsupported(tmp_dir, dvc, mocker, run_copy):
+def test_push_pull_unsupported(tmp_dir, dvc, mocker, run_copy, local_remote, caplog):
     tmp_dir.gen("foo", "foo")
     run_copy("foo", "bar", name="copy-foo-bar")
     mocker.patch.object(
         dvc.cloud, "get_remote_odb", side_effect=RunCacheNotSupported("foo")
     )
-    with pytest.raises(RunCacheNotSupported):
+    with caplog.at_level(logging.DEBUG):
         dvc.push(run_cache=True)
-    with pytest.raises(RunCacheNotSupported):
+        assert "failed to push run cache" in caplog.text
+    with caplog.at_level(logging.DEBUG):
         dvc.pull(run_cache=True)
+        assert "failed to pull run cache" in caplog.text