[Storage] Incremental copy through page blob does not accept None values for seal_blob and blob_tags_string (#41124)

Author: weirongw23-msft

sdk/storage/azure-storage-blob/CHANGELOG.md

@@ -4,6 +4,9 @@
 
 ### Features Added
 
+### Bugs Fixed
+- Fixed an issue where `BlobClient`'s `start_copy_from_url` with `incremental_copy=True` results in `TypeError`.
+
 ## 12.26.0b1 (2025-05-06)
 
 ### Features Added

sdk/storage/azure-storage-blob/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/storage/azure-storage-blob",
-  "Tag": "python/storage/azure-storage-blob_56ef3e2a11"
+  "Tag": "python/storage/azure-storage-blob_01573d6981"
 }

sdk/storage/azure-storage-blob/azure/storage/blob/_blob_client_helpers.py

@@ -658,19 +658,20 @@ def _start_copy_from_url_options(  # pylint:disable=too-many-statements
 
     options = {
         'copy_source': source_url,
-        'seal_blob': kwargs.pop('seal_destination_blob', None),
         'timeout': timeout,
         'modified_access_conditions': dest_mod_conditions,
-        'blob_tags_string': blob_tags_string,
         'headers': headers,
         'cls': return_response_headers,
     }
+
     if not incremental_copy:
         source_mod_conditions = get_source_conditions(kwargs)
         dest_access_conditions = get_access_conditions(kwargs.pop('destination_lease', None))
         options['source_modified_access_conditions'] = source_mod_conditions
         options['lease_access_conditions'] = dest_access_conditions
         options['tier'] = tier.value if tier else None
+        options['seal_blob'] = kwargs.pop('seal_destination_blob', None)
+        options['blob_tags_string'] = blob_tags_string
     options.update(kwargs)
     return options
 

sdk/storage/azure-storage-blob/tests/test_page_blob.py

@@ -1981,49 +1981,51 @@ def test_create_blob_with_md5_large(self, **kwargs):
 
         # Assert
 
-    @pytest.mark.skip(reason="Requires further investigation. Failing for unexpected kwarg seal_blob")
     @BlobPreparer()
+    @recorded_by_proxy
     def test_incremental_copy_blob(self, **kwargs):
         storage_account_name = kwargs.pop("storage_account_name")
         storage_account_key = kwargs.pop("storage_account_key")
 
         bsc = BlobServiceClient(self.account_url(storage_account_name, "blob"), credential=storage_account_key, max_page_size=4 * 1024)
         self._setup(bsc)
-        source_blob = self._create_blob(bsc, length=2048)
-        data = self.get_random_bytes(512)
-        resp1 = source_blob.upload_page(data, offset=0, length=512)
-        resp2 = source_blob.upload_page(data, offset=1024, length=512)
-        source_snapshot_blob = source_blob.create_snapshot()
-
-        snapshot_blob = BlobClient.from_blob_url(
-            source_blob.url, credential=source_blob.credential, snapshot=source_snapshot_blob)
-        sas_token = self.generate_sas(
-            generate_blob_sas,
-            snapshot_blob.account_name,
-            snapshot_blob.container_name,
-            snapshot_blob.blob_name,
-            snapshot=snapshot_blob.snapshot,
-            account_key=snapshot_blob.credential.account_key,
-            permission=BlobSasPermissions(read=True),
-            expiry=datetime.utcnow() + timedelta(hours=1),
-        )
-        sas_blob = BlobClient.from_blob_url(snapshot_blob.url, credential=sas_token)
 
-        # Act
-        dest_blob = bsc.get_blob_client(self.container_name, 'dest_blob')
-        copy = dest_blob.start_copy_from_url(sas_blob.url, incremental_copy=True)
+        try:
+            source_blob = self._create_blob(bsc, length=2048)
+            data = self.get_random_bytes(512)
+            resp1 = source_blob.upload_page(data, offset=0, length=512)
+            resp2 = source_blob.upload_page(data, offset=1024, length=512)
+            source_snapshot_blob = source_blob.create_snapshot()
+
+            snapshot_blob = BlobClient.from_blob_url(
+                source_blob.url, credential=source_blob.credential, snapshot=source_snapshot_blob)
+            sas_token = self.generate_sas(
+                generate_blob_sas,
+                snapshot_blob.account_name,
+                snapshot_blob.container_name,
+                snapshot_blob.blob_name,
+                snapshot=snapshot_blob.snapshot,
+                account_key=snapshot_blob.credential.account_key,
+                permission=BlobSasPermissions(read=True),
+                expiry=datetime.utcnow() + timedelta(hours=1),
+            )
+            sas_blob = BlobClient.from_blob_url(snapshot_blob.url, credential=sas_token)
 
-        # Assert
-        assert copy is not None
-        assert copy['copy_id'] is not None
-        assert copy['copy_status'] == 'pending'
+            # Act
+            dest_blob = bsc.get_blob_client(self.container_name, 'dest_blob')
+            copy = dest_blob.start_copy_from_url(sas_blob.url, incremental_copy=True)
 
-        copy_blob = self._wait_for_async_copy(dest_blob)
-        assert copy_blob.copy.status == 'success'
-        assert copy_blob.copy.destination_snapshot is not None
+            # Assert
+            assert copy is not None
+            assert copy['copy_id'] is not None
+            assert copy['copy_status'] == 'pending'
 
-        # strip off protocol
-        assert copy_blob.copy.source.endswith(sas_blob.url[5:])
+            copy_blob = self._wait_for_async_copy(dest_blob)
+            assert copy_blob.copy.status == 'success'
+            assert copy_blob.copy.destination_snapshot is not None
+        finally:
+            bsc.delete_container(self.container_name)
+            bsc.delete_container(self.source_container_name)
 
     @BlobPreparer()
     @recorded_by_proxy

sdk/storage/azure-storage-blob/tests/test_page_blob_async.py

@@ -1965,7 +1965,6 @@ async def test_create_blob_with_md5_large(self, **kwargs):
 
         # Assert
 
-    @pytest.mark.skip(reason="Requires further investigation. Failing for unexpected kwarg seal_blob")
     @BlobPreparer()
     @recorded_by_proxy_async
     async def test_incremental_copy_blob(self, **kwargs):
@@ -1974,42 +1973,44 @@ async def test_incremental_copy_blob(self, **kwargs):
 
         bsc = BlobServiceClient(self.account_url(storage_account_name, "blob"), credential=storage_account_key, max_page_size=4 * 1024)
         await self._setup(bsc)
-        source_blob = await self._create_blob(bsc, 2048)
-        data = self.get_random_bytes(512)
-        resp1 = await source_blob.upload_page(data, offset=0, length=512)
-        resp2 = await source_blob.upload_page(data, offset=1024, length=512)
-        source_snapshot_blob = await source_blob.create_snapshot()
-
-        snapshot_blob = BlobClient.from_blob_url(
-            source_blob.url, credential=source_blob.credential, snapshot=source_snapshot_blob)
-        sas_token = self.generate_sas(
-            generate_blob_sas,
-            snapshot_blob.account_name,
-            snapshot_blob.container_name,
-            snapshot_blob.blob_name,
-            snapshot=snapshot_blob.snapshot,
-            account_key=snapshot_blob.credential.account_key,
-            permission=BlobSasPermissions(read=True),
-            expiry=datetime.utcnow() + timedelta(hours=1),
-        )
-        sas_blob = BlobClient.from_blob_url(snapshot_blob.url, credential=sas_token)
 
+        try:
+            source_blob = await self._create_blob(bsc, 2048)
+            data = self.get_random_bytes(512)
+            resp1 = await source_blob.upload_page(data, offset=0, length=512)
+            resp2 = await source_blob.upload_page(data, offset=1024, length=512)
+            source_snapshot_blob = await source_blob.create_snapshot()
+
+            snapshot_blob = BlobClient.from_blob_url(
+                source_blob.url, credential=source_blob.credential, snapshot=source_snapshot_blob)
+            sas_token = self.generate_sas(
+                generate_blob_sas,
+                snapshot_blob.account_name,
+                snapshot_blob.container_name,
+                snapshot_blob.blob_name,
+                snapshot=snapshot_blob.snapshot,
+                account_key=snapshot_blob.credential.account_key,
+                permission=BlobSasPermissions(read=True),
+                expiry=datetime.utcnow() + timedelta(hours=1),
+            )
+            sas_blob = BlobClient.from_blob_url(snapshot_blob.url, credential=sas_token)
 
-        # Act
-        dest_blob = bsc.get_blob_client(self.container_name, 'dest_blob')
-        copy = await dest_blob.start_copy_from_url(sas_blob.url, incremental_copy=True)
 
-        # Assert
-        assert copy is not None
-        assert copy['copy_id'] is not None
-        assert copy['copy_status'] == 'pending'
+            # Act
+            dest_blob = bsc.get_blob_client(self.container_name, 'dest_blob')
+            copy = await dest_blob.start_copy_from_url(sas_blob.url, incremental_copy=True)
 
-        copy_blob = await self._wait_for_async_copy(dest_blob)
-        assert copy_blob.copy.status == 'success'
-        assert copy_blob.copy.destination_snapshot is not None
+            # Assert
+            assert copy is not None
+            assert copy['copy_id'] is not None
+            assert copy['copy_status'] == 'pending'
 
-        # strip off protocol
-        assert copy_blob.copy.source.endswith(sas_blob.url[5:])
+            copy_blob = await self._wait_for_async_copy(dest_blob)
+            assert copy_blob.copy.status == 'success'
+            assert copy_blob.copy.destination_snapshot is not None
+        finally:
+            await bsc.delete_container(self.container_name)
+            await bsc.delete_container(self.source_container_name)
 
     @BlobPreparer()
     @recorded_by_proxy_async