@pytest.mark.live_test_only
@GlobalStorageAccountPreparer()
@AsyncStorageTestCase.await_prepared_test
async def test_upload_to_url_bytes_with_sas(self, resource_group, location, storage_account, storage_account_key):
    (await self._setup(storage_account.name, storage_account_key))
    data = ((b'12345678' * 1024) * 1024)
    blob_name = self._get_blob_reference()
    blob = self.bsc.get_blob_client(self.container_name, blob_name)
    token = generate_blob_sas(blob.account_name, blob.container_name, blob.blob_name, snapshot=blob.snapshot, account_key=blob.credential.account_key, permission=BlobSasPermissions(write=True), expiry=(datetime.utcnow() + timedelta(hours=1)))
    sas_blob = BlobClient.from_blob_url(blob.url, credential=token)
    uploaded = (await upload_blob_to_url(sas_blob.url, data))
    self.assertIsNotNone(uploaded)
    content = (await (await blob.download_blob()).readall())
    self.assertEqual(data, content)
@pytest.mark.live_test_only
@GlobalStorageAccountPreparer()
@AsyncStorageTestCase.await_prepared_test
async def test_upload_to_url_bytes_with_sas(self, resource_group, location, storage_account, storage_account_key):
    (await self._setup(storage_account.name, storage_account_key))
    data = ((b'12345678' * 1024) * 1024)
    blob_name = self._get_blob_reference()
    blob = self.bsc.get_blob_client(self.container_name, blob_name)
    token = generate_blob_sas(blob.account_name, blob.container_name, blob.blob_name, snapshot=blob.snapshot, account_key=blob.credential.account_key, permission=BlobSasPermissions(write=True), expiry=(datetime.utcnow() + timedelta(hours=1)))
    sas_blob = BlobClient.from_blob_url(blob.url, credential=token)
    uploaded = (await upload_blob_to_url(sas_blob.url, data))
    self.assertIsNotNone(uploaded)
    content = (await (await blob.download_blob()).readall())
    self.assertEqual(data, content)