[Storage][102] CRC64 content validation - part 5 - datalake (#46034)

Author: jalauzon-msft

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

@@ -33,7 +33,7 @@
     ImmutabilityPolicy,
     StandardBlobTier,
 )
-from azure.storage.blob._shared.policies import StorageContentValidation
+from azure.storage.blob._shared.validation import calculate_content_md5
 
 
 # ------------------------------------------------------------------------------
@@ -555,7 +555,7 @@ def test_upload_blob_from_url_with_source_content_md5(self, **kwargs):
         source_blob = self._create_blob(data=b"This is test data to be copied over.")
         source_blob_props = source_blob.get_blob_properties()
         source_md5 = source_blob_props.content_settings.content_md5
-        bad_source_md5 = StorageContentValidation.get_content_md5(b"this is bad data")
+        bad_source_md5 = calculate_content_md5(b"this is bad data")
         sas = self.generate_sas(
             generate_blob_sas,
             account_name=storage_account_name,

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

@@ -580,7 +580,7 @@ async def test_upload_blob_from_url_with_source_content_md5(self, **kwargs):
         source_blob = await self._create_blob(data=b"This is test data to be copied over.")
         source_blob_props = await source_blob.get_blob_properties()
         source_md5 = source_blob_props.content_settings.content_md5
-        bad_source_md5 = StorageContentValidation.get_content_md5(b"this is bad data")
+        bad_source_md5 = calculate_content_md5(b"this is bad data")
         sas = self.generate_sas(
             generate_blob_sas,
             account_name=storage_account_name,

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

@@ -465,10 +465,10 @@ def test_download_blob_large_chunks(self, **kwargs):
         blob.upload_blob(data, overwrite=True, max_concurrency=5)
 
         # Act
-        downloader = blob.download_blob(validate_content='crc64', max_concurrency=3)
+        downloader = blob.download_blob(validate_content='crc64', max_concurrency=5)
         content = downloader.read()
 
-        downloader = blob.download_blob(offset=5 * 1024 * 1024, length=25 *1024 * 1024)
+        downloader = blob.download_blob(offset=5 * 1024 * 1024, length=25 * 1024 * 1024, validate_content='crc64')
         partial = downloader.read()
 
         # Assert

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

@@ -8,7 +8,7 @@
 
 import pytest
 from azure.core.exceptions import ResourceExistsError
-from azure.storage.blob import BlobBlock, BlobType
+from azure.storage.blob import BlobBlock, BlobType, ContainerClient as SyncContainerClient
 from azure.storage.blob.aio import (
     BlobClient,
     BlobServiceClient,
@@ -46,11 +46,16 @@ async def _setup(self, account_name):
         except ResourceExistsError:
             pass
 
-    # TODO: Figure out how to get this to run automatically
-    async def _teardown(self):
+    def teardown_method(self, _):
+        # Use sync client as teardown_method must be sync
         if self.container:
+            sync_credential = self.get_credential(SyncContainerClient, is_async=False)
+            sync_container = SyncContainerClient.from_container_url(
+                self.container.url,
+                credential=sync_credential)
+
             try:
-                await self.container.delete_container()
+                sync_container.delete_container()
             except:
                 pass
 
@@ -74,6 +79,9 @@ async def test_encryption_blocked_crc64(self, **kwargs):
         with pytest.raises(ValueError):
             await blob.upload_blob(b'123', validate_content='crc64')
 
+        # Needed for teardown
+        self.container = None
+
     @BlobPreparer()
     @pytest.mark.parametrize('a', [BlobType.BLOCKBLOB, BlobType.PAGEBLOB, BlobType.APPENDBLOB])  # a: blob_type
     @pytest.mark.parametrize('b', [True, "auto", 'md5', 'crc64'])  # b: validate_content
@@ -106,8 +114,6 @@ async def test_upload_blob(self, a, b, **kwargs):
         await blob.upload_blob(str_iter, blob_type=a, length=len(str_data_encoded), encoding='utf-8', validate_content=b, overwrite=True, raw_request_hook=assert_method)
         assert await (await blob.download_blob()).read() == str_data_encoded
 
-        await self._teardown()
-
     @BlobPreparer()
     @pytest.mark.parametrize('a', [BlobType.BLOCKBLOB, BlobType.PAGEBLOB, BlobType.APPENDBLOB])  # a: blob_type
     @pytest.mark.parametrize('b', [True, 'md5', 'crc64'])  # b: validate_content
@@ -143,8 +149,6 @@ async def test_upload_blob_chunks(self, a, b, **kwargs):
         await blob.upload_blob(str_iter, blob_type=a, length=len(str_data_encoded), encoding='utf-8', validate_content=b, overwrite=True, raw_request_hook=assert_method)
         assert await (await blob.download_blob()).read() == str_data_encoded
 
-        await self._teardown()
-
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
     @GenericTestProxyParametrize1()
@@ -169,7 +173,6 @@ async def test_upload_blob_substream(self, a, **kwargs):
         # Assert
         content = await blob.download_blob()
         assert await content.read() == data
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'auto', 'md5', 'crc64'])  # a: validate_content
@@ -200,7 +203,6 @@ def generator():
         # Assert
         content = await blob.download_blob()
         assert await content.read() == data1 + data2.encode('utf-8-sig') + data1
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
@@ -221,7 +223,6 @@ async def test_stage_block_streaming(self, a, **kwargs):
         # Assert
         result = await blob.download_blob()
         assert await result.read() == content
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
@@ -274,7 +275,6 @@ def generator():
         # Assert
         content = await blob.download_blob()
         assert await content.readall() == data1 + data2.encode('utf-16') + data1
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
@@ -294,7 +294,6 @@ async def test_append_block_streaming(self, a, **kwargs):
 
         result = await blob.download_blob()
         assert await result.read() == content
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
@@ -317,7 +316,6 @@ async def test_append_block_streaming_large(self, a, **kwargs):
 
         result = await blob.download_blob()
         assert await result.read() == data1 + data2 + data3
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'auto', 'md5', 'crc64'])  # a: validate_content
@@ -341,7 +339,6 @@ async def test_upload_page(self, a, **kwargs):
         # Assert
         content = await blob.download_blob(offset=0, length=len(data1) + len(data2_encoded))
         assert await content.read() == data1 + data2_encoded
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'auto', 'md5', 'crc64'])  # a: validate_content
@@ -368,7 +365,6 @@ async def test_download_blob(self, a, **kwargs):
         # Assert
         assert content == data
         assert stream.read() == data
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
@@ -403,7 +399,6 @@ async def test_download_blob_chunks(self, a, **kwargs):
         assert content == data
         assert stream.read() == data
         assert read_content == data
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
@@ -432,7 +427,6 @@ async def test_download_blob_chunks_partial(self, a, **kwargs):
         # Assert
         assert content == data[10:1010]
         assert stream.read() == data[10:1010]
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.live_test_only
@@ -447,16 +441,15 @@ async def test_download_blob_large_chunks(self, **kwargs):
         await blob.upload_blob(data, overwrite=True, max_concurrency=5)
 
         # Act
