content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.cli;
  
  import alluxio.conf.InstancedConfiguration;
  import alluxio.conf.PropertyKey;
  import alluxio.underfs.ContentHashable;
  import alluxio.underfs.Fingerprint;
  import alluxio.underfs.UfsDirectoryStatus;
  import alluxio.underfs.UfsFileStatus;
  import alluxio.underfs.UfsStatus;
  import alluxio.underfs.UnderFileSystem;
  import alluxio.underfs.options.CreateOptions;
  import alluxio.underfs.options.DeleteOptions;
  import alluxio.underfs.options.ListOptions;
  import alluxio.underfs.options.MkdirsOptions;
  import alluxio.underfs.options.OpenOptions;
  import alluxio.util.CommonUtils;
  import alluxio.util.UnderFileSystemUtils;
  import alluxio.util.WaitForOptions;
  import alluxio.util.io.PathUtils;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.concurrent.atomic.AtomicReference;
  
  /**
   * Examples for under filesystem common operations.
   * The class should contain all the Alluxio ufs semantics.
   */
  public final class UnderFileSystemCommonOperations {
    private static final Logger LOG = LoggerFactory.getLogger(UnderFileSystemCommonOperations.class);
    private static final byte[] TEST_BYTES = "TestBytes".getBytes();
  
    private static final String FILE_CONTENT_LENGTH_INCORRECT
        = "The content length of the written file is %s but expected %s";
    private static final String FILE_CONTENT_INCORRECT
        = "The content of the written file is incorrect";
    private static final String FILE_CONTENT_HASH_DOES_NOT_MATCH_UFS
        = "Content hash computed during file upload does not match content hash on UFS";
    private static final String FILE_EXISTS_CHECK_SHOULD_SUCCEED
        = "Should succeed in UnderFileSystem.exists() check, but failed";
    private static final String FILE_EXISTS_CHECK_SHOULD_FAILED
        = "Should failed in UnderFileSystem.exists() check, but succeed";
    private static final String FILE_STATUS_RESULT_INCORRECT
        = "The result of UnderFileSystem.getFileStatus() is incorrect";
    private static final String IS_FAIL_CHECK_SHOULD_SUCCEED
        = "Should succeed in UnderFileSystem.isFile() check, but failed";
    private static final String IS_FAIL_CHECK_SHOULD_FAILED
        = "Should failed in UnderFileSystem.isFile() check, but succeed";
    private static final String IS_DIRECTORY_CHECK_SHOULD_SUCCEED
        = "Should succeed in UnderFileSystem.isDirectory() check, but failed";
    private static final String IS_DIRECTORY_CHECK_SHOULD_FAILED
        = "Should failed in UnderFileSystem.isDirectory() check, but succeed";
    private static final String LIST_STATUS_RESULT_INCORRECT
        = "The result of UnderFileSystem.listStatus() is incorrect";
    private static final int RETRY_TIMEOUT_MS = 180000;
    private static final int RETRY_INTERVAL_MS = 1000;
  
    private final InstancedConfiguration mConfiguration;
    private final UnderFileSystem mUfs;
    private final String mUfsPath;
    // A child directory of the ufs path to run tests against
    private final String mTopLevelTestDirectory;
  
    /**
     * @param ufsPath the under filesystem path
     * @param topLevelTestDirectory the top level test directory
     * @param ufs the under filesystem
     * @param configuration the instance configuration
     */
    public UnderFileSystemCommonOperations(String ufsPath, String topLevelTestDirectory,
        UnderFileSystem ufs, InstancedConfiguration configuration) {
      mUfsPath = ufsPath;
      mTopLevelTestDirectory = topLevelTestDirectory;
      mUfs = ufs;
      mConfiguration = configuration;
    }
  
    /**
     * Test for creating file atomic.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata"})
    public void createAtomicTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createAtomic");
      OutputStream stream = mUfs.create(testFile, CreateOptions.defaults(mConfiguration)
          .setEnsureAtomic(true));
      stream.write(TEST_BYTES);
      if (mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
      stream.close();
      checkContentHash(testFile, stream);
      if (!mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for creating empty file.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata"})
    public void createEmptyTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createEmpty");
      createEmptyFile(testFile);
      if (!mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for creating file without parent.
     */
    @RelatedS3Operations(operations = {})
    public void createNoParentTest() throws IOException {
      // Run the test only for local UFS. Other UFSs succeed if no parents are present
      if (!UnderFileSystemUtils.isLocal(mUfs)) {
        return;
      }
      boolean haveIOException = false;
      try {
        String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createNoParent/testFile");
        OutputStream o = mUfs.create(testFile, CreateOptions.defaults(mConfiguration)
            .setCreateParent(false));
        o.close();
      } catch (IOException e) { // Expected to have IOException
        haveIOException = true;
      }
      if (!haveIOException) {
        throw new IOException("Expected to have IOException but do not have");
      }
    }
  
    /**
     * Test for creating file with parent.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata"})
    public void createParentTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createParent/testFile");
      OutputStream o = mUfs.create(testFile, CreateOptions.defaults(mConfiguration)
          .setCreateParent(true));
      o.close();
      checkContentHash(testFile, o);
      if (!mUfs.exists(testFile)) {
        throw new IOException(FILE_EXISTS_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for creating and opening file.
     */
    @RelatedS3Operations(operations = {"upload", "getObject"})
    public void createOpenTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createOpen");
      createTestBytesFile(testFile);
      byte[] buf = new byte[TEST_BYTES.length];
      int bytesRead = mUfs.open(testFile).read(buf);
      if (TEST_BYTES.length != bytesRead || !Arrays.equals(buf, TEST_BYTES)) {
        throw new IOException(FILE_CONTENT_INCORRECT);
      }
    }
  
    /**
     * Test for creating and opening empty file.
     */
    @RelatedS3Operations(operations = {"upload", "getObject"})
    public void createOpenEmptyTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createOpenEmpty");
      createEmptyFile(testFile);
      byte[] buf = new byte[0];
      int bytesRead = mUfs.open(testFile).read(buf);
      boolean bytesReadCorrect = bytesRead == 0;
      if (UnderFileSystemUtils.isHdfs(mUfs) && bytesRead == -1) {
        // TODO(adit): Consider making the return value uniform across UFSs
        bytesReadCorrect = true;
      }
      if (!bytesReadCorrect) {
        throw new IOException(String.format(FILE_CONTENT_LENGTH_INCORRECT, bytesRead, 0));
      }
    }
  
    /**
     * Test for creating file and opening at position.
     */
    @RelatedS3Operations(operations = {"upload", "getObject"})
    public void createOpenAtPositionTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createOpenAtPosition");
      prepareMultiBlockFile(testFile);
      int[] offsets = {0, 256, 511, 512, 513, 768, 1024, 1025};
      for (int offset : offsets) {
        InputStream inputStream = mUfs.open(testFile, OpenOptions.defaults().setOffset(offset));
        if (TEST_BYTES[offset % TEST_BYTES.length] != inputStream.read()) {
          throw new IOException(FILE_CONTENT_INCORRECT);
        }
        inputStream.close();
      }
    }
  
    /**
     * Test for creating and opening large file.
     */
    @RelatedS3Operations(operations = {"upload", "getObject"})
    public void createOpenLargeTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createOpenLarge");
      int numCopies = prepareMultiBlockFile(testFile);
      InputStream inputStream = mUfs.open(testFile);
      byte[] buf = new byte[numCopies * TEST_BYTES.length];
      int offset = 0;
      int noReadCount = 0;
      while (offset < buf.length && noReadCount < 3) {
        int bytesRead;
        try {
          bytesRead = inputStream.read(buf, offset, buf.length - offset);
        } catch (Exception e) {
          LOG.info("Failed to read from file {}: {}", testFile, e.toString());
          bytesRead = -1;
        }
        if (bytesRead != -1) {
          noReadCount = 0;
          for (int i = 0; i < bytesRead; ++i) {
            if (TEST_BYTES[(offset + i) % TEST_BYTES.length] != buf[offset + i]) {
              throw new IOException(FILE_CONTENT_INCORRECT);
            }
          }
          offset += bytesRead;
        } else {
          ++noReadCount;
        }
      }
      if (noReadCount > 3) {
        throw new IOException("Too many retries in reading the written large file");
      }
    }
  
    /**
     * Test for creating and open existing large file.
     */
    @RelatedS3Operations(operations = {"upload", "getObject"})
    public void createOpenExistingLargeFileTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createOpenExistingLargeFile");
      int numCopies = prepareMultiBlockFile(testFile);
      InputStream inputStream = mUfs.openExistingFile(testFile);
      byte[] buf = new byte[numCopies * TEST_BYTES.length];
      int offset = 0;
      while (offset < buf.length) {
        int bytesRead = inputStream.read(buf, offset, buf.length - offset);
        if (bytesRead == -1) {
          break;
        }
        for (int i = 0; i < bytesRead; ++i) {
          if (TEST_BYTES[(offset + i) % TEST_BYTES.length] != buf[offset + i]) {
            throw new IOException(FILE_CONTENT_INCORRECT);
          }
        }
        offset += bytesRead;
      }
      if (buf.length != offset) {
        throw new IOException(FILE_CONTENT_INCORRECT);
      }
    }
  
    /**
     * Test for create file, open and seek.
     */
    @RelatedS3Operations(operations = {"upload", "getObject"})
    public void createOpenSkip() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "createOpenSkip");
      prepareMultiBlockFile(testFile);
      int[] offsets = {0, 256, 511, 512, 513, 768, 1024, 1025};
      for (int offset : offsets) {
        InputStream inputStream = mUfs.open(testFile, OpenOptions.defaults());
        long bytesSkipped = 0;
        while (bytesSkipped != offset) {
          bytesSkipped += inputStream.skip(offset - bytesSkipped);
        }
        if (TEST_BYTES[offset % TEST_BYTES.length] != inputStream.read()) {
          throw new IOException(FILE_CONTENT_INCORRECT);
        }
        inputStream.close();
      }
    }
  
    /**
     * Test for deleting file.
     */
    @RelatedS3Operations(operations = {"upload", "deleteObject", "getObjectMetadata"})
    public void deleteFileTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "deleteFile");
      createEmptyFile(testFile);
      mUfs.deleteFile(testFile);
      if (mUfs.exists(testFile)) {
        throw new IOException(FILE_EXISTS_CHECK_SHOULD_FAILED);
      }
      if (mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
    }
  
    /**
     * Test for deleting directory.
     */
    @RelatedS3Operations(operations = {"putObject", "deleteObjects",
        "listObjectsV2", "getObjectMetadata"})
    public void deleteDirTest() throws IOException {
      String testDirEmpty = PathUtils.concatPath(mTopLevelTestDirectory, "deleteDirTestDirEmpty");
      String testDirNonEmpty = PathUtils
          .concatPath(mTopLevelTestDirectory, "deleteDirTestDirNonEmpty1");
      String testDirNonEmptyChildDir
          = PathUtils.concatPath(testDirNonEmpty, "deleteDirTestDirNonEmpty2");
      String testDirNonEmptyChildFile
          = PathUtils.concatPath(testDirNonEmpty, "deleteDirTestDirNonEmptyF");
      String testDirNonEmptyChildDirFile =
          PathUtils.concatPath(testDirNonEmptyChildDir, "deleteDirTestDirNonEmptyChildDirF");
      mUfs.mkdirs(testDirEmpty, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      mUfs.mkdirs(testDirNonEmpty, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      mUfs.mkdirs(testDirNonEmptyChildDir, MkdirsOptions.defaults(mConfiguration)
          .setCreateParent(false));
      createEmptyFile(testDirNonEmptyChildFile);
      createEmptyFile(testDirNonEmptyChildDirFile);
      mUfs.deleteDirectory(testDirEmpty, DeleteOptions.defaults().setRecursive(false));
      if (mUfs.isDirectory(testDirEmpty)) {
        throw new IOException("Directory is deleted "
            + "but succeed in UnderFileSystem.isDirectory() check");
      }
      try {
        mUfs.deleteDirectory(testDirNonEmpty, DeleteOptions.defaults().setRecursive(false));
      } catch (IOException e) {
        // Some File systems may throw IOException
      }
      if (!mUfs.isDirectory(testDirNonEmpty)) {
        throw new IOException("Created directory should succeed "
            + "in UnderFileSystem.isDirectory() check, but failed");
      }
      mUfs.deleteDirectory(testDirNonEmpty, DeleteOptions.defaults().setRecursive(true));
      if (mUfs.isDirectory(testDirNonEmpty) || mUfs.isDirectory(testDirNonEmptyChildDir)
          || mUfs.isFile(testDirNonEmptyChildFile) || mUfs.isFile(testDirNonEmptyChildDirFile)) {
        throw new IOException("Deleted file or directory still exist");
      }
    }
  
    /**
     * Test for deleting large directory.
     */
    @RelatedS3Operations(operations = {"putObject", "deleteObjects",
        "listObjectsV2", "getObjectMetadata"})
    public void deleteLargeDirectoryTest() throws Exception {
      LargeDirectoryConfig config = prepareLargeDirectory();
      mUfs.deleteExistingDirectory(config.getTopLevelDirectory(),
          DeleteOptions.defaults().setRecursive(true));
  
      String[] children = config.getChildren();
      for (String child : children) {
        // Retry for some time to allow list operations eventual consistency for S3 and GCS.
        // See http://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html and
        // https://cloud.google.com/storage/docs/consistency for more details.
        CommonUtils.waitFor("deleted path does not exist", () -> {
          try {
            return !mUfs.isFile(child) && !mUfs.isDirectory(child);
          } catch (Exception e) {
            return false;
          }
        }, WaitForOptions.defaults().setTimeoutMs(RETRY_TIMEOUT_MS).setInterval(RETRY_INTERVAL_MS));
      }
    }
  
    /**
     * Test for creating and deleting file conjunction.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata", "deleteObject"})
    public void createDeleteFileConjuctionTest() throws IOException {
      String testFile = PathUtils
          .concatPath(mTopLevelTestDirectory, "deleteThenCreateNonexistingFile");
      createTestBytesFile(testFile);
      if (!mUfs.exists(testFile)) {
        throw new IOException(FILE_EXISTS_CHECK_SHOULD_SUCCEED);
      }
      if (!mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
  
      mUfs.deleteExistingFile(testFile);
      if (mUfs.exists(testFile)) {
        throw new IOException(FILE_EXISTS_CHECK_SHOULD_FAILED);
      }
  
      OutputStream o = mUfs.createNonexistingFile(testFile);
      o.write(TEST_BYTES);
      o.close();
      if (!mUfs.exists(testFile)) {
        throw new IOException(FILE_EXISTS_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for creating and deleting existing directory.
     */
    @RelatedS3Operations(operations = {"putObject", "deleteObjects",
        "listObjectsV2", "getObjectMetadata"})
    public void createThenDeleteExistingDirectoryTest() throws IOException {
      LargeDirectoryConfig config = prepareLargeDirectory();
      if (!mUfs.deleteExistingDirectory(config.getTopLevelDirectory(),
          DeleteOptions.defaults().setRecursive(true))) {
        throw new IOException("Failed to delete existing directory");
      }
    }
  
    /**
     * Test for checking file existence.
     */
    @RelatedS3Operations(operations = {"putObject", "getObjectMetadata"})
    public void existsTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "testFile");
      if (mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
      createEmptyFile(testFile);
      if (!mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
      String testDir = PathUtils.concatPath(mTopLevelTestDirectory, "testDir");
      if (mUfs.isDirectory(testDir)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_FAILED);
      }
      mUfs.mkdirs(testDir, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      if (!mUfs.isDirectory(testDir)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for getting directory status.
     */
    @RelatedS3Operations(operations = {"putObject", "getObjectMetadata"})
    public void getDirectoryStatusTest() throws IOException {
      String testDir = PathUtils.concatPath(mTopLevelTestDirectory, "testDir");
      mUfs.mkdirs(testDir);
  
      UfsStatus status = mUfs.getStatus(testDir);
      if (!(status instanceof UfsDirectoryStatus)) {
        throw new IOException("Failed to get ufs directory status");
      }
    }
  
    /**
     * Test for getting existing directory status.
     */
    @RelatedS3Operations(operations = {"putObject", "getObjectMetadata"})
    public void createThenGetExistingDirectoryStatusTest() throws IOException {
      String testDir = PathUtils.concatPath(mTopLevelTestDirectory, "testDir");
      mUfs.mkdirs(testDir);
      UfsStatus status = mUfs.getExistingStatus(testDir);
      if (!(status instanceof UfsDirectoryStatus)) {
        throw new IOException("Failed to get ufs directory status");
      }
    }
  
    /**
     * Test for getting file size.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata"})
    public void getFileSizeTest() throws IOException {
      String testFileEmpty = PathUtils.concatPath(mTopLevelTestDirectory, "testFileEmpty");
      String testFileNonEmpty = PathUtils.concatPath(mTopLevelTestDirectory, "testFileNonEmpty");
      createEmptyFile(testFileEmpty);
      createTestBytesFile(testFileNonEmpty);
      if (mUfs.getFileStatus(testFileEmpty).getContentLength() != 0
          || mUfs.getFileStatus(testFileNonEmpty).getContentLength() != TEST_BYTES.length) {
        throw new IOException(FILE_STATUS_RESULT_INCORRECT);
      }
    }
  
    /**
     * Test for getting existing file status.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata"})
    public void createThenGetExistingFileStatusTest() throws IOException {
      String testFileNonEmpty = PathUtils.concatPath(mTopLevelTestDirectory, "testFileNonEmpty");
      String testFileLarge = PathUtils.concatPath(mTopLevelTestDirectory, "testFileLarge");
      createTestBytesFile(testFileNonEmpty);
      int numCopies = prepareMultiBlockFile(testFileLarge);
      if (TEST_BYTES.length != mUfs.getExistingFileStatus(testFileNonEmpty).getContentLength()
          || TEST_BYTES.length * numCopies
          != mUfs.getExistingFileStatus(testFileLarge).getContentLength()) {
        throw new IOException(FILE_STATUS_RESULT_INCORRECT);
      }
    }
  
    /**
     * Test for getting file status.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata"})
    public void getFileStatusTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "testFile");
      createEmptyFile(testFile);
  
      UfsStatus status = mUfs.getStatus(testFile);
      if (!(status instanceof UfsFileStatus)) {
        throw new IOException("Failed to get ufs file status");
      }
    }
  
    /**
     * Test for getting existing status.
     */
    @RelatedS3Operations(operations = {})
    public void createThenGetExistingStatusTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "testFile");
      createTestBytesFile(testFile);
  
      UfsStatus status = mUfs.getExistingStatus(testFile);
      if (!(status instanceof UfsFileStatus)) {
        throw new IOException("Failed to get ufs file status");
      }
    }
  
    /**
     * Test for getting modification time.
     */
    @RelatedS3Operations(operations = {"upload", "getObjectMetadata"})
    public void getModTimeTest() throws IOException {
      long slack = 5000; // Some file systems may report nearest second.
      long start = System.currentTimeMillis();
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "testFile");
      createTestBytesFile(testFile);
      long end = System.currentTimeMillis();
      long modTime = mUfs.getFileStatus(testFile).getLastModifiedTime();
      if (modTime < start - slack || modTime > end + slack) {
        throw new IOException(FILE_STATUS_RESULT_INCORRECT);
      }
    }
  
    /**
     * Test for getting status of non-existent directory.
     */
    @RelatedS3Operations(operations = {"getObjectMetadata"})
    public void getNonExistingDirectoryStatusTest() throws IOException {
      String testDir = PathUtils.concatPath(mTopLevelTestDirectory, "nonExistentDir");
      try {
        mUfs.getDirectoryStatus(testDir);
      } catch (FileNotFoundException e) {
        return;
      }
      throw new IOException(
          "Get status on a non-existent directory did not through " + FileNotFoundException.class);
    }
  
    /**
     * Test for getting status of non-existent file.
     */
    @RelatedS3Operations(operations = {"getObjectMetadata"})
    public void getNonExistingFileStatusTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "nonExistentFile");
      try {
        mUfs.getFileStatus(testFile);
      } catch (FileNotFoundException e) {
        return;
      }
      throw new IOException(
          "Get file status on a non-existent file did not through " + FileNotFoundException.class);
    }
  
    /**
     * Test for getting status of non-existent path.
     */
    @RelatedS3Operations(operations = {"getObjectMetadata"})
    public void getNonExistingPathStatusTest() throws IOException {
      String testPath = PathUtils.concatPath(mTopLevelTestDirectory, "nonExistentPath");
      try {
        mUfs.getStatus(testPath);
      } catch (FileNotFoundException e) {
        return;
      }
      throw new IOException(
          "Get status on a non-existent path did not through " + FileNotFoundException.class);
    }
  
    /**
     * Test for checking file is actual file.
     */
    @RelatedS3Operations(operations = {"putObject", "deleteObject"})
    public void isFileTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "testFile");
      String testDir = PathUtils.concatPath(mTopLevelTestDirectory, "testDir");
      if (mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
      createEmptyFile(testFile);
      mUfs.mkdirs(testDir, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      if (!mUfs.isFile(testFile)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
      if (mUfs.isFile(testDir)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
    }
  
    /**
     * Test for listing status of the root directory of the UFS.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void listStatusRootTest() throws IOException {
      UfsStatus[] rootList = mUfs.listStatus("/");
      if (rootList == null || rootList.length == 0) {
        throw new IOException("Unable to list UFS root path");
      }
    }
  
    /**
     * Test for listing status.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void listStatusTest() throws IOException {
      String testDirNonEmpty = PathUtils.concatPath(mTopLevelTestDirectory, "testDirNonEmpty1");
      String testDirNonEmptyChildDir = PathUtils.concatPath(testDirNonEmpty, "testDirNonEmpty2");
      String testDirNonEmptyChildFile = PathUtils.concatPath(testDirNonEmpty, "testDirNonEmptyF");
      String testDirNonEmptyChildDirFile =
          PathUtils.concatPath(testDirNonEmptyChildDir, "testDirNonEmptyChildDirF");
      mUfs.mkdirs(testDirNonEmpty, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      mUfs.mkdirs(testDirNonEmptyChildDir, MkdirsOptions.defaults(mConfiguration)
          .setCreateParent(false));
      createEmptyFile(testDirNonEmptyChildFile);
      createEmptyFile(testDirNonEmptyChildDirFile);
      String[] expectedResTopDir = new String[] {"testDirNonEmpty2", "testDirNonEmptyF"};
      // Some file systems may prefix with a slash
      String[] expectedResTopDir2 = new String[] {"/testDirNonEmpty2", "/testDirNonEmptyF"};
      Arrays.sort(expectedResTopDir);
      Arrays.sort(expectedResTopDir2);
      UfsStatus[] resTopDirStatus = mUfs.listStatus(testDirNonEmpty);
      String[] resTopDir = UfsStatus.convertToNames(resTopDirStatus);
      Arrays.sort(resTopDir);
      if (!Arrays.equals(expectedResTopDir, resTopDir)
          && !Arrays.equals(expectedResTopDir2, resTopDir)) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
      if (!mUfs.listStatus(testDirNonEmptyChildDir)[0].getName().equals("testDirNonEmptyChildDirF")
              || mUfs.listStatus(testDirNonEmptyChildDir)[0].getName()
              .equals("/testDirNonEmptyChildDirF")) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
      for (int i = 0; i < resTopDir.length; ++i) {
        if (mUfs.isDirectory(PathUtils.concatPath(testDirNonEmpty, resTopDirStatus[i].getName()))
            != resTopDirStatus[i].isDirectory()) {
          throw new IOException("UnderFileSystem.isDirectory() result is different from expected");
        }
      }
    }
  
    /**
     * Test for listing empty directory.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void listStatusEmptyTest() throws IOException {
      String testDir = PathUtils.concatPath(mTopLevelTestDirectory, "listStatusEmpty");
      mUfs.mkdirs(testDir);
      UfsStatus[] res = mUfs.listStatus(testDir);
      if (res.length != 0) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
    }
  
    /**
     * Test for listing status on file.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void listStatusFileTest() throws IOException {
      String testFile = PathUtils.concatPath(mTopLevelTestDirectory, "listStatusFile");
      createEmptyFile(testFile);
      if (mUfs.listStatus(testFile) != null) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
    }
  
    /**
     * Test for listing large directory.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void listLargeDirectoryTest() throws Exception {
      LargeDirectoryConfig config = prepareLargeDirectory();
      String[] children = config.getChildren();
  
      // Retry for some time to allow list operations eventual consistency for S3 and GCS.
      // See http://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html and
      // https://cloud.google.com/storage/docs/consistency for more details.
      AtomicReference<UfsStatus[]> results = new AtomicReference<>();
      CommonUtils.waitFor("list large directory", () -> {
        try {
          results.set(mUfs.listStatus(config.getTopLevelDirectory()));
          return children.length == results.get().length;
        } catch (Exception e) {
          return false;
        }
      }, WaitForOptions.defaults().setTimeoutMs(RETRY_TIMEOUT_MS).setInterval(RETRY_INTERVAL_MS));
  
      String[] resultNames = UfsStatus.convertToNames(results.get());
      Arrays.sort(resultNames);
      for (int i = 0; i < children.length; ++i) {
        if (!resultNames[i].equals(CommonUtils.stripPrefixIfPresent(children[i],
            PathUtils.normalizePath(config.getTopLevelDirectory(), "/")))) {
          throw new IOException(LIST_STATUS_RESULT_INCORRECT);
        }
      }
    }
  
    /**
     * Test for listing status recursively.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void listStatusRecursiveTest() throws IOException {
      String root = mTopLevelTestDirectory;
      // TODO(andrew): Should this directory be created in LocalAlluxioCluster creation code?
      mUfs.mkdirs(root);
      // Empty lsr should be empty
      if (mUfs.listStatus(root).length != 0) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
  
      // Create a tree of subdirectories and files
      String sub1 = PathUtils.concatPath(root, "sub1");
      String sub2 = PathUtils.concatPath(root, "sub2");
      String sub11 = PathUtils.concatPath(sub1, "sub11");
      String file11 = PathUtils.concatPath(sub11, "file11");
      String file2 = PathUtils.concatPath(sub2, "file2");
      String file = PathUtils.concatPath(root, "file");
      // lsr of nonexistent path should be null
      if (mUfs.listStatus(sub1, ListOptions.defaults().setRecursive(true)) != null) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
  
      mUfs.mkdirs(sub1, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      mUfs.mkdirs(sub2, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      mUfs.mkdirs(sub11, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      createEmptyFile(file11);
      createEmptyFile(file2);
      createEmptyFile(file);
  
      // lsr from root should return paths relative to the root
      String[] expectedResRoot =
          {"sub1", "sub2", "sub1/sub11", "sub1/sub11/file11", "sub2/file2", "file"};
      UfsStatus[] actualResRootStatus =
          mUfs.listStatus(root, ListOptions.defaults().setRecursive(true));
      String[] actualResRoot = UfsStatus.convertToNames(actualResRootStatus);
      Arrays.sort(expectedResRoot);
      Arrays.sort(actualResRoot);
      if (!Arrays.equals(expectedResRoot, actualResRoot)) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
      for (int i = 0; i < actualResRoot.length; ++i) {
        if (mUfs.isDirectory(PathUtils.concatPath(root, actualResRootStatus[i].getName()))
            != actualResRootStatus[i].isDirectory()) {
          throw new IOException("UnderFileSystem.isDirectory() result is different from expected");
        }
      }
      // lsr from sub1 should return paths relative to sub1
      String[] expectedResSub1 = {"sub11", "sub11/file11"};
      Arrays.sort(expectedResSub1);
      // Case A: the path sub1 does not end with a trailing /
      String[] actualResSub1a =
          UfsStatus.convertToNames(mUfs.listStatus(sub1, ListOptions.defaults().setRecursive(true)));
      Arrays.sort(actualResSub1a);
      if (!Arrays.equals(expectedResSub1, actualResSub1a)
          || (mUfs.listStatus(file, ListOptions.defaults().setRecursive(true)) != null)) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
      // Case B: the path sub1 ends with a trailing /
      String[] actualResSub1b = UfsStatus.convertToNames(
          mUfs.listStatus(sub1 + "/", ListOptions.defaults().setRecursive(true)));
      Arrays.sort(actualResSub1b);
      if (!Arrays.equals(expectedResSub1, actualResSub1b)
          || (mUfs.listStatus(file, ListOptions.defaults().setRecursive(true)) != null)) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
    }
  
    /**
     * Test for creating directory.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void mkdirsTest() throws IOException {
      // make sure the underfs address dir exists already
      mUfs.mkdirs(mTopLevelTestDirectory);
      // empty lsr should be empty
      if (mUfs.listStatus(mTopLevelTestDirectory).length != 0) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
  
      String testDirTop = PathUtils.concatPath(mTopLevelTestDirectory, "testDirTop");
      String testDir1 = PathUtils.concatPath(mTopLevelTestDirectory, "1");
      String testDir2 = PathUtils.concatPath(testDir1, "2");
      String testDir3 = PathUtils.concatPath(testDir2, "3");
      String testDirDeep = PathUtils.concatPath(testDir3, "testDirDeep");
      mUfs.mkdirs(testDirTop, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      mUfs.mkdirs(testDirDeep, MkdirsOptions.defaults(mConfiguration).setCreateParent(true));
      if (!(mUfs.isDirectory(testDirTop) && mUfs.isDirectory(testDir1) && mUfs.isDirectory(testDir2)
          && mUfs.isDirectory(testDir3) && mUfs.isDirectory(testDirDeep))) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for checking directory in object storage.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void objectCommonPrefixesIsDirectoryTest() throws IOException {
      // Only run test for an object store
      if (!mUfs.isObjectStorage()) {
        return;
      }
  
      ObjectStorePreConfig config = prepareObjectStore();
  
      String baseDirectoryPath = config.getBaseDirectoryPath();
      if (!mUfs.isDirectory(baseDirectoryPath)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_SUCCEED);
      }
  
      for (String subDirName : config.getSubDirectoryNames()) {
        String subDirPath = PathUtils.concatPath(baseDirectoryPath, subDirName);
        if (!mUfs.isDirectory(subDirPath)) {
          throw new IOException(IS_DIRECTORY_CHECK_SHOULD_SUCCEED);
        }
      }
    }
  
    /**
     * Test for listing status non recursively in object storage.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void objectCommonPrefixesListStatusNonRecursiveTest() throws IOException {
      // Only run test for an object store
      if (!mUfs.isObjectStorage()) {
        return;
      }
  
      ObjectStorePreConfig config = prepareObjectStore();
  
      String baseDirectoryPath = config.getBaseDirectoryPath();
      UfsStatus[] results = mUfs.listStatus(baseDirectoryPath);
      if (config.getSubDirectoryNames().length + config.getFileNames().length != results.length) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
      // Check for direct children files
      for (String fileName : config.getFileNames()) {
        int foundIndex = -1;
        for (int i = 0; i < results.length; ++i) {
          if (results[i].getName().equals(fileName)) {
            foundIndex = i;
          }
        }
        if (foundIndex < 0 || !results[foundIndex].isFile()) {
          throw new IOException(LIST_STATUS_RESULT_INCORRECT);
        }
      }
      // Check if pseudo-directories were inferred
      for (String subDirName : config.getSubDirectoryNames()) {
        int foundIndex = -1;
        for (int i = 0; i < results.length; ++i) {
          if (results[i].getName().equals(subDirName)) {
            foundIndex = i;
          }
        }
        if (foundIndex < 0 || !results[foundIndex].isDirectory()) {
          throw new IOException(LIST_STATUS_RESULT_INCORRECT);
        }
      }
    }
  
    /**
     * Test for listing status recursively in object storage.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void objectCommonPrefixesListStatusRecursiveTest() throws IOException {
      // Only run test for an object store
      if (!mUfs.isObjectStorage()) {
        return;
      }
  
      ObjectStorePreConfig config = prepareObjectStore();
  
      String baseDirectoryPath = config.getBaseDirectoryPath();
      UfsStatus[] results =
          mUfs.listStatus(baseDirectoryPath, ListOptions.defaults().setRecursive(true));
      String[] fileNames = config.getFileNames();
      String[] subDirNames = config.getSubDirectoryNames();
      if (subDirNames.length + fileNames.length + subDirNames.length * fileNames.length
          != results.length) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
      // Check for direct children files
      for (String fileName : fileNames) {
        int foundIndex = -1;
        for (int i = 0; i < results.length; ++i) {
          if (results[i].getName().equals(fileName)) {
            foundIndex = i;
          }
        }
        if (foundIndex < 0 || !results[foundIndex].isFile()) {
          throw new IOException(LIST_STATUS_RESULT_INCORRECT);
        }
      }
      for (String subDirName : subDirNames) {
        // Check if pseudo-directories were inferred
        int dirIndex = -1;
        for (int i = 0; i < results.length; ++i) {
          if (results[i].getName().equals(subDirName)) {
            dirIndex = i;
          }
        }
        if (dirIndex < 0 || !results[dirIndex].isDirectory()) {
          throw new IOException(LIST_STATUS_RESULT_INCORRECT);
        }
        // Check for indirect children
        for (String fileName : config.getFileNames()) {
          int fileIndex = -1;
          for (int i = 0; i < results.length; ++i) {
            if (results[i].getName().equals(String.format("%s/%s", subDirName, fileName))) {
              fileIndex = i;
            }
          }
          if (fileIndex < 0 || !results[fileIndex].isFile()) {
            throw new IOException(LIST_STATUS_RESULT_INCORRECT);
          }
        }
      }
    }
  
    /**
     * Test for listing status recursively in nested directory in object storage.
     */
    @RelatedS3Operations(operations = {"putObject", "listObjectsV2", "getObjectMetadata"})
    public void objectNestedDirsListStatusRecursiveTest() throws IOException {
      // Only run test for an object store
      if (!mUfs.isObjectStorage()) {
        return;
      }
  
      String root = mTopLevelTestDirectory;
      int nesting = 5;
  
      String path = root;
      for (int i = 0; i < nesting; i++) {
        path = PathUtils.concatPath(path, "dir" + i);
      }
      String file1 = PathUtils.concatPath(path, "file.txt");
  
      // Empty lsr should be empty
      if (mUfs.listStatus(root, ListOptions.defaults().setRecursive(true)).length != 0) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
  
      String fileKey = file1.substring(PathUtils.normalizePath(mUfsPath, "/").length());
      if (!mUfs.mkdirs(fileKey)) {
        throw new IOException("Failed to create empty object");
      }
  
      path = "";
      ArrayList<String> paths = new ArrayList<>();
      for (int i = 0; i < nesting; i++) {
        if (i == 0) {
          path = "dir" + i;
        } else {
          path = PathUtils.concatPath(path, "dir" + i);
        }
        paths.add(path);
      }
      path = PathUtils.concatPath(path, "file.txt");
      paths.add(path);
  
      String[] expectedStatus = paths.toArray(new String[0]);
      String[] actualStatus =
          UfsStatus.convertToNames(mUfs.listStatus(root, ListOptions.defaults().setRecursive(true)));
      Arrays.sort(expectedStatus);
      Arrays.sort(actualStatus);
      if (expectedStatus.length != actualStatus.length
          || !Arrays.equals(expectedStatus, actualStatus)) {
        throw new IOException(LIST_STATUS_RESULT_INCORRECT);
      }
    }
  
    /**
     * Test for renaming file.
     */
    @RelatedS3Operations(operations = {"upload", "copyObject", "deleteObject", "getObjectMetadata"})
    public void renameFileTest() throws IOException {
      String testFileSrc = PathUtils.concatPath(mTopLevelTestDirectory, "renameFileSrc");
      String testFileDst = PathUtils.concatPath(mTopLevelTestDirectory, "renameFileDst");
      createEmptyFile(testFileSrc);
      mUfs.renameFile(testFileSrc, testFileDst);
      if (mUfs.isFile(testFileSrc)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
      if (!mUfs.isFile(testFileDst)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for renaming renamable file.
     */
    @RelatedS3Operations(operations = {"upload", "copyObject", "deleteObject", "getObjectMetadata"})
    public void renameRenamableFileTest() throws IOException {
      String testFileSrc = PathUtils.concatPath(mTopLevelTestDirectory, "renameFileSrc");
      String testFileDst = PathUtils.concatPath(mTopLevelTestDirectory, "renameFileDst");
      prepareMultiBlockFile(testFileSrc);
      mUfs.renameRenamableFile(testFileSrc, testFileDst);
      if (mUfs.isFile(testFileSrc)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
      if (!mUfs.isFile(testFileDst)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for renaming directory.
     */
    @RelatedS3Operations(operations = {"putObject", "upload", "copyObject",
        "listObjectsV2", "getObjectMetadata"})
    public void renameDirectoryTest() throws IOException {
      String testDirSrc = PathUtils.concatPath(mTopLevelTestDirectory, "renameDirectorySrc");
      String testDirSrcChild = PathUtils.concatPath(testDirSrc, "testFile");
      String testDirDst = PathUtils.concatPath(mTopLevelTestDirectory, "renameDirectoryDst");
      String testDirDstChild = PathUtils.concatPath(testDirDst, "testFile");
      mUfs.mkdirs(testDirSrc, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      createEmptyFile(testDirSrcChild);
      mUfs.renameDirectory(testDirSrc, testDirDst);
      if (mUfs.isDirectory(testDirSrc)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_FAILED);
      }
      if (mUfs.isFile(testDirSrcChild)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
      if (!mUfs.isDirectory(testDirDst)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_SUCCEED);
      }
      if (!mUfs.isFile(testDirDstChild)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for renaming deep directory.
     */
    @RelatedS3Operations(operations = {"putObject", "upload", "copyObject",
        "listObjectsV2", "getObjectMetadata"})
    public void renameDirectoryDeepTest() throws IOException {
      String testDirSrc = PathUtils.concatPath(mTopLevelTestDirectory, "renameDirectoryDeepSrc");
      String testDirSrcChild = PathUtils.concatPath(testDirSrc, "testFile");
      String testDirSrcNested = PathUtils.concatPath(testDirSrc, "testNested");
      String testDirSrcNestedChild = PathUtils.concatPath(testDirSrcNested, "testNestedFile");
  
      String testDirDst = PathUtils.concatPath(mTopLevelTestDirectory, "renameDirectoryDeepDst");
      String testDirDstChild = PathUtils.concatPath(testDirDst, "testFile");
      String testDirDstNested = PathUtils.concatPath(testDirDst, "testNested");
      String testDirDstNestedChild = PathUtils.concatPath(testDirDstNested, "testNestedFile");
  
      mUfs.mkdirs(testDirSrc, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      createEmptyFile(testDirSrcChild);
      mUfs.mkdirs(testDirSrcNested, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      createEmptyFile(testDirSrcNestedChild);
  
      mUfs.renameDirectory(testDirSrc, testDirDst);
  
      if (mUfs.isDirectory(testDirSrc) || mUfs.isDirectory(testDirSrcNested)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_FAILED);
      }
      if (mUfs.isFile(testDirSrcChild) || mUfs.isFile(testDirSrcNestedChild)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
  
      if (!mUfs.isDirectory(testDirDst) || !mUfs.isDirectory(testDirDstNested)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_SUCCEED);
      }
      if (!mUfs.isFile(testDirDstChild) || !mUfs.isFile(testDirDstNestedChild)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for renaming renamable directory.
     */
    @RelatedS3Operations(operations = {"putObject", "upload", "copyObject",
        "listObjectsV2", "getObjectMetadata"})
    public void renameRenamableDirectoryTest() throws IOException {
      String testDirSrc = PathUtils.concatPath(mTopLevelTestDirectory, "renameRenamableDirectorySrc");
      String testDirSrcChild = PathUtils.concatPath(testDirSrc, "testFile");
      String testDirSrcNested = PathUtils.concatPath(testDirSrc, "testNested");
      String testDirSrcNestedChild = PathUtils.concatPath(testDirSrcNested, "testNestedFile");
  
      String testDirDst = PathUtils.concatPath(mTopLevelTestDirectory, "renameRenamableDirectoryDst");
      String testDirDstChild = PathUtils.concatPath(testDirDst, "testFile");
      String testDirDstNested = PathUtils.concatPath(testDirDst, "testNested");
      String testDirDstNestedChild = PathUtils.concatPath(testDirDstNested, "testNestedFile");
  
      mUfs.mkdirs(testDirSrc, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      prepareMultiBlockFile(testDirSrcChild);
      mUfs.mkdirs(testDirSrcNested, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
      prepareMultiBlockFile(testDirSrcNestedChild);
  
      mUfs.renameRenamableDirectory(testDirSrc, testDirDst);
  
      if (mUfs.isDirectory(testDirSrc) || mUfs.isDirectory(testDirSrcNested)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_FAILED);
      }
      if (mUfs.isFile(testDirSrcChild) || mUfs.isFile(testDirSrcNestedChild)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_FAILED);
      }
  
      if (!mUfs.isDirectory(testDirDst) || !mUfs.isDirectory(testDirDstNested)) {
        throw new IOException(IS_DIRECTORY_CHECK_SHOULD_SUCCEED);
      }
      if (!mUfs.isFile(testDirDstChild) || !mUfs.isFile(testDirDstNestedChild)) {
        throw new IOException(IS_FAIL_CHECK_SHOULD_SUCCEED);
      }
    }
  
    /**
     * Test for renaming large directory.
     */
    @RelatedS3Operations(operations = {"putObject", "upload", "copyObject",
        "listObjectsV2", "getObjectMetadata"})
    public void renameLargeDirectoryTest() throws Exception {
      LargeDirectoryConfig config = prepareLargeDirectory();
      String dstTopLevelDirectory = PathUtils.concatPath(mTopLevelTestDirectory, "topLevelDirMoved");
      mUfs.renameDirectory(config.getTopLevelDirectory(), dstTopLevelDirectory);
      // 1. Check the src directory no longer exists
      String[] srcChildren = config.getChildren();
      for (String src : srcChildren) {
        // Retry for some time to allow list operations eventual consistency for S3 and GCS.
        // See http://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html and
        // https://cloud.google.com/storage/docs/consistency for more details.
        CommonUtils.waitFor("list after delete consistency", () -> {
          try {
            return !mUfs.exists(src);
          } catch (IOException e) {
            return false;
          }
        }, WaitForOptions.defaults().setTimeoutMs(RETRY_TIMEOUT_MS).setInterval(RETRY_INTERVAL_MS));
      }
      // 2. Check the dst directory exists
      CommonUtils.waitFor("list after create consistency", () -> {
        UfsStatus[] results;
        try {
          results = mUfs.listStatus(dstTopLevelDirectory);
          if (srcChildren.length != results.length) {
            return false;
          }
          // Check nested files and directories in dst exist
          String[] resultNames = UfsStatus.convertToNames(results);
          Arrays.sort(resultNames);
          for (int i = 0; i < srcChildren.length; ++i) {
            if (!resultNames[i].equals(CommonUtils.stripPrefixIfPresent(srcChildren[i],
                PathUtils.normalizePath(config.getTopLevelDirectory(), "/")))) {
              throw new IOException("Destination is different from source after rename directory");
            }
          }
        } catch (IOException e) {
          return false;
        }
        return true;
      }, WaitForOptions.defaults().setTimeoutMs(RETRY_TIMEOUT_MS).setInterval(RETRY_INTERVAL_MS));
    }
  
    private void createEmptyFile(String path) throws IOException {
      OutputStream o = mUfs.create(path);
      o.close();
      checkContentHash(path, o);
    }
  
    private void createTestBytesFile(String path) throws IOException {
      OutputStream o = mUfs.create(path);
      o.write(TEST_BYTES);
      o.close();
      checkContentHash(path, o);
    }
  
    // should be called after the stream is closed, to be sure the content hash computed
    // by the stream is equal to the content hash on the UFS
    private void checkContentHash(String path, OutputStream stream) throws IOException {
      if (stream instanceof ContentHashable) {
        if (((ContentHashable) stream).getContentHash().isPresent()) {
          String ufsHash = mUfs.getParsedFingerprint(path).getTag(Fingerprint.Tag.CONTENT_HASH);
          String streamHash = mUfs.getParsedFingerprint(path,
              ((ContentHashable) stream).getContentHash().get())
              .getTag(Fingerprint.Tag.CONTENT_HASH);
          if (!streamHash.equals(ufsHash)) {
            throw new IOException(FILE_CONTENT_HASH_DOES_NOT_MATCH_UFS);
          }
        }
      }
    }
  
    // Prepare directory tree for pagination tests
    private LargeDirectoryConfig prepareLargeDirectory() throws IOException {
      final String filePrefix = "a_";
      final String folderPrefix = "b_";
  
      String topLevelDirectory = PathUtils.concatPath(mTopLevelTestDirectory, "topLevelDir");
  
      final int numFiles = 100;
  
      String[] children = new String[numFiles + numFiles];
  
      // Make top level directory
      mUfs.mkdirs(topLevelDirectory, MkdirsOptions.defaults(mConfiguration).setCreateParent(false));
  
      // Make the children files
      for (int i = 0; i < numFiles; ++i) {
        children[i] = PathUtils.concatPath(topLevelDirectory, filePrefix + String.format("%04d", i));
        createEmptyFile(children[i]);
      }
      // Make the children folders
      for (int i = 0; i < numFiles; ++i) {
        children[numFiles + i] =
            PathUtils.concatPath(topLevelDirectory, folderPrefix + String.format("%04d", i));
        mUfs.mkdirs(children[numFiles + i], MkdirsOptions.defaults(mConfiguration)
            .setCreateParent(false));
      }
  
      return new LargeDirectoryConfig(topLevelDirectory, children);
    }
  
    /**
     * Test configuration for pagination tests.
     */
    private class LargeDirectoryConfig {
      private String mTopLevelDirectory;
      // Children for top level directory
      private String[] mChildren;
  
      /**
       * Constructs {@link LargeDirectoryConfig} for pagination tests.
       *
       * @param topLevelDirectory the top level directory of the directory tree for pagination tests
       * @param children the children files of the directory tree for pagination tests
       */
      LargeDirectoryConfig(String topLevelDirectory, String[] children) {
        mTopLevelDirectory = topLevelDirectory;
        mChildren = children;
      }
  
      /**
       * @return the top level directory of the directory tree for pagination tests
       */
      String getTopLevelDirectory() {
        return mTopLevelDirectory;
      }
  
      /**
       * @return the children files of the directory tree for pagination tests
       */
      String[] getChildren() {
        return mChildren;
      }
    }
  
    /**
     * Prepares a multi-block file by making copies of TEST_BYTES.
     *
     * @param testFile path of file to create
     * @return the number of copies of TEST_BYTES made
     */
    private int prepareMultiBlockFile(String testFile) throws IOException {
      OutputStream outputStream = mUfs.create(testFile);
      // Write multiple blocks of data
      int numBlocks = 3;
      // Test block size is small enough for 'int'
      int blockSize = (int) mConfiguration.getBytes(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT);
      int numCopies = numBlocks * blockSize / TEST_BYTES.length;
      for (int i = 0; i < numCopies; ++i) {
        outputStream.write(TEST_BYTES);
      }
      outputStream.close();
      checkContentHash(testFile, outputStream);
      return numCopies;
    }
  
    /**
     * Prepare an object store for testing by creating a set of files and directories directly (not
     * through Alluxio). No breadcrumbs are created for directories.
     *
     * @return configuration for the pre-populated objects
     */
    private ObjectStorePreConfig prepareObjectStore() throws IOException {
      // Base directory for list status
      String baseDirectoryName = "base";
      String baseDirectoryPath = PathUtils.concatPath(mTopLevelTestDirectory, baseDirectoryName);
      String baseDirectoryKey =
          baseDirectoryPath.substring(PathUtils.normalizePath(mUfsPath, "/").length());
      // Pseudo-directories to be inferred
      String[] subDirectories = {"a", "b", "c"};
      // Every directory (base and pseudo) has these files
      String[] childrenFiles = {"sample1.jpg", "sample2.jpg", "sample3.jpg"};
      // Populate children of base directory
      for (String child : childrenFiles) {
        String path = String.format("%s/%s", baseDirectoryKey, child);
        OutputStream stream = mUfs.create(path);
        stream.close();
        checkContentHash(path, stream);
      }
      // Populate children of sub-directories
      for (String subdir : subDirectories) {
        for (String child : childrenFiles) {
          String path = String.format("%s/%s/%s", baseDirectoryKey, subdir, child);
          OutputStream stream = mUfs.create(path);
          stream.close();
          checkContentHash(path, stream);
        }
      }
      return new ObjectStorePreConfig(baseDirectoryPath, childrenFiles, subDirectories);
    }
  
    /**
     * Test configuration for pre-populating an object store.
     */
    private class ObjectStorePreConfig {
      private String mBaseDirectoryPath;
      private String[] mSubDirectoryNames;
      private String[] mFileNames;
  
      /**
       * Constructs {@link ObjectStorePreConfig} for pre-population an object store.
       *
       * @param baseDirectoryKey the base directory key for pre-populating of an object store
       * @param childrenFiles the children files for pre-populating an object store
       * @param subDirectories the sub-directories for pre-populating an object store
       */
      ObjectStorePreConfig(String baseDirectoryKey, String[] childrenFiles, String[] subDirectories) {
        mBaseDirectoryPath = baseDirectoryKey;
        mFileNames = childrenFiles;
        mSubDirectoryNames = subDirectories;
      }
  
      /**
       * @return base directory path for pre-populating an object store
       */
      String getBaseDirectoryPath() {
        return mBaseDirectoryPath;
      }
  
      /**
       * @return filenames for pre-populating an object store
       */
      String[] getFileNames() {
        return mFileNames;
      }
  
      /**
       * @return names of sub-directories for pre-populating an object store
       */
      String[] getSubDirectoryNames() {
        return mSubDirectoryNames;
      }
    }
  }
  