loader: don't forget to set meta.isdir

Leftover from #8273

Fixes #8300

.github/workflows/tests.yaml

@@ -43,6 +43,9 @@ jobs:
       matrix:
         os: [ubuntu-20.04, windows-latest, macos-latest]
         pyv: ["3.8", "3.9", "3.10"]
+        pytest-filter: 
+        - "import or plot or live or experiment"
+        - "not (import or plot or live or experiment)"
         include:
         - {os: ubuntu-latest, pyv: "3.11-dev"}
 
@@ -75,10 +78,7 @@ jobs:
         AIOHTTP_NO_EXTENSIONS: ${{ matrix.pyv == '3.11-dev' && '1' }}
     - name: run tests
       timeout-minutes: 40
-      run: >-
-        python -m tests -n=auto
-        --cov-report=xml --cov-report=term
-        ${{ env.extra_test_args }}
+      run: pytest -nauto -ra --durations 100 --cov=dvc --cov-report=xml --cov-report=term -k "${{ matrix.pytest-filter }}"
     - name: upload coverage report
       uses: codecov/codecov-action@v3
       with:

dvc/api/__init__.py

@@ -1,9 +1,15 @@
-from .data import (  # noqa, pylint: disable=redefined-builtin
-    get_url,
-    open,
-    read,
-)
+from dvc.fs.dvc import _DvcFileSystem as DvcFileSystem
+
+from .data import open  # pylint: disable=redefined-builtin
+from .data import get_url, read
 from .experiments import make_checkpoint
 from .params import params_show
 
-__all__ = ["get_url", "make_checkpoint", "open", "params_show", "read"]
+__all__ = [
+    "get_url",
+    "make_checkpoint",
+    "open",
+    "params_show",
+    "read",
+    "DvcFileSystem",
+]

dvc/dependency/__init__.py

@@ -18,7 +18,7 @@ SCHEMA: Mapping[str, Any] = {
 }
 
 
-def _get(stage, p, info):
+def _get(stage, p, info, fs_config=None):
     if info and info.get(RepoDependency.PARAM_REPO):
         repo = info.pop(RepoDependency.PARAM_REPO)
         return RepoDependency(repo, stage, p, info)
@@ -27,7 +27,7 @@ def _get(stage, p, info):
         params = info.pop(ParamsDependency.PARAM_PARAMS)
         return ParamsDependency(stage, p, params)
 
-    return Dependency(stage, p, info)
+    return Dependency(stage, p, info, fs_config=fs_config)
 
 
 def loadd_from(stage, d_list):
@@ -38,10 +38,18 @@ def loadd_from(stage, d_list):
     return ret
 
 
-def loads_from(stage, s_list, erepo=None):
+def loads_from(stage, s_list, erepo=None, fs_config=None):
     assert isinstance(s_list, list)
     info = {RepoDependency.PARAM_REPO: erepo} if erepo else {}
-    return [_get(stage, s, info.copy()) for s in s_list]
+    return [
+        _get(
+            stage,
+            s,
+            info.copy(),
+            fs_config=fs_config,
+        )
+        for s in s_list
+    ]
 
 
 def _merge_params(s_list):

dvc/dependency/param.py

@@ -47,7 +47,7 @@ class ParamsDependency(Dependency):
         super().__init__(stage, path, repo=repo)
         self.hash_info = hash_info
 
-    def dumpd(self):
+    def dumpd(self, **kwargs):
         ret = super().dumpd()
         if not self.hash_info:
             ret[self.PARAM_PARAMS] = self.params or {}

dvc/dependency/repo.py

@@ -59,7 +59,7 @@ class RepoDependency(Dependency):
     def save(self):
         pass
 
-    def dumpd(self):
+    def dumpd(self, **kwargs):
         return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo}
 
     def download(self, to, jobs=None):

dvc/dvcfile.py

