@Test
  public void testNormalFlow() throws Exception {
    cleanup();
    PartitionId partitionId = new MockPartitionId(testPartition, MockClusterMap.DEFAULT_PARTITION_CLASS);
    BlobId blobId = new BlobId(BLOB_ID_V6, BlobIdType.NATIVE, dataCenterId, accountId, containerId, partitionId, false,
        BlobDataType.DATACHUNK);
    byte[] uploadData = TestUtils.getRandomBytes(blobSize);
    InputStream inputStream = new ByteArrayInputStream(uploadData);
    long now = System.currentTimeMillis();
    CloudBlobMetadata cloudBlobMetadata =
        new CloudBlobMetadata(blobId, now, now + 60000, blobSize, CloudBlobMetadata.EncryptionOrigin.VCR, vcrKmsContext,
            cryptoAgentFactory, blobSize, (short) 0);

    // attempt undelete before uploading blob
    try {
      undeleteBlobWithRetry(blobId, (short) 1);
      fail("Undelete of a non existent blob should fail.");
    } catch (CloudStorageException cex) {
      assertEquals(cex.getStatusCode(), HttpConstants.StatusCodes.NOTFOUND);
    }

    assertTrue("Expected upload to return true",
        AzureTestUtils.uploadBlobWithRetry(blobId, blobSize, cloudBlobMetadata, inputStream, cloudRequestAgent,
            azureDest));

    // Get blob should return the same data
    verifyDownloadMatches(blobId, uploadData);

    // Try to upload same blob again
    assertFalse("Expected duplicate upload to return false",
        AzureTestUtils.uploadBlobWithRetry(blobId, blobSize, cloudBlobMetadata, new ByteArrayInputStream(uploadData),
            cloudRequestAgent, azureDest));

    // ttl update
    long expirationTime = Utils.Infinite_Time;
    try {
      updateBlobExpirationWithRetry(blobId, expirationTime);
    } catch (Exception ex) {
      fail("Expected update to be successful");
    }
    CloudBlobMetadata metadata =
        getBlobMetadataWithRetry(Collections.singletonList(blobId), partitionId.toPathString(), cloudRequestAgent,
            azureDest).get(blobId.getID());
    assertEquals(expirationTime, metadata.getExpirationTime());

    // delete blob
    long deletionTime = now + 10000;
    //TODO add a test case here to verify life version after delete.
    assertTrue("Expected deletion to return true", cloudRequestAgent.doWithRetries(
        () -> azureDest.deleteBlob(blobId, deletionTime, (short) 0, dummyCloudUpdateValidator), "DeleteBlob",
        partitionId.toPathString()));
    metadata =
        getBlobMetadataWithRetry(Collections.singletonList(blobId), partitionId.toPathString(), cloudRequestAgent,
            azureDest).get(blobId.getID());
    assertEquals(deletionTime, metadata.getDeletionTime());

    // undelete blob
    assertEquals(undeleteBlobWithRetry(blobId, (short) 1), 1);
    metadata =
        getBlobMetadataWithRetry(Collections.singletonList(blobId), partitionId.toPathString(), cloudRequestAgent,
            azureDest).get(blobId.getID());
    assertEquals(metadata.getDeletionTime(), Utils.Infinite_Time);
    assertEquals(metadata.getLifeVersion(), 1);

    // undelete with a higher life version updates life version.
    assertEquals(undeleteBlobWithRetry(blobId, (short) 2), 2);
    metadata =
        getBlobMetadataWithRetry(Collections.singletonList(blobId), partitionId.toPathString(), cloudRequestAgent,
            azureDest).get(blobId.getID());
    assertEquals(metadata.getDeletionTime(), Utils.Infinite_Time);
    assertEquals(metadata.getLifeVersion(), 2);

    // delete after undelete. Set the deletion time to some value which is before the retentionPeriodDays time period.
    long newDeletionTime = now - TimeUnit.DAYS.toMillis(retentionPeriodDays + 1);
    //TODO add a test case here to verify life version after delete.
    assertTrue("Expected deletion to return true", cloudRequestAgent.doWithRetries(
        () -> azureDest.deleteBlob(blobId, newDeletionTime, (short) 3, dummyCloudUpdateValidator), "DeleteBlob",
        partitionId.toPathString()));
    metadata =
        getBlobMetadataWithRetry(Collections.singletonList(blobId), partitionId.toPathString(), cloudRequestAgent,
            azureDest).get(blobId.getID());
    assertEquals(newDeletionTime, metadata.getDeletionTime());
    // delete changes life version.
    assertEquals(metadata.getLifeVersion(), 3);

    // compact partition
    azureDest.compactPartition(partitionId.toPathString());
    assertTrue("Expected empty set after purge",
        getBlobMetadataWithRetry(Collections.singletonList(blobId), partitionId.toPathString(), cloudRequestAgent,
            azureDest).isEmpty());

    // Get blob should fail after purge
    try {
      verifyDownloadMatches(blobId, uploadData);
      fail("download blob should fail after data is purged");
    } catch (CloudStorageException csex) {
    }
  }
Test normal operations.
@throws Exception on error
ambry-cloud/src/test/java/com/github/ambry/cloud/azure/AzureIntegrationTest.java