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

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import alluxio.SystemPropertyRule;
import alluxio.annotation.dora.DoraTestTodoItem;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.Configuration;
import alluxio.conf.InstancedConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.master.MasterInquireClient;
import alluxio.security.User;
import alluxio.uri.MultiMasterAuthority;
import alluxio.uri.ZookeeperAuthority;

import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;

import java.io.Closeable;
import java.io.IOException;
import java.security.Principal;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import javax.security.auth.Subject;

public class FileSystemFactoryTest {
  @Before
  public void before() {
    Configuration.reloadProperties();
  }

  @After
  public void after() {
    FileSystem.Factory.FILESYSTEM_CACHE.purge();
  }

  @Test
  public void testCloseRemovesFromCache() throws Exception {
    FileSystem fs1 = FileSystem.Factory.get();
    fs1.close();
    assertTrue("FileSystem should be marked as closed", fs1.isClosed());
    FileSystem fs2 = FileSystem.Factory.get();
    assertFalse("FileSystem shouldn't be closed", fs2.isClosed());
    assertNotSame("Should have different references", fs1, fs2);
  }

  @Test
  public void singleMasterFileSystemCacheTest()  {
    fileSystemCacheTest();
  }

  @Test
  @DoraTestTodoItem(action = DoraTestTodoItem.Action.FIX, owner = "Jiacheng",
      comment = "fix master HA or remove if no longer needed")
  @Ignore
  public void multiMasterFileSystemCacheTest()  {
    try (Closeable p = new SystemPropertyRule(PropertyKey.MASTER_RPC_ADDRESSES.getName(),
        "192.168.0.1:1234,192.168.0.2:1445,192.168.0.3:9943").toResource()) {
      Configuration.reloadProperties();
      AlluxioConfiguration conf = Configuration.global();
      MasterInquireClient.ConnectDetails connectDetails =
          MasterInquireClient.Factory.getConnectDetails(conf);
      // Make sure we have a MultiMaster authority
      assertTrue(connectDetails.toAuthority() instanceof MultiMasterAuthority);
      fileSystemCacheTest();
    } catch (IOException e) {
      fail("Unable to set system properties");
    }
  }

  @Test
  public void zkFileSystemCacheTest()  {
    Map<String, String> sysProps = new HashMap<>();
    sysProps.put(PropertyKey.ZOOKEEPER_ENABLED.getName(), Boolean.toString(true));
    sysProps.put(PropertyKey.ZOOKEEPER_ADDRESS.getName(), "zk@192.168.0.5");
    sysProps.put(PropertyKey.ZOOKEEPER_ELECTION_PATH.getName(), "/alluxio/leader");

    try (Closeable p = new SystemPropertyRule(sysProps).toResource()) {
      Configuration.reloadProperties();
      AlluxioConfiguration conf = Configuration.global();
      MasterInquireClient.ConnectDetails connectDetails =
          MasterInquireClient.Factory.getConnectDetails(conf);
      // Make sure we have a Zookeeper authority
      assertTrue(connectDetails.toAuthority() instanceof ZookeeperAuthority);
      fileSystemCacheTest();
    } catch (IOException e) {
      fail("Unable to set system properties");
    }
  }

  @Test
  public void nullSubjectTest()  {
    assertThrows(NullPointerException.class, () -> FileSystem.Factory.get(null));
  }

  @Test
  public void uncachedFileSystemDoesntAffectCache() throws Exception {
    FileSystem fs1 = FileSystem.Factory.get();
    InstancedConfiguration conf = Configuration.copyGlobal();
    conf.set(PropertyKey.USER_WORKER_LIST_REFRESH_INTERVAL, "1sec");
    FileSystem fs2 = FileSystem.Factory.create(conf);
    fs2.close();
    FileSystem fs3 = FileSystem.Factory.get();
    assertSame("closing custom config should result in same FileSystem",
        getDelegatedFileSystem(fs1), getDelegatedFileSystem(fs3));
    assertFalse("FileSystem should not be closed", fs1.isClosed());
  }

  // Helper method to get the underlying delegated file system from cache
  private FileSystem getDelegatedFileSystem(FileSystem fs) {
    return ((FileSystemCache.InstanceCachingFileSystem) fs).mDelegatedFileSystem;
  }

  private void fileSystemCacheTest()  {
    FileSystem.Factory.FILESYSTEM_CACHE.purge();
    FileSystem fs1 = FileSystem.Factory.get();
    FileSystem fs2 = FileSystem.Factory.get();
    assertSame("Second client should have been retrieved from cache.",
        getDelegatedFileSystem(fs1), getDelegatedFileSystem(fs2));
    fs2 = FileSystem.Factory.get(new Subject());
    assertSame("Passing empty subject should have given the same cached client",
        getDelegatedFileSystem(fs1), getDelegatedFileSystem(fs2));
    fs2 = FileSystem.Factory.get(createTestSubject("alluxio-test"));
    assertNotSame("Passing filled subject should have given a new client",
        getDelegatedFileSystem(fs1), getDelegatedFileSystem(fs2));
    fs1 = FileSystem.Factory.get(createTestSubject("alluxio-test"));
    assertSame("Second subject with same credentials should return cached client",
        getDelegatedFileSystem(fs1), getDelegatedFileSystem(fs2));
    fs2 = FileSystem.Factory.get(createTestSubject("alluxio-test-2"));
    assertNotSame("Passing filled subject should have given a new client",
        getDelegatedFileSystem(fs1), getDelegatedFileSystem(fs2));
  }

  private Subject createTestSubject(String username) {
    User user = new User(username);
    Set<Principal> principals = new HashSet<>();
    principals.add(user);
    return new Subject(false, principals, new HashSet<>(), new HashSet<>());
  }
}