@@ -198,7 +198,7 @@ class SingleStageFile(FileMixin):
         if self.verify:
             check_dvcfile_path(self.repo, self.path)
         logger.debug("Saving information to '%s'.", relpath(self.path))
-        dump_yaml(self.path, serialize.to_single_stage_file(stage))
+        dump_yaml(self.path, serialize.to_single_stage_file(stage, **kwargs))
         self.repo.scm_context.track_file(self.relpath)
 
     def remove_stage(self, stage):  # pylint: disable=unused-argument

dvc/fs/dvc.py

@@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Any, Callable, Optional, Tuple, Type, Union
 from fsspec.spec import AbstractFileSystem
 from funcy import cached_property, wrap_prop, wrap_with
 
+from dvc.utils.fs import makedirs
 from dvc_objects.fs.base import FileSystem
 from dvc_objects.fs.callbacks import DEFAULT_CALLBACK
 from dvc_objects.fs.path import Path
@@ -218,23 +219,20 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         dvc_fs = self._datafss.get(repo_key)
         return repo, dvc_fs, subkey
 
-    def open(
-        self, path, mode="r", encoding="utf-8", **kwargs
+    def _open(
+        self, path, mode="rb", **kwargs
     ):  # pylint: disable=arguments-renamed, arguments-differ
-        if "b" in mode:
-            encoding = None
-
         key = self._get_key_from_relative(path)
         fs_path = self._from_key(key)
         try:
-            return self.repo.fs.open(fs_path, mode=mode, encoding=encoding)
+            return self.repo.fs.open(fs_path, mode=mode)
         except FileNotFoundError:
             _, dvc_fs, subkey = self._get_subrepo_info(key)
             if not dvc_fs:
                 raise
 
         dvc_path = _get_dvc_path(dvc_fs, subkey)
-        return dvc_fs.open(dvc_path, mode=mode, encoding=encoding, **kwargs)
+        return dvc_fs.open(dvc_path, mode=mode)
 
     def isdvc(self, path, **kwargs):
         key = self._get_key_from_relative(path)
@@ -299,6 +297,11 @@ class _DvcFileSystem(AbstractFileSystem):  # pylint:disable=abstract-method
         key = self._get_key_from_relative(rpath)
         fs_path = self._from_key(key)
         fs = self.repo.fs
+
+        if self.isdir(rpath):
+            makedirs(lpath, exist_ok=True)
+            return None
+
         try:
             fs.get_file(fs_path, lpath, callback=callback, **kwargs)
             return

dvc/output.py

@@ -1,10 +1,10 @@
 import logging
 import os
 from collections import defaultdict
-from typing import TYPE_CHECKING, Dict, Optional, Set, Type
+from typing import TYPE_CHECKING, Dict, List, Optional, Set, Tuple, Type
 from urllib.parse import urlparse
 
-from funcy import collecting, project
+from funcy import cached_property, collecting, project
 from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo
 
 from dvc import prompt
@@ -34,6 +34,8 @@ from .utils import relpath
 from .utils.fs import path_isin
 
 if TYPE_CHECKING:
+    from dvc_data.hashfile.obj import HashFile
+    from dvc_data.index import DataIndexKey
     from dvc_objects.db import ObjectDB
 
 logger = logging.getLogger(__name__)
@@ -84,6 +86,7 @@ def loadd_from(stage, d_list):
         live = d.pop(Output.PARAM_LIVE, False)
         remote = d.pop(Output.PARAM_REMOTE, None)
         annot = {field: d.pop(field, None) for field in ANNOTATION_FIELDS}
+        files = d.pop(Output.PARAM_FILES, None)
         ret.append(
             _get(
                 stage,
@@ -97,6 +100,7 @@ def loadd_from(stage, d_list):
                 live=live,
                 remote=remote,
                 **annot,
+                files=files,
             )
         )
     return ret
@@ -242,6 +246,7 @@ class Output:
     PARAM_PATH = "path"
     PARAM_CACHE = "cache"
     PARAM_CHECKPOINT = "checkpoint"
+    PARAM_FILES = "files"
     PARAM_METRIC = "metric"
     PARAM_METRIC_TYPE = "type"
     PARAM_METRIC_XPATH = "xpath"
@@ -290,6 +295,8 @@ class Output:
         meta=None,
         remote=None,
         repo=None,
+        fs_config=None,
+        files: List[Dict[str, Any]] = None,
     ):
         self.annot = Annotation(
             desc=desc, type=type, labels=labels or [], meta=meta or {}
@@ -299,10 +306,12 @@ class Output:
         # NOTE: when version_aware is not passed into get_cloud_fs, it will be
         # set based on whether or not path is versioned
         fs_kwargs = {"version_aware": True} if meta.version_id else {}
+
+        if fs_config is not None:
+            fs_kwargs.update(**fs_config)
+
         fs_cls, fs_config, fs_path = get_cloud_fs(
-            self.repo,
-            url=path,
-            **fs_kwargs,
+            self.repo, url=path, **fs_kwargs
         )
         self.fs = fs_cls(**fs_config)
 
@@ -336,6 +345,7 @@ class Output:
         # should be absolute and don't contain remote:// refs.
         self.stage = stage
         self.meta = meta
+        self.files = files
         self.use_cache = False if self.IS_DEPENDENCY else cache
         self.metric = False if self.IS_DEPENDENCY else metric
         self.plot = False if self.IS_DEPENDENCY else plot
@@ -344,7 +354,7 @@ class Output:
         self.live = live
 
         self.fs_path = self._parse_path(self.fs, fs_path)
-        self.obj = None
+        self.obj: Optional["HashFile"] = None
 
         self.remote = remote
 
@@ -363,6 +373,8 @@ class Output:
             name=self.hash_name,
             value=getattr(self.meta, self.hash_name, None),
         )