-        downloader = await blob.download_blob(validate_content='crc64', max_concurrency=3)
+        downloader = await blob.download_blob(validate_content='crc64', max_concurrency=5)
         content = await downloader.read()
 
-        downloader = await blob.download_blob(offset=5 * 1024 * 1024, length=25 * 1024 * 1024)
+        downloader = await blob.download_blob(offset=5 * 1024 * 1024, length=25 * 1024 * 1024, validate_content='crc64')
         partial = await downloader.read()
 
         # Assert
         assert content == data
         assert partial == data[5 * 1024 * 1024: 30 * 1024 * 1024]
-        await self._teardown()
 
     @BlobPreparer()
     @pytest.mark.parametrize('a', [True, 'md5', 'crc64'])  # a: validate_content
@@ -488,4 +481,3 @@ async def test_download_blob_chars(self, a, **kwargs):
 
         result += await stream.readall()
         assert result == data
-        await self._teardown()

sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_data_lake_file_client.py

@@ -427,15 +427,11 @@ def upload_data(
             If a date is passed in without timezone info, it is assumed to be UTC.
             Specify this header to perform the operation only if
             the resource has not been modified since the specified date/time.
-        :keyword bool validate_content:
-            If true, calculates an MD5 hash for each chunk of the file. The storage
-            service checks the hash of the content that has arrived with the hash
-            that was sent. This is primarily valuable for detecting bitflips on
-            the wire if using http instead of https, as https (the default), will
-            already validate. Note that this MD5 hash is not stored with the
-            blob. Also note that if enabled, the memory-efficient upload algorithm
-            will not be used because computing the MD5 hash requires buffering
-            entire blocks, and doing so defeats the purpose of the memory-efficient algorithm.
+        :keyword validate_content:
+            Enables checksum validation for the transfer. Any checksum calculated is NOT stored with the blob.
+            Choose "auto" (let the SDK choose the best algorithm), "crc64", or "md5". The use of bool is deprecated.
+            NOTE: The use of "auto" or "crc64" requires the `azure-storage-extensions` package to be installed.
+        :paramtype validate_content: Union[bool, Literal['auto', 'crc64', 'md5']]
         :keyword str etag:
             An ETag value, or the wildcard character (*). Used to check if the resource has changed,
             and act according to the condition specified by the `match_condition` parameter.
@@ -497,13 +493,11 @@ def append_data(
         :type length: int or None
         :keyword bool flush:
             If true, will commit the data after it is appended.
-        :keyword bool validate_content:
-            If true, calculates an MD5 hash of the block content. The storage
-            service checks the hash of the content that has arrived
-            with the hash that was sent. This is primarily valuable for detecting
-            bitflips on the wire if using http instead of https as https (the default)
-            will already validate. Note that this MD5 hash is not stored with the
-            file.
+        :keyword validate_content:
+            Enables checksum validation for the transfer. Any checksum calculated is NOT stored with the blob.
+            Choose "auto" (let the SDK choose the best algorithm), "crc64", or "md5". The use of bool is deprecated.
+            NOTE: The use of "auto" or "crc64" requires the `azure-storage-extensions` package to be installed.
+        :paramtype validate_content: Union[bool, Literal['auto', 'crc64', 'md5']]
         :keyword lease_action:
             Used to perform lease operations along with appending data.
 
@@ -714,6 +708,11 @@ def download_file(
         :paramtype progress_hook: ~typing.Callable[[int, int], None]
         :keyword bool decompress: If True, any compressed content, identified by the Content-Encoding header, will be
             decompressed automatically before being returned. Default value is True.
+        :keyword validate_content:
+            Enables checksum validation for the transfer. Any checksum calculated is NOT stored with the blob.
+            Choose "auto" (let the SDK choose the best algorithm), "crc64", or "md5". The use of bool is deprecated.
+            NOTE: The use of "auto" or "crc64" requires the `azure-storage-extensions` package to be installed.
+        :paramtype validate_content: Union[bool, Literal['auto', 'crc64', 'md5']]
         :keyword int timeout:
             Sets the server-side timeout for the operation in seconds. For more details see
             https://learn.microsoft.com/rest/api/storageservices/setting-timeouts-for-blob-service-operations.

sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_data_lake_file_client.pyi

@@ -149,7 +149,7 @@ class DataLakeFileClient(PathClient):
         if_unmodified_since: Optional[datetime] = None,
         etag: Optional[str] = None,
         match_condition: Optional[MatchConditions] = None,
-        validate_content: Optional[bool] = None,
+        validate_content: Optional[Union[bool, Literal['auto', 'crc64', 'md5']]] = None,
         cpk: Optional[CustomerProvidedEncryptionKey] = None,
         max_concurrency: Optional[int] = None,
         chunk_size: Optional[int] = None,
@@ -166,7 +166,7 @@ class DataLakeFileClient(PathClient):
         length: Optional[int] = None,
         *,
         flush: Optional[bool] = None,
-        validate_content: Optional[bool] = None,
+        validate_content: Optional[Union[bool, Literal['auto', 'crc64', 'md5']]] = None,
         lease_action: Optional[Literal["acquire", "auto-renew", "release", "acquire-release"]] = None,
         lease_duration: int = -1,
         lease: Optional[Union[DataLakeLeaseClient, str]] = None,
@@ -207,6 +207,7 @@ class DataLakeFileClient(PathClient):
         cpk: Optional[CustomerProvidedEncryptionKey] = None,
         max_concurrency: Optional[int] = None,
         progress_hook: Optional[Callable[[int, int], None]] = None,
+        validate_content: Optional[Union[bool, Literal['auto', 'crc64', 'md5']]] = None,
         timeout: Optional[int] = None,
         **kwargs: Any
     ) -> StorageStreamDownloader: ...

sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_data_lake_file_client_helpers.py

@@ -24,6 +24,7 @@
 from ._shared.response_handlers import return_response_headers
 from ._shared.uploads import IterStreamer
 from ._shared.uploads_async import AsyncIterStreamer
+from ._shared.validation import parse_validation_option
 
 if TYPE_CHECKING:
     from ._generated.operations import PathOperations
@@ -47,14 +48,16 @@ def _append_data_options(
     if isinstance(data, bytes):
         data = data[:length]
 
+    validate_content = parse_validation_option(kwargs.pop('validate_content', None))
+
     cpk_info = get_cpk_info(scheme, kwargs)
     kwargs.update(get_lease_action_properties(kwargs))
 
     options = {
         'body': data,
         'position': offset,
         'content_length': length,
-        'validate_content': kwargs.pop('validate_content', False),
+        'validate_content': validate_content,
         'cpk_info': cpk_info,
         'timeout': kwargs.pop('timeout', None),
         'cls': return_response_headers
@@ -122,7 +125,7 @@ def _upload_options(
     else:
         raise TypeError(f"Unsupported data type: {type(data)}")
 
-    validate_content = kwargs.pop('validate_content', False)
+    validate_content = parse_validation_option(kwargs.pop('validate_content', None))
     content_settings = kwargs.pop('content_settings', None)
     metadata = kwargs.pop('metadata', None)
     max_concurrency = kwargs.pop('max_concurrency', None)

sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/base_client_async.py

@@ -36,12 +36,11 @@
 from .parser import DEVSTORE_ACCOUNT_KEY, _get_development_storage_endpoint
 from .policies import (
     QueueMessagePolicy,
-    StorageContentValidation,
     StorageHeadersPolicy,
     StorageHosts,
     StorageRequestHook,
 )
-from .policies_async import AsyncStorageBearerTokenCredentialPolicy, AsyncStorageResponseHook
+from .policies_async import AsyncStorageBearerTokenCredentialPolicy, AsyncContentValidationPolicy, AsyncStorageResponseHook
 from .response_handlers import PartialBatchErrorException, process_storage_error
 from .._shared_access_signature import _is_credential_sastoken
 
@@ -130,7 +129,7 @@ def _create_pipeline(
             QueueMessagePolicy(),
             config.proxy_policy,
             config.user_agent_policy,
-            StorageContentValidation(),
+            AsyncContentValidationPolicy(),
             ContentDecodePolicy(response_encoding="utf-8"),
             AsyncRedirectPolicy(**kwargs),
             StorageHosts(hosts=hosts, **kwargs),