feat(webdav): add bearer_token_command for dynamic token acquisition (#10917)

Co-authored-by: GreenHatHG <[email protected]>
Co-authored-by: skshetry <[email protected]>

.pre-commit-config.yaml

@@ -22,7 +22,7 @@ repos:
       - id: sort-simple-yaml
       - id: trailing-whitespace
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: 'v0.14.6'
+    rev: 'v0.14.7'
     hooks:
       - id: ruff-check
         args: [--fix, --exit-non-zero-on-fix]

dvc/config_schema.py

@@ -151,6 +151,7 @@ WEBDAV_COMMON = {
     "password": str,
     "ask_password": Bool,
     "token": str,
+    "bearer_token_command": str,
     "custom_auth_header": str,
     "cert_path": str,
     "key_path": str,

pyproject.toml

@@ -124,7 +124,7 @@ tests = [
     "sqlalchemy>=1,<3", # optional dependency for `import-db`
     "pandas>=1", # optional dependency for `import-db`
 ]
-webdav = ["dvc-webdav>=3,<4"]
+webdav = ["dvc-webdav>=3.0.1,<4"]
 webhdfs = ["dvc-webhdfs>=3.1,<4"]
 webhdfs_kerberos = ["dvc-webhdfs[kerberos]>=3.1,<4"]
 

tests/unit/remote/test_webdav.py

@@ -153,3 +153,31 @@ def test_remote_with_jobs(dvc, base_url, fs_cls):
     assert config["user"] == "admin"
     assert f"{scheme}://{user}@example.com" in config["host"]
     assert cls is fs_cls
+
+
+def test_bearer_token_command(mocker):
+    mock_auth_obj = mocker.Mock()
+    mock_get_auth = mocker.patch(
+        "dvc_webdav.get_bearer_auth", return_value=mock_auth_obj
+    )
+
+    command = "my-token-cmd.sh"
+    timeout = 10
+
+    # Configuration with potential conflicts
+    config = {
+        "url": url,
+        "token": "static-token-value",
+        "user": "conflict-user",
+        "password": "conflict-password",
+        "bearer_token_command": command,
+        "timeout": timeout,
+    }
+
+    fs = WebDAVFileSystem(**config)
+
+    # the factory function was invoked with the correct arguments from the config.
+    mock_get_auth.assert_called_once_with(command, timeout)
+
+    # the generated auth object is correctly injected into `fs_args`.
+    assert fs.fs_args["auth"] == mock_auth_obj