index: mark remote as read_only for imports

Fixes #10206

.github/workflows/build.yaml

@@ -39,7 +39,7 @@ jobs:
         SETUPTOOLS_SCM_PRETEND_VERSION_FOR_DVC: ${{ env.version }}
     - run: twine check --strict dist/*
 
-    - uses: actions/upload-artifact@v3
+    - uses: actions/upload-artifact@v4
       with:
         name: Packages
         path: dist/
@@ -58,7 +58,7 @@ jobs:
       id-token: write
 
     steps:
-    - uses: actions/download-artifact@v3
+    - uses: actions/download-artifact@v4
       with:
         name: Packages
         path: dist
@@ -83,7 +83,7 @@ jobs:
       id-token: write
 
     steps:
-    - uses: actions/download-artifact@v3
+    - uses: actions/download-artifact@v4
       with:
         name: Packages
         path: dist

.github/workflows/codeql.yml

@@ -31,10 +31,10 @@ jobs:
       uses: actions/checkout@v4
 
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v2
+      uses: github/codeql-action/init@v3
       with:
         languages: ${{ matrix.language }}
         queries: security-extended
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v2
+      uses: github/codeql-action/analyze@v3

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.1.6'
+    rev: 'v0.1.9'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]

dvc/_debug.py

@@ -23,10 +23,11 @@ def viztracer_profile(
     tracer = viztracer.VizTracer(max_stack_depth=depth, log_async=log_async)
 
     tracer.start()
-    yield
-    tracer.stop()
-
-    tracer.save(path() if callable(path) else path)
+    try:
+        yield
+    finally:
+        tracer.stop()
+        tracer.save(path() if callable(path) else path)
 
 
 @contextmanager
@@ -45,31 +46,33 @@ def yappi_profile(
     yappi.set_clock_type("wall" if wall_clock else "cpu")
 
     yappi.start()
-    yield
-    yappi.stop()
+    try:
+        yield
+    finally:
+        yappi.stop()
 
-    threads = yappi.get_thread_stats()
-    stats = {}
-    if separate_threads:
-        for thread in threads:
-            ctx_id = thread.id
-            stats[ctx_id] = yappi.get_func_stats(ctx_id=ctx_id)
-    else:
-        stats[None] = yappi.get_func_stats()
-
-    fpath = path() if callable(path) else path
-    for ctx_id, st in stats.items():
-        if fpath:
-            out = f"{fpath}-{ctx_id}" if ctx_id is not None else fpath
-            st.save(out, type="callgrind")
+        threads = yappi.get_thread_stats()
+        stats = {}
+        if separate_threads:
+            for thread in threads:
+                ctx_id = thread.id
+                stats[ctx_id] = yappi.get_func_stats(ctx_id=ctx_id)
         else:
-            if ctx_id is not None:
-                print(f"\nThread {ctx_id}")  # noqa: T201
-            st.print_all()
-            if ctx_id is None:
-                threads.print_all()
+            stats[None] = yappi.get_func_stats()
 
-    yappi.clear_stats()
+        fpath = path() if callable(path) else path
+        for ctx_id, st in stats.items():
+            if fpath:
+                out = f"{fpath}-{ctx_id}" if ctx_id is not None else fpath
+                st.save(out, type="callgrind")
+            else:
+                if ctx_id is not None:
+                    print(f"\nThread {ctx_id}")  # noqa: T201
+                st.print_all()
+                if ctx_id is None:
+                    threads.print_all()
+
+        yappi.clear_stats()
 
 
 @contextmanager
@@ -85,13 +88,15 @@ def instrument(html_output=False):
     profiler = Profiler()
 
     profiler.start()
-    yield
-    profiler.stop()
+    try:
+        yield
+    finally:
+        profiler.stop()
 
-    if html_output:
-        profiler.open_in_browser()
-        return
-    print(profiler.output_text(unicode=True, color=True))  # noqa: T201
+        if html_output:
+            profiler.open_in_browser()
+        else:
+            print(profiler.output_text(unicode=True, color=True))  # noqa: T201
 
 
 @contextmanager
@@ -102,13 +107,14 @@ def profile(dump_path: Optional[str] = None):
     prof = cProfile.Profile()
     prof.enable()
 
-    yield
-
-    prof.disable()
-    if not dump_path:
-        prof.print_stats(sort="cumtime")
-        return
-    prof.dump_stats(dump_path)
+    try:
+        yield
+    finally:
+        prof.disable()
+        if dump_path:
+            prof.dump_stats(dump_path)
+        else:
+            prof.print_stats(sort="cumtime")
 
 
 @contextmanager

dvc/cachemgr.py

@@ -27,7 +27,7 @@ def _get_odb(
     cls, config, fs_path = get_cloud_fs(repo.config, **settings)
     fs = fs or cls(**config)
     if prefix:
-        fs_path = fs.path.join(fs_path, *prefix)
+        fs_path = fs.join(fs_path, *prefix)
     if hash_name:
         config["hash_name"] = hash_name
     return get_odb(fs, fs_path, state=repo.state, **config)
@@ -86,7 +86,7 @@ class CacheManager:
         return FileStorage(
             key=(),
             fs=self.local.fs,
-            path=self.local.fs.path.join(self.default_local_cache_dir, self.FS_DIR),
+            path=self.local.fs.join(self.default_local_cache_dir, self.FS_DIR),
         )
 
     def _init_odb(self, schemes):

dvc/commands/dag.py

@@ -86,7 +86,7 @@ def _collect_targets(repo, target, outs):
             targets.extend([str(out) for out in stage.outs])
             continue
 
-        for out in outs_trie.itervalues(prefix=repo.fs.path.parts(path)):
+        for out in outs_trie.itervalues(prefix=repo.fs.parts(path)):
             targets.extend(str(out))
 
     return targets

dvc/commands/gc.py

@@ -86,6 +86,7 @@ class CmdGC(CmdBase):
             num=self.args.num,
             not_in_remote=self.args.not_in_remote,
             dry=self.args.dry,
+            skip_failed=self.args.skip_failed,
         )
         return 0
 
@@ -188,6 +189,12 @@ def add_parser(subparsers, parent_parser):
         help="Remote storage to collect garbage in",
         metavar="<name>",
     )
+    gc_parser.add_argument(
+        "--skip-failed",
+        action="store_true",
+        default=False,
+        help="Skip revisions that fail when collected.",
+    )
     gc_parser.add_argument(
         "-f",
         "--force",

dvc/config.py

@@ -14,7 +14,7 @@ from .utils.objects import cached_property
 
 if TYPE_CHECKING:
     from dvc.fs import FileSystem
-    from dvc.types import DictStrAny, StrPath
+    from dvc.types import DictStrAny
 
 logger = logger.getChild(__name__)
 
@@ -89,8 +89,8 @@ class Config(dict):
 
     def __init__(
         self,
-        dvc_dir: Optional["StrPath"] = None,
-        local_dvc_dir: Optional["StrPath"] = None,
+        dvc_dir: Optional[str] = None,
+        local_dvc_dir: Optional[str] = None,
         validate: bool = True,
         fs: Optional["FileSystem"] = None,
         config: Optional["DictStrAny"] = None,
@@ -105,7 +105,7 @@ class Config(dict):
         self.fs = fs or self.wfs
 
         if dvc_dir:
-            self.dvc_dir = self.fs.path.abspath(dvc_dir)
+            self.dvc_dir = self.fs.abspath(dvc_dir)
 
         self.local_dvc_dir = local_dvc_dir
         if not fs and not local_dvc_dir:
@@ -145,10 +145,10 @@ class Config(dict):
         }
 
         if self.dvc_dir is not None:
-            files["repo"] = self.fs.path.join(self.dvc_dir, self.CONFIG)
+            files["repo"] = self.fs.join(self.dvc_dir, self.CONFIG)
 
         if self.local_dvc_dir is not None:
-            files["local"] = self.wfs.path.join(self.local_dvc_dir, self.CONFIG_LOCAL)
+            files["local"] = self.wfs.join(self.local_dvc_dir, self.CONFIG_LOCAL)
 
         return files
 
@@ -302,11 +302,11 @@ class Config(dict):
             return path.def_path
 
         if os.path.expanduser(path) != path:
-            return localfs.path.as_posix(path)
+            return localfs.as_posix(path)
 
         if isinstance(path, RelPath) or not os.path.isabs(path):
             path = relpath(path, conf_dir)
-            return localfs.path.as_posix(path)
+            return localfs.as_posix(path)
 
         return path
 

dvc/config_schema.py

@@ -240,6 +240,9 @@ SCHEMA = {
                     "ssl_verify": Any(Bool, str),
                     "token": str,
                     "use_https": Bool,
+                    "user": str,
+                    "password": str,
+                    "data_proxy_target": str,
                     Optional("verify", default=False): Bool,
                     **REMOTE_COMMON,
                 },

dvc/data_cloud.py

@@ -35,11 +35,9 @@ class Remote:
 
         path = self.path
         if self.worktree:
-            path = self.fs.path.join(
-                path, ".dvc", CacheManager.FILES_DIR, DEFAULT_ALGORITHM
-            )
+            path = self.fs.join(path, ".dvc", CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
         else:
-            path = self.fs.path.join(path, CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
+            path = self.fs.join(path, CacheManager.FILES_DIR, DEFAULT_ALGORITHM)
         return get_odb(self.fs, path, hash_name=DEFAULT_ALGORITHM, **self.config)
 
     @cached_property

dvc/dependency/base.py

@@ -33,7 +33,7 @@ class Dependency(Output):
         if self.fs.version_aware:
             old_fs_path = self.fs_path
             try:
-                self.fs_path = self.fs.path.version_path(self.fs_path, None)
+                self.fs_path = self.fs.version_path(self.fs_path, None)
                 if self.changed_meta():
                     return {str(self): "update available"}
             finally:
@@ -42,9 +42,9 @@ class Dependency(Output):
 
     def update(self, rev=None):
         if self.fs.version_aware:
-            self.fs_path = self.fs.path.version_path(self.fs_path, rev)
+            self.fs_path = self.fs.version_path(self.fs_path, rev)
             self.meta = self.get_meta()
-            self.fs_path = self.fs.path.version_path(self.fs_path, self.meta.version_id)
+            self.fs_path = self.fs.version_path(self.fs_path, self.meta.version_id)
 
     def download(self, to, jobs=None):
         fs_download(self.fs, self.fs_path, to.fs_path, jobs=jobs)
@@ -52,7 +52,7 @@ class Dependency(Output):
     def save(self):
         super().save()
         if self.fs.version_aware:
-            self.fs_path = self.fs.path.version_path(self.fs_path, self.meta.version_id)
+            self.fs_path = self.fs.version_path(self.fs_path, self.meta.version_id)
 
     def dumpd(self, **kwargs):
         if self.fs.version_aware:

dvc/dependency/param.py

@@ -54,12 +54,15 @@ def read_param_file(
                 continue
         return ret
 
-    from dpath import merge
+    from copy import deepcopy
 
-    for key_path in key_paths:
+    from dpath import merge
+    from funcy import distinct
+
+    for key_path in distinct(key_paths):
         merge(
             ret,
-            dpath.search(config, key_path, separator="."),
+            deepcopy(dpath.search(config, key_path, separator=".")),
             separator=".",
         )
     return ret

dvc/dvcfile.py

@@ -294,7 +294,7 @@ class ProjectFile(FileMixin):
     def resolver(self) -> "DataResolver":
         from .parsing import DataResolver
 
-        wdir = self.repo.fs.path.parent(self.path)
+        wdir = self.repo.fs.parent(self.path)
         return DataResolver(self.repo, wdir, self.contents)
 
     @cached_property

dvc/exceptions.py

@@ -357,3 +357,15 @@ class ArtifactNotFoundError(DvcException):
 
         desc = f" @ {stage or version}" if (stage or version) else ""
         super().__init__(f"Unable to find artifact '{name}{desc}'")
+
+
+class RevCollectionError(DvcException):
+    """Thrown if a revision failed to be collected.
+
+    Args:
+        rev (str): revision that failed (or "workspace").
+    """
+
+    def __init__(self, rev):
+        self.rev = rev
+        super().__init__(f"Failed to collect '{rev}'")

dvc/fs/__init__.py

@@ -25,7 +25,6 @@ from dvc_objects.fs.errors import (  # noqa: F401
     ConfigError,
     RemoteMissingDepsError,
 )
-from dvc_objects.fs.path import Path  # noqa: F401
 
 from .callbacks import Callback  # noqa: F401
 from .data import DataFileSystem  # noqa: F401
@@ -54,23 +53,20 @@ def download(
     from .callbacks import TqdmCallback
 
     with TqdmCallback(
-        desc=f"Downloading {fs.path.name(fs_path)}",
+        desc=f"Downloading {fs.name(fs_path)}",
         unit="files",
     ) as cb:
         # NOTE: We use dvc-objects generic.copy over fs.get since it makes file
         # download atomic and avoids fsspec glob/regex path expansion.
         if fs.isdir(fs_path):
             from_infos = [
-                path
-                for path in fs.find(fs_path)
-                if not path.endswith(fs.path.flavour.sep)
+                path for path in fs.find(fs_path) if not path.endswith(fs.flavour.sep)
             ]
             if not from_infos:
                 localfs.makedirs(to, exist_ok=True)
                 return 0
             to_infos = [
-                localfs.path.join(to, *fs.path.relparts(info, fs_path))
-                for info in from_infos
+                localfs.join(to, *fs.relparts(info, fs_path)) for info in from_infos
             ]
         else:
             from_infos = [fs_path]

dvc/fs/data.py

@@ -29,6 +29,9 @@ class DataFileSystem(FileSystem):
 
         return _DataFileSystem(**self.fs_args)
 
+    def getcwd(self):
+        return self.fs.getcwd()
+
     def isdvc(self, path, **kwargs):
         return self.fs.isdvc(path, **kwargs)
 

dvc/fs/dvc.py

@@ -13,7 +13,6 @@ from funcy import wrap_with
 
 from dvc.log import logger
 from dvc_objects.fs.base import FileSystem
-from dvc_objects.fs.path import Path
 
 from .data import DataFileSystem
 
@@ -71,7 +70,7 @@ def _merge_info(repo, key, fs_info, dvc_info):
 
 
 def _get_dvc_path(dvc_fs, subkey):
-    return dvc_fs.path.join(*subkey) if subkey else ""
+    return dvc_fs.join(*subkey) if subkey else ""
 
 
 class _DVCFileSystem(AbstractFileSystem):
@@ -131,55 +130,88 @@ class _DVCFileSystem(AbstractFileSystem):
             ...    rev="main",
             ... )
         """
-        from pygtrie import Trie
-
         super().__init__()
+        self._repo = repo
+        self._repo_factory = repo_factory
+        self._traverse_subrepos = subrepos
         self._repo_stack = ExitStack()
-        if repo is None:
-            url = url if url is not None else fo
-            repo = self._make_repo(
-                url=url,
-                rev=rev,
-                subrepos=subrepos,
-                config=config,
-                remote=remote,
-                remote_config=remote_config,
-            )
-            assert repo is not None
+        self._repo_kwargs = {
+            "url": url if url is not None else fo,
+            "rev": rev,
+            "subrepos": subrepos,
+            "config": config,
+            "remote": remote,
+            "remote_config": remote_config,
+        }
 
-            repo_factory = repo._fs_conf["repo_factory"]
-            self._repo_stack.enter_context(repo)
+    def getcwd(self):
+        relparts: Tuple[str, ...] = ()
+        assert self.repo is not None
+        if self.repo.fs.isin(self.repo.fs.getcwd(), self.repo.root_dir):
+            relparts = self.repo.fs.relparts(self.repo.fs.getcwd(), self.repo.root_dir)
+        return self.root_marker + self.sep.join(relparts)
 
-        if not repo_factory:
+    @classmethod
+    def join(cls, *parts: str) -> str:
+        return posixpath.join(*parts)
+
+    @classmethod
+    def parts(cls, path: str) -> Tuple[str, ...]:
+        ret = []
+        while True:
+            path, part = posixpath.split(path)
+
+            if part:
+                ret.append(part)
+                continue
+
+            if path:
+                ret.append(path)
+
+            break
+
+        ret.reverse()
+
+        return tuple(ret)
+
+    def normpath(self, path: str) -> str:
+        return posixpath.normpath(path)
+
+    def abspath(self, path: str) -> str:
+        if not posixpath.isabs(path):
+            path = self.join(self.getcwd(), path)
+        return self.normpath(path)
+
+    def relpath(self, path: str, start: Optional[str] = None) -> str:
+        if start is None:
+            start = "."
+        return posixpath.relpath(self.abspath(path), start=self.abspath(start))
+
+    def relparts(self, path: str, start: Optional[str] = None) -> Tuple[str, ...]:
+        return self.parts(self.relpath(path, start=start))
+
+    @functools.cached_property
+    def repo(self):
+        if self._repo:
+            return self._repo
+
+        repo = self._make_repo(**self._repo_kwargs)
+
+        self._repo_stack.enter_context(repo)
+        self._repo = repo
+        return repo
+
+    @functools.cached_property
+    def repo_factory(self):
+        if self._repo_factory:
+            return self._repo_factory
+
+        if self._repo:
             from dvc.repo import Repo
 
-            self.repo_factory: RepoFactory = Repo
-        else:
-            self.repo_factory = repo_factory
+            return Repo
 
-        def _getcwd():
-            relparts: Tuple[str, ...] = ()
-            assert repo is not None
-            if repo.fs.path.isin(repo.fs.path.getcwd(), repo.root_dir):
-                relparts = repo.fs.path.relparts(repo.fs.path.getcwd(), repo.root_dir)
-            return self.root_marker + self.sep.join(relparts)
-
-        self.path = Path(self.sep, getcwd=_getcwd)
-        self.repo = repo
-        self.hash_jobs = repo.fs.hash_jobs
-        self._traverse_subrepos = subrepos
-
-        self._subrepos_trie = Trie()
-        """Keeps track of each and every path with the corresponding repo."""
-
-        key = self._get_key(self.repo.root_dir)
-        self._subrepos_trie[key] = repo
-
-        self._datafss = {}
-        """Keep a datafs instance of each repo."""
-
-        if hasattr(repo, "dvc_dir"):
-            self._datafss[key] = DataFileSystem(index=repo.index.data["repo"])
+        return self.repo._fs_conf["repo_factory"]
 
     @functools.cached_property
     def fsid(self) -> str:
@@ -194,20 +226,43 @@ class _DVCFileSystem(AbstractFileSystem):
 
     def _get_key(self, path: "StrPath") -> Key:
         path = os.fspath(path)
-        parts = self.repo.fs.path.relparts(path, self.repo.root_dir)
+        parts = self.repo.fs.relparts(path, self.repo.root_dir)
         if parts == (os.curdir,):
             return ()
         return parts
 
+    @functools.cached_property
+    def _subrepos_trie(self):
+        """Keeps track of each and every path with the corresponding repo."""
+
+        from pygtrie import Trie
+
+        trie = Trie()
+        key = self._get_key(self.repo.root_dir)
+        trie[key] = self.repo
+        return trie
+
     def _get_key_from_relative(self, path) -> Key:
         path = self._strip_protocol(path)
-        parts = self.path.relparts(path, self.root_marker)
+        parts = self.relparts(path, self.root_marker)
         if parts and parts[0] == os.curdir:
             return parts[1:]
         return parts
 
     def _from_key(self, parts: Key) -> str:
-        return self.repo.fs.path.join(self.repo.root_dir, *parts)
+        return self.repo.fs.join(self.repo.root_dir, *parts)
+
+    @functools.cached_property
+    def _datafss(self):
+        """Keep a datafs instance of each repo."""
+
+        datafss = {}
+
+        if hasattr(self.repo, "dvc_dir"):
+            key = self._get_key(self.repo.root_dir)
+            datafss[key] = DataFileSystem(index=self.repo.index.data["repo"])
+
+        return datafss
 
     @property
     def repo_url(self):
@@ -261,7 +316,7 @@ class _DVCFileSystem(AbstractFileSystem):
 
         from dvc.repo import Repo
 
-        repo_path = self.repo.fs.path.join(dir_path, Repo.DVC_DIR)
+        repo_path = self.repo.fs.join(dir_path, Repo.DVC_DIR)
         return self.repo.fs.isdir(repo_path)
 
     def _get_subrepo_info(
@@ -315,7 +370,7 @@ class _DVCFileSystem(AbstractFileSystem):
             dvc_path = _get_dvc_path(dvc_fs, subkey)
             with suppress(FileNotFoundError, NotADirectoryError):
                 for info in dvc_fs.ls(dvc_path, detail=True):
-                    dvc_infos[dvc_fs.path.name(info["name"])] = info
+                    dvc_infos[dvc_fs.name(info["name"])] = info
                 dvc_exists = True
 
         fs_exists = False
@@ -328,7 +383,7 @@ class _DVCFileSystem(AbstractFileSystem):
                 for info in repo.dvcignore.ls(
                     fs, fs_path, detail=True, ignore_subrepos=ignore_subrepos
                 ):
-                    fs_infos[fs.path.name(info["name"])] = info
+                    fs_infos[fs.name(info["name"])] = info
                 fs_exists = True
             except (FileNotFoundError, NotADirectoryError):
                 pass
@@ -347,7 +402,7 @@ class _DVCFileSystem(AbstractFileSystem):
             if not dvcfiles and _is_dvc_file(name):
                 continue
 
-            entry_path = self.path.join(path, name)
+            entry_path = self.join(path, name)
             info = _merge_info(
                 repo, (*subkey, name), fs_infos.get(name), dvc_infos.get(name)
             )
@@ -395,7 +450,7 @@ class _DVCFileSystem(AbstractFileSystem):
         # NOTE: if some parent in fs_path turns out to be a file, it means
         # that the whole repofs branch doesn't exist.
         if dvc_info and not fs_info:
-            for parent in fs.path.parents(fs_path):
+            for parent in fs.parents(fs_path):
                 try:
                     if fs.info(parent)["type"] != "directory":
                         dvc_info = None
@@ -477,6 +532,9 @@ class DVCFileSystem(FileSystem):
     def fs(self) -> "_DVCFileSystem":
         return _DVCFileSystem(**self.fs_args)
 
+    def getcwd(self):
+        return self.fs.getcwd()
+
     @property
     def fsid(self) -> str:
         return self.fs.fsid
@@ -484,10 +542,6 @@ class DVCFileSystem(FileSystem):
     def isdvc(self, path, **kwargs) -> bool:
         return self.fs.isdvc(path, **kwargs)
 
-    @property
-    def path(self) -> Path:
-        return self.fs.path
-
     @property
     def repo(self) -> "Repo":
         return self.fs.repo

dvc/fs/git.py

@@ -46,9 +46,11 @@ class GitFileSystem(FileSystem):
 
         return FsspecGitFileSystem(**self.fs_args)
 
-    @functools.cached_property
-    def path(self):
-        return self.fs.path
+    def getcwd(self):
+        return self.fs.getcwd()
+
+    def chdir(self, path):
+        self.fs.chdir(path)
 
     @property
     def rev(self) -> str:

dvc/ignore.py

@@ -51,8 +51,8 @@ class DvcIgnorePatterns(DvcIgnore):
 
     @classmethod
     def from_file(cls, path, fs, name):
-        assert fs.path.isabs(path)
-        dirname = fs.path.normpath(fs.path.dirname(path))
+        assert fs.isabs(path)
+        dirname = fs.normpath(fs.dirname(path))
         with fs.open(path, encoding="utf-8") as fobj:
             path_spec_lines = [
                 PatternInfo(line, f"{name}:{line_no + 1}:{line}")
@@ -194,7 +194,7 @@ class DvcIgnoreFilter:
         )
 
     def _get_key(self, path):
-        parts = self.fs.path.relparts(path, self.root_dir)
+        parts = self.fs.relparts(path, self.root_dir)
         if parts == (os.curdir,):
             return ()
         return parts
@@ -204,13 +204,13 @@ class DvcIgnoreFilter:
         old_pattern = trie.longest_prefix(key).value
         matches = old_pattern.matches(dirname, DvcIgnore.DVCIGNORE_FILE, False)
 
-        path = self.fs.path.join(dirname, DvcIgnore.DVCIGNORE_FILE)
+        path = self.fs.join(dirname, DvcIgnore.DVCIGNORE_FILE)
         if not matches and self.fs.exists(path):
-            name = self.fs.path.relpath(path, self.root_dir)
+            name = self.fs.relpath(path, self.root_dir)
             new_pattern = DvcIgnorePatterns.from_file(path, self.fs, name)
             if old_pattern:
                 plist, prefix = merge_patterns(
-                    self.fs.path.flavour,
+                    self.fs.flavour,
                     old_pattern.pattern_list,
                     old_pattern.dirname,
                     new_pattern.pattern_list,
@@ -239,7 +239,7 @@ class DvcIgnoreFilter:
                     dnames = []
 
             for dname in dnames:
-                self._update_sub_repo(self.fs.path.join(dirname, dname), ignore_trie)
+                self._update_sub_repo(self.fs.join(dirname, dname), ignore_trie)
 
     def _update_sub_repo(self, path, ignore_trie: Trie):
         from dvc.repo import Repo
@@ -247,18 +247,18 @@ class DvcIgnoreFilter:
         if path == self.root_dir:
             return
 
-        dvc_dir = self.fs.path.join(path, Repo.DVC_DIR)
+        dvc_dir = self.fs.join(path, Repo.DVC_DIR)
         if not self.fs.exists(dvc_dir):
             return
 
-        root, dname = self.fs.path.split(path)
+        root, dname = self.fs.split(path)
         key = self._get_key(root)
         pattern_info = PatternInfo(f"/{dname}/", f"in sub_repo:{dname}")
         new_pattern = DvcIgnorePatterns([pattern_info], root, self.fs.sep)
         old_pattern = ignore_trie.longest_prefix(key).value
         if old_pattern:
             plist, prefix = merge_patterns(
-                self.fs.path.flavour,
+                self.fs.flavour,
                 old_pattern.pattern_list,
                 old_pattern.dirname,
                 new_pattern.pattern_list,
@@ -269,7 +269,7 @@ class DvcIgnoreFilter:
             ignore_trie[key] = new_pattern
 
     def __call__(self, root, dirs, files, ignore_subrepos=True):
-        abs_root = self.fs.path.abspath(root)
+        abs_root = self.fs.abspath(root)
         ignore_pattern = self._get_trie_pattern(
             abs_root, dnames=dirs, ignore_subrepos=ignore_subrepos
         )
@@ -283,7 +283,7 @@ class DvcIgnoreFilter:
         nondirs = []
 
         for entry in fs.ls(path, detail=True, **kwargs):
-            name = fs.path.name(entry["name"])
+            name = fs.name(entry["name"])
             fs_dict[name] = entry
             if entry["type"] == "directory":
                 dirs.append(name)
@@ -338,7 +338,7 @@ class DvcIgnoreFilter:
         else:
             ignores_trie = self._ignores_trie_subrepos
 
-        if not self.fs.path.isin_or_eq(dirname, self.root_dir):
+        if not self.fs.isin_or_eq(dirname, self.root_dir):
             # outside of the repo
             return None
 
@@ -349,12 +349,12 @@ class DvcIgnoreFilter:
             return ignore_pattern
 
         prefix_key = ignores_trie.longest_prefix(key).key or ()
-        prefix = self.fs.path.join(self.root_dir, *prefix_key)
+        prefix = self.fs.join(self.root_dir, *prefix_key)
 
         dirs = list(
             takewhile(
                 lambda path: path != prefix,
-                (parent for parent in localfs.path.parents(dirname)),
+                (parent for parent in localfs.parents(dirname)),
             )
         )
         dirs.reverse()
@@ -370,7 +370,7 @@ class DvcIgnoreFilter:
     ):
         if self._outside_repo(path):
             return False
-        dirname, basename = self.fs.path.split(self.fs.path.normpath(path))
+        dirname, basename = self.fs.split(self.fs.normpath(path))
         ignore_pattern = self._get_trie_pattern(dirname, None, ignore_subrepos)
         if ignore_pattern:
             return ignore_pattern.matches(dirname, basename, is_dir)
@@ -378,7 +378,7 @@ class DvcIgnoreFilter:
 
     def is_ignored_dir(self, path: str, ignore_subrepos: bool = True) -> bool:
         # only used in LocalFileSystem
-        path = self.fs.path.abspath(path)
+        path = self.fs.abspath(path)
         if path == self.root_dir:
             return False
 
@@ -386,18 +386,18 @@ class DvcIgnoreFilter:
 
     def is_ignored_file(self, path: str, ignore_subrepos: bool = True) -> bool:
         # only used in LocalFileSystem
-        path = self.fs.path.abspath(path)
+        path = self.fs.abspath(path)
         return self._is_ignored(path, False, ignore_subrepos=ignore_subrepos)
 
     def _outside_repo(self, path):
-        return not self.fs.path.isin_or_eq(path, self.root_dir)
+        return not self.fs.isin_or_eq(path, self.root_dir)
 
     def check_ignore(self, target):
         # NOTE: can only be used in `dvc check-ignore`, see
         # https://github.com/iterative/dvc/issues/5046
-        full_target = self.fs.path.abspath(target)
+        full_target = self.fs.abspath(target)
         if not self._outside_repo(full_target):
-            dirname, basename = self.fs.path.split(self.fs.path.normpath(full_target))
+            dirname, basename = self.fs.split(self.fs.normpath(full_target))
             pattern = self._get_trie_pattern(dirname)
             if pattern:
                 matches = pattern.matches(

dvc/output.py

@@ -378,7 +378,7 @@ class Output:
         if (
             self.repo
             and self.fs.protocol == "local"
-            and not self.fs.path.isabs(self.def_path)
+            and not self.fs.isabs(self.def_path)
         ):
             self.fs = self.repo.fs
 
@@ -411,7 +411,7 @@ class Output:
         self.remote = remote
 
         if self.fs.version_aware:
-            _, version_id = self.fs.path.coalesce_version(
+            _, version_id = self.fs.coalesce_version(
                 self.def_path, self.meta.version_id
             )
             self.meta.version_id = version_id
@@ -465,7 +465,7 @@ class Output:
             parsed.scheme != "remote"
             and self.stage
             and self.stage.repo.fs == fs
-            and not fs.path.isabs(fs_path)
+            and not fs.isabs(fs_path)
         ):
             # NOTE: we can path either from command line or .dvc file,
             # so we should expect both posix and windows style paths.
@@ -473,9 +473,9 @@ class Output:
             #
             # FIXME: if we have Windows path containing / or posix one with \
             # then we have #2059 bug and can't really handle that.
-            fs_path = fs.path.join(self.stage.wdir, fs_path)
+            fs_path = fs.join(self.stage.wdir, fs_path)
 
-        return fs.path.abspath(fs.path.normpath(fs_path))
+        return fs.abspath(fs.normpath(fs_path))
 
     def __repr__(self):
         return f"{type(self).__name__}: {self.def_path!r}"
@@ -491,14 +491,14 @@ class Output:
         ):
             return str(self.def_path)
 
-        if not self.fs.path.isin(self.fs_path, self.repo.root_dir):
+        if not self.fs.isin(self.fs_path, self.repo.root_dir):
             return self.fs_path
 
-        cur_dir = self.fs.path.getcwd()
-        if self.fs.path.isin(cur_dir, self.repo.root_dir):
-            return self.fs.path.relpath(self.fs_path, cur_dir)
+        cur_dir = self.fs.getcwd()
+        if self.fs.isin(cur_dir, self.repo.root_dir):
+            return self.fs.relpath(self.fs_path, cur_dir)
 
-        return self.fs.path.relpath(self.fs_path, self.repo.root_dir)
+        return self.fs.relpath(self.fs_path, self.repo.root_dir)
 
     def clear(self):
         self.hash_info = HashInfo.from_dict({})
@@ -515,10 +515,10 @@ class Output:
         if urlparse(self.def_path).scheme == "remote":
             return False
 
-        if self.fs.path.isabs(self.def_path):
+        if self.fs.isabs(self.def_path):
             return False
 
-        return self.repo and self.fs.path.isin(
+        return self.repo and self.fs.isin(
             self.fs_path,
             self.repo.root_dir,
         )
@@ -607,11 +607,11 @@ class Output:
     def index_key(self) -> Tuple[str, "DataIndexKey"]:
         if self.is_in_repo:
             workspace = "repo"
-            key = self.repo.fs.path.relparts(self.fs_path, self.repo.root_dir)
+            key = self.repo.fs.relparts(self.fs_path, self.repo.root_dir)
         else:
             workspace = self.fs.protocol
-            no_drive = self.fs.path.flavour.splitdrive(self.fs_path)[1]
-            key = self.fs.path.parts(no_drive)[1:]
+            no_drive = self.fs.flavour.splitdrive(self.fs_path)[1]
+            key = self.fs.parts(no_drive)[1:]
         return workspace, key
 
     def changed_checksum(self):
@@ -799,7 +799,7 @@ class Output:
                         callback=cb,
                     )
             if relink:
-                rel = self.fs.path.relpath(filter_info or self.fs_path)
+                rel = self.fs.relpath(filter_info or self.fs_path)
                 with CheckoutCallback(desc=f"Checking out {rel}", unit="files") as cb:
                     self._checkout(
                         filter_info or self.fs_path,
@@ -814,7 +814,7 @@ class Output:
                 self.set_exec()
 
     def _commit_granular_dir(self, filter_info, hardlink) -> Optional["HashFile"]:
-        prefix = self.fs.path.parts(self.fs.path.relpath(filter_info, self.fs_path))
+        prefix = self.fs.parts(self.fs.relpath(filter_info, self.fs_path))
         staging, _, obj = self._build(
             self.cache,
             self.fs_path,
@@ -862,7 +862,7 @@ class Output:
             ret.update(split_file_meta_from_cloud(meta_d))
 
         if self.is_in_repo:
-            path = self.fs.path.as_posix(relpath(self.fs_path, self.stage.wdir))
+            path = self.fs.as_posix(relpath(self.fs_path, self.stage.wdir))
         else:
             path = self.def_path
 
@@ -949,7 +949,7 @@ class Output:
             return None
 
         assert obj
-        fs_path = self.fs.path
+        fs_path = self.fs
         if filter_info and filter_info != self.fs_path:
             prefix = fs_path.relparts(filter_info, self.fs_path)
             assert isinstance(obj, Tree)
@@ -1154,7 +1154,7 @@ class Output:
         assert obj is None or isinstance(obj, Tree)
         if filter_info and filter_info != self.fs_path:
             assert obj
-            prefix = self.fs.path.parts(self.fs.path.relpath(filter_info, self.fs_path))
+            prefix = self.fs.parts(self.fs.relpath(filter_info, self.fs_path))
             return obj.filter(prefix)
         return obj
 
@@ -1297,10 +1297,7 @@ class Output:
     def unstage(self, path: str) -> Tuple["Meta", "Tree"]:
         from pygtrie import Trie
 
-        from dvc_objects.fs.path import Path
-
-        assert isinstance(self.fs.path, Path)
-        rel_key = tuple(self.fs.path.parts(self.fs.path.relpath(path, self.fs_path)))
+        rel_key = tuple(self.fs.parts(self.fs.relpath(path, self.fs_path)))
 
         if self.hash_info:
             tree = self.get_dir_cache()
@@ -1333,11 +1330,8 @@ class Output:
     ) -> Tuple["Meta", "Tree"]:
         from pygtrie import Trie
 
-        from dvc_objects.fs.path import Path
-
-        assert isinstance(self.fs.path, Path)
         append_only = True
-        rel_key = tuple(self.fs.path.parts(self.fs.path.relpath(path, self.fs_path)))
+        rel_key = tuple(self.fs.parts(self.fs.relpath(path, self.fs_path)))
 
         if self.hash_info:
             tree = self.get_dir_cache()

dvc/parsing/__init__.py

@@ -143,11 +143,11 @@ class DataResolver:
         self.parsing_config = repo.config.get("parsing", {})
 
         if os.path.isabs(wdir):
-            wdir = fs.path.relpath(wdir)
+            wdir = fs.relpath(wdir)
             wdir = "" if wdir == os.curdir else wdir
 
         self.wdir = wdir
-        self.relpath = fs.path.normpath(fs.path.join(self.wdir, "dvc.yaml"))
+        self.relpath = fs.normpath(fs.join(self.wdir, "dvc.yaml"))
 
         vars_ = d.get(VARS_KWD, [])
         check_interpolations(vars_, VARS_KWD, self.relpath)
@@ -248,7 +248,7 @@ class EntryDefinition:
             wdir = to_str(context.resolve_str(wdir))
         except (ContextError, ParseError) as exc:
             format_and_raise(exc, f"'{self.where}.{name}.wdir'", self.relpath)
-        return self.resolver.fs.path.join(self.wdir, wdir)
+        return self.resolver.fs.join(self.wdir, wdir)
 
     def resolve(self, **kwargs):
         try:

dvc/parsing/context.py

@@ -384,7 +384,7 @@ class Context(CtxDict):
 
     def merge_from(self, fs, item: str, wdir: str, overwrite=False):
         path, _, keys_str = item.partition(":")
-        path = fs.path.normpath(fs.path.join(wdir, path))
+        path = fs.normpath(fs.join(wdir, path))
 
         select_keys = lfilter(bool, keys_str.split(",")) if keys_str else None
         if path in self.imports:
@@ -429,7 +429,7 @@ class Context(CtxDict):
         default: Optional[str] = None,
     ):
         if default:
-            to_import = fs.path.join(wdir, default)
+            to_import = fs.join(wdir, default)
             if fs.exists(to_import):
                 self.merge_from(fs, default, wdir)
             else:

dvc/repo/__init__.py

@@ -12,7 +12,12 @@ from typing import (
     Union,
 )
 
-from dvc.exceptions import NotDvcRepoError, OutputNotFoundError
+from dvc.exceptions import (
+    DvcException,
+    NotDvcRepoError,
+    OutputNotFoundError,
+    RevCollectionError,
+)
 from dvc.ignore import DvcIgnoreFilter
 from dvc.log import logger
 from dvc.utils.objects import cached_property
@@ -114,7 +119,7 @@ class Repo:
         try:
             root_dir = self.find_root(root_dir, fs)
             fs = fs or localfs
-            dvc_dir = fs.path.join(root_dir, self.DVC_DIR)
+            dvc_dir = fs.join(root_dir, self.DVC_DIR)
         except NotDvcRepoError:
             if not uninitialized:
                 raise
@@ -207,7 +212,7 @@ class Repo:
                 self.fs.makedirs(self.tmp_dir, exist_ok=True)
 
                 self.lock = make_lock(
-                    self.fs.path.join(self.tmp_dir, "lock"),
+                    self.fs.join(self.tmp_dir, "lock"),
                     tmp_dir=self.tmp_dir,
                     hardlink_lock=self.config["core"].get("hardlink_lock", False),
                     friendly=True,
@@ -251,7 +256,7 @@ class Repo:
         )
 
     @cached_property
-    def local_dvc_dir(self):
+    def local_dvc_dir(self) -> Optional[str]:
         from dvc.fs import GitFileSystem, LocalFileSystem
 
         if not self.dvc_dir:
@@ -263,10 +268,10 @@ class Repo:
         if not isinstance(self.fs, GitFileSystem):
             return None
 
-        relparts = ()
+        relparts: Tuple[str, ...] = ()
         if self.root_dir != "/":
             # subrepo
-            relparts = self.fs.path.relparts(self.root_dir, "/")
+            relparts = self.fs.relparts(self.root_dir, "/")
 
         dvc_dir = os.path.join(
             self.scm.root_dir,
@@ -380,6 +385,7 @@ class Repo:
             except KeyError:
                 pass
         self.data_index.commit()
+        self.data_index.close()
         self._reset()
 
     def get_data_index_entry(
@@ -395,7 +401,7 @@ class Repo:
             index = subrepo.index.data[workspace]
         else:
             index = self.index.data[workspace]
-            key = self.fs.path.relparts(path, self.root_dir)
+            key = self.fs.relparts(path, self.root_dir)
 
         try:
             return index, index[key]
@@ -411,18 +417,18 @@ class Repo:
 
         fs = fs or localfs
         root = root or os.curdir
-        root_dir = fs.path.abspath(root)
+        root_dir = fs.abspath(root)
 
         if not fs.isdir(root_dir):
             raise NotDvcRepoError(f"directory '{root}' does not exist")
 
         while True:
-            dvc_dir = fs.path.join(root_dir, cls.DVC_DIR)
+            dvc_dir = fs.join(root_dir, cls.DVC_DIR)
             if fs.isdir(dvc_dir):
                 return root_dir
             if isinstance(fs, LocalFileSystem) and os.path.ismount(root_dir):
                 break
-            parent = fs.path.parent(root_dir)
+            parent = fs.parent(root_dir)
             if parent == root_dir:
                 break
             root_dir = parent
@@ -440,7 +446,7 @@ class Repo:
 
         fs = fs or localfs
         root_dir = cls.find_root(root, fs=fs)
-        return fs.path.join(root_dir, cls.DVC_DIR)
+        return fs.join(root_dir, cls.DVC_DIR)
 
     @staticmethod
     def init(root_dir=os.curdir, no_scm=False, force=False, subdir=False) -> "Repo":
@@ -492,6 +498,7 @@ class Repo:
         revs=None,
         num=1,
         push: bool = False,
+        skip_failed: bool = False,
     ):
         """Get the stages related to the given target and collect
         the `info` of its outputs.
@@ -510,7 +517,7 @@ class Repo:
         """
         used = defaultdict(set)
 
-        for _ in self.brancher(
+        for rev in self.brancher(
             revs=revs,
             all_branches=all_branches,
             all_tags=all_tags,
@@ -519,17 +526,23 @@ class Repo:
             commit_date=commit_date,
             num=num,
         ):
-            for odb, objs in self.index.used_objs(
-                targets,
-                remote=remote,
-                force=force,
-                jobs=jobs,
-                recursive=recursive,
-                with_deps=with_deps,
-                push=push,
-            ).items():
-                used[odb].update(objs)
-
+            try:
+                for odb, objs in self.index.used_objs(
+                    targets,
+                    remote=remote,
+                    force=force,
+                    jobs=jobs,
+                    recursive=recursive,
+                    with_deps=with_deps,
+                    push=push,
+                ).items():
+                    used[odb].update(objs)
+            except DvcException as exc:
+                rev = rev or "workspace"
+                if skip_failed:
+                    logger.warning("Failed to collect '%s', skipping", rev)
+                else:
+                    raise RevCollectionError(rev) from exc
         if used_run_cache:
             for odb, objs in self.stage_cache.get_used_objs(
                 used_run_cache, remote=remote, force=force, jobs=jobs
@@ -542,19 +555,19 @@ class Repo:
         # using `outs_graph` to ensure graph checks are run
         outs = outs or self.index.outs_graph
 
-        abs_path = self.fs.path.abspath(path)
+        abs_path = self.fs.abspath(path)
         fs_path = abs_path
 
         def func(out):
             def eq(one, two):
                 return one == two
 
-            match = eq if strict else out.fs.path.isin_or_eq
+            match = eq if strict else out.fs.isin_or_eq
 
             if out.protocol == "local" and match(fs_path, out.fs_path):
                 return True
 
-            if recursive and out.fs.path.isin(out.fs_path, fs_path):
+            if recursive and out.fs.isin(out.fs_path, fs_path):
                 return True
 
             return False
@@ -566,7 +579,7 @@ class Repo:
         return matched
 
     def is_dvc_internal(self, path):
-        path_parts = self.fs.path.normpath(path).split(self.fs.sep)
+        path_parts = self.fs.normpath(path).split(self.fs.sep)
         return self.DVC_DIR in path_parts
 
     @cached_property
@@ -613,10 +626,10 @@ class Repo:
         cache_dir = self.config["core"].get("site_cache_dir") or site_cache_dir()
 
         if isinstance(self.fs, GitFileSystem):
-            relparts = ()
+            relparts: Tuple[str, ...] = ()
             if self.root_dir != "/":
                 # subrepo
-                relparts = self.fs.path.relparts(self.root_dir, "/")
+                relparts = self.fs.relparts(self.root_dir, "/")
             root_dir = os.path.join(self.scm.root_dir, *relparts)
         else:
             root_dir = self.root_dir

dvc/repo/add.py

@@ -181,7 +181,7 @@ def _add_transfer(
 
 def _add(stage: "Stage", source: Optional[str] = None, no_commit: bool = False) -> None:
     out = stage.outs[0]
-    path = out.fs.path.abspath(source) if source else None
+    path = out.fs.abspath(source) if source else None
     try:
         stage.add_outs(path, no_commit=no_commit)
     except CacheLinkError:

dvc/repo/artifacts.py

@@ -229,7 +229,7 @@ class Artifacts:
                 dvc_studio_config=dvc_studio_config,
                 **kwargs,
             ).items():
-                to_info = localfs.path.join(out, path)
+                to_info = localfs.join(out, path)
                 if localfs.exists(to_info) and not force:
                     hint = "\nTo override it, re-run with '--force'."
                     raise FileExistsLocallyError(  # noqa: TRY301
@@ -254,7 +254,7 @@ class Artifacts:
                 fs, from_infos, localfs, to_infos, callback=cb, batch_size=jobs
             )
 
-        return len(to_infos), relpath(localfs.path.commonpath(to_infos))
+        return len(to_infos), relpath(localfs.commonpath(to_infos))
 
     @classmethod
     def get(  # noqa: PLR0913

dvc/repo/brancher.py

@@ -60,12 +60,12 @@ def brancher(
     from dvc.fs import LocalFileSystem
 
     repo_root_parts: Tuple[str, ...] = ()
-    if self.fs.path.isin(self.root_dir, self.scm.root_dir):
-        repo_root_parts = self.fs.path.relparts(self.root_dir, self.scm.root_dir)
+    if self.fs.isin(self.root_dir, self.scm.root_dir):
+        repo_root_parts = self.fs.relparts(self.root_dir, self.scm.root_dir)
 
     cwd_parts: Tuple[str, ...] = ()
-    if self.fs.path.isin(self.fs.path.getcwd(), self.scm.root_dir):
-        cwd_parts = self.fs.path.relparts(self.fs.path.getcwd(), self.scm.root_dir)
+    if self.fs.isin(self.fs.getcwd(), self.scm.root_dir):
+        cwd_parts = self.fs.relparts(self.fs.getcwd(), self.scm.root_dir)
 
     saved_fs = self.fs
     saved_root = self.root_dir
@@ -125,18 +125,18 @@ def _switch_fs(
     logger.trace("switching fs to revision %s", rev[:7])
     assert isinstance(repo.scm, Git)
     fs = GitFileSystem(scm=repo.scm, rev=rev)
-    root_dir = repo.fs.path.join("/", *repo_root_parts)
+    root_dir = repo.fs.join("/", *repo_root_parts)
     if not fs.exists(root_dir):
         raise NotDvcRepoError(f"Commit '{rev[:7]}' does not contain a DVC repo")
 
     repo.fs = fs
     repo.root_dir = root_dir
-    repo.dvc_dir = fs.path.join(root_dir, repo.DVC_DIR)
+    repo.dvc_dir = fs.join(root_dir, repo.DVC_DIR)
     repo._reset()
 
     if cwd_parts:
-        cwd = repo.fs.path.join("/", *cwd_parts)
-        repo.fs.path.chdir(cwd)
+        cwd = repo.fs.join("/", *cwd_parts)
+        repo.fs.chdir(cwd)
 
 
 @contextmanager
@@ -148,12 +148,12 @@ def switch(repo: "Repo", rev: str) -> Iterator[str]:
         rev = resolve_rev(repo.scm, rev)
 
     repo_root_parts: Tuple[str, ...] = ()
-    if repo.fs.path.isin(repo.root_dir, repo.scm.root_dir):
-        repo_root_parts = repo.fs.path.relparts(repo.root_dir, repo.scm.root_dir)
+    if repo.fs.isin(repo.root_dir, repo.scm.root_dir):
+        repo_root_parts = repo.fs.relparts(repo.root_dir, repo.scm.root_dir)
 
     cwd_parts: Tuple[str, ...] = ()
-    if repo.fs.path.isin(repo.fs.path.getcwd(), repo.scm.root_dir):
-        cwd_parts = repo.fs.path.relparts(repo.fs.path.getcwd(), repo.scm.root_dir)
+    if repo.fs.isin(repo.fs.getcwd(), repo.scm.root_dir):
+        cwd_parts = repo.fs.relparts(repo.fs.getcwd(), repo.scm.root_dir)
 
     saved_fs = repo.fs
     saved_root = repo.root_dir

dvc/repo/checkout.py

@@ -81,7 +81,7 @@ def _check_can_delete(
         if cache_fs.exists(cache_path):
             continue
 
-        entry_paths.append(fs.path.join(path, *(entry.key or ())))
+        entry_paths.append(fs.join(path, *(entry.key or ())))
 
     if not entry_paths:
         return
@@ -172,7 +172,7 @@ def checkout(  # noqa: C901
         )
 
         for out_path in out_paths:
-            if self.fs.path.isin_or_eq(dest_path, out_path):
+            if self.fs.isin_or_eq(dest_path, out_path):
                 failed.add(out_path)
 
     with ui.progress(
@@ -195,7 +195,7 @@ def checkout(  # noqa: C901
 
     typ_map = {ADD: "added", DELETE: "deleted", MODIFY: "modified"}
     for key, typ in out_changes.items():
-        out_path = self.fs.path.join(self.root_dir, *key)
+        out_path = self.fs.join(self.root_dir, *key)
 
         if out_path in failed:
             self.fs.remove(out_path, recursive=True)

dvc/repo/data.py

@@ -205,7 +205,7 @@ class Status(TypedDict):
 
 def _transform_git_paths_to_dvc(repo: "Repo", files: Iterable[str]) -> List[str]:
     """Transform files rel. to Git root to DVC root, and drop outside files."""
-    rel = repo.fs.path.relpath(repo.root_dir, repo.scm.root_dir).rstrip("/")
+    rel = repo.fs.relpath(repo.root_dir, repo.scm.root_dir).rstrip("/")
 
     # if we have repo root in a different location than scm's root,
     # i.e. subdir repo, all git_paths need to be transformed rel. to the DVC
@@ -215,11 +215,11 @@ def _transform_git_paths_to_dvc(repo: "Repo", files: Iterable[str]) -> List[str]
         length = len(prefix)
         files = (file[length:] for file in files if file.startswith(prefix))
 
-    start = repo.fs.path.relpath(repo.fs.path.getcwd(), repo.root_dir)
+    start = repo.fs.relpath(repo.fs.getcwd(), repo.root_dir)
     if start in (os.curdir, ""):
         return list(files)
     # we need to convert repo relative paths to curdir relative.
-    return [repo.fs.path.relpath(file, start) for file in files]
+    return [repo.fs.relpath(file, start) for file in files]
 
 
 def status(

dvc/repo/experiments/stash.py

@@ -171,7 +171,7 @@ class ApplyStash(Stash):
         right_fs = self.scm.get_fs(right_rev)
         paths = [path for path in left_fs.find("/") if not right_fs.exists(path)]
         dest_paths = [
-            localfs.path.join(self.scm.root_dir, left_fs.path.relpath(path, "/"))
+            localfs.join(self.scm.root_dir, left_fs.relpath(path, "/"))
             for path in paths
         ]
         for src, dest in zip(paths, dest_paths):

dvc/repo/fetch.py

@@ -219,7 +219,7 @@ def _log_unversioned(data: List["DataIndex"]) -> Tuple[List["DataIndex"], int]:
         index.storage_map = fs_index.storage_map
         for key, entry in fs_index.iteritems():
             if entry.meta and not entry.meta.isdir and entry.meta.version_id is None:
-                unversioned.append(fs.unstrip_protocol(fs.path.join(remote.path, *key)))
+                unversioned.append(fs.unstrip_protocol(fs.join(remote.path, *key)))
             else:
                 index[key] = entry
         fs_index.close()

dvc/repo/gc.py

@@ -68,6 +68,7 @@ def gc(  # noqa: PLR0913, C901
     num: Optional[int] = None,
     not_in_remote: bool = False,
     dry: bool = False,
+    skip_failed: bool = False,
 ):
     # require `workspace` to be true to come into effect.
     # assume `workspace` to be enabled if any of `all_tags`, `all_commits`,
@@ -113,6 +114,7 @@ def gc(  # noqa: PLR0913, C901
                 jobs=jobs,
                 revs=[rev] if rev else None,
                 num=num or 1,
+                skip_failed=skip_failed,
             ).items():
                 if odb not in odb_to_obj_ids:
                     odb_to_obj_ids[odb] = set()

dvc/repo/graph.py

@@ -126,7 +126,7 @@ def build_graph(stages, outs_trie=None):
     outs_trie = outs_trie or build_outs_trie(stages)
 
     for stage in stages:
-        out = outs_trie.shortest_prefix(localfs.path.parts(stage.path)).value
+        out = outs_trie.shortest_prefix(localfs.parts(stage.path)).value
         if out:
             raise StagePathAsOutputError(stage, str(out))
 
@@ -139,7 +139,7 @@ def build_graph(stages, outs_trie=None):
             continue
 
         for dep in stage.deps:
-            dep_key = dep.fs.path.parts(dep.fs_path)
+            dep_key = dep.fs.parts(dep.fs_path)
             overlapping = [n.value for n in outs_trie.prefixes(dep_key)]
             if outs_trie.has_subtrie(dep_key):
                 overlapping.extend(outs_trie.values(prefix=dep_key))
@@ -165,7 +165,7 @@ def build_outs_graph(graph, outs_trie):
         if stage.is_db_import:
             continue
         for dep in stage.deps:
-            dep_key = dep.fs.path.parts(dep.fs_path)
+            dep_key = dep.fs.parts(dep.fs_path)
             overlapping = [n.value for n in outs_trie.prefixes(dep_key)]
             if outs_trie.has_subtrie(dep_key):
                 overlapping.extend(outs_trie.values(prefix=dep_key))

dvc/repo/index.py

@@ -97,7 +97,7 @@ def collect_files(
     for root, dirs, files in walk_iter:
         dvcfile_filter = partial(is_dvcfile_and_not_ignored, root)
         for file in filter(dvcfile_filter, files):
-            file_path = fs.path.join(root, file)
+            file_path = fs.join(root, file)
             try:
                 index = Index.from_file(repo, file_path)
             except DvcException as exc:
@@ -129,10 +129,25 @@ def _load_data_from_outs(index, prefix, outs):
         for key_len in range(1, len(key)):
             parents.add((ws, key[:key_len]))
 
+        loaded = None
+        if out.files:
+            loaded = True
+            for okey, ometa, ohi in out.get_obj():
+                for key_len in range(1, len(okey)):
+                    parents.add((ws, (*key, *okey[:key_len])))
+
+                fkey = (*key, *okey)
+                index[(*prefix, ws, *fkey)] = DataIndexEntry(
+                    key=fkey,
+                    meta=ometa,
+                    hash_info=ohi,
+                )
+
         entry = DataIndexEntry(
             key=key,
             meta=out.meta,
             hash_info=out.hash_info,
+            loaded=loaded,
         )
 
         if (
@@ -201,12 +216,12 @@ def _load_storage_from_out(storage_map, key, out):
                 FileStorage(
                     key,
                     fs_cache.fs,
-                    fs_cache.fs.path.join(
+                    fs_cache.fs.join(
                         fs_cache.path, dep.fs.protocol, tokenize(dep.fs_path)
                     ),
                 )
             )
-        storage_map.add_remote(FileStorage(key, dep.fs, dep.fs_path))
+        storage_map.add_remote(FileStorage(key, dep.fs, dep.fs_path, read_only=True))
 
 
 class Index:
@@ -414,7 +429,7 @@ class Index:
             by_workspace[workspace].add(key)
 
         for path in _collect_top_level_metrics(self.repo):
-            key = self.repo.fs.path.relparts(path, self.repo.root_dir)
+            key = self.repo.fs.relparts(path, self.repo.root_dir)
             by_workspace["repo"].add(key)
 
         return dict(by_workspace)
@@ -432,7 +447,7 @@ class Index:
             param_paths = chain(param_paths, [default_file])
 
         for path in param_paths:
-            key = self.repo.fs.path.relparts(path, self.repo.root_dir)
+            key = self.repo.fs.relparts(path, self.repo.root_dir)
             by_workspace["repo"].add(key)
 
         return dict(by_workspace)
@@ -451,7 +466,7 @@ class Index:
             by_workspace[workspace].add(key)
 
         for path in self._plot_sources:
-            key = self.repo.fs.path.parts(path)
+            key = self.repo.fs.parts(path)
             by_workspace["repo"].add(key)
 
         return dict(by_workspace)
@@ -701,8 +716,8 @@ class IndexView:
             if not out.use_cache:
                 continue
             workspace, key = out.index_key
-            if filter_info and out.fs.path.isin(filter_info, out.fs_path):
-                key = key + out.fs.path.relparts(filter_info, out.fs_path)
+            if filter_info and out.fs.isin(filter_info, out.fs_path):
+                key = key + out.fs.relparts(filter_info, out.fs_path)
             entry = self._index.data[workspace].get(key)
             if entry and entry.meta and entry.meta.isdir:
                 prefixes[workspace].recursive.add(key)
@@ -718,8 +733,8 @@ class IndexView:
                 continue
 
             workspace, key = out.index_key
-            if filter_info and out.fs.path.isin(filter_info, out.fs_path):
-                key = key + out.fs.path.relparts(filter_info, out.fs_path)
+            if filter_info and out.fs.isin(filter_info, out.fs_path):
+                key = key + out.fs.relparts(filter_info, out.fs_path)
             ret[workspace].add(key)
 
         return dict(ret)
@@ -782,7 +797,7 @@ def build_data_index(  # noqa: C901, PLR0912
     data = DataIndex()
     parents = set()
     for key in index.data_keys.get(workspace, set()):
-        out_path = fs.path.join(path, *key)
+        out_path = fs.join(path, *key)
 
         for key_len in range(1, len(key)):
             parents.add(key[:key_len])
@@ -827,7 +842,7 @@ def build_data_index(  # noqa: C901, PLR0912
             callback.relative_update(1)
 
     for key in parents:
-        parent_path = fs.path.join(path, *key)
+        parent_path = fs.join(path, *key)
         if not fs.exists(parent_path):
             continue
         direntry = DataIndexEntry(key=key, meta=Meta(isdir=True), loaded=True)

dvc/repo/ls.py

@@ -89,7 +89,7 @@ def _ls(
         if not recursive:
             files.update(dirs)
 
-        parts = fs.path.relparts(root, fs_path)
+        parts = fs.relparts(root, fs_path)
         if parts == (".",):
             parts = ()
 

dvc/repo/ls_url.py

@@ -18,7 +18,7 @@ def ls_url(url, *, fs_config=None, recursive=False, config=None):
 
         for info in files.values():
             ls_info = {
-                "path": fs.path.relpath(info["name"], fs_path),
+                "path": fs.relpath(info["name"], fs_path),
                 "isdir": info["type"] == "directory",
                 "size": info.get("size"),
             }

dvc/repo/metrics/show.py

@@ -17,7 +17,6 @@ from funcy import ldistinct
 from scmrepo.exceptions import SCMError
 
 from dvc.log import logger
-from dvc.repo import locked
 from dvc.scm import NoSCMError
 from dvc.utils import as_posix, expand_paths
 from dvc.utils.collections import ensure_list
@@ -35,10 +34,10 @@ logger = logger.getChild(__name__)
 def _collect_top_level_metrics(repo: "Repo") -> Iterator[str]:
     top_metrics = repo.index._metrics
     for dvcfile, metrics in top_metrics.items():
-        wdir = repo.fs.path.relpath(repo.fs.path.parent(dvcfile), repo.root_dir)
+        wdir = repo.fs.relpath(repo.fs.parent(dvcfile), repo.root_dir)
         for file in metrics:
-            path = repo.fs.path.join(wdir, as_posix(file))
-            yield repo.fs.path.normpath(path)
+            path = repo.fs.join(wdir, as_posix(file))
+            yield repo.fs.normpath(path)
 
 
 def _extract_metrics(metrics, path: str):
@@ -101,7 +100,7 @@ def _collect_metrics(
 
     if not targets or outs_only:
         outs = metrics_from_target(repo, stages) if stages else repo.index.metrics
-        relpath = repo.fs.path.relpath
+        relpath = repo.fs.relpath
         metrics.extend(relpath(out.fs_path, repo.root_dir) for out in outs)
 
     if not targets and not outs_only and not stages:
@@ -128,8 +127,8 @@ class Result(TypedDict, total=False):
 
 
 def to_relpath(fs: "FileSystem", root_dir: str, d: Result) -> Result:
-    relpath = fs.path.relpath
-    cwd = fs.path.getcwd()
+    relpath = fs.relpath
+    cwd = fs.getcwd()
 
     start = relpath(cwd, root_dir)
     data = d.get("data")
@@ -156,7 +155,7 @@ def _gather_metrics(
     fs = repo.dvcfs
     for fs_path, result in _read_metrics(fs, files, cache=True):
         repo_path = fs_path.lstrip(fs.root_marker)
-        repo_os_path = os.sep.join(fs.path.parts(repo_path))
+        repo_os_path = os.sep.join(fs.parts(repo_path))
         if not isinstance(result, Exception):
             data.update({repo_os_path: FileResult(data=result)})
             continue
@@ -185,7 +184,6 @@ def _hide_workspace(
     return res
 
 
-@locked
 def show(
     repo: "Repo",
     targets: Optional[List[str]] = None,

dvc/repo/open_repo.py

@@ -71,7 +71,7 @@ def erepo_factory(url, root_dir, cache_config):
         _config = cache_config.copy()
         if os.path.isdir(url):
             fs = fs or localfs
-            repo_path = os.path.join(url, *fs.path.relparts(path, root_dir))
+            repo_path = os.path.join(url, *fs.relparts(path, root_dir))
             _config.update(_get_remote_config(repo_path))
         return Repo(path, fs=fs, config=_config, **_kwargs)
 

dvc/repo/params/show.py

@@ -5,7 +5,6 @@ from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Tuple, Un
 
 from dvc.dependency.param import ParamsDependency, read_param_file
 from dvc.log import logger
-from dvc.repo import locked
 from dvc.repo.metrics.show import FileResult, Result
 from dvc.stage import PipelineStage
 from dvc.utils import as_posix, expand_paths
@@ -21,10 +20,10 @@ logger = logger.getChild(__name__)
 def _collect_top_level_params(repo: "Repo") -> Iterator[str]:
     top_params = repo.index._params
     for dvcfile, params in top_params.items():
-        wdir = repo.fs.path.relpath(repo.fs.path.parent(dvcfile), repo.root_dir)
+        wdir = repo.fs.relpath(repo.fs.parent(dvcfile), repo.root_dir)
         for file in params:
-            path = repo.fs.path.join(wdir, as_posix(file))
-            yield repo.fs.path.normpath(path)
+            path = repo.fs.join(wdir, as_posix(file))
+            yield repo.fs.normpath(path)
 
 
 def params_from_target(
@@ -55,7 +54,7 @@ def _collect_params(
 
     if not targets or stages:
         deps = params_from_target(repo, stages) if stages else repo.index.params
-        relpath = repo.fs.path.relpath
+        relpath = repo.fs.relpath
         params.extend(
             {relpath(dep.fs_path, repo.root_dir): list(dep.params)} for dep in deps
         )
@@ -93,7 +92,7 @@ def _collect_vars(repo, params, stages=None) -> Dict:
                 # to reduce noise and duplication, they are skipped
 
                 # `file` is relative
-                abspath = repo.fs.path.abspath(file)
+                abspath = repo.fs.abspath(file)
                 repo_path = repo.dvcfs.from_os_path(abspath)
                 if repo_path in params:
                     continue
@@ -138,7 +137,7 @@ def _gather_params(
     fs = repo.dvcfs
     for fs_path, result in _read_params(fs, files_keypaths, cache=True):
         repo_path = fs_path.lstrip(fs.root_marker)
-        repo_os_path = os.sep.join(fs.path.parts(repo_path))
+        repo_os_path = os.sep.join(fs.parts(repo_path))
         if not isinstance(result, Exception):
             data.update({repo_os_path: FileResult(data=result)})
             continue
@@ -158,7 +157,6 @@ def _gather_params(
     return data
 
 
-@locked
 def show(
     repo: "Repo",
     targets: Optional[List[str]] = None,

dvc/repo/plots/__init__.py

@@ -386,7 +386,7 @@ def _matches(targets, config_file, plot_id):
 
 
 def _normpath(path):
-    # TODO dvcfs.path.normopath normalizes to windows path on Windows
+    # TODO dvcfs.normopath normalizes to windows path on Windows
     # even though other methods work as expected
     import posixpath
 
@@ -399,7 +399,7 @@ def _relpath(fs, path):
     # and invoking from some subdir `dvcfile.relpath` returns strange long
     # relative paths
     # ("../../../../../../dvc.yaml") - investigate
-    return fs.path.relpath(fs.path.join("/", fs.from_os_path(path)), fs.path.getcwd())
+    return fs.relpath(fs.join("/", fs.from_os_path(path)), fs.getcwd())
 
 
 def _collect_output_plots(repo, targets, props, onerror: Optional[Callable] = None):
@@ -413,7 +413,7 @@ def _collect_output_plots(repo, targets, props, onerror: Optional[Callable] = No
         if _matches(targets, config_path, str(plot)):
             unpacked = unpack_if_dir(
                 fs,
-                _normpath(fs.path.join(wdir_relpath, plot.def_path)),
+                _normpath(fs.join(wdir_relpath, plot.def_path)),
                 props={**plot_props, **props},
                 onerror=onerror,
             )
@@ -438,7 +438,7 @@ def _adjust_sources(fs, plot_props, config_dir):
     old_y = new_plot_props.pop("y", {})
     new_y = {}
     for filepath, val in old_y.items():
-        new_y[_normpath(fs.path.join(config_dir, filepath))] = val
+        new_y[_normpath(fs.join(config_dir, filepath))] = val
     new_plot_props["y"] = new_y
     return new_plot_props
 
@@ -452,13 +452,13 @@ def _resolve_definitions(
     onerror: Optional[Callable[[Any], Any]] = None,
 ):
     config_path = os.fspath(config_path)
-    config_dir = fs.path.dirname(config_path)
+    config_dir = fs.dirname(config_path)
     result: Dict[str, Dict] = {}
     for plot_id, plot_props in definitions.items():
         if plot_props is None:
             plot_props = {}
         if _id_is_path(plot_props):
-            data_path = _normpath(fs.path.join(config_dir, plot_id))
+            data_path = _normpath(fs.join(config_dir, plot_id))
             if _matches(targets, config_path, plot_id):
                 unpacked = unpack_if_dir(
                     fs,

dvc/repo/push.py

@@ -6,26 +6,49 @@ from dvc.ui import ui
 from . import locked
 
 
+def _rebuild(idx, path, fs, cb):
+    from dvc_data.index import DataIndex, DataIndexEntry, Meta
+
+    new = DataIndex()
+    items = list(idx.items())
+
+    cb.set_size(len(items))
+    for key, entry in items:
+        if entry.meta and entry.meta.isdir:
+            meta = Meta(isdir=True)
+        else:
+            try:
+                meta = Meta.from_info(fs.info(fs.join(path, *key)), fs.protocol)
+            except FileNotFoundError:
+                meta = None
+
+        if meta:
+            new.add(DataIndexEntry(key=key, meta=meta))
+
+        cb.relative_update(1)
+
+    return new
+
+
 def _update_meta(index, **kwargs):
-    from dvc.repo.index import build_data_index
     from dvc.repo.worktree import _merge_push_meta, worktree_view_by_remotes
 
     stages = set()
     for remote_name, idx in worktree_view_by_remotes(index, push=True, **kwargs):
         remote = index.repo.cloud.get_remote(remote_name)
 
-        with ui.progress("Collecting", unit="entry") as pb:
-            new = build_data_index(
-                idx,
-                remote.path,
-                remote.fs,
-                callback=pb.as_callback(),
-            )
+        if not remote.fs.version_aware:
+            continue
+
+        with ui.progress(
+            desc=f"Collecting {remote.path} on {remote.fs.protocol}",
+            unit="entry",
+            leave=True,
+        ) as pb:
+            cb = pb.as_callback()
+            new = _rebuild(idx.data["repo"], remote.path, remote.fs, cb)
 
         for out in idx.outs:
-            if not remote.fs.version_aware:
-                continue
-
             _merge_push_meta(out, new, remote.name)
             stages.add(out.stage)
 
@@ -100,6 +123,7 @@ def push(  # noqa: PLR0913
     with ui.progress(
         desc="Collecting",
         unit="entry",
+        leave=True,
     ) as pb:
         data = collect(
             [idx.data["repo"] for idx in indexes.values()],
@@ -114,7 +138,8 @@ def push(  # noqa: PLR0913
     try:
         with ui.progress(
             desc="Pushing",
-            unit="file",
+            bar_format="{desc}",
+            leave=True,
         ) as pb:
             push_transferred, push_failed = ipush(
                 data,

dvc/repo/stage.py

@@ -219,7 +219,7 @@ class StageLoad:
         self, path: Optional[str] = None, name: Optional[str] = None
     ) -> str:
         if path:
-            return self.repo.fs.path.abspath(path)
+            return self.repo.fs.abspath(path)
 
         path = PROJECT_FILE
         logger.debug("Assuming '%s' to be a stage inside '%s'", name, path)
@@ -348,7 +348,7 @@ class StageLoad:
         if recursive and self.fs.isdir(target):
             from dvc.repo.graph import collect_inside_path
 
-            path = self.fs.path.abspath(target)
+            path = self.fs.abspath(target)
             return collect_inside_path(path, graph or self.repo.index.graph)
 
         stages = self.from_target(target, glob=glob)
@@ -393,7 +393,7 @@ class StageLoad:
             if not (recursive and self.fs.isdir(target)):
                 try:
                     (out,) = self.repo.find_outs_by_path(target, strict=False)
-                    return [StageInfo(out.stage, self.fs.path.abspath(target))]
+                    return [StageInfo(out.stage, self.fs.abspath(target))]
                 except OutputNotFoundError:
                     pass
 

dvc/repo/trie.py

@@ -9,7 +9,7 @@ def build_outs_trie(stages):
 
     for stage in stages:
         for out in stage.outs:
-            out_key = out.fs.path.parts(out.fs_path)
+            out_key = out.fs.parts(out.fs_path)
 
             # Check for dup outs
             if out_key in outs:

dvc/repo/worktree.py

@@ -131,10 +131,11 @@ def _merge_push_meta(  # noqa: C901
             entries.append(entry)
             if entry.meta is not None and entry.meta.isdir:
                 continue
-            fs_path = repo.fs.path.join(repo.root_dir, *subkey)
-            meta, hash_info = old_tree.get(
-                repo.fs.path.relparts(fs_path, out.fs_path)
-            ) or (None, None)
+            fs_path = repo.fs.join(repo.root_dir, *subkey)
+            meta, hash_info = old_tree.get(repo.fs.relparts(fs_path, out.fs_path)) or (
+                None,
+                None,
+            )
             entry.hash_info = hash_info
             if entry.meta:
                 entry.meta.remote = remote
@@ -354,11 +355,11 @@ def _get_update_diff_index(
             # downloading
             if out.isdir():
                 if not entry.meta.isdir:
-                    fs_path = repo.fs.path.join(repo.root_dir, *entry.key)
+                    fs_path = repo.fs.join(repo.root_dir, *entry.key)
                     tree = out.obj
                     assert isinstance(tree, Tree)
                     _, entry.hash_info = tree.get(  # type: ignore[misc]
-                        repo.fs.path.relparts(fs_path, out.fs_path)
+                        repo.fs.relparts(fs_path, out.fs_path)
                     )
             else:
                 entry.hash_info = out.hash_info

dvc/rwlock.py

@@ -43,10 +43,10 @@ class RWLockFileFormatError(DvcException):
 
 @contextmanager
 def _edit_rwlock(lock_dir, fs, hardlink):
-    path = fs.path.join(lock_dir, RWLOCK_FILE)
+    path = fs.join(lock_dir, RWLOCK_FILE)
 
     rwlock_guard = make_lock(
-        fs.path.join(lock_dir, RWLOCK_LOCK),
+        fs.join(lock_dir, RWLOCK_LOCK),
         tmp_dir=lock_dir,
         hardlink_lock=hardlink,
     )
@@ -82,7 +82,7 @@ def _check_blockers(tmp_dir, lock, info, *, mode, waiters):  # noqa: C901, PLR09
     to_release = defaultdict(list)
     for path, infos in lock[mode].items():
         for waiter_path in waiters:
-            if localfs.path.overlaps(waiter_path, path):
+            if localfs.overlaps(waiter_path, path):
                 break
         else:
             continue

dvc/stage/__init__.py

@@ -656,7 +656,7 @@ class Stage(params.StageParams):
 
     def filter_outs(self, fs_path) -> Iterable["Output"]:
         def _func(o):
-            return o.fs.path.isin_or_eq(fs_path, o.fs_path)
+            return o.fs.isin_or_eq(fs_path, o.fs_path)
 
         return filter(_func, self.outs) if fs_path else self.outs
 

dvc/stage/cache.py

@@ -181,9 +181,9 @@ class StageCache:
 
         path = self._get_cache_path(cache_key, cache_value)
         local_fs = self.repo.cache.legacy.fs
-        parent = local_fs.path.parent(path)
+        parent = local_fs.parent(path)
         self.repo.cache.legacy.makedirs(parent)
-        tmp = local_fs.path.join(parent, fs.utils.tmp_fname())
+        tmp = local_fs.join(parent, fs.utils.tmp_fname())
         assert os.path.exists(parent)
         assert os.path.isdir(parent)
         dump_yaml(tmp, cache)
@@ -236,7 +236,7 @@ class StageCache:
         from_fs = from_odb.fs
         to_fs = to_odb.fs
         func = fs.generic.log_exceptions(fs.generic.copy)
-        runs = from_fs.path.join(from_odb.path, "runs")
+        runs = from_fs.join(from_odb.path, "runs")
 
         http_odb = next(
             (odb for odb in (from_odb, to_odb) if isinstance(odb.fs, HTTPFileSystem)),
@@ -252,20 +252,20 @@ class StageCache:
             return ret
 
         for src in from_fs.find(runs):
-            rel = from_fs.path.relpath(src, from_odb.path)
+            rel = from_fs.relpath(src, from_odb.path)
             if not isinstance(to_fs, LocalFileSystem):
-                rel = from_fs.path.as_posix(rel)
+                rel = from_fs.as_posix(rel)
 
-            dst = to_fs.path.join(to_odb.path, rel)
-            key = to_fs.path.parent(dst)
+            dst = to_fs.join(to_odb.path, rel)
+            key = to_fs.parent(dst)
             # check if any build cache already exists for this key
             # TODO: check if MaxKeys=1 or something like that applies
             # or otherwise this will take a lot of time!
             if to_fs.exists(key) and first(to_fs.find(key)):
                 continue
 
-            src_name = from_fs.path.name(src)
-            parent_name = from_fs.path.name(from_fs.path.parent(src))
+            src_name = from_fs.name(src)
+            parent_name = from_fs.name(from_fs.parent(src))
             with TqdmCallback(
                 desc=src_name,
                 bytes=True,

dvc/stage/utils.py

@@ -184,9 +184,9 @@ def resolve_wdir(wdir, path):
 
 
 def resolve_paths(fs, path, wdir=None):
-    path = fs.path.abspath(path)
+    path = fs.abspath(path)
     wdir = wdir or os.curdir
-    wdir = fs.path.abspath(fs.path.join(fs.path.dirname(path), wdir))
+    wdir = fs.abspath(fs.join(fs.dirname(path), wdir))
     return path, wdir
 
 

dvc/testing/cloud.py

@@ -32,6 +32,14 @@ class Cloud(ABC):
     def write_bytes(self, contents):
         raise NotImplementedError
 
+    @abstractmethod
+    def unlink(self, missing_ok: bool = False) -> None:
+        pass
+
+    @abstractmethod
+    def rmdir(self, recursive: bool = True) -> None:
+        pass
+
     def read_text(self, encoding=None, errors=None):
         if not encoding:
             encoding = locale.getpreferredencoding(False)

dvc/testing/remote_tests.py

@@ -171,7 +171,7 @@ class TestRemoteVersionAware:
         (stage,) = tmp_dir.dvc_gen("foo", "foo")
         run_copy("foo", "foo_copy", name="copy")
 
-        dvc.push()
+        assert dvc.push()
         assert (remote_version_aware / "foo").read_text() == "foo"
         assert (remote_version_aware / "foo_copy").read_text() == "foo"
         foo_dvc = (tmp_dir / "foo.dvc").read_text()
@@ -185,26 +185,26 @@ class TestRemoteVersionAware:
         remove(tmp_dir / "foo")
         remove(tmp_dir / "foo_copy")
 
-        dvc.pull()
+        assert dvc.pull()
         assert (tmp_dir / "foo").read_text() == "foo"
         assert (tmp_dir / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
-        dvc.push()
+        assert not dvc.push()
         assert (remote_version_aware / "foo").read_text() == "foo"
         assert (remote_version_aware / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
         dvc.reproduce()
-        dvc.push()
+        assert not dvc.push()
         assert (remote_version_aware / "foo").read_text() == "foo"
         assert (remote_version_aware / "foo_copy").read_text() == "foo"
         assert (tmp_dir / "foo.dvc").read_text() == foo_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
-    def test_dir(self, tmp_dir, dvc, run_copy, remote_version_aware):
+    def test_dir(self, tmp_dir, dvc, run_copy, remote_version_aware):  # noqa: PLR0915
         (stage,) = tmp_dir.dvc_gen(
             {
                 "data_dir": {
@@ -215,7 +215,8 @@ class TestRemoteVersionAware:
             }
         )
 
-        dvc.push()
+        assert not dvc.fetch()
+        assert dvc.push()
 
         data_dir_dvc = (tmp_dir / "data_dir.dvc").read_text()
         assert "files" in data_dir_dvc
@@ -230,7 +231,7 @@ class TestRemoteVersionAware:
         remove(dvc.cache.local.path)
         remove(tmp_dir / "data_dir")
 
-        dvc.pull()
+        assert dvc.pull()
         assert (tmp_dir / "data_dir" / "data").read_text() == "data"
         assert (
             tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
@@ -240,20 +241,47 @@ class TestRemoteVersionAware:
         run_copy("data_dir", "data_dir_copy", name="copy")
         dvc_lock = (tmp_dir / "dvc.lock").read_text()
 
-        dvc.push()
+        assert dvc.push()
+        assert (remote_version_aware / "data_dir").exists()
+        assert (remote_version_aware / "data_dir" / "data").exists()
+        assert (remote_version_aware / "data_dir_copy").exists()
+        assert (remote_version_aware / "data_dir_copy" / "data").exists()
         assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
         assert (tmp_dir / "dvc.lock").read_text() != dvc_lock
         dvc_lock = (tmp_dir / "dvc.lock").read_text()
 
-        dvc.push()
+        assert not dvc.push()
+        assert (remote_version_aware / "data_dir").exists()
+        assert (remote_version_aware / "data_dir" / "data").exists()
+        assert (remote_version_aware / "data_dir_copy").exists()
+        assert (remote_version_aware / "data_dir_copy" / "data").exists()
         assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
         assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
 
         dvc.cache.local.clear()
         remove(tmp_dir / "data_dir")
-        dvc.push()
+        remove(tmp_dir / "data_dir_copy")
+        assert not dvc.push()
         assert (remote_version_aware / "data_dir").exists()
         assert (remote_version_aware / "data_dir" / "data").exists()
+        assert (remote_version_aware / "data_dir_copy").exists()
+        assert (remote_version_aware / "data_dir_copy" / "data").exists()
+        assert (tmp_dir / "data_dir.dvc").read_text() == data_dir_dvc
+        assert (tmp_dir / "dvc.lock").read_text() == dvc_lock
+
+        (remote_version_aware / "data_dir").rmdir()
+        (remote_version_aware / "data_dir_copy").rmdir()
+        assert not (remote_version_aware / "data_dir").exists()
+        assert not (remote_version_aware / "data_dir_copy").exists()
+        assert dvc.pull()
+        assert (tmp_dir / "data_dir" / "data").read_text() == "data"
+        assert (
+            tmp_dir / "data_dir" / "data_sub_dir" / "data_sub"
+        ).read_text() == "data_sub"
+        assert (tmp_dir / "data_dir_copy" / "data").read_text() == "data"
+        assert (
+            tmp_dir / "data_dir_copy" / "data_sub_dir" / "data_sub"
+        ).read_text() == "data_sub"
 
 
 class TestRemoteWorktree:

dvc/testing/workspace_tests.py

@@ -163,8 +163,8 @@ class TestImportURLVersionAware:
 
 
 def match_files(fs, entries, expected):
-    entries_content = {(fs.path.normpath(d["path"]), d["isdir"]) for d in entries}
-    expected_content = {(fs.path.normpath(d["path"]), d["isdir"]) for d in expected}
+    entries_content = {(fs.normpath(d["path"]), d["isdir"]) for d in entries}
+    expected_content = {(fs.normpath(d["path"]), d["isdir"]) for d in expected}
     assert entries_content == expected_content
 
 
@@ -177,7 +177,7 @@ class TestLsUrl:
         match_files(
             fs,
             result,
-            [{"path": fs.path.join(fs_path, fname), "isdir": False}],
+            [{"path": fs.join(fs_path, fname), "isdir": False}],
         )
 
     def test_dir(self, cloud):

dvc/utils/__init__.py

@@ -287,7 +287,7 @@ def resolve_paths(repo, out, always_local=False):
         # urlparse interprets windows drive letters as URL scheme
         scheme = ""
 
-    if scheme or not localfs.path.isin_or_eq(abspath, repo.root_dir):
+    if scheme or not localfs.isin_or_eq(abspath, repo.root_dir):
         wdir = os.getcwd()
     elif contains_symlink_up_to(dirname, repo.root_dir) or (
         os.path.isdir(abspath) and localfs.is_symlink(abspath)

dvc/utils/plots.py

@@ -9,7 +9,7 @@ def get_plot_id(config_plot_id: str, config_file_path: str = ""):
 
 
 def group_definitions_by_id(
-    definitions: Dict[str, Dict]
+    definitions: Dict[str, Dict],
 ) -> Dict[str, Tuple[str, Dict]]:
     """
     Format ID and extracts plot_definition for each plot.

dvc/utils/serialize/__init__.py

@@ -21,7 +21,7 @@ PARSERS.update(
 
 
 def load_path(fs_path, fs, **kwargs):
-    suffix = fs.path.suffix(fs_path).lower()
+    suffix = fs.suffix(fs_path).lower()
     loader = LOADERS[suffix]
     return loader(fs_path, fs=fs, **kwargs)
 

pyproject.toml

@@ -35,7 +35,7 @@ dependencies = [
     "configobj>=5.0.6",
     "distro>=1.3",
     "dpath<3,>=2.1.0",
-    "dvc-data>=2.24,<2.25",
+    "dvc-data>=3.5,<3.6",
     "dvc-http>=2.29.0",
     "dvc-render>=1.0.0,<2",
     "dvc-studio-client>=0.17.1,<1",
@@ -45,7 +45,7 @@ dependencies = [
     "flufl.lock>=5,<8",
     "funcy>=1.14",
     "grandalf<1,>=0.7",
-    "gto>=1.4.0,<2",
+    "gto>=1.6.0,<2",
     "hydra-core>=1.1",
     "iterative-telemetry>=0.0.7",
     "networkx>=2.5",
@@ -59,7 +59,7 @@ dependencies = [
     "requests>=2.22",
     "rich>=12",
     "ruamel.yaml>=0.17.11",
-    "scmrepo>=1.5.0,<2",
+    "scmrepo>=2.0.2,<3",
     "shortuuid>=0.5",
     "shtab<2,>=1.3.4",
     "tabulate>=0.8.7",
@@ -71,13 +71,13 @@ dependencies = [
 
 [project.optional-dependencies]
 all = ["dvc[azure,gdrive,gs,hdfs,oss,s3,ssh,webdav,webhdfs]"]
-azure = ["dvc-azure>=2.23.0"]
+azure = ["dvc-azure>=3,<4"]
 dev = ["dvc[azure,gdrive,gs,hdfs,lint,oss,s3,ssh,tests,webdav,webhdfs]"]
-gdrive = ["dvc-gdrive==2.20"]
-gs = ["dvc-gs==2.22.1"]
-hdfs = ["dvc-hdfs==2.19"]
+gdrive = ["dvc-gdrive>=3,<4"]
+gs = ["dvc-gs>=3,<4"]
+hdfs = ["dvc-hdfs>=3,<4"]
 lint = [
-    "mypy==1.7.1",
+    "mypy==1.8.0",
     "pandas-stubs",
     "types-colorama",
     "types-psutil",
@@ -87,10 +87,10 @@ lint = [
     "types-toml",
     "types-tqdm",
 ]
-oss = ["dvc-oss>=2.22.0"]
-s3 = ["dvc-s3==2.23.0"]
-ssh = ["dvc-ssh>=2.22.1,<3"]
-ssh_gssapi = ["dvc-ssh[gssapi]>=2.22.1,<3"]
+oss = ["dvc-oss>=3,<4"]
+s3 = ["dvc-s3>=3,<4"]
+ssh = ["dvc-ssh>=3,<4"]
+ssh_gssapi = ["dvc-ssh[gssapi]>=3,<4"]
 testing = [
   "pytest-benchmark[histogram]",
   "pytest-test-utils",
@@ -116,9 +116,9 @@ tests = [
     "sqlalchemy>=1,<3", # optional dependency for `import-db`
     "pandas>=1",
 ]
-webdav = ["dvc-webdav==2.19.1"]
-webhdfs = ["dvc-webhdfs==2.19"]
-webhdfs_kerberos = ["dvc-webhdfs[kerberos]==2.19"]
+webdav = ["dvc-webdav>=3,<4"]
+webhdfs = ["dvc-webhdfs>=3.1,<4"]
+webhdfs_kerberos = ["dvc-webhdfs[kerberos]>=3.1,<4"]
 
 [project.urls]
 Documentation = "https://dvc.org/doc"

tests/func/repro/test_repro.py

@@ -1331,10 +1331,10 @@ def test_repro_keep_going(mocker, tmp_dir, dvc, copy_script):
         stage1, upstream=[bar_stage], force=False, interactive=False
     )
     foo_call = mocker.call(foo_stage, upstream=[], force=False, interactive=False)
-    assert spy.call_args_list in (
-        [bar_call, stage1_call, foo_call],
-        [foo_call, bar_call, stage1_call],
-    )
+    assert len(spy.call_args_list) == 3
+    assert foo_call in spy.call_args_list
+    assert bar_call in spy.call_args_list
+    assert stage1_call in spy.call_args_list
 
 
 def test_repro_ignore_errors(mocker, tmp_dir, dvc, copy_script):
@@ -1362,10 +1362,11 @@ def test_repro_ignore_errors(mocker, tmp_dir, dvc, copy_script):
         force=False,
         interactive=False,
     )
-    assert spy.call_args_list in (
-        [bar_call, stage1_call, foo_call, stage2_call],
-        [foo_call, bar_call, stage1_call, stage2_call],
-    )
+    assert len(spy.call_args_list) == 4
+    assert foo_call in spy.call_args_list
+    assert bar_call in spy.call_args_list
+    assert stage1_call in spy.call_args_list
+    assert stage2_call in spy.call_args_list
 
 
 @pytest.mark.parametrize("persist", [True, False])

tests/func/test_data_cloud.py

@@ -481,12 +481,13 @@ def test_push_pull_fetch_pipeline_stages(tmp_dir, dvc, run_copy, local_remote):
 
 
 def test_pull_partial(tmp_dir, dvc, local_remote):
-    tmp_dir.dvc_gen({"foo": {"bar": {"baz": "baz"}, "spam": "spam"}})
+    other_files = {f"spam{i}": f"spam{i}" for i in range(10)}
+    tmp_dir.dvc_gen({"foo": {"bar": {"baz": "baz"}, **other_files}})
     dvc.push()
     clean(["foo"], dvc)
 
     stats = dvc.pull(os.path.join("foo", "bar"))
-    assert stats["fetched"] == 3
+    assert stats["fetched"] == 2
     assert (tmp_dir / "foo").read_text() == {"bar": {"baz": "baz"}}
 
 

tests/func/test_fs.py

@@ -13,15 +13,15 @@ def test_cleanfs_subrepo(tmp_dir, dvc, scm, monkeypatch):
 
     path = subrepo_dir.fs_path
 
-    assert dvc.fs.exists(dvc.fs.path.join(path, "foo"))
-    assert dvc.fs.isfile(dvc.fs.path.join(path, "foo"))
-    assert dvc.fs.exists(dvc.fs.path.join(path, "dir"))
-    assert dvc.fs.isdir(dvc.fs.path.join(path, "dir"))
+    assert dvc.fs.exists(dvc.fs.join(path, "foo"))
+    assert dvc.fs.isfile(dvc.fs.join(path, "foo"))
+    assert dvc.fs.exists(dvc.fs.join(path, "dir"))
+    assert dvc.fs.isdir(dvc.fs.join(path, "dir"))
 
-    assert subrepo.fs.exists(subrepo.fs.path.join(path, "foo"))
-    assert subrepo.fs.isfile(subrepo.fs.path.join(path, "foo"))
-    assert subrepo.fs.exists(subrepo.fs.path.join(path, "dir"))
-    assert subrepo.fs.isdir(subrepo.fs.path.join(path, "dir"))
+    assert subrepo.fs.exists(subrepo.fs.join(path, "foo"))
+    assert subrepo.fs.isfile(subrepo.fs.join(path, "foo"))
+    assert subrepo.fs.exists(subrepo.fs.join(path, "dir"))
+    assert subrepo.fs.isdir(subrepo.fs.join(path, "dir"))
 
 
 def test_walk_dont_ignore_subrepos(tmp_dir, scm, dvc):

tests/func/test_gc.py

@@ -7,7 +7,7 @@ import textwrap
 import pytest
 
 from dvc.cli import main
-from dvc.exceptions import CollectCacheError, InvalidArgumentError
+from dvc.exceptions import CollectCacheError, InvalidArgumentError, RevCollectionError
 from dvc.fs import LocalFileSystem
 from dvc.utils.fs import remove
 from dvc_data.hashfile.db.local import LocalHashFileDB
@@ -111,8 +111,9 @@ def test_gc_no_dir_cache(tmp_dir, dvc):
 
     remove(dir_stage.outs[0].cache_path)
 
-    with pytest.raises(CollectCacheError):
+    with pytest.raises(RevCollectionError) as exc:
         dvc.gc(workspace=True)
+    assert type(exc.value.__cause__) == CollectCacheError
 
     assert _count_files(dvc.cache.local.path) == 4
     dvc.gc(force=True, workspace=True)
@@ -439,3 +440,13 @@ def test_gc_logging(caplog, dvc, good_and_bad_cache):
     assert "Removed 3 objects from repo cache." in caplog.text
     assert "No unused 'local' cache to remove." in caplog.text
     assert "No unused 'legacy' cache to remove." in caplog.text
+
+
+def test_gc_skip_failed(tmp_dir, dvc):
+    with open("dvc.yaml", mode="w") as f:
+        f.write("\ninvalid")
+
+    with pytest.raises(RevCollectionError):
+        dvc.gc(force=True, workspace=True)
+
+    dvc.gc(force=True, workspace=True, skip_failed=True)

tests/func/test_repo_index.py

@@ -1,3 +1,4 @@
+import os
 from itertools import chain
 
 import pytest
@@ -338,3 +339,53 @@ params:
     tmp_dir.gen("dvc.yaml", top_level_params)
     index = Index.from_repo(dvc)
     assert index.param_keys == {"repo": {("classifier", "custom_params_file.yaml")}}
+
+
+def test_data_index(tmp_dir, dvc, local_cloud, erepo_dir):
+    tmp_dir.dvc_gen(
+        {
+            "foo": b"foo",
+            "dir": {"bar": b"bar", "subdir": {"baz": b"baz"}},
+        }
+    )
+
+    with erepo_dir.chdir():
+        erepo_dir.dvc_gen("efoo", b"efoo", commit="create efoo")
+        erepo_dir.dvc_gen(
+            "edir",
+            {"ebar": b"ebar", "esubdir": {"ebaz": b"ebaz"}},
+            commit="create edir",
+        )
+
+    dvc.imp(os.fspath(erepo_dir), "efoo")
+    dvc.imp(os.fspath(erepo_dir), "edir")
+
+    local_cloud.gen("ifoo", b"ifoo")
+    local_cloud.gen("idir", {"ibar": b"ibar", "isubdir": {"ibaz": b"ibaz"}})
+
+    dvc.imp_url(str(local_cloud / "ifoo"))
+    dvc.imp_url(str(local_cloud / "idir"))
+
+    index = Index.from_repo(dvc)
+    assert index.data_keys == {
+        "local": set(),
+        "repo": {("dir",), ("edir",), ("efoo",), ("foo",), ("idir",), ("ifoo",)},
+    }
+
+    data = index.data["repo"]
+    assert set(data.keys()) == {
+        ("dir",),
+        ("edir",),
+        ("efoo",),
+        ("foo",),
+        ("idir",),
+        ("ifoo",),
+    }
+
+    assert not data.storage_map[("foo",)].remote
+    assert not data.storage_map[("dir",)].remote
+
+    assert data.storage_map[("efoo",)].remote.read_only
+    assert data.storage_map[("edir",)].remote.read_only
+    assert data.storage_map[("ifoo",)].remote.read_only
+    assert data.storage_map[("idir",)].remote.read_only

tests/func/test_virtual_directory.py

@@ -166,32 +166,27 @@ def test_virtual_update_noop(tmp_dir, dvc, remote):
 
 
 def test_partial_checkout_and_update(M, tmp_dir, dvc, remote):
-    tmp_dir.gen({"dir": {"foo": "foo", "subdir": {"lorem": "lorem"}}})
+    dir1 = {f"{i}.txt": f"dir1 {i}" for i in range(10)}
+    dir2 = {f"{i}.txt": f"dir2 {i}" for i in range(10)}
+    tmp_dir.gen({"dir": {"foo": "foo", "subdir": dir1, "subdir2": dir2}})
 
     (stage,) = dvc.add("dir")
     out = stage.outs[0]
 
-    assert out.hash_info == HashInfo(
-        name="md5", value="22a16c9bf84b3068bc2206d88a6b5776.dir"
-    )
-    assert out.meta == Meta(isdir=True, size=8, nfiles=2)
+    assert out.hash_info == HashInfo("md5", "9899dc38082ee0ec33de077df62a4a12.dir")
+    assert out.meta == Meta(isdir=True, size=123, nfiles=21)
 
-    assert dvc.push() == 3
+    assert dvc.push() == 22
     dvc.cache.local.clear()
     shutil.rmtree("dir")
 
-    assert dvc.pull("dir/subdir") == M.dict(
-        added=[join("dir", "")],
-        fetched=3,
-    )
-    assert (tmp_dir / "dir").read_text() == {"subdir": {"lorem": "lorem"}}
+    assert dvc.pull("dir/subdir") == M.dict(added=[join("dir", "")], fetched=11)
+    assert (tmp_dir / "dir").read_text() == {"subdir": dir1}
 
-    tmp_dir.gen({"dir": {"subdir": {"ipsum": "ipsum"}}})
-    (stage,) = dvc.add("dir/subdir/ipsum")
+    tmp_dir.gen({"dir": {"subdir": {"file": "file"}}})
+    (stage,) = dvc.add(join("dir", "subdir", "file"))
 
     out = stage.outs[0]
-    assert out.hash_info == HashInfo(
-        name="md5", value="06d953a10e0b0ffacba04876a9351e39.dir"
-    )
-    assert out.meta == Meta(isdir=True, size=13, nfiles=3)
+    assert out.hash_info == HashInfo("md5", "e7531c3930f28a00edc25e6cef91db03.dir")
+    assert out.meta == Meta(isdir=True, size=127, nfiles=22)
     assert dvc.push() == 2

tests/unit/command/test_gc.py

@@ -26,6 +26,7 @@ def test_(dvc, scm, mocker):
             "--projects",
             "project1",
             "project2",
+            "--skip-failed",
         ]
     )
     assert cli_args.func == CmdGC
@@ -51,6 +52,7 @@ def test_(dvc, scm, mocker):
         num=None,
         not_in_remote=False,
         dry=True,
+        skip_failed=True,
     )
 
     cli_args = parse_args(["gc"])

tests/unit/fs/test_path.py

@@ -1,29 +0,0 @@
-import pytest
-
-from dvc.fs import Path
-
-
-@pytest.mark.parametrize("prefix", ["", "/"])
-@pytest.mark.parametrize("postfix", ["", "/"])
-@pytest.mark.parametrize(
-    "path,expected",
-    [
-        ("path", ("path",)),
-        ("some/path", ("some", "path")),
-    ],
-)
-def test_parts_posix(prefix, postfix, path, expected):
-    assert Path("/").parts(prefix + path + postfix) == tuple(prefix) + expected
-
-
-@pytest.mark.parametrize("postfix", ["", "\\"])
-@pytest.mark.parametrize(
-    "path,expected",
-    [
-        ("path", ("path",)),
-        ("c:\\path", ("c:", "\\", "path")),
-        ("some\\path", ("some", "path")),
-    ],
-)
-def test_parts_nt(postfix, path, expected):
-    assert Path("\\").parts(path + postfix) == expected

tests/unit/test_ignore.py

@@ -173,11 +173,10 @@ def test_match_ignore_from_file(
     )
     dvcignore_dirname = os.path.dirname(dvcignore_path)
 
-    fs = mocker.MagicMock()
-    fs.path = localfs.path
-    fs.sep = localfs.sep
-    mocker.patch.object(fs, "open", mocker.mock_open(read_data="\n".join(patterns)))
-    ignore_file = DvcIgnorePatterns.from_file(dvcignore_path, fs, "mocked")
+    mocker.patch.object(
+        localfs, "open", mocker.mock_open(read_data="\n".join(patterns))
+    )
+    ignore_file = DvcIgnorePatterns.from_file(dvcignore_path, localfs, "mocked")
 
     assert (
         ignore_file.matches(dvcignore_dirname, file_to_ignore_relpath) == expected_match