added sync versions of error code for copy blob tests and recordings

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_de13a7bd1f"
+  "Tag": "python/storage/azure-storage-blob_d03ceb7e4b"
 }

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

@@ -340,6 +340,30 @@ def test_append_block_from_url(self, **kwargs):
             destination_blob_client.append_block_from_url(source_blob_client.url + '?' + sas,
                                                           source_length=LARGE_BLOB_SIZE)
 
+    @BlobPreparer()
+    @recorded_by_proxy
+    def test_append_blob_async_copy_source_error_and_status_code(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        # Arrange
+        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)
+        dest_blob = self._create_blob(bsc)
+
+        # Act
+        with pytest.raises(HttpResponseError) as e:
+            dest_blob.append_block_from_url(source_blob.url)
+
+        # Assert
+        assert e.value.response.headers["x-ms-copy-source-status-code"] == "409"
+        assert e.value.response.headers["x-ms-copy-source-error-code"] == "PublicAccessNotPermitted"
+        assert "copysourceerrormessage:Public access is not permitted on this storage account." in e.value.message
+
     @BlobPreparer()
     @recorded_by_proxy
     def test_append_block_from_url_and_validate_content_md5(self, **kwargs):

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

@@ -624,6 +624,33 @@ def test_put_block_with_immutability_policy(self, **kwargs):
 
         return variables
 
