dvc: bump to 2.3.0

* fs: temporarily disable prefix-based search for fsspec-implementations

* tests: add the other hash

.pre-commit-config.yaml

@@ -1,3 +1,6 @@
+ci:
+  skip: [pylint, dvc-pre-commit]
+
 repos:
   - hooks:
       - id: black

dvc/config_schema.py

@@ -145,7 +145,7 @@ SCHEMA = {
                     "session_token": str,
                     Optional("listobjects", default=False): Bool,  # obsoleted
                     Optional("use_ssl", default=True): Bool,
-                    Optional("ssl_verify", default=True): Bool,
+                    "ssl_verify": Any(Bool, str),
                     "sse": str,
                     "sse_kms_key_id": str,
                     "acl": str,

dvc/fs/azure.py

@@ -84,14 +84,12 @@ class AzureFileSystem(ObjectFSWrapper):
 
         url = config.get("url")
         self.path_info = self.PATH_CLS(url)
-        self.bucket = self.path_info.bucket
 
-        if not self.bucket:
+        if not self.path_info.bucket:
             container = self._az_config.get("storage", "container_name", None)
             url = f"azure://{container}"
 
         self.path_info = self.PATH_CLS(url)
-        self.bucket = self.path_info.bucket
 
     def _prepare_credentials(self, **config):
         from azure.identity.aio import DefaultAzureCredential
@@ -117,8 +115,21 @@ class AzureFileSystem(ObjectFSWrapper):
         login_info["client_id"] = config.get("client_id")
         login_info["client_secret"] = config.get("client_secret")
 
-        if not any(
+        if not (login_info["account_name"] or login_info["connection_string"]):
+            raise AzureAuthError(
+                "Authentication to Azure Blob Storage requires either "
+                "account_name or connection_string.\nLearn more about "
+                "configuration settings at "
+                + format_link("https://man.dvc.org/remote/modify")
+            )
+
+        any_secondary = any(
             value for key, value in login_info.items() if key != "account_name"
+        )
+        if (
+            login_info["account_name"]
+            and not any_secondary
+            and not config.get("allow_anonymous_login")
         ):
             login_info["credential"] = DefaultAzureCredential(
                 exclude_interactive_browser_credential=False
@@ -167,10 +178,6 @@ class AzureFileSystem(ObjectFSWrapper):
         try:
             with _temp_event_loop():
                 file_system = AzureBlobFileSystem(**self.fs_args)
-                if self.bucket not in [
-                    container.rstrip("/") for container in file_system.ls("/")
-                ]:
-                    file_system.mkdir(self.bucket)
         except (ValueError, AzureError) as e:
             raise AzureAuthError(
                 f"Authentication to Azure Blob Storage via {self.login_method}"

dvc/fs/fsspec_wrapper.py

@@ -11,6 +11,8 @@ from .base import BaseFileSystem
 
 # pylint: disable=no-member
 class FSSpecWrapper(BaseFileSystem):
+    TRAVERSE_PREFIX_LEN = 2
+
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.fs_args = {"skip_instance_cache": True}

dvc/fs/oss.py

@@ -45,7 +45,6 @@ class OSSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         url = config.get("url")
         self.path_info = self.PATH_CLS(url) if url else None
 
-        self.bucket = self.path_info.bucket
         self.endpoint = config.get("oss_endpoint") or os.getenv("OSS_ENDPOINT")
 
         self.key_id = (
@@ -68,25 +67,17 @@ class OSSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         logger.debug(f"key id: {self.key_id}")
         logger.debug(f"key secret: {self.key_secret}")
 
-        auth = oss2.Auth(self.key_id, self.key_secret)
-        bucket = oss2.Bucket(auth, self.endpoint, self.bucket)
+        return oss2.Auth(self.key_id, self.key_secret)
 
-        # Ensure bucket exists
-        try:
-            bucket.get_bucket_info()
-        except oss2.exceptions.NoSuchBucket:
-            bucket.create_bucket(
-                oss2.BUCKET_ACL_PUBLIC_READ,
-                oss2.models.BucketCreateConfig(
-                    oss2.BUCKET_STORAGE_CLASS_STANDARD
-                ),
-            )
-        return bucket
+    def _get_bucket(self, bucket):
+        import oss2
+
+        return oss2.Bucket(self.oss_service, self.endpoint, bucket)
 
     def _generate_download_url(self, path_info, expires=3600):
-        assert path_info.bucket == self.bucket
-
-        return self.oss_service.sign_url("GET", path_info.path, expires)
+        return self._get_bucket(path_info.bucket).sign_url(
+            "GET", path_info.path, expires
+        )
 
     def exists(self, path_info) -> bool:
         paths = self._list_paths(path_info)
@@ -96,7 +87,7 @@ class OSSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         import oss2
 
         for blob in oss2.ObjectIterator(
-            self.oss_service, prefix=path_info.path
+            self._get_bucket(path_info.bucket), prefix=path_info.path
         ):
             yield blob.key
 
@@ -114,16 +105,17 @@ class OSSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
             raise NotImplementedError
 
         logger.debug(f"Removing oss://{path_info}")
-        self.oss_service.delete_object(path_info.path)
+        self._get_bucket(path_info.bucket).delete_object(path_info.path)
 
     def _upload_fobj(self, fobj, to_info):
-        self.oss_service.put_object(to_info.path, fobj)
+        self._get_bucket(to_info.bucket).put_object(to_info.path, fobj)
 
     def _upload(
         self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
     ):
         with Tqdm(desc=name, disable=no_progress_bar, bytes=True) as pbar:
-            self.oss_service.put_object_from_file(
+            bucket = self._get_bucket(to_info.bucket)
+            bucket.put_object_from_file(
                 to_info.path, from_file, progress_callback=pbar.update_to
             )
 
@@ -133,8 +125,9 @@ class OSSFileSystem(BaseFileSystem):  # pylint:disable=abstract-method
         with Tqdm(desc=name, disable=no_progress_bar, bytes=True) as pbar:
             import oss2
 
+            bucket = self._get_bucket(from_info.bucket)
             oss2.resumable_download(
-                self.oss_service,
+                bucket,
                 from_info.path,
                 to_file,
                 progress_callback=pbar.update_to,

dvc/fs/s3.py

@@ -103,7 +103,7 @@ class BaseS3FileSystem(ObjectFSWrapper):
         client = login_info["client_kwargs"]
         client["region_name"] = config.get("region")
         client["endpoint_url"] = config.get("endpointurl")
-        client["verify"] = config.get("ssl_verify", True)
+        client["verify"] = config.get("ssl_verify")
 
         # encryptions
         additional = login_info["s3_additional_kwargs"]

dvc/objects/db/base.py

@@ -399,8 +399,15 @@ class ObjectDB:
         # hashes_exist() (see ssh, local)
         assert self.fs.TRAVERSE_PREFIX_LEN >= 2
 
+        # During the tests, for ensuring that the traverse behavior
+        # is working we turn on this option. It will ensure the
+        # list_hashes_traverse() is called.
+        always_traverse = getattr(self.fs, "_ALWAYS_TRAVERSE", False)
+
         hashes = set(hashes)
-        if len(hashes) == 1 or not self.fs.CAN_TRAVERSE:
+        if (
+            len(hashes) == 1 or not self.fs.CAN_TRAVERSE
+        ) and not always_traverse:
             remote_hashes = self.list_hashes_exists(hashes, jobs, name)
             return remote_hashes
 
@@ -418,7 +425,7 @@ class ObjectDB:
             )
         else:
             traverse_weight = traverse_pages
-        if len(hashes) < traverse_weight:
+        if len(hashes) < traverse_weight and not always_traverse:
             logger.debug(
                 "Large remote ('{}' hashes < '{}' traverse weight), "
                 "using object_exists for remaining hashes".format(

dvc/version.py

@@ -6,7 +6,7 @@
 import os
 import subprocess
 
-_BASE_VERSION = "2.2.0"
+_BASE_VERSION = "2.3.0"
 
 
 def _generate_version(base_version):

setup.py

@@ -73,7 +73,7 @@ install_requires = [
     "flufl.lock>=3.2,<4",
     "win-unicode-console>=0.5; sys_platform == 'win32'",
     "pywin32>=225; sys_platform == 'win32'",
-    "networkx>=2.1",
+    "networkx~=2.5",
     "psutil>=5.8.0",
     "pydot>=1.2.4",
     "speedcopy>=2.0.1; python_version < '3.8' and sys_platform == 'win32'",

test_requirements.txt

@@ -2,7 +2,7 @@ wheel==0.36.2
 
 # Test requirements
 pytest==6.2.4
-pytest-cov==2.12.0
+pytest-cov==2.12.1
 pytest-xdist==2.2.1
 pytest-mock==3.6.1
 pytest-lazy-fixture==0.6.3
@@ -31,7 +31,7 @@ pydocstyle==6.1.1
 jaraco.windows==5.5.0
 
 # pylint requirements
-pylint==2.8.2
+pylint==2.8.3
 # we use this to suppress pytest-related false positives in our tests.
 pylint-pytest==1.0.3
 # we use this to suppress some messages in tests, eg: foo/bar naming,

tests/func/test_data_cloud.py

@@ -41,6 +41,14 @@ all_clouds = [
     )
 ]
 
+# Clouds that implement the general methods that can be tested
+# for functional tests that require extensive apis (e.g traversing
+# via walk_files())
+full_clouds = [
+    pytest.lazy_fixture(cloud)
+    for cloud in ["s3", "gs", "azure", "ssh", "hdfs"]
+]
+
 
 @pytest.mark.needs_internet
 @pytest.mark.parametrize("remote", all_clouds, indirect=True)
@@ -548,3 +556,23 @@ def test_pull_partial(tmp_dir, dvc, local_remote):
     stats = dvc.pull(os.path.join("foo", "bar"))
     assert stats["fetched"] == 1
     assert (tmp_dir / "foo").read_text() == {"bar": {"baz": "baz"}}
+
+
+@pytest.mark.parametrize("remote", full_clouds, indirect=True)
+def test_pull_00_prefix(tmp_dir, dvc, remote, monkeypatch):
+    # Related: https://github.com/iterative/dvc/issues/6089
+
+    fs_type = type(dvc.cloud.get_remote("upstream").fs)
+    monkeypatch.setattr(fs_type, "_ALWAYS_TRAVERSE", True, raising=False)
+    monkeypatch.setattr(fs_type, "LIST_OBJECT_PAGE_SIZE", 256, raising=False)
+
+    # foo's md5 checksum is 00411460f7c92d2124a67ea0f4cb5f85
+    # bar's md5 checksum is 0000000018e6137ac2caab16074784a6
+    tmp_dir.dvc_gen({"foo": "363", "bar": "jk8ssl"})
+
+    dvc.push()
+    clean(["foo", "bar"], dvc)
+
+    stats = dvc.pull()
+    assert stats["fetched"] == 2
+    assert set(stats["added"]) == {"foo", "bar"}

tests/remotes/azure.py

@@ -28,19 +28,12 @@ class Azure(Base, CloudURLInfo):
     @cached_property
     def service_client(self):
         # pylint: disable=no-name-in-module
-        from azure.core.exceptions import ResourceNotFoundError
         from azure.storage.blob import BlobServiceClient
 
         service_client = BlobServiceClient.from_connection_string(
             self.CONNECTION_STRING
         )
 
-        container_client = service_client.get_container_client(self.bucket)
-        try:  # verify that container exists
-            container_client.get_container_properties()
-        except ResourceNotFoundError:
-            container_client.create_container()
-
         return service_client
 
     @property
@@ -98,7 +91,16 @@ def azure_server(test_config, docker_compose, docker_services):
 
 @pytest.fixture
 def azure(azure_server):
+    from azure.core.exceptions import ResourceNotFoundError
+
     url = f"azure://{TEST_AZURE_CONTAINER}/{uuid.uuid4()}"
     ret = Azure(url)
     ret.config = {"url": url, "connection_string": azure_server}
+
+    container = ret.service_client.get_container_client(TEST_AZURE_CONTAINER)
+    try:  # verify that container exists
+        container.get_container_properties()
+    except ResourceNotFoundError:
+        container.create_container()
+
     return ret

tests/remotes/oss.py

@@ -64,6 +64,8 @@ def oss_server(test_config, docker_compose, docker_services):
 
 @pytest.fixture
 def oss(oss_server):
+    import oss2
+
     url = OSS.get_url()
     ret = OSS(url)
     ret.config = {
@@ -72,6 +74,19 @@ def oss(oss_server):
         "oss_key_secret": EMULATOR_OSS_ACCESS_KEY_SECRET,
         "oss_endpoint": oss_server,
     }
+
+    auth = oss2.Auth(
+        EMULATOR_OSS_ACCESS_KEY_ID, EMULATOR_OSS_ACCESS_KEY_SECRET
+    )
+    bucket = oss2.Bucket(auth, oss_server, TEST_OSS_REPO_BUCKET)
+    try:
+        bucket.get_bucket_info()
+    except oss2.exceptions.NoSuchBucket:
+        bucket.create_bucket(
+            oss2.BUCKET_ACL_PUBLIC_READ,
+            oss2.models.BucketCreateConfig(oss2.BUCKET_STORAGE_CLASS_STANDARD),
+        )
+
     return ret
 
 

tests/unit/fs/test_azure.py

@@ -3,7 +3,7 @@ from concurrent.futures import ThreadPoolExecutor
 
 import pytest
 
-from dvc.fs.azure import AzureFileSystem, _temp_event_loop
+from dvc.fs.azure import AzureAuthError, AzureFileSystem, _temp_event_loop
 from dvc.path_info import PathInfo
 
 container_name = "container-name"
@@ -67,3 +67,66 @@ def test_temp_event_loop():
 
         future = executor.submit(wrapped_procedure)
         assert future.result() == "yeey"
+
+
+def test_azure_login_methods():
+    def get_login_method(config):
+        fs = AzureFileSystem(**config)
+        # pylint: disable=pointless-statement
+        return fs.login_method
+
+    with pytest.raises(AzureAuthError):
+        get_login_method({})
+
+    assert (
+        get_login_method({"connection_string": "test"}) == "connection string"
+    )
+    assert get_login_method({"account_name": "test"}).startswith(
+        "default credentials"
+    )
+    assert (
+        get_login_method(
+            {"account_name": "test", "allow_anonymous_login": True}
+        )
+        == "anonymous login"
+    )
+
+    with pytest.raises(AzureAuthError):
+        get_login_method(
+            {"tenant_id": "test", "client_id": "test", "client_secret": "test"}
+        )
+
+    assert (
+        get_login_method(
+            {
+                "account_name": "test",
+                "tenant_id": "test",
+                "client_id": "test",
+                "client_secret": "test",
+            }
+        )
+        == "AD service principal"
+    )
+
+    assert (
+        get_login_method({"account_name": "test", "account_key": "test"})
+        == "account key"
+    )
+    assert (
+        get_login_method({"account_name": "test", "sas_token": "test"})
+        == "SAS token"
+    )
+    assert (
+        get_login_method(
+            {
+                "connection_string": "test",
+                "account_name": "test",
+                "sas_token": "test",
+            }
+        )
+        == "connection string"
+    )
+    assert (
+        get_login_method({"connection_string": "test", "sas_token": "test"})
+        == "connection string"
+    )

tests/unit/fs/test_s3.py

@@ -37,7 +37,15 @@ def test_verify_ssl_default_param(dvc):
     config = {"url": url}
     fs = S3FileSystem(**config)
 
-    assert fs.fs_args["client_kwargs"]["verify"]
+    assert "client_kwargs" not in fs.fs_args
+
+    config = {
+        "url": url,
+        "endpointurl": "https://my.custom.s3:1234",
+    }
+    fs = S3FileSystem(**config)
+
+    assert "verify" not in fs.fs_args["client_kwargs"]
 
 
 def test_s3_config_credentialpath(dvc, monkeypatch):
@@ -72,6 +80,29 @@ def test_ssl_verify_bool_param(dvc):
     assert fs.fs_args["client_kwargs"]["verify"] == config["ssl_verify"]
 
 
+def test_ssl_verify_path_param(dvc):
+    config = {"url": url, "ssl_verify": "/path/to/custom/cabundle.pem"}
+    fs = S3FileSystem(**config)
+
+    assert fs.fs_args["client_kwargs"]["verify"] == config["ssl_verify"]
+
+
+def test_ssl_verify_none_param(dvc):
+    config = {"url": url, "ssl_verify": None}
+    fs = S3FileSystem(**config)
+
+    assert "client_kwargs" not in fs.fs_args
+
+    config = {
+        "url": url,
+        "endpointurl": "https://my.custom.s3:1234",
+        "ssl_verify": None,
+    }
+    fs = S3FileSystem(**config)
+
+    assert "verify" not in fs.fs_args["client_kwargs"]
+
+
 def test_grants(dvc):
     config = {
         "url": url,

tests/unit/test_config.py

@@ -1,4 +1,5 @@
 import os
+import textwrap
 
 import pytest
 
@@ -33,3 +34,39 @@ def test_get_fs(tmp_dir, scm):
     assert config._get_fs("local") == config.wfs
     assert config._get_fs("global") == config.wfs
     assert config._get_fs("system") == config.wfs
+
+
+def test_s3_ssl_verify(tmp_dir, dvc):
+    config = Config(validate=False)
+    with config.edit() as conf:
+        conf["remote"]["remote-name"] = {"url": "s3://bucket/dvc"}
+
+    assert "ssl_verify" not in config["remote"]["remote-name"]
+
+    with config.edit() as conf:
+        section = conf["remote"]["remote-name"]
+        section["ssl_verify"] = False
+
+    assert (tmp_dir / ".dvc" / "config").read_text() == textwrap.dedent(
+        """\
+        [core]
+            no_scm = True
+        ['remote "remote-name"']
+            url = s3://bucket/dvc
+            ssl_verify = False
+        """
+    )
+
+    with config.edit() as conf:
+        section = conf["remote"]["remote-name"]
+        section["ssl_verify"] = "/path/to/custom/cabundle.pem"
+
+    assert (tmp_dir / ".dvc" / "config").read_text() == textwrap.dedent(
+        """\
+        [core]
+            no_scm = True
+        ['remote "remote-name"']
+            url = s3://bucket/dvc
+            ssl_verify = /path/to/custom/cabundle.pem
+        """
+    )