fs: get_cloud_fs: accept config instead of repo

Config is a more immutable thing and using `repo` is confusing and might be error
prone in the bigger picture.

Related to #9754

dvc/cachemgr.py

@@ -24,7 +24,7 @@ def _get_odb(
     if not settings:
         return None
 
-    cls, config, fs_path = get_cloud_fs(repo, **settings)
+    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)

dvc/data_cloud.py

@@ -90,7 +90,7 @@ def get_remote(
         if name:
             from dvc.fs import get_cloud_fs
 
-            cls, config, fs_path = get_cloud_fs(self.repo, name=name)
+            cls, config, fs_path = get_cloud_fs(self.repo.config, name=name)
 
             if config.get("worktree"):
                 version_aware = config.get("version_aware")

dvc/fs/__init__.py

@@ -87,7 +87,7 @@ def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
 def parse_external_url(url, config=None):
     remote_config = dict(config) if config else {}
     remote_config["url"] = url
-    fs_cls, fs_config, fs_path = get_cloud_fs(None, **remote_config)
+    fs_cls, fs_config, fs_path = get_cloud_fs({}, **remote_config)
     fs = fs_cls(**fs_config)
     return fs, fs_path
 
@@ -133,18 +133,14 @@ def _resolve_remote_refs(config, remote_conf):
         return remote_conf
 
     base = get_fs_config(config, name=parsed.netloc)
-    cls, _, _ = _get_cloud_fs(config, **base)
+    cls, _, _ = get_cloud_fs(config, **base)
     relpath = parsed.path.lstrip("/").replace("/", cls.sep)
     url = cls.sep.join((base["url"], relpath))
     return {**base, **remote_conf, "url": url}
 
 
-def get_cloud_fs(repo, **kwargs):
-    repo_config = repo.config if repo else {}
-    return _get_cloud_fs(repo_config, **kwargs)
-
-
-def _get_cloud_fs(repo_config, **kwargs):
+def get_cloud_fs(repo_config, **kwargs):
+    repo_config = repo_config or {}
     core_config = repo_config.get("core", {})
 
     remote_conf = get_fs_config(repo_config, **kwargs)

dvc/output.py

@@ -354,7 +354,11 @@ def __init__(  # noqa: PLR0913
         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)
+        fs_cls, fs_config, fs_path = get_cloud_fs(
+            self.repo.config if self.repo else {},
+            url=path,
+            **fs_kwargs,
+        )
         self.fs = fs_cls(**fs_config)
 
         if (
@@ -1017,7 +1021,9 @@ def transfer(
         if odb is None:
             odb = self.cache
 
-        cls, config, from_info = get_cloud_fs(self.repo, url=source)
+        cls, config, from_info = get_cloud_fs(
+            self.repo.config if self.repo else {}, url=source
+        )
         from_fs = cls(**config)
 
         # When running import-url --to-remote / add --to-remote/-o ... we

tests/unit/fs/test_fs.py

@@ -1,5 +1,3 @@
-from dataclasses import dataclass
-
 import pytest
 from dvc_http import HTTPFileSystem, HTTPSFileSystem
 from dvc_s3 import S3FileSystem
@@ -63,14 +61,8 @@ def test_remote_url():
     }
 
 
-@dataclass
-class FakeRepo:
-    config: dict
-
-
 def test_get_cloud_fs():
-    repo = FakeRepo({})
-    cls, config, path = get_cloud_fs(repo, url="ssh://example.com:/dir/path")
+    cls, config, path = get_cloud_fs({}, url="ssh://example.com:/dir/path")
     assert cls is SSHFileSystem
     assert config == {"host": "example.com", "verify": False}
     assert path == "/dir/path"

tests/unit/fs/test_tree.py

@@ -13,11 +13,11 @@ def test_get_cloud_fs(tmp_dir, dvc):
     first = f"{base}/first"
     second = f"{first}/second"
 
-    _, _, path = get_cloud_fs(dvc, name="base")
+    _, _, path = get_cloud_fs(dvc.config, name="base")
     assert path == base
-    _, _, path = get_cloud_fs(dvc, name="first")
+    _, _, path = get_cloud_fs(dvc.config, name="first")
     assert path == first
-    _, _, path = get_cloud_fs(dvc, name="second")
+    _, _, path = get_cloud_fs(dvc.config, name="second")
     assert path == second
 
 
@@ -34,9 +34,9 @@ def test_get_cloud_fs_validate(tmp_dir, dvc):
         default=False,
     )
 
-    assert get_cloud_fs(dvc, name="base")[1]["host"] == "example.com"
-    assert get_cloud_fs(dvc, name="first")[1]["host"] == "example.com"
-    assert get_cloud_fs(dvc, name="first")[1]["type"] == ["symlink"]
+    assert get_cloud_fs(dvc.config, name="base")[1]["host"] == "example.com"
+    assert get_cloud_fs(dvc.config, name="first")[1]["host"] == "example.com"
+    assert get_cloud_fs(dvc.config, name="first")[1]["type"] == ["symlink"]
 
     with pytest.raises(ConfigError):
-        get_cloud_fs(dvc, name="second")
+        get_cloud_fs(dvc.config, name="second")

tests/unit/remote/test_remote.py

@@ -12,7 +12,7 @@ def test_remote_with_hash_jobs(dvc):
     }
     dvc.config["core"]["checksum_jobs"] = 200
 
-    cls, config, _ = get_cloud_fs(dvc, name="with_hash_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="with_hash_jobs")
     fs = cls(**config)
     assert fs.hash_jobs == 100
 
@@ -23,7 +23,7 @@ def test_remote_with_jobs(dvc):
         "jobs": 100,
     }
 
-    cls, config, _ = get_cloud_fs(dvc, name="with_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="with_jobs")
     fs = cls(**config)
     assert fs.jobs == 100
 
@@ -32,15 +32,15 @@ def test_remote_without_hash_jobs(dvc):
     dvc.config["remote"]["without_hash_jobs"] = {"url": "s3://bucket/name"}
     dvc.config["core"]["checksum_jobs"] = 200
 
-    cls, config, _ = get_cloud_fs(dvc, name="without_hash_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="without_hash_jobs")
     fs = cls(**config)
     assert fs.hash_jobs == 200
 
 
 def test_remote_without_hash_jobs_default(dvc):
     dvc.config["remote"]["without_hash_jobs"] = {"url": "s3://bucket/name"}
 
-    cls, config, _ = get_cloud_fs(dvc, name="without_hash_jobs")
+    cls, config, _ = get_cloud_fs(dvc.config, name="without_hash_jobs")
     fs = cls(**config)
     assert fs.hash_jobs == fs.HASH_JOBS
 

tests/unit/remote/test_webdav.py

@@ -145,14 +145,14 @@ def test_remote_with_jobs(dvc, base_url, fs_cls):
     remote_config = {"url": base_url}
 
     dvc.config["remote"]["dav"] = remote_config
-    cls, config, _ = get_cloud_fs(dvc, name="dav")
+    cls, config, _ = get_cloud_fs(dvc.config, name="dav")
     assert config["user"] == user
     assert f"{scheme}://{user}@example.com" in config["host"]
     assert cls is fs_cls
 
     # config from remote takes priority
     remote_config.update({"user": "admin"})
-    cls, config, _ = get_cloud_fs(dvc, name="dav")
+    cls, config, _ = get_cloud_fs(dvc.config, name="dav")
     assert config["user"] == "admin"
     assert f"{scheme}://{user}@example.com" in config["host"]
     assert cls is fs_cls