/*
 * 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.master.file;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import alluxio.AlluxioURI;
import alluxio.AuthenticatedClientUserResource;
import alluxio.AuthenticatedUserRule;
import alluxio.Constants;
import alluxio.annotation.dora.DoraTestTodoItem;
import alluxio.conf.Configuration;
import alluxio.exception.AccessControlException;
import alluxio.exception.BlockInfoException;
import alluxio.exception.DirectoryNotEmptyException;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.FileAlreadyExistsException;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.exception.status.FailedPreconditionException;
import alluxio.grpc.Command;
import alluxio.grpc.CommandType;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.GetStatusPOptions;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.LoadMetadataPType;
import alluxio.grpc.MountPOptions;
import alluxio.grpc.SetAclAction;
import alluxio.grpc.SetAclPOptions;
import alluxio.grpc.SetAttributePOptions;
import alluxio.master.file.contexts.CreateDirectoryContext;
import alluxio.master.file.contexts.CreateFileContext;
import alluxio.master.file.contexts.DeleteContext;
import alluxio.master.file.contexts.ExistsContext;
import alluxio.master.file.contexts.GetStatusContext;
import alluxio.master.file.contexts.ListStatusContext;
import alluxio.master.file.contexts.MountContext;
import alluxio.master.file.contexts.RenameContext;
import alluxio.master.file.contexts.SetAclContext;
import alluxio.master.file.contexts.SetAttributeContext;
import alluxio.master.metastore.InodeStore;
import alluxio.security.authorization.AclEntry;
import alluxio.security.authorization.Mode;
import alluxio.util.IdUtils;
import alluxio.util.io.FileUtils;
import alluxio.wire.FileInfo;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.io.Closeable;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@DoraTestTodoItem(action = DoraTestTodoItem.Action.REMOVE, owner = "jiacheng",
    comment = "remove FileSystemMaster unit testcases")
@Ignore
@RunWith(Parameterized.class)
public class FileSystemMasterFsOptsTest extends FileSystemMasterTestBase {

  public FileSystemMasterFsOptsTest(InodeStore.Factory factory) {
    mInodeStoreFactory = factory;
  }

  @Test
  public void createFileUsesOperationTime() throws Exception {
    AlluxioURI path = new AlluxioURI("/test");
    mFileSystemMaster.createFile(path, CreateFileContext.defaults().setOperationTimeMs(100));
    FileInfo info = mFileSystemMaster.getFileInfo(path, GetStatusContext.defaults());
    assertEquals(100, info.getLastModificationTimeMs());
    assertEquals(100, info.getLastAccessTimeMs());
  }

  @Test
  public void createFileWithOverwrite() throws Exception {
    AlluxioURI path = new AlluxioURI("/test");
    mFileSystemMaster.createFile(path, CreateFileContext.defaults());
    // create without overwrite
    Exception e = assertThrows(FileAlreadyExistsException.class, () -> {
      mFileSystemMaster.createFile(path, CreateFileContext.defaults());
    });
    assertTrue(e.getMessage()
        .contains(ExceptionMessage.CANNOT_OVERWRITE_FILE_WITHOUT_OVERWRITE.getMessage(path)));

    // create with overwrite
    CreateFileContext createFileContextWithOverwrite = CreateFileContext.defaults();
    createFileContextWithOverwrite.getOptions().setOverwrite(true);
    mFileSystemMaster.createFile(path, createFileContextWithOverwrite);
    FileInfo info = mFileSystemMaster.getFileInfo(path, GetStatusContext.defaults());

    // overwrite an existed directory
    AlluxioURI testpath = new AlluxioURI("/test2");
    mFileSystemMaster.createDirectory(testpath, CreateDirectoryContext.defaults());

    e = assertThrows(FileAlreadyExistsException.class, () -> {
      mFileSystemMaster.createFile(testpath, createFileContextWithOverwrite);
    });
    assertTrue(e.getMessage()
          .contains(ExceptionMessage.CANNOT_OVERWRITE_DIRECTORY.getMessage(testpath)));
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method.
   */
  @Test
  public void deleteFile() throws Exception {
    // cannot delete root
    try {
      mFileSystemMaster.delete(ROOT_URI,
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)
              .setDeleteMountPoint(true)));
      fail("Should not have been able to delete the root");
    } catch (InvalidPathException e) {
      assertEquals(ExceptionMessage.DELETE_ROOT_DIRECTORY.getMessage(), e.getMessage());
    }

    // delete the file
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    mFileSystemMaster.delete(NESTED_FILE_URI, DeleteContext.defaults());

    try {
      mBlockMaster.getBlockInfo(blockId);
      fail("Expected blockInfo to fail");
    } catch (BlockInfoException e) {
      // expected
    }

    // Update the heartbeat of removedBlockId received from worker 1.
    Command heartbeat1 = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat1);
    assertFalse(mBlockMaster.isBlockLost(blockId));

    // verify the file is deleted
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE_URI));

    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());
    // Create ufs file.
    Files.createDirectory(Paths.get(ufsMount.join("dir1").getPath()));
    Files.createFile(Paths.get(ufsMount.join("dir1").join("file1").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());

    AlluxioURI uri = new AlluxioURI("/mnt/local/dir1");
    mFileSystemMaster.listStatus(uri, ListStatusContext
        .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS)));
    mFileSystemMaster.delete(new AlluxioURI("/mnt/local/dir1/file1"),
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setAlluxioOnly(true)));

    // ufs file still exists
    assertTrue(Files.exists(Paths.get(ufsMount.join("dir1").join("file1").getPath())));
    // verify the file is deleted
    mThrown.expect(FileDoesNotExistException.class);
    mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local/dir1/file1"), GetStatusContext
        .mergeFrom(GetStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER)));
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method with a
   * non-empty directory.
   */
  @Test
  public void deleteNonemptyDirectory() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    String dirName = mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).getName();
    try {
      mFileSystemMaster.delete(NESTED_URI, DeleteContext.defaults());
      fail("Deleting a non-empty directory without setting recursive should fail");
    } catch (DirectoryNotEmptyException e) {
      String expectedMessage =
          ExceptionMessage.DELETE_NONEMPTY_DIRECTORY_NONRECURSIVE.getMessage(dirName);
      assertEquals(expectedMessage, e.getMessage());
    }

    // Now delete with recursive set to true.
    mFileSystemMaster.delete(NESTED_URI,
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a directory.
   */
  @Test
  public void deleteDir() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    // delete the dir
    mFileSystemMaster.delete(NESTED_URI,
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));

    // verify the dir is deleted
    assertEquals(-1, mFileSystemMaster.getFileId(NESTED_URI));

    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());
    // Create ufs file.
    Files.createDirectory(Paths.get(ufsMount.join("dir1").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());
    // load the dir1 to alluxio
    mFileSystemMaster.listStatus(new AlluxioURI("/mnt/local"), ListStatusContext
        .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS)));
    mFileSystemMaster.delete(new AlluxioURI("/mnt/local/dir1"), DeleteContext
        .mergeFrom(DeletePOptions.newBuilder().setRecursive(true).setAlluxioOnly(true)));
    // ufs directory still exists
    assertTrue(Files.exists(Paths.get(ufsMount.join("dir1").getPath())));
    // verify the directory is deleted
    Files.delete(Paths.get(ufsMount.join("dir1").getPath()));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/mnt/local/dir1")));
  }

  @Test
  public void deleteRecursiveClearsInnerInodesAndEdges() throws Exception {
    createFileWithSingleBlock(new AlluxioURI("/a/b/c/d/e"));
    createFileWithSingleBlock(new AlluxioURI("/a/b/x/y/z"));
    mFileSystemMaster.delete(new AlluxioURI("/a/b"),
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
    assertEquals(1, mInodeStore.allEdges().size());
    assertEquals(2, mInodeStore.allInodes().size());
  }

  @Test
  public void deleteDirRecursiveWithPermissions() throws Exception {
    // userA has permissions to delete directory and nested file
    createFileWithSingleBlock(NESTED_FILE_URI);
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mFileSystemMaster.delete(NESTED_URI,
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
    }
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_URI));
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE_URI));
  }

  @Test
  public void deleteDirRecursiveWithReadOnlyCheck() throws Exception {
    AlluxioURI rootPath = new AlluxioURI("/mnt/");
    mFileSystemMaster.createDirectory(rootPath, CreateDirectoryContext.defaults());
    // Create ufs file.
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    Files.createDirectory(Paths.get(ufsMount.join("dir1").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.mergeFrom(MountPOptions.newBuilder().setReadOnly(true)));
    mThrown.expect(AccessControlException.class);
    // Will throw AccessControlException because /mnt/local is a readonly mount point
    mFileSystemMaster.delete(rootPath,
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
  }

  @Test
  public void deleteDirRecursiveWithInsufficientPermissions() throws Exception {
    // userA has permissions to delete directory but not one of the nested files
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0700).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE2_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mFileSystemMaster.delete(NESTED_URI,
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
      fail("Deleting a directory w/ insufficient permission on child should fail");
    } catch (FailedPreconditionException e) {
      assertTrue(e.getMessage().contains("/nested/test/file (Permission denied"));
      assertTrue(e.getMessage().contains("/nested/test (Directory not empty)"));
    }
    // Then the nested file and the dir will be left
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE_URI));
    // File with permission will be deleted
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE2_URI));
  }

  @Test
  public void deleteDirRecursiveNoPermOnFile() throws Exception {
    // The structure looks like below
    // /nested
    // /nested/test/file
    // /nested/test/file2
    // /nested/test2/file
    // /nested/test2/file2
    // /nested/test2/dir
    // /nested/test2/dir/file
    // userA has no permission on /nested/test/file
    // So deleting the root will fail on:
    // /nested/, /nested/test/, /nested/test/file
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/file"));
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/file2"));
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/dir/file"));

    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0700).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE2_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/file2"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/dir/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));

    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mFileSystemMaster.delete(new AlluxioURI("/nested"),
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
      fail("Deleting a directory w/ insufficient permission on child should fail");
    } catch (FailedPreconditionException e) {
      assertTrue(e.getMessage().contains("/nested/test/file (Permission denied"));
      assertTrue(e.getMessage().contains("/nested/test (Directory not empty)"));
    }
    // The existing files/dirs will be: /, /nested/, /nested/test/, /nested/test/file
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(ROOT_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested")));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE_URI));
    // The other files should be deleted successfully
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE2_URI));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/file")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/file2")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/dir")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/dir/file")));
  }

  @Test
  public void deleteDirRecursiveNoPermOnFileDiffOrder() throws Exception {
    // The structure looks like below
    // /nested
    // /nested/test/file
    // /nested/test/file2
    // /nested/test2/file
    // /nested/test2/file2
    // /nested/test2/dir
    // /nested/test2/dir/file
    // userA has no permission on /nested/test/file2
    // So deleting the root will fail on:
    // /nested/, /nested/test/, /nested/test/file2
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/file"));
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/file2"));
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/dir/file"));

    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE2_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0700).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/file2"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/dir/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));

    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mFileSystemMaster.delete(new AlluxioURI("/nested"),
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
      fail("Deleting a directory w/ insufficient permission on child should fail");
    } catch (FailedPreconditionException e) {
      assertTrue(e.getMessage().contains("/nested/test/file2 (Permission denied"));
      assertTrue(e.getMessage().contains("/nested/test (Directory not empty)"));
    }
    // The existing files/dirs will be: /, /nested/, /nested/test/, /nested/test/file
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(ROOT_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested")));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE2_URI));
    // The other files should be deleted successfully
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE_URI));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/file")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/file2")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/dir")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/dir/file")));
  }

  @Test
  public void deleteNestedDirRecursiveNoPermOnFile() throws Exception {
    // The structure looks like below
    // /nested
    // /nested/nested
    // /nested/nested/test/file
    // /nested/nested/test/file2
    // userA has no permission on /nested/nested/test/file
    // So deleting the root will fail on:
    // /nested/, /nested/nested, /nested/nested/test/, /nested/nested/test/file
    createFileWithSingleBlock(new AlluxioURI("/nested/nested/test/file"));
    createFileWithSingleBlock(new AlluxioURI("/nested/nested/test/file2"));

    mFileSystemMaster.setAttribute(new AlluxioURI("/nested"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/nested"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/nested/test"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/nested/test/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0700).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/nested/test/file2"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));

    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mFileSystemMaster.delete(new AlluxioURI("/nested"),
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
      fail("Deleting a directory w/ insufficient permission on child should fail");
    } catch (FailedPreconditionException e) {
      assertTrue(e.getMessage().contains("/nested/nested/test/file (Permission denied"));
      assertTrue(e.getMessage().contains("/nested/nested/test (Directory not empty)"));
      assertTrue(e.getMessage().contains("/nested/nested (Directory not empty)"));
      assertTrue(e.getMessage().contains("/nested (Directory not empty)"));
    }
    // The existing files/dirs will be: /, /nested/, /nested/nested/,
    // /nested/nested/test/, /nested/test/file
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(ROOT_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested")));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/nested")));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/nested/test")));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/nested/test/file")));
    // The other files should be deleted successfully
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/nested/test/file2")));
  }

  @Test
  public void deleteDirRecursiveNoPermOnDir() throws Exception {
    // The structure looks like below
    // /nested/
    //
    // /nested/test/
    // /nested/test/file
    // /nested/test/file2
    // /nested/test/dir
    // /nested/test/dir/file
    //
    // /nested/test2/
    // /nested/test2/file
    // /nested/test2/file2
    // /nested/test2/dir
    // /nested/test2/dir/file
    // userA has no permission on /nested/test/file
    // So deleting the root will fail on:
    // /nested/, /nested/test/, /nested/test/file
    createFileWithSingleBlock(NESTED_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE2_URI);
    createFileWithSingleBlock(NESTED_DIR_FILE_URI);
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/file"));
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/file2"));
    createFileWithSingleBlock(new AlluxioURI("/nested/test2/dir/file"));

    // No permission on the dir, therefore all the files will be kept
    // although the user has permission over those nested files
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0700).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(NESTED_FILE2_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(NESTED_DIR_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test/dir/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    // The user has permission over everything under this dir, so everything will be removed
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/file2"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    mFileSystemMaster.setAttribute(new AlluxioURI("/nested/test2/dir/file"), SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));

    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mFileSystemMaster.delete(new AlluxioURI("/nested"),
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)));
      fail("Deleting a directory w/ insufficient permission on child should fail");
    } catch (FailedPreconditionException e) {
      assertTrue(e.getMessage().contains("/nested/test (Permission denied"));
      assertTrue(e.getMessage().contains("/nested (Directory not empty)"));
    }
    // The existing files/dirs will be: /, /nested/, /nested/test/
    // and everything under /nested/test/
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(ROOT_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested")));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test/file2")));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_DIR_URI));
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_DIR_FILE_URI));
    // The other files should be deleted successfully
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/file")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/file2")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/dir")));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(new AlluxioURI("/nested/test2/dir/file")));
  }

  /**
   * Tests the {@link FileSystemMaster#rename(AlluxioURI, AlluxioURI, RenameContext)} method.
   */
  @Test
  public void rename() throws Exception {
    mFileSystemMaster.createFile(NESTED_FILE_URI, mNestedFileContext);

    // try to rename a file to root
    try {
      mFileSystemMaster.rename(NESTED_FILE_URI, ROOT_URI, RenameContext.defaults());
      fail("Renaming to root should fail.");
    } catch (InvalidPathException e) {
      assertEquals(ExceptionMessage.RENAME_CANNOT_BE_TO_ROOT.getMessage(), e.getMessage());
    }

    // move root to another path
    try {
      mFileSystemMaster.rename(ROOT_URI, TEST_URI, RenameContext.defaults());
      fail("Should not be able to rename root");
    } catch (InvalidPathException e) {
      assertEquals(ExceptionMessage.ROOT_CANNOT_BE_RENAMED.getMessage(), e.getMessage());
    }

    // move to existing path
    try {
      mFileSystemMaster.rename(NESTED_FILE_URI, NESTED_URI, RenameContext.defaults());
      fail("Should not be able to overwrite existing file.");
    } catch (FileAlreadyExistsException e) {
      assertEquals(String
          .format("Cannot rename because destination already exists. src: %s dst: %s",
              NESTED_FILE_URI.getPath(), NESTED_URI.getPath()), e.getMessage());
    }

    // move a nested file to a root file
    mFileSystemMaster.rename(NESTED_FILE_URI, TEST_URI, RenameContext.defaults());
    assertEquals(mFileSystemMaster.getFileInfo(TEST_URI, GET_STATUS_CONTEXT).getPath(),
        TEST_URI.getPath());

    // move a file where the dst is lexicographically earlier than the source
    AlluxioURI newDst = new AlluxioURI("/abc_test");
    mFileSystemMaster.rename(TEST_URI, newDst, RenameContext.defaults());
    assertEquals(mFileSystemMaster.getFileInfo(newDst, GET_STATUS_CONTEXT).getPath(),
        newDst.getPath());
  }

  /**
   * Tests the {@link FileSystemMaster#getFileInfo(AlluxioURI, GetStatusContext)} method.
   */
  @Test
  public void getFileInfo() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    long fileId;
    FileInfo info;

    fileId = mFileSystemMaster.getFileId(ROOT_URI);
    info = mFileSystemMaster.getFileInfo(fileId);
    assertEquals(ROOT_URI.getPath(), info.getPath());
    assertEquals(ROOT_URI.getPath(),
        mFileSystemMaster.getFileInfo(ROOT_URI, GET_STATUS_CONTEXT).getPath());

    fileId = mFileSystemMaster.getFileId(NESTED_URI);
    info = mFileSystemMaster.getFileInfo(fileId);
    assertEquals(NESTED_URI.getPath(), info.getPath());
    assertEquals(NESTED_URI.getPath(),
        mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).getPath());

    fileId = mFileSystemMaster.getFileId(NESTED_FILE_URI);
    info = mFileSystemMaster.getFileInfo(fileId);
    assertEquals(NESTED_FILE_URI.getPath(), info.getPath());
    assertEquals(NESTED_FILE_URI.getPath(),
        mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).getPath());

    // Test non-existent id.
    try {
      mFileSystemMaster.getFileInfo(fileId + 1234);
      fail("getFileInfo() for a non-existent id should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }

    // Test non-existent URIs.
    try {
      mFileSystemMaster.getFileInfo(ROOT_FILE_URI, GET_STATUS_CONTEXT);
      fail("getFileInfo() for a non-existent URI should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
    try {
      mFileSystemMaster.getFileInfo(TEST_URI, GET_STATUS_CONTEXT);
      fail("getFileInfo() for a non-existent URI should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
    try {
      mFileSystemMaster.getFileInfo(NESTED_URI.join("DNE"), GET_STATUS_CONTEXT);
      fail("getFileInfo() for a non-existent URI should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
  }

  @Test
  public void getFileInfoWithLoadMetadata() throws Exception {
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());

    // Create ufs file.
    Files.createFile(Paths.get(ufsMount.join("file").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());

    // 3 directories exist.
    assertEquals(3, countPaths());

    // getFileInfo should load metadata automatically.
    AlluxioURI uri = new AlluxioURI("/mnt/local/file");
    assertEquals(uri.getPath(),
        mFileSystemMaster.getFileInfo(uri, GET_STATUS_CONTEXT).getPath());

    // getFileInfo should have loaded another file, so now 4 paths exist.
    assertEquals(4, countPaths());
  }

  @Test
  public void getFileIdWithLoadMetadata() throws Exception {
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());

    // Create ufs file.
    Files.createFile(Paths.get(ufsMount.join("file").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());

    // 3 directories exist.
    assertEquals(3, countPaths());

    // getFileId should load metadata automatically.
    AlluxioURI uri = new AlluxioURI("/mnt/local/file");
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(uri));

    // getFileId should have loaded another file, so now 4 paths exist.
    assertEquals(4, countPaths());
  }

  @Test
  public void listStatusWithLoadMetadataNever() throws Exception {
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());

    // Create ufs file.
    Files.createDirectory(Paths.get(ufsMount.join("dir1").getPath()));
    Files.createFile(Paths.get(ufsMount.join("dir1").join("file1").getPath()));
    Files.createFile(Paths.get(ufsMount.join("dir1").join("file2").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());

    // 3 directories exist.
    assertEquals(3, countPaths());

    // getFileId should load metadata automatically.
    AlluxioURI uri = new AlluxioURI("/mnt/local/dir1");
    try {
      mFileSystemMaster.listStatus(uri, ListStatusContext
          .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER)));
      fail("Exception expected");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }

    assertEquals(3, countPaths());
  }

  @Test
  public void listStatusWithLoadMetadataOnce() throws Exception {
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());

    // Create ufs file.
    Files.createDirectory(Paths.get(ufsMount.join("dir1").getPath()));
    Files.createFile(Paths.get(ufsMount.join("dir1").join("file1").getPath()));
    Files.createFile(Paths.get(ufsMount.join("dir1").join("file2").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());

    // 3 directories exist.
    assertEquals(3, countPaths());

    // getFileId should load metadata automatically.
    AlluxioURI uri = new AlluxioURI("/mnt/local/dir1");
    List<FileInfo> fileInfoList =
        mFileSystemMaster.listStatus(uri, ListStatusContext.defaults());
    Set<String> paths = new HashSet<>();
    for (FileInfo fileInfo : fileInfoList) {
      paths.add(fileInfo.getPath());
    }
    assertEquals(2, paths.size());
    assertTrue(paths.contains("/mnt/local/dir1/file1"));
    assertTrue(paths.contains("/mnt/local/dir1/file2"));
    // listStatus should have loaded another 3 files (dir1, dir1/file1, dir1/file2), so now 6
    // paths exist.
    assertEquals(6, countPaths());
  }

  @Test
  public void listStatusWithLoadMetadataAlways() throws Exception {
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());

    // Create ufs file.
    Files.createDirectory(Paths.get(ufsMount.join("dir1").getPath()));
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());

    // 3 directories exist.
    assertEquals(3, countPaths());

    // getFileId should load metadata automatically.
    AlluxioURI uri = new AlluxioURI("/mnt/local/dir1");
    List<FileInfo> fileInfoList =
        mFileSystemMaster.listStatus(uri, ListStatusContext.defaults());
    assertEquals(0, fileInfoList.size());
    // listStatus should have loaded another files (dir1), so now 4 paths exist.
    assertEquals(4, countPaths());

    // Add two files.
    Files.createFile(Paths.get(ufsMount.join("dir1").join("file1").getPath()));
    Files.createFile(Paths.get(ufsMount.join("dir1").join("file2").getPath()));

    fileInfoList =
        mFileSystemMaster.listStatus(uri, ListStatusContext.defaults());
    assertEquals(0, fileInfoList.size());
    // No file is loaded since dir1 has been loaded once.
    assertEquals(4, countPaths());

    fileInfoList = mFileSystemMaster.listStatus(uri, ListStatusContext
        .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS)));
    Set<String> paths = new HashSet<>();
    for (FileInfo fileInfo : fileInfoList) {
      paths.add(fileInfo.getPath());
    }
    assertEquals(2, paths.size());
    assertTrue(paths.contains("/mnt/local/dir1/file1"));
    assertTrue(paths.contains("/mnt/local/dir1/file2"));
    // listStatus should have loaded another 2 files (dir1/file1, dir1/file2), so now 6
    // paths exist.
    assertEquals(6, countPaths());
  }

  /**
   * Tests listing status on a non-persisted directory.
   * @deprecated
   */
  @Deprecated
  public void listStatusWithLoadMetadataNonPersistedDir() throws Exception {
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());

    // Create ufs file.
    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());

    // 3 directories exist.
    assertEquals(3, countPaths());

    // Create a drectory in alluxio which is not persisted.
    AlluxioURI folder = new AlluxioURI("/mnt/local/folder");
    mFileSystemMaster.createDirectory(folder, CreateDirectoryContext.defaults());

    assertFalse(
        mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local/folder"), GET_STATUS_CONTEXT)
            .isPersisted());

    // Create files in ufs.
    Files.createDirectory(Paths.get(ufsMount.join("folder").getPath()));
    Files.createFile(Paths.get(ufsMount.join("folder").join("file1").getPath()));
    Files.createFile(Paths.get(ufsMount.join("folder").join("file2").getPath()));

    // getStatus won't mark folder as persisted.
    assertFalse(
        mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local/folder"), GET_STATUS_CONTEXT)
            .isPersisted());

    List<FileInfo> fileInfoList =
        mFileSystemMaster.listStatus(folder, ListStatusContext.defaults());
    assertEquals(2, fileInfoList.size());
    // listStatus should have loaded files (folder, folder/file1, folder/file2), so now 6 paths
    // exist.
    assertEquals(6, countPaths());

    Set<String> paths = new HashSet<>();
    for (FileInfo f : fileInfoList) {
      paths.add(f.getPath());
    }
    assertEquals(2, paths.size());
    assertTrue(paths.contains("/mnt/local/folder/file1"));
    assertTrue(paths.contains("/mnt/local/folder/file2"));

    assertTrue(
        mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local/folder"), GET_STATUS_CONTEXT)
            .isPersisted());
  }

  @Test
  public void listStatus() throws Exception {
    final int files = 10;
    List<FileInfo> infos;
    List<String> filenames;

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }
    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext
        .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER)));
    assertEquals(files, infos.size());
    // Copy out filenames to use List contains.
    filenames = new ArrayList<>();
    for (FileInfo info : infos) {
      filenames.add(info.getPath());
    }
    // Compare all filenames.
    for (int i = 0; i < files; i++) {
      assertTrue(
          filenames.contains(ROOT_URI.join("file" + String.format("%05d", i)).toString()));
    }

    // Test single file.
    createFileWithSingleBlock(ROOT_FILE_URI);
    infos = mFileSystemMaster.listStatus(ROOT_FILE_URI, ListStatusContext
        .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER)));
    assertEquals(1, infos.size());
    assertEquals(ROOT_FILE_URI.getPath(), infos.get(0).getPath());

    // Test files in nested directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(NESTED_URI.join("file" + String.format("%05d", i)));
    }
    infos = mFileSystemMaster.listStatus(NESTED_URI, ListStatusContext
        .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER)));
    assertEquals(files, infos.size());
    // Copy out filenames to use List contains.
    filenames = new ArrayList<>();
    for (FileInfo info : infos) {
      filenames.add(info.getPath());
    }
    // Compare all filenames.
    for (int i = 0; i < files; i++) {
      assertTrue(
          filenames.contains(NESTED_URI.join("file" + String.format("%05d", i)).toString()));
    }

    // Test non-existent URIs.
    try {
      mFileSystemMaster.listStatus(NESTED_URI.join("DNE"), ListStatusContext
          .mergeFrom(ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER)));
      fail("listStatus() for a non-existent URI should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
  }

  @Test
  public void listStatusRecursive() throws Exception {
    final int files = 10;
    List<FileInfo> infos;
    List<String> filenames;

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }
    // Test files in nested directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(NESTED_URI.join("file" + String.format("%05d", i)));
    }

    // Test recursive listStatus
    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
        .newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS).setRecursive(true)));
    // 10 files in each directory, 2 levels of directories
    assertEquals(files + files + 2, infos.size());

    filenames = new ArrayList<>();
    for (FileInfo info : infos) {
      filenames.add(info.getPath());
    }
    for (int i = 0; i < files; i++) {
      assertTrue(
          filenames.contains(ROOT_URI.join("file" + String.format("%05d", i)).toString()));
    }
    for (int i = 0; i < files; i++) {
      assertTrue(
          filenames.contains(NESTED_URI.join("file" + String.format("%05d", i)).toString()));
    }
  }

  @Test
  public void listStatusRecursivePermissions() throws Exception {
    final int files = 10;
    List<FileInfo> infos;
    List<String> filenames;

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }
    // Test files in nested directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(NESTED_URI.join("file" + String.format("%05d", i)));
    }

    // Test with permissions
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext.mergeFrom(SetAttributePOptions
        .newBuilder().setMode(new Mode((short) 0400).toProto()).setRecursive(true)));
    try (Closeable r = new AuthenticatedUserRule("test_user1", Configuration.global())
        .toResource()) {
      // Test recursive listStatus
      infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
          .newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS).setRecursive(true)));

      // 10 files in the root directory, 2 level of directories
      assertEquals(files + 2, infos.size());
    }
  }

  @Test
  public void listStatusRecursiveLoadMetadata() throws Exception {
    final int files = 10;
    List<FileInfo> infos;

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }

    FileUtils.createFile(Paths.get(mUnderFS).resolve("ufsfile1").toString());
    // Test interaction between recursive and loadMetadata
    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
        .newBuilder().setLoadMetadataType(LoadMetadataPType.ONCE).setRecursive(false)));

    assertEquals(files + 1  , infos.size());

    FileUtils.createFile(Paths.get(mUnderFS).resolve("ufsfile2").toString());
    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
        .newBuilder().setLoadMetadataType(LoadMetadataPType.ONCE).setRecursive(false)));
    assertEquals(files + 1  , infos.size());

    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
        .newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS).setRecursive(false)));
    assertEquals(files + 2, infos.size());

    // Test files in nested directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(NESTED_URI.join("file" + String.format("%05d", i)));
    }

    FileUtils.createFile(Paths.get(mUnderFS).resolve("nested/test/ufsnestedfile1").toString());
    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
        .newBuilder().setLoadMetadataType(LoadMetadataPType.ONCE).setRecursive(true)));
    // 2 sets of files, 2 files inserted at root, 2 directories nested and test,
    // 1 file ufsnestedfile1
    assertEquals(files + files + 2 + 2 + 1, infos.size());

    FileUtils.createFile(Paths.get(mUnderFS).resolve("nested/test/ufsnestedfile2").toString());
    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
        .newBuilder().setLoadMetadataType(LoadMetadataPType.ONCE).setRecursive(true)));
    assertEquals(files + files + 2 + 2 + 1, infos.size());

    infos = mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
        .newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS).setRecursive(true)));
    assertEquals(files + files + 2 + 2 + 2, infos.size());
  }

  /**
   * Tests that an exception is thrown when trying to create a file in a non-existing directory
   * without setting the {@code recursive} flag.
   */
  @Test
  public void renameUnderNonexistingDir() throws Exception {
    mThrown.expect(FileDoesNotExistException.class);
    mThrown.expectMessage(ExceptionMessage.PATH_DOES_NOT_EXIST.getMessage("/nested/test"));
    CreateFileContext context = CreateFileContext
        .mergeFrom(CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB));
    mFileSystemMaster.createFile(TEST_URI, context);

    // nested dir
    mFileSystemMaster.rename(TEST_URI, NESTED_FILE_URI, RenameContext.defaults());
  }

  @Test
  public void renameToNonExistentParent() throws Exception {
    CreateFileContext context = CreateFileContext.mergeFrom(
        CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB).setRecursive(true));
    mFileSystemMaster.createFile(NESTED_URI, context);

    try {
      mFileSystemMaster.rename(NESTED_URI, new AlluxioURI("/testDNE/b"), RenameContext.defaults());
      fail("Rename to a non-existent parent path should not succeed.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
  }

  /**
   * Tests that an exception is thrown when trying to rename a file to a prefix of the original
   * file.
   */
  @Test
  public void renameToSubpath() throws Exception {
    mFileSystemMaster.createFile(NESTED_URI, mNestedFileContext);
    mThrown.expect(InvalidPathException.class);
    mThrown.expectMessage("Traversal failed for path /nested/test/file. "
        + "Component 2(test) is a file, not a directory");
    mFileSystemMaster.rename(NESTED_URI, NESTED_FILE_URI, RenameContext.defaults());
  }

  @Test
  public void setDefaultAclforFile() throws Exception {
    SetAclContext context = SetAclContext.defaults();
    createFileWithSingleBlock(NESTED_FILE_URI);

    Set<String> newEntries = Sets.newHashSet("default:user::rwx",
        "default:group::rwx", "default:other::r-x");

    mThrown.expect(UnsupportedOperationException.class);
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);
  }

  @Test
  public void setDefaultAcl() throws Exception {
    SetAclContext context = SetAclContext.defaults();
    createFileWithSingleBlock(NESTED_FILE_URI);
    Set<String> entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertEquals(0, entries.size());

    // replace
    Set<String> newEntries = Sets.newHashSet("default:user::rwx",
        "default:group::rwx", "default:other::r-x");
    mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.REPLACE,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertEquals(newEntries, entries);

    // replace
    newEntries = Sets.newHashSet("default:user::rw-", "default:group::r--", "default:other::r--");
    mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.REPLACE,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);
    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertEquals(newEntries, entries);

    // modify existing
    newEntries = Sets.newHashSet("default:user::rwx", "default:group::rw-", "default:other::r-x");
    mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertEquals(newEntries, entries);

    // modify add
    Set<String> oldEntries = new HashSet<>(entries);
    newEntries = Sets.newHashSet("default:user:usera:---", "default:group:groupa:--x");
    mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertTrue(entries.containsAll(oldEntries));
    assertTrue(entries.containsAll(newEntries));
    assertTrue(entries.contains("default:mask::rwx"));

    // modify existing and add
    newEntries = Sets.newHashSet("default:user:usera:---", "default:group:groupa:--x",
        "default:other::r-x");
    mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertTrue(entries.containsAll(newEntries));

    // remove default
    mFileSystemMaster
        .setAcl(NESTED_URI, SetAclAction.REMOVE_DEFAULT, Collections.emptyList(), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertEquals(0, entries.size());

    // remove
    newEntries =
        Sets.newHashSet("default:user:usera:---", "default:user:userb:rwx",
            "default:group:groupa:--x", "default:group:groupb:-wx");
    mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);
    oldEntries = new HashSet<>(entries);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    assertTrue(entries.containsAll(oldEntries));

    Set<String> deleteEntries = Sets.newHashSet("default:user:userb:rwx",
        "default:group:groupa:--x");
    mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.REMOVE,
        deleteEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).convertDefaultAclToStringEntries());
    Set<String> remainingEntries = new HashSet<>(newEntries);
    assertTrue(remainingEntries.removeAll(deleteEntries));
    assertTrue(entries.containsAll(remainingEntries));

    final Set<String> finalEntries = entries;
    assertTrue(deleteEntries.stream().noneMatch(finalEntries::contains));
  }

  @Test
  public void setAcl() throws Exception {
    SetAclContext context = SetAclContext.defaults();
    createFileWithSingleBlock(NESTED_FILE_URI);

    Set<String> entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertEquals(3, entries.size());

    // replace
    Set<String> newEntries = Sets.newHashSet("user::rwx", "group::rwx", "other::rwx");
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.REPLACE,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertEquals(newEntries, entries);

    // replace
    newEntries = Sets.newHashSet("user::rw-", "group::r--", "other::r--");
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.REPLACE,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertEquals(newEntries, entries);

    // modify existing
    newEntries = Sets.newHashSet("user::rwx", "group::r--", "other::r-x");
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertEquals(newEntries, entries);

    // modify add
    Set<String> oldEntries = new HashSet<>(entries);
    newEntries = Sets.newHashSet("user:usera:---", "group:groupa:--x");
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertTrue(entries.containsAll(oldEntries));
    assertTrue(entries.containsAll(newEntries));
    // check if the mask got updated correctly
    assertTrue(entries.contains("mask::r-x"));

    // modify existing and add
    newEntries = Sets.newHashSet("user:usera:---", "group:groupa:--x", "other::r-x");
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertTrue(entries.containsAll(newEntries));

    // remove all
    mFileSystemMaster
        .setAcl(NESTED_FILE_URI, SetAclAction.REMOVE_ALL, Collections.emptyList(), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertEquals(3, entries.size());

    // remove
    newEntries =
        Sets.newHashSet("user:usera:---", "user:userb:rwx", "group:groupa:--x", "group:groupb:-wx");
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.MODIFY,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);
    oldEntries = new HashSet<>(entries);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertTrue(entries.containsAll(oldEntries));

    Set<String> deleteEntries = Sets.newHashSet("user:userb:rwx", "group:groupa:--x");
    mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.REMOVE,
        deleteEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    Set<String> remainingEntries = new HashSet<>(newEntries);
    assertTrue(remainingEntries.removeAll(deleteEntries));
    assertTrue(entries.containsAll(remainingEntries));

    final Set<String> finalEntries = entries;
    assertTrue(deleteEntries.stream().noneMatch(finalEntries::contains));
  }

  @Test
  public void setRecursiveAcl() throws Exception {
    final int files = 10;
    SetAclContext context =
        SetAclContext.mergeFrom(SetAclPOptions.newBuilder().setRecursive(true));

    // Test files in root directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(ROOT_URI.join("file" + String.format("%05d", i)));
    }
    // Test files in nested directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(NESTED_URI.join("file" + String.format("%05d", i)));
    }

    // Test files in nested directory.
    for (int i = 0; i < files; i++) {
      createFileWithSingleBlock(NESTED_DIR_URI.join("file" + String.format("%05d", i)));
    }

    // replace
    Set<String> newEntries = Sets.newHashSet("user::rw-", "group::r-x", "other::-wx");
    mFileSystemMaster.setAcl(ROOT_URI, SetAclAction.REPLACE,
        newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()), context);

    List<FileInfo> infos =
        mFileSystemMaster.listStatus(ROOT_URI, ListStatusContext.mergeFrom(ListStatusPOptions
            .newBuilder().setLoadMetadataType(LoadMetadataPType.ONCE).setRecursive(true)));
    assertEquals(files * 3 + 3, infos.size());
    for (FileInfo info : infos) {
      assertEquals(newEntries, Sets.newHashSet(info.convertAclToStringEntries()));
    }
  }

  @Test
  public void exists() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI, mNestedFileContext);

    //Test existing file
    assertEquals(true, mFileSystemMaster.exists(NESTED_FILE_URI, ExistsContext.defaults()));

    //Test non-existent file
    assertEquals(false, mFileSystemMaster.exists(NESTED_FILE2_URI, ExistsContext.defaults()));

    //Test file without parent permission
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0700).toProto())));
    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mThrown.expect(AccessControlException.class);
      mFileSystemMaster.exists(NESTED_FILE_URI, ExistsContext.defaults());
    }

    //Test non-existent file without parent permission
    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      mThrown.expect(AccessControlException.class);
      mFileSystemMaster.exists(NESTED_FILE2_URI, ExistsContext.defaults());
    }
  }
}
