fix webdav tests and then add some (#6189)

dvc/fs/webdav.py

@@ -49,17 +49,19 @@ class WebDAVFileSystem(FSSpecWrapper):  # pylint:disable=abstract-method
         }
 
     def _prepare_credentials(self, **config):
-        self.user = user = config.get("user", None)
-        self.password = password = config.get("password", None)
+        user = config.get("user", None)
+        password = config.get("password", None)
 
         headers = {}
         token = config.get("token")
+        auth = None
         if token:
             headers.update({"Authorization": f"Bearer {token}"})
-        elif user and not password and config.get("ask_password"):
-            self.password = password = ask_password(config["host"], self.user)
+        elif user:
+            if not password and config.get("ask_password"):
+                password = ask_password(config["host"], user)
+            auth = (user, password)
 
-        auth = (user, password) if user and password else None
         return {"headers": headers, "auth": auth}
 
     @wrap_prop(threading.Lock())

tests/unit/remote/test_webdav.py

@@ -1,25 +1,91 @@
+from unittest.mock import patch
+
 import pytest
 
 from dvc.fs import get_cloud_fs
 from dvc.fs.webdav import WebDAVFileSystem, WebDAVSFileSystem
+from tests.utils.asserts import issubset
 
-# Test configuration
+url_fmt = "{scheme}://{user}@example.com/public.php/webdav"
 url = "webdav://example.com/public.php/webdav"
 user = "username"
-userurl = f"webdav://{user}@example.com/public.php/webdav"
 password = "password"
-https_url = "webdavs://example.com/public.php/webdav"
+token = "4MgjsNM5aSJjxIKM"
+
+
+def test_common():
+    fs = WebDAVFileSystem(
+        url=url,
+        cert_path="cert/path",
+        key_path="key/path",
+        ssl_verify="bundle.pem",
+        timeout=10,
+        prefix="/public.php/webdav",
+        user=None,
+        password=None,
+        ask_password=False,
+        token=None,
+    )
+    assert issubset(
+        {
+            "headers": {},
+            "auth": None,
+            "base_url": url,
+            "cert": ("cert/path", "key/path"),
+            "verify": "bundle.pem",
+            "timeout": 10,
+        },
+        fs.fs_args,
+    )
+    assert fs.prefix == "/public.php/webdav"
 
 
 def test_user():
     fs = WebDAVFileSystem(url=url, user=user)
-    assert fs.user == user
+    assert issubset({"auth": (user, None), "headers": {}}, fs.fs_args)
 
 
 def test_password():
     config = {"url": url, "user": user, "password": password}
     fs = WebDAVFileSystem(**config)
-    assert fs.password == password
+    assert issubset(
+        {
+            "headers": {},
+            "auth": (user, password),
+        },
+        fs.fs_args,
+    )
+
+
+def test_token():
+    config = {"token": token, "url": url}
+    fs = WebDAVFileSystem(**config)
+    assert issubset(
+        {"headers": {"Authorization": f"Bearer {token}"}, "auth": None},
+        fs.fs_args,
+    )
+
+
+@patch("dvc.fs.webdav.ask_password")
+def test_ask_password(ask_password_mocked):
+    ask_password_mocked.return_value = "pass"
+    host = "host"
+
+    # it should not ask for password as password is set
+    config = {
+        "url": url,
+        "user": user,
+        "password": password,
+        "ask_password": True,
+        "host": host,
+    }
+    fs = WebDAVFileSystem(**config)
+    assert issubset({"auth": (user, password), "headers": {}}, fs.fs_args)
+
+    config.pop("password")
+    fs = WebDAVFileSystem(**config)
+    assert issubset({"auth": (user, "pass"), "headers": {}}, fs.fs_args)
+    ask_password_mocked.assert_called_once_with(host, user)
 
 
 def test_ssl_verify_custom_cert():
@@ -34,18 +100,24 @@ def test_ssl_verify_custom_cert():
 
 @pytest.mark.parametrize(
     "base_url, fs_cls",
-    [(url, WebDAVFileSystem), (https_url, WebDAVSFileSystem)],
+    [
+        (url_fmt.format(scheme="webdav", user=user), WebDAVFileSystem),
+        (url_fmt.format(scheme="webdavs", user=user), WebDAVSFileSystem),
+    ],
 )
 def test_remote_with_jobs(dvc, base_url, fs_cls):
-    remote_config = {"url": base_url, "user": user}
+    scheme = "http" + ("s" if fs_cls is WebDAVSFileSystem else "")
+    remote_config = {"url": base_url}
 
     dvc.config["remote"]["dav"] = remote_config
     cls, config, _ = get_cloud_fs(dvc, 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")
     assert config["user"] == "admin"
+    assert f"{scheme}://{user}@example.com" in config["host"]
     assert cls is fs_cls

tests/utils/__init__.py

@@ -2,10 +2,15 @@ import csv
 import os
 from contextlib import contextmanager
 
+import pytest
 from funcy import first
 
 from dvc.scm import Git
 
+# rewrite assertions in assert, pytest does not rewrite for other modules
+# than tests itself.
+pytest.register_assert_rewrite("tests.utils.asserts")
+
 
 def get_gitignore_content():
     with open(Git.GITIGNORE) as gitignore:

tests/utils/asserts.py

@@ -1,7 +1,12 @@
-from typing import Any
+from typing import Any, Dict
 from unittest.mock import ANY, Mock
 
 
+def issubset(subset: Dict, superset: Dict) -> bool:
+    assert {**superset, **subset} == superset
+    return True
+
+
 def called_once_with_subset(m: Mock, *args: Any, **kwargs: Any) -> bool:
     m.assert_called_once()
     m_args, m_kwargs = m.call_args