/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.fs.viewfs;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.AbstractFileSystem;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsConstants;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.MiniDFSNNTopology;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

/**
 * Test for viewfs with LinkFallback mount table entries.
 */
public class TestViewFsLinkFallback {
  private static FileSystem fsDefault;
  private FileSystem fsTarget;
  private static MiniDFSCluster cluster;
  private static URI viewFsDefaultClusterUri;
  private Path targetTestRoot;

  @BeforeClass
  public static void clusterSetupAtBeginning()
      throws IOException, URISyntaxException {
    int nameSpacesCount = 3;
    int dataNodesCount = 3;
    int fsIndexDefault = 0;
    Configuration conf = new Configuration();
    FileSystem[] fsHdfs = new FileSystem[nameSpacesCount];
    conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY,
        true);
    cluster = new MiniDFSCluster.Builder(conf)
        .nnTopology(MiniDFSNNTopology.simpleFederatedTopology(
            nameSpacesCount))
        .numDataNodes(dataNodesCount)
        .build();
    cluster.waitClusterUp();

    for (int i = 0; i < nameSpacesCount; i++) {
      fsHdfs[i] = cluster.getFileSystem(i);
    }
    fsDefault = fsHdfs[fsIndexDefault];
    viewFsDefaultClusterUri = new URI(FsConstants.VIEWFS_SCHEME,
        Constants.CONFIG_VIEWFS_DEFAULT_MOUNT_TABLE, "/", null, null);

  }

  @AfterClass
  public static void clusterShutdownAtEnd() throws Exception {
    if (cluster != null) {
      cluster.shutdown();
    }
  }

  @Before
  public void setUp() throws Exception {
    fsTarget = fsDefault;
    initializeTargetTestRoot();
  }

  private void initializeTargetTestRoot() throws IOException {
    targetTestRoot = fsDefault.makeQualified(new Path("/"));
    for (FileStatus status : fsDefault.listStatus(targetTestRoot)) {
      fsDefault.delete(status.getPath(), true);
    }
  }

  /**
   * Tests that directory making should be successful when the parent directory
   * is same as the existent fallback directory. The new dir should be created
   * in fallback instead failing.
   */
  @Test
  public void testMkdirOfLinkParentWithFallbackLinkWithSameMountDirectoryTree()
      throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS, false);
    ConfigUtil.addLink(conf, "/user1/hive/warehouse/partition-0",
        new Path(targetTestRoot.toString()).toUri());
    Path dir1 = new Path(targetTestRoot,
        "fallbackDir/user1/hive/warehouse/partition-0");
    fsTarget.mkdirs(dir1);
    Path fallbackTarget = new Path(targetTestRoot, "fallbackDir");
    ConfigUtil.addLinkFallback(conf, fallbackTarget.toUri());
    AbstractFileSystem vfs =
        AbstractFileSystem.get(viewFsDefaultClusterUri, conf);
    Path p = new Path("/user1/hive/warehouse/test");
    Path test = Path.mergePaths(fallbackTarget, p);
    assertFalse(fsTarget.exists(test));
    vfs.mkdir(p, null, true);
    assertTrue(fsTarget.exists(test));
  }

  /**
   * Tests that directory making should be successful when attempting to create
   * the root directory as it's already exist.
   */
  @Test
  public void testMkdirOfRootWithFallbackLinkAndMountWithSameDirTree()
      throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS, false);
    ConfigUtil
        .addLink(conf, "/user1", new Path(targetTestRoot.toString()).toUri());
    Path dir1 = new Path(targetTestRoot, "fallbackDir/user1");
    fsTarget.mkdirs(dir1);
    Path fallbackTarget = new Path(targetTestRoot, "fallbackDir");
    ConfigUtil.addLinkFallback(conf, fallbackTarget.toUri());
    AbstractFileSystem vfs =
        AbstractFileSystem.get(viewFsDefaultClusterUri, conf);
    Path p = new Path("/");
    Path test = Path.mergePaths(fallbackTarget, p);
    assertTrue(fsTarget.exists(test));
    vfs.mkdir(p, null, true);
    assertTrue(fsTarget.exists(test));
  }

  /**
   * Tests the making of a new directory which is not matching to any of
   * internal directory under the root.
   */
  @Test
  public void testMkdirOfNewDirWithOutMatchingToMountOrFallbackDirTree()
      throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS, false);
    ConfigUtil.addLink(conf, "/user1/hive/warehouse/partition-0",
        new Path(targetTestRoot.toString()).toUri());
    Path fallbackTarget = new Path(targetTestRoot, "fallbackDir");
    fsTarget.mkdirs(fallbackTarget);
    ConfigUtil.addLinkFallback(conf, fallbackTarget.toUri());
    AbstractFileSystem vfs =
        AbstractFileSystem.get(viewFsDefaultClusterUri, conf);
    // user2 does not exist in fallback
    Path p = new Path("/user2");
    Path test = Path.mergePaths(fallbackTarget, p);
    assertFalse(fsTarget.exists(test));
    vfs.mkdir(p, null, true);
    assertTrue(fsTarget.exists(test));
  }

  /**
   * Tests that when the parent dirs does not exist in fallback but the parent
   * dir is same as mount internal directory, then we create parent structure
   * (mount internal directory tree structure) in fallback.
   */
  @Test
  public void testMkdirWithFallbackLinkWithMountPathMatchingDirExist()
      throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS, false);
    ConfigUtil.addLink(conf, "/user1/hive",
        new Path(targetTestRoot.toString()).toUri());
    Path fallbackTarget = new Path(targetTestRoot, "fallbackDir");
    fsTarget.mkdirs(fallbackTarget);
    ConfigUtil.addLinkFallback(conf, fallbackTarget.toUri());

    AbstractFileSystem vfs =
        AbstractFileSystem.get(viewFsDefaultClusterUri, conf);
    //user1 does not exist in fallback
    Path immediateLevelToInternalDir = new Path("/user1/test");
    Path test = Path.mergePaths(fallbackTarget, immediateLevelToInternalDir);
    assertFalse(fsTarget.exists(test));
    vfs.mkdir(immediateLevelToInternalDir, null, true);
    assertTrue(fsTarget.exists(test));
  }

  /**
   * Tests that when the parent dirs does not exist in fallback but the
   * immediate parent dir is not same as mount internal directory, then we
   * create parent structure (mount internal directory tree structure) in
   * fallback.
   */
  @Test
  public void testMkdirOfDeepTreeWithFallbackLinkAndMountPathMatchingDirExist()
      throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS, false);
    ConfigUtil.addLink(conf, "/user1/hive",
        new Path(targetTestRoot.toString()).toUri());
    Path fallbackTarget = new Path(targetTestRoot, "fallbackDir");
    fsTarget.mkdirs(fallbackTarget);
    ConfigUtil.addLinkFallback(conf, fallbackTarget.toUri());

    AbstractFileSystem vfs =
        AbstractFileSystem.get(viewFsDefaultClusterUri, conf);
    //user1 does not exist in fallback
    Path multipleLevelToInternalDir = new Path("/user1/test/test");
    Path test = Path.mergePaths(fallbackTarget, multipleLevelToInternalDir);
    assertFalse(fsTarget.exists(test));
    vfs.mkdir(multipleLevelToInternalDir, null, true);
    assertTrue(fsTarget.exists(test));
  }

  /**
   * Tests that mkdir with createParent false should still create parent in
   * fallback when the same mount dir exist.
   */
  @Test
  public void testMkdirShouldCreateParentDirInFallbackWhenMountDirExist()
      throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS, false);
    ConfigUtil.addLink(conf, "/user1/hive/test",
        new Path(targetTestRoot.toString()).toUri());
    Path fallbackTarget = new Path(targetTestRoot, "fallbackDir");
    fsTarget.mkdirs(fallbackTarget);
    ConfigUtil.addLinkFallback(conf, fallbackTarget.toUri());
    AbstractFileSystem vfs = AbstractFileSystem.get(viewFsDefaultClusterUri,
        conf);
    //user1/hive/test1 does not exist in fallback
    Path multipleLevelToInternalDir = new Path("/user1/hive/test1");
    Path test = Path.mergePaths(fallbackTarget, multipleLevelToInternalDir);
    assertFalse(fsTarget.exists(test));
    // user1/hive exist in viewFS.
    assertNotNull(vfs.getFileStatus(new Path("/user1/hive")));
    // user1/hive does not exists in fallback.
    assertFalse(fsTarget.exists(test.getParent()));

    vfs.mkdir(multipleLevelToInternalDir, FsPermission.getDirDefault(), false);
    assertTrue(fsTarget.exists(test));

  }

  /**
   * Tests that mkdir should fail with IOE when there is a problem with
   * fallbackfs.
   */
  @Test
  public void testMkdirShouldFailWhenFallbackFSNotAvailable()
      throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS, false);
    ConfigUtil.addLink(conf, "/user1/test",
        new Path(targetTestRoot.toString()).toUri());
    Path fallbackTarget = new Path(targetTestRoot, "fallbackDir");
    fsTarget.mkdirs(fallbackTarget);
    ConfigUtil.addLinkFallback(conf, fallbackTarget.toUri());
    AbstractFileSystem vfs = AbstractFileSystem.get(viewFsDefaultClusterUri,
        conf);
    //user1/test1 does not exist in fallback
    Path nextLevelToInternalDir = new Path("/user1/test1");
    Path test = Path.mergePaths(fallbackTarget, nextLevelToInternalDir);
    assertFalse(fsTarget.exists(test));
    // user1 exists in viewFS mount.
    assertNotNull(vfs.getFileStatus(new Path("/user1")));
    // user1 does not exists in fallback.
    assertFalse(fsTarget.exists(test.getParent()));
    cluster.shutdownNameNodes();
    try {
      // /user1/test1 does not exist in mount internal dir tree, it would
      // attempt to create in fallback.
      vfs.mkdir(nextLevelToInternalDir, FsPermission.getDirDefault(),
          false);
      Assert.fail("It should throw IOE when fallback fs not available.");
    } catch (IOException e) {
      cluster.restartNameNodes();
      // should succeed when fallback fs is back to normal.
      vfs.mkdir(nextLevelToInternalDir, FsPermission.getDirDefault(),
          false);
    }
    assertTrue(fsTarget.exists(test));
  }

}