+        if self.hash_info and self.hash_info.isdir:
+            self.meta.isdir = True
 
     def _parse_path(self, fs, fs_path):
         parsed = urlparse(self.def_path)
@@ -408,6 +420,7 @@ class Output:
         self.hash_info = HashInfo.from_dict({})
         self.meta = Meta.from_dict({})
         self.obj = None
+        self.files = None
 
     @property
     def protocol(self):
@@ -487,6 +500,17 @@ class Output:
 
         return self.fs.exists(self.fs_path)
 
+    @cached_property
+    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)
+        else:
+            workspace = self.fs.protocol
+            no_drive = self.fs.path.flavour.splitdrive(self.fs_path)[1]
+            key = self.fs.path.parts(no_drive)[1:]
+        return workspace, key
+
     def changed_checksum(self):
         return self.hash_info != self.get_hash()
 
@@ -596,6 +620,7 @@ class Output:
                 dry_run=True,
             )
             self.hash_info = obj.hash_info
+            self.files = None
             if not self.IS_DEPENDENCY:
                 logger.debug(
                     "Output '%s' doesn't use cache. Skipping saving.", self
@@ -612,6 +637,7 @@ class Output:
             ignore=self.dvcignore,
         )
         self.hash_info = self.obj.hash_info
+        self.files = None
 
     def set_exec(self):
         if self.isfile() and self.meta.isexec:
@@ -693,8 +719,10 @@ class Output:
         )
         return checkout_obj
 
-    def dumpd(self):
-        ret = {**self.hash_info.to_dict(), **self.meta.to_dict()}
+    def dumpd(self, **kwargs):
+        meta = self.meta.to_dict()
+        meta.pop("isdir", None)
+        ret = {**self.hash_info.to_dict(), **meta}
 
         if self.is_in_repo:
             path = self.fs.path.as_posix(
@@ -737,6 +765,18 @@ class Output:
         if self.remote:
             ret[self.PARAM_REMOTE] = self.remote
 
+        if (
+            self.use_cache
+            and self.is_in_repo
+            and self.hash_info.isdir
+            and (kwargs.get("with_files") or self.files is not None)
+        ):
+            if self.obj:
+                obj = self.obj
+            else:
+                obj = self.get_obj()
+            ret[self.PARAM_FILES] = obj.as_list(with_meta=True)
+
         return ret
 
     def verify_metric(self):
@@ -769,14 +809,22 @@ class Output:
         ) as cb:
             self.fs.get(self.fs_path, to.fs_path, batch_size=jobs, callback=cb)
 
