ls-url: refactor tests so they work with all supported plugins.

Add `config` argument to ls_url() (only used in testing) to be able
to set up the FSs that do require it.
This commit is contained in:
Ronan Lamy
2022-10-04 14:37:29 +03:00
committed by Ruslan Kuprieiev
parent bd88000e2f
commit 8ff92cc7b0
3 changed files with 19 additions and 28 deletions
+6 -4
View File
@@ -2,14 +2,16 @@ from dvc.exceptions import URLMissingError
from dvc.fs import get_cloud_fs
def parse_external_url(url):
fs_cls, fs_config, fs_path = get_cloud_fs(None, url=url)
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 = fs_cls(**fs_config)
return fs, fs_path
def ls_url(url, recursive=False):
fs, fs_path = parse_external_url(url)
def ls_url(url, config=None, recursive=False):
fs, fs_path = parse_external_url(url, config=config)
return _ls_url(fs, fs_path, orig_url=url, recursive=recursive)
-5
View File
@@ -51,7 +51,6 @@ import sys
from contextlib import contextmanager
from functools import partialmethod
from dvc.fs import LocalFileSystem
from dvc.utils import serialize
from dvc.utils.fs import makedirs
@@ -71,10 +70,6 @@ class TmpDir(pathlib.Path):
def config(self):
return {"url": self.url}
@property
def fs(self):
return LocalFileSystem(**self.config)
def __new__(cls, *args, **kwargs):
if cls is TmpDir:
cls = ( # pylint: disable=self-cls-assignment
+13 -19
View File
@@ -3,7 +3,7 @@ import os
import pytest
from dvc.exceptions import URLMissingError
from dvc.repo.ls_url import _ls_url
from dvc.repo.ls_url import ls_url, parse_external_url
class TestImport:
@@ -153,21 +153,20 @@ class TestLsUrl:
@pytest.mark.parametrize("fname", ["foo", "foo.dvc", "dir/foo"])
def test_file(self, cloud, fname):
cloud.gen({fname: "foo contents"})
fs = cloud.fs
fs_path = fs.path.join(cloud.fs_path, fname)
result = _ls_url(fs, fs_path)
match_files(fs, result, [{"path": fs_path, "isdir": False}])
fs, fs_path = parse_external_url(cloud.url, cloud.config)
result = ls_url(str(cloud / fname), cloud.config)
match_files(
fs,
result,
[{"path": fs.path.join(fs_path, fname), "isdir": False}],
)
def test_dir(self, cloud):
cloud.gen(
{"dir/foo": "foo contents", "dir/subdir/bar": "bar contents"}
)
fs = cloud.fs
fs_path = fs.path.join(cloud.fs_path, "dir")
result = _ls_url(fs, fs_path)
fs, _ = parse_external_url(cloud.url, cloud.config)
result = ls_url(str(cloud / "dir"), cloud.config)
match_files(
fs,
result,
@@ -181,11 +180,8 @@ class TestLsUrl:
cloud.gen(
{"dir/foo": "foo contents", "dir/subdir/bar": "bar contents"}
)
fs = cloud.fs
fs_path = fs.path.join(cloud.fs_path, "dir")
result = _ls_url(fs, fs_path, recursive=True)
fs, _ = parse_external_url(cloud.url, cloud.config)
result = ls_url(str(cloud / "dir"), cloud.config, recursive=True)
match_files(
fs,
result,
@@ -196,7 +192,5 @@ class TestLsUrl:
)
def test_nonexistent(self, cloud):
fs = cloud.fs
fs_path = fs.path.join(cloud.fs_path, "nonexistent")
with pytest.raises(URLMissingError):
_ls_url(fs, fs_path)
ls_url(str(cloud / "dir"), cloud.config)