+    @BlobPreparer()
+    @recorded_by_proxy
+    def test_stage_block_from_url_copy_source_error_and_status_code(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        # Arrange
+        self._setup(storage_account_name, storage_account_key)
+        source_blob_name = "sourceblob"
+        source_blob = self.bsc.get_blob_client(self.container_name, source_blob_name)
+        target_blob_name = "targetblob"
+        target_blob = self.bsc.get_blob_client(self.container_name, target_blob_name)
+        split = 4 * 1024
+
+        # Act
+        with pytest.raises(HttpResponseError) as e:
+            target_blob.stage_block_from_url(
+                block_id=1,
+                source_url=source_blob.url,
+                source_offset=0,
+                source_length=split)
+
+        # Assert
+        assert e.value.response.headers["x-ms-copy-source-status-code"] == "409"
+        assert e.value.response.headers["x-ms-copy-source-error-code"] == "PublicAccessNotPermitted"
+        assert "copysourceerrormessage:Public access is not permitted on this storage account." in e.value.message
+
     @BlobPreparer()
     @recorded_by_proxy
     def test_put_block_list_invalid_block_id(self, **kwargs):

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

@@ -1693,6 +1693,61 @@ def test_abort_copy_blob_with_synchronous_copy_fails(self, **kwargs):
         # Assert
         assert copy_resp['copy_status'] == 'success'
 
+    @BlobPreparer()
+    @recorded_by_proxy
+    def test_copy_blob_async_copy_source_error_and_status_code(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        # Arrange
+        self._setup(storage_account_name, storage_account_key)
+        source_blob_name = "sourceblob"
+        source_blob = self.bsc.get_blob_client(self.container_name, source_blob_name)
+        target_blob_name = "targetblob"
+        target_blob = self.bsc.get_blob_client(self.container_name, target_blob_name)
+
+        sas_token = self.generate_sas(
+            generate_blob_sas,
+            source_blob.account_name,
+            source_blob.container_name,
+            source_blob.blob_name,
+            account_key=source_blob.credential.account_key,
+            permission=BlobSasPermissions(read=True),
+            expiry=datetime.utcnow() + timedelta(hours=1),
+        )
+        blob = BlobClient.from_blob_url(source_blob.url, credential=sas_token)
+
+        # Act
+        with pytest.raises(HttpResponseError) as e:
+            target_blob.start_copy_from_url(blob.url)
+
+        # Assert
+        assert e.value.response.headers["x-ms-copy-source-error-code"] == "BlobNotFound"
+        assert e.value.response.headers["x-ms-copy-source-status-code"] == "404"
+        assert "copysourceerrormessage:The specified blob does not exist." in e.value.message
+
+    @BlobPreparer()
+    @recorded_by_proxy
+    def test_copy_blob_copy_source_error_and_status_code(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        # Arrange
+        self._setup(storage_account_name, storage_account_key)
+        source_blob_name = "sourceblob"
+        source_blob = self.bsc.get_blob_client(self.container_name, source_blob_name)
+        target_blob_name = "targetblob"
+        target_blob = self.bsc.get_blob_client(self.container_name, target_blob_name)
+
+        # Act
+        with pytest.raises(HttpResponseError) as e:
+            target_blob.start_copy_from_url(source_blob.url, requires_sync=True)
+
+        # Assert
+        assert e.value.response.headers["x-ms-copy-source-status-code"] == "409"
+        assert e.value.response.headers["x-ms-copy-source-error-code"] == "PublicAccessNotPermitted"
+        assert "copysourceerrormessage:Public access is not permitted on this storage account." in e.value.message
+
     @BlobPreparer()
     @recorded_by_proxy
     def test_snapshot_blob(self, **kwargs):

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

@@ -577,6 +577,30 @@ def test_upload_pages_from_url(self, **kwargs):
         assert blob_properties.get('etag') == source_with_special_chars_resp.get('etag')
         assert blob_properties.get('last_modified') == source_with_special_chars_resp.get('last_modified')
 
+    @BlobPreparer()
+    @recorded_by_proxy
+    def test_upload_pages_from_url_copy_source_error_and_status_code(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        # Arrange
+        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)
+        dest_blob = self._create_blob(bsc)
+
+        # Act
+        with pytest.raises(HttpResponseError) as e:
+            dest_blob.upload_blob_from_url(source_blob.url)
+
+        # Assert
+        assert e.value.response.headers["x-ms-copy-source-status-code"] == "409"
+        assert e.value.response.headers["x-ms-copy-source-error-code"] == "PublicAccessNotPermitted"
+        assert "copysourceerrormessage:Public access is not permitted on this storage account." in e.value.message
+
     @BlobPreparer()
     @recorded_by_proxy
     def test_upload_pages_from_url_with_oauth(self, **kwargs):

sdk/storage/azure-storage-file-share/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/storage/azure-storage-file-share",
-  "Tag": "python/storage/azure-storage-file-share_50b174c48c"
+  "Tag": "python/storage/azure-storage-file-share_a600655193"
 }

sdk/storage/azure-storage-file-share/tests/test_file.py

@@ -1566,6 +1566,30 @@ def test_update_range_from_file_url_with_oauth(self, **kwargs):
         destination_file_client.upload_range_from_url(source_blob_client.url, offset=0, length=512, source_offset=0,
                                                       source_authorization=token)
 
+    @FileSharePreparer()
+    @recorded_by_proxy
+    def test_update_range_from_file_url_copy_source_error_and_status_code(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        self._setup(storage_account_name, storage_account_key)
+        source_file_name = 'testfile'
+        source_file_client = self._create_file(file_name=source_file_name)
+        data = b'abcdefghijklmnop' * 32
+        source_file_client.upload_range(data, offset=0, length=512)
+
+        destination_file_name = 'filetoupdate'
+        destination_file_client = self._create_empty_file(file_name=destination_file_name)
+
+        with pytest.raises(HttpResponseError) as e:
+            destination_file_client.upload_range_from_url(
+                source_file_client.url, offset=0, length=512, source_offset=0)
+
+        assert e.value.response.headers["x-ms-copy-source-status-code"] == "401"
+        assert e.value.response.headers["x-ms-copy-source-error-code"] == "NoAuthenticationInformation"
+        assert ("copysourceerrormessage:Server failed to authenticate the request. Please refer to the information in "
+                "the www-authenticate header.") in e.value.message
+
     @FileSharePreparer()
     @recorded_by_proxy
     def test_update_range_from_file_url_with_lease(self, **kwargs):
@@ -2584,6 +2608,30 @@ def test_abort_copy_file_with_synchronous_copy_fails(self, **kwargs):
         # Assert
         assert copy_resp['copy_status'] == 'success'
 
+    @FileSharePreparer()
+    @recorded_by_proxy
+    def test_copy_file_async_copy_source_error_and_status_code(self, **kwargs):
+        storage_account_name = kwargs.pop("storage_account_name")
+        storage_account_key = kwargs.pop("storage_account_key")
+
+        self._setup(storage_account_name, storage_account_key)
+
+        # Act
+        file_client = ShareFileClient(
+            self.account_url(storage_account_name, "file"),
+            share_name=self.share_name,
+            file_path="targetfile",
+            credential=storage_account_key)
+
+        with pytest.raises(ResourceNotFoundError) as e:
+            file_client.start_copy_from_url("https://error.file.core.windows.net/")
+
+        # Assert
+        assert e.value.response.headers["x-ms-copy-source-status-code"] == "400"
+        assert e.value.response.headers["x-ms-copy-source-error-code"] == "InvalidQueryParameterValue"
+        assert ("copysourceerrormessage:Value for one of the query parameters specified in the "
+                "request URI is invalid.") in e.value.message
+
     @FileSharePreparer()
     @recorded_by_proxy
     def test_unicode_get_file_unicode_name(self, **kwargs):