-    def get_obj(self, filter_info=None, **kwargs):
+    def get_obj(
+        self, filter_info: Optional[str] = None, **kwargs
+    ) -> Optional["HashFile"]:
+        obj: Optional["HashFile"] = None
         if self.obj:
             obj = self.obj
         elif self.hash_info:
-            try:
-                obj = oload(self.odb, self.hash_info)
-            except (FileNotFoundError, ObjectFormatError):
-                return None
+            if self.files:
+                tree = Tree.from_list(self.files, hash_name=self.hash_name)
+                tree.digest()
+                obj = tree
+            else:
+                try:
+                    obj = oload(self.odb, self.hash_info)
+                except (FileNotFoundError, ObjectFormatError):
+                    return None
         else:
             return None
 
@@ -896,6 +944,7 @@ class Output:
         )
 
         self.hash_info = obj.hash_info
+        self.files = None
         return obj
 
     def get_files_number(self, filter_info=None):
@@ -966,6 +1015,7 @@ class Output:
 
         obj = self.get_obj()
         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)
             )
@@ -1102,6 +1152,7 @@ class Output:
         self.odb.add(merged.path, merged.fs, merged.oid)
 
         self.hash_info = merged.hash_info
+        self.files = None
         self.meta = Meta(
             size=du(self.odb, merged),
             nfiles=len(merged),
@@ -1140,10 +1191,17 @@ ARTIFACT_SCHEMA = {
     Output.PARAM_CHECKPOINT: bool,
 }
 
+DIR_FILES_SCHEMA: Dict[str, Any] = {
+    **CHECKSUMS_SCHEMA,
+    **META_SCHEMA,
+    Required(Tree.PARAM_RELPATH): str,
+}
+
 SCHEMA = {
     **ARTIFACT_SCHEMA,
     **ANNOTATION_SCHEMA,
     Output.PARAM_CACHE: bool,
     Output.PARAM_METRIC: Output.METRIC_SCHEMA,
     Output.PARAM_REMOTE: str,
+    Output.PARAM_FILES: [DIR_FILES_SCHEMA],
 }

dvc/repo/fetch.py

@@ -34,8 +34,8 @@ def _fetch_worktree(repo, remote):
             if not out.is_in_repo:
                 continue
 
-            key = repo.fs.path.relparts(out.fs_path, repo.root_dir)
-            entry = repo.index.data["repo"][key]
+            workspace, key = out.index_key
+            entry = repo.index.data[workspace][key]
             out.hash_info = entry.hash_info
             out.meta = entry.meta
 

dvc/repo/imp_url.py

@@ -26,6 +26,7 @@ def imp_url(
     labels=None,
     meta=None,
     jobs=None,
+    fs_config=None,
 ):
     from dvc.dvcfile import Dvcfile
     from dvc.stage import Stage, create_stage, restore_fields
@@ -61,6 +62,7 @@ def imp_url(
         deps=[url],
         outs=[out],
         erepo=erepo,
+        fs_config=fs_config,
     )
     restore_fields(stage)
 

dvc/repo/index.py

@@ -183,15 +183,7 @@ class Index:
             if not out.use_cache:
                 continue
 
-            if out.is_in_repo:
-                workspace = "repo"
-                key = self.repo.fs.path.relparts(
-                    out.fs_path, self.repo.root_dir
-                )
-            else:
-                workspace = out.fs.protocol
-                no_drive = out.fs.path.flavour.splitdrive(out.fs_path)[1]
-                key = out.fs.path.parts(no_drive)[1:]
+            workspace, key = out.index_key
 
             try:
                 remote = self.repo.cloud.get_remote_odb(out.remote)
@@ -202,7 +194,7 @@ class Index:
 
             data_index[key] = DataIndexEntry(
                 meta=out.meta,
-                obj=out.obj,
+                obj=out.get_obj() if out.files else out.obj,
                 hash_info=out.hash_info,
                 odb=out.odb,
                 cache=out.odb,

dvc/repo/push.py

@@ -10,11 +10,14 @@ if TYPE_CHECKING:
 
 
 def _push_worktree(repo, remote):
-    from dvc_data.index import build, checkout
+    from dvc_data.hashfile.tree import tree_from_index
+    from dvc_data.index import build, checkout, collect
 
     index = repo.index.data["repo"]
-    checkout(index, remote.path, remote.fs)
+    checkout(index, remote.path, remote.fs, force=True)
     build(index, remote.path, remote.fs)
+    if any(out.isdir() for out in repo.index.outs):
+        collect(index, remote.path, remote.fs)
 
     for stage in repo.index.stages:
         for out in stage.outs:
@@ -24,12 +27,30 @@ def _push_worktree(repo, remote):
             if not out.is_in_repo:
                 continue
 
-            key = repo.fs.path.relparts(out.fs_path, repo.root_dir)
-            entry = repo.index.data["repo"][key]
-            out.hash_info = entry.hash_info
-            out.meta = entry.meta
-
-        stage.dvcfile.dump(stage)
+            workspace, key = out.index_key
+            index = repo.index.data[workspace]
+            entry = index[key]
+            if out.isdir():
+                old_tree = out.get_obj()
+                entry.hash_info = old_tree.hash_info
+                entry.meta = out.meta
+                for subkey, entry in index.iteritems(key):
+                    if entry.meta.isdir:
+                        continue
+                    fs_path = repo.fs.path.join(repo.root_dir, *subkey)
+                    _, hash_info = old_tree.get(
+                        repo.fs.path.relparts(fs_path, out.fs_path)
+                    )
+                    entry.hash_info = hash_info
+                tree_meta, new_tree = tree_from_index(index, key)
+                new_tree.digest()
+                out.obj = new_tree
+                out.hash_info = new_tree.hash_info
+                out.meta = tree_meta
+            else:
+                out.hash_info = entry.hash_info
+                out.meta = entry.meta
+        stage.dvcfile.dump(stage, with_files=True)
 
     return len(index)
 

dvc/schema.py

@@ -4,11 +4,10 @@ from voluptuous import Any, Optional, Required, Schema
 
 from dvc import dependency, output
 from dvc.annotations import ANNOTATION_SCHEMA
-from dvc.output import CHECKSUMS_SCHEMA, Output
+from dvc.output import CHECKSUMS_SCHEMA, DIR_FILES_SCHEMA, META_SCHEMA, Output
 from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
 from dvc.parsing.versions import SCHEMA_KWD, lockfile_version_schema
 from dvc.stage.params import StageParams
-from dvc_data.hashfile.meta import Meta
 
 STAGES = "stages"
 SINGLE_STAGE_SCHEMA = {
@@ -26,10 +25,9 @@ SINGLE_STAGE_SCHEMA = {
 
 DATA_SCHEMA = {
     **CHECKSUMS_SCHEMA,
+    **META_SCHEMA,
     Required("path"): str,
-    Meta.PARAM_SIZE: int,
-    Meta.PARAM_NFILES: int,
-    Meta.PARAM_ISEXEC: bool,
+    Output.PARAM_FILES: [DIR_FILES_SCHEMA],
 }
 LOCK_FILE_STAGE_SCHEMA = {
     Required(StageParams.PARAM_CMD): Any(str, list),
@@ -57,7 +55,6 @@ OUT_PSTAGE_DETAILED_SCHEMA = {
 }
 
 PLOTS = "plots"
-PLOTS_SCHEMA = dict
 PLOT_PROPS = {
     Output.PARAM_PLOT_TEMPLATE: str,
     Output.PARAM_PLOT_X: str,
@@ -113,6 +110,15 @@ def either_or(primary, fallback, fallback_includes=None):
     return validator
 
 
+PLOT_DEFINITION = {
+    Output.PARAM_PLOT_X: str,
+    Output.PARAM_PLOT_Y: Any(str, [str], {str: Any(str, [str])}),
+    Output.PARAM_PLOT_X_LABEL: str,
+    Output.PARAM_PLOT_Y_LABEL: str,
+    Output.PARAM_PLOT_TITLE: str,
+    Output.PARAM_PLOT_TEMPLATE: str,
+}
+PLOTS_SCHEMA = {str: Any(PLOT_DEFINITION, None)}
 FOREACH_IN = {
     Required(FOREACH_KWD): Any(dict, list, str),
     Required(DO_KWD): STAGE_DEFINITION,

dvc/stage/__init__.py

@@ -88,7 +88,7 @@ def create_stage(cls, repo, path, external=False, **kwargs):
     if not external:
         check_no_externals(stage)
     fill_stage_dependencies(
-        stage, **project(kwargs, ["deps", "erepo", "params"])
+        stage, **project(kwargs, ["deps", "erepo", "params", "fs_config"])
     )
     check_circular_dependency(stage)
     check_duplicated_arguments(stage)
@@ -461,8 +461,8 @@ class Stage(params.StageParams):
     def reload(self):
         return self.dvcfile.stage
 
-    def dumpd(self):
-        return get_dump(self)
+    def dumpd(self, **kwargs):
+        return get_dump(self, **kwargs)
 
     def compute_md5(self):
         # `dvc add`ed files don't need stage md5

dvc/stage/loader.py

@@ -71,6 +71,8 @@ class StageLoader(Mapping):
             item.meta = Meta.from_dict(info)
             hash_value = getattr(item.meta, item.hash_name, None)
             item.hash_info = HashInfo(item.hash_name, hash_value)
+            if item.hash_info and item.hash_info.isdir:
+                item.meta.isdir = True
 
     @classmethod
     def load_stage(cls, dvcfile, name, stage_data, lock_data=None):

dvc/stage/serialize.py

@@ -156,10 +156,12 @@ def to_single_stage_lockfile(stage: "Stage") -> dict:
     assert stage.cmd
 
     def _dumpd(item):
+        meta_d = item.meta.to_dict()
+        meta_d.pop("isdir", None)
         ret = [
             (item.PARAM_PATH, item.def_path),
             *item.hash_info.to_dict().items(),
-            *item.meta.to_dict().items(),
+            *meta_d.items(),
         ]
 
         return OrderedDict(ret)
@@ -186,8 +188,8 @@ def to_lockfile(stage: "PipelineStage") -> dict:
     return {stage.name: to_single_stage_lockfile(stage)}
 
 
-def to_single_stage_file(stage: "Stage"):
-    state = stage.dumpd()
+def to_single_stage_file(stage: "Stage", **kwargs):
+    state = stage.dumpd(**kwargs)
 
     # When we load a stage we parse yaml with a fast parser, which strips
     # off all the comments and formatting. To retain those on update we do

dvc/stage/utils.py

@@ -108,12 +108,16 @@ def _load_live_output(
     return outs
 
 
-def fill_stage_dependencies(stage, deps=None, erepo=None, params=None):
+def fill_stage_dependencies(
+    stage, deps=None, erepo=None, params=None, fs_config=None
+):
     from dvc.dependency import loads_from, loads_params
 
     assert not stage.deps
     stage.deps = []
-    stage.deps += loads_from(stage, deps or [], erepo=erepo)
+    stage.deps += loads_from(
+        stage, deps or [], erepo=erepo, fs_config=fs_config
+    )
     stage.deps += loads_params(stage, params or [])
 
 
@@ -222,7 +226,7 @@ def resolve_paths(fs, path, wdir=None):
     return path, wdir
 
 
-def get_dump(stage):
+def get_dump(stage, **kwargs):
     return {
         key: value
         for key, value in {
@@ -231,8 +235,8 @@ def get_dump(stage):
             stage.PARAM_CMD: stage.cmd,
             stage.PARAM_WDIR: resolve_wdir(stage.wdir, stage.path),
             stage.PARAM_FROZEN: stage.frozen,
-            stage.PARAM_DEPS: [d.dumpd() for d in stage.deps],
-            stage.PARAM_OUTS: [o.dumpd() for o in stage.outs],
+            stage.PARAM_DEPS: [d.dumpd(**kwargs) for d in stage.deps],
+            stage.PARAM_OUTS: [o.dumpd(**kwargs) for o in stage.outs],
             stage.PARAM_ALWAYS_CHANGED: stage.always_changed,
             stage.PARAM_META: stage.meta,
         }.items()

dvc/testing/test_api.py

@@ -1,4 +1,7 @@
+import pytest
+
 from dvc import api
+from dvc.api import DvcFileSystem
 from dvc.utils.fs import remove
 
 
@@ -18,3 +21,103 @@ class TestAPI:
 
         with api.open("foo") as fd:
             assert fd.read() == "foo-text"
+
+    @pytest.mark.parametrize(
+        "clear_cache", [True, False], ids=["cache", "no_cache"]
+    )
+    @pytest.mark.parametrize(
+        "fs_kwargs",
+        [
+            {},
+            {"url": "{path}"},
+            {"url": "{path}", "rev": "master"},
+            {"url": "file://{posixpath}"},
+            {"url": "file://{posixpath}", "rev": "master"},
+        ],
+        ids=["current", "local", "local_rev", "git", "git_rev"],
+    )
+    def test_filesystem(
+        self,
+        M,
+        tmp_dir,
+        make_tmp_dir,
+        scm,
+        dvc,
+        remote,
+        fs_kwargs,
+        clear_cache,
+    ):
+        tmp_dir.scm_gen({"scripts": {"script1": "script1"}}, commit="scripts")
+        tmp_dir.dvc_gen({"data": {"foo": "foo", "bar": "bar"}}, commit="data")
+        dvc.push()
+
+        if clear_cache:
+            remove(dvc.odb.repo.path)
+
+        if url := fs_kwargs.get("url"):
+            fs_kwargs["url"] = url.format(
+                path=tmp_dir, posixpath=tmp_dir.as_posix()
+            )
+
+        fs = DvcFileSystem(**fs_kwargs)
+
+        assert fs.ls("/", detail=False) == M.unordered(
+            "/.gitignore", "/scripts", "/data"
+        )
+        assert fs.ls("scripts", detail=False) == ["scripts/script1"]
+        assert fs.ls("data", detail=False) == M.unordered(
+            "data/foo", "data/bar"
+        )
+
+        data_info = M.dict(
+            name="/data",
+            type="directory",
+            dvc_info=M.dict(isdvc=True, isout=True),
+        )
+        scripts_info = M.dict(name="/scripts", type="directory", isexec=False)
+
+        assert sorted(fs.ls("/"), key=lambda i: i["name"]) == [
+            M.dict(name="/.gitignore", type="file", isexec=False),
+            data_info,
+            scripts_info,
+        ]
+
+        with pytest.raises(FileNotFoundError):
+            fs.info("/not-existing-path")
+
+        assert fs.info("/") == M.dict(name="/", isexec=False, type="directory")
+        assert fs.info("/data") == data_info
+        assert fs.info("/scripts") == scripts_info
+        assert fs.info("/data/foo") == M.dict(name="/data/foo", type="file")
+        assert fs.info("/scripts/script1") == M.dict(
+            name="/scripts/script1", type="file"
+        )
+
+        assert not fs.isdvc("/")
+        assert fs.isdvc("/data")
+        assert fs.isdvc("/data/foo")
+        assert not fs.isdvc("/scripts")
+        assert not fs.isdvc("/scripts/script1")
+
+        with pytest.raises((IsADirectoryError, PermissionError)):
+            fs.open("data")
+        with pytest.raises((IsADirectoryError, PermissionError)):
+            fs.open("scripts")
+        with fs.open("/data/foo") as fobj:
+            assert fobj.read() == b"foo"
+        with fs.open("/scripts/script1") as fobj:
+            assert fobj.read() == b"script1"
+
+        tmp = make_tmp_dir("temp-download")
+        fs.get_file("data/foo", tmp / "foo")
+        assert (tmp / "foo").read_text() == "foo"
+
+        fs.get_file("scripts/script1", tmp / "script1")
+        assert (tmp / "script1").read_text() == "script1"
+
+        fs.get("/", (tmp / "all").fs_path, recursive=True)
+        assert (tmp / "all").read_text() == {
+            ".gitignore": "/data\n",
+            "data": {"bar": "bar", "foo": "foo"},
+            "scripts": {"script1": "script1"},
+        }

setup.cfg

@@ -66,7 +66,7 @@ install_requires =
     dvc-render==0.0.10
     dvc-task==0.1.2
     dvclive>=0.10.0
-    dvc-data==0.7.1
+    dvc-data==0.8.0
     dvc-http==2.19.1
     hydra-core>=1.1.0
 
@@ -116,7 +116,6 @@ tests =
     # https://github.com/docker/docker-py/issues/2902
     pytest-docker==0.11.0; python_version < '3.10' or sys_platform != 'win32'
     flaky==3.7.0
-    mock==4.0.3
     pytest-timeout==2.1.0
     filelock==3.8.0
     # pylint requirements
@@ -146,6 +145,8 @@ exclude =
 [options.entry_points]
 console_scripts =
     dvc = dvc.cli:main
+fsspec.specs =
+    dvc = dvc.api:DvcFileSystem
 
 [flake8]
 ignore=

tests/func/test_import_url.py

@@ -7,7 +7,7 @@ import pytest
 
 from dvc.annotations import Annotation
 from dvc.cli import main
-from dvc.dependency.base import DependencyDoesNotExistError
+from dvc.dependency.base import Dependency, DependencyDoesNotExistError
 from dvc.exceptions import InvalidArgumentError
 from dvc.stage import Stage
 from dvc.testing.test_workspace import TestImport as _TestImport
@@ -330,3 +330,21 @@ def test_imp_url_with_annotations(M, tmp_dir, dvc, local_workspace):
     )
     assert stage.outs[0].annot == Annotation(**annot)
     assert (tmp_dir / "foo.dvc").parse() == M.dict(outs=[M.dict(**annot)])
+
+
+def test_import_url_fs_config(tmp_dir, dvc, workspace, mocker):
+    import dvc.fs as dvc_fs
+
+    workspace.gen("foo", "foo")
+
+    url = "remote://workspace/foo"
+    get_fs_config = mocker.spy(dvc_fs, "get_fs_config")
+    dep_init = mocker.spy(Dependency, "__init__")
+    dvc.imp_url(url, fs_config={"jobs": 42})
+
+    dep_init_kwargs = dep_init.call_args[1]
+    assert dep_init_kwargs.get("fs_config") == {"jobs": 42}
+
+    assert get_fs_config.call_args_list[0][1] == {"url": "foo"}
+    assert get_fs_config.call_args_list[1][1] == {"url": url, "jobs": 42}
+    assert get_fs_config.call_args_list[2][1] == {"name": "workspace"}