fix(test_wasb.py): SAS token tests failing with azure-storage-blob 12.30.0

providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/wasb.py

@@ -30,6 +30,7 @@
 import os
 from typing import TYPE_CHECKING, Any, cast
 
+from azure.core.credentials import AzureSasCredential
 from azure.core.exceptions import HttpResponseError, ResourceExistsError, ResourceNotFoundError
 from azure.identity import ClientSecretCredential
 from azure.identity.aio import (
@@ -203,7 +204,9 @@ def get_conn(self) -> BlobServiceClient:
         if sas_token:
             if sas_token.startswith("https"):
                 return BlobServiceClient(account_url=sas_token, **extra)
-            return BlobServiceClient(account_url=f"{account_url.rstrip('/')}/{sas_token}", **extra)
+            return BlobServiceClient(
+                account_url=account_url, credential=AzureSasCredential(sas_token), **extra
+            )
 
         # Fall back to old auth (password) or use managed identity if not provided.
         credential: str | TokenCredential | None = conn.password
@@ -671,7 +674,7 @@ async def get_async_conn(self) -> AsyncBlobServiceClient:
                 self.blob_service_client = AsyncBlobServiceClient(account_url=sas_token, **extra)
             else:
                 self.blob_service_client = AsyncBlobServiceClient(
-                    account_url=f"{account_url.rstrip('/')}/{sas_token}", **extra
+                    account_url=account_url, credential=AzureSasCredential(sas_token), **extra
                 )
             return self.blob_service_client
 

providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_wasb.py

@@ -23,6 +23,7 @@
 from unittest.mock import create_autospec
 
 import pytest
+from azure.core.credentials import AzureSasCredential
 from azure.core.exceptions import ResourceNotFoundError
 from azure.storage.blob import BlobServiceClient, ContainerClient
 from azure.storage.blob._models import BlobProperties
@@ -45,6 +46,11 @@
 ACCESS_KEY_STRING = "AccountName=name;skdkskd"
 PROXIES = {"http": "http_proxy_uri", "https": "https_proxy_uri"}
 
+# https://learn.microsoft.com/en-us/rest/api/storageservices/create-service-sas
+# The hook wraps plain SAS tokens in AzureSasCredential for the SDK to sign requests.
+SAS_TOKEN = "sv=2021-08-06&ss=b&srt=sco&sp=r&sig=samplesignature"
+HTTPS_SAS_TOKEN = f"https://login.blob.core.windows.net/?{SAS_TOKEN}"
+
 
 @pytest.fixture
 def mocked_blob_service_client():
@@ -147,24 +153,24 @@ def setup_method(self, create_mock_connections):
                 conn_id="sas_conn_id",
                 conn_type=self.connection_type,
                 login=self.login,
-                extra={"sas_token": "token", "proxies": self.proxies},
+                extra={"sas_token": SAS_TOKEN, "proxies": self.proxies},
             ),
             Connection(
                 conn_id=self.extra__wasb__sas_conn_id,
                 conn_type=self.connection_type,
                 login=self.login,
-                extra={"extra__wasb__sas_token": "token", "proxies": self.proxies},
+                extra={"extra__wasb__sas_token": SAS_TOKEN, "proxies": self.proxies},
             ),
             Connection(
                 conn_id=self.http_sas_conn_id,
                 conn_type=self.connection_type,
-                extra={"sas_token": "https://login.blob.core.windows.net/token", "proxies": self.proxies},
+                extra={"sas_token": HTTPS_SAS_TOKEN, "proxies": self.proxies},
             ),
             Connection(
                 conn_id=self.extra__wasb__http_sas_conn_id,
                 conn_type=self.connection_type,
                 extra={
-                    "extra__wasb__sas_token": "https://login.blob.core.windows.net/token",
+                    "extra__wasb__sas_token": HTTPS_SAS_TOKEN,
                     "proxies": self.proxies,
                 },
             ),
@@ -304,8 +310,12 @@ def test_sas_token_provided_and_active_directory_id_used_as_host(
         self, mocked_connection, mocked_blob_service_client
     ):
         WasbHook(wasb_conn_id="testconn").get_conn()
+        called_credential = mocked_blob_service_client.call_args[1]["credential"]
+        assert isinstance(called_credential, AzureSasCredential)
+        assert called_credential.signature == "SAStoken"
         mocked_blob_service_client.assert_called_once_with(
-            account_url="https://testaccountname.blob.core.windows.net/SAStoken",
+            account_url="https://testaccountname.blob.core.windows.net/",
+            credential=called_credential,
             sas_token="SAStoken",
         )
 
@@ -361,10 +371,14 @@ def test_sas_token_connection(self, conn_id_str, extra_key):
         hook_conn = hook.get_connection(hook.conn_id)
         sas_token = hook_conn.extra_dejson[extra_key]
         assert isinstance(conn, BlobServiceClient)
-        assert conn.url.startswith("https://")
-        if hook_conn.login:
-            assert hook_conn.login in conn.url
-        assert conn.url.endswith(sas_token + "/")
+        if sas_token.startswith("https"):
+            # HTTPS SAS URL: the full URL is passed to BlobServiceClient as-is;
+            # credential handling is left to the SDK.
+            assert sas_token in conn.url
+        else:
+            # Plain SAS token: the hook wraps it in AzureSasCredential.
+            assert isinstance(conn.credential, AzureSasCredential)
+            assert conn.credential.signature == sas_token
 
     @pytest.mark.parametrize(
         argnames="conn_id_str",