/*
 * 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.AlluxioURI;
import alluxio.Constants;
import alluxio.annotation.SuppressFBWarnings;
import alluxio.client.ReadType;
import alluxio.client.WriteType;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemContext;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.WorkerNetAddress;
import alluxio.util.io.PathUtils;

import com.beust.jcommander.IStringConverter;
import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.internal.Lists;
import com.google.common.base.Preconditions;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Driver to run Alluxio tests.
 */
@ThreadSafe
public final class TestRunner {
  /** Read types to test. */
  private static final List<ReadType> READ_TYPES =
      Arrays.asList(ReadType.CACHE_PROMOTE, ReadType.CACHE, ReadType.NO_CACHE);

  /** Write types to test. */
  private static final List<WriteType> WRITE_TYPES =
      Arrays.asList(WriteType.CACHE_THROUGH, WriteType.THROUGH);

  @Parameter(names = "--directory",
      description = "Alluxio path for the tests working directory.")
  private String mDirectory = AlluxioURI.SEPARATOR;

  @Parameter(names = {"-h", "--help"}, description = "Prints usage information", help = true)
  private boolean mHelp;

  @Parameter(names = "--operation", description = "The operation to test, either BASIC or "
      + "BASIC_NON_BYTE_BUFFER. By default both operations are tested.")
  private String mOperation;

  @Parameter(names = "--readType",
      description = "The read type to use, one of NO_CACHE, CACHE, CACHE_PROMOTE. "
      + "By default all readTypes are tested.")
  private String mReadType;

  @Parameter(names = "--writeType",
      description = "The write type to use, one of MUST_CACHE, CACHE_THROUGH, "
      + "THROUGH, ASYNC_THROUGH. By default all writeTypes are tested.")
  private String mWriteType;

  @Parameter(names = {"--workers", "--worker"},
      description = "Alluxio worker addresses to run tests on. "
          + "If not specified, random ones will be used.",
      converter = WorkerAddressConverter.class)
  @Nullable
  @SuppressFBWarnings(value = "UWF_NULL_FIELD", justification = "Injected field")
  private List<WorkerNetAddress> mWorkerAddresses = null;

  /**
   * The operation types to test.
   */
  enum OperationType {
    /**
     * Basic operations.
     */
    BASIC,
    /**
     * Basic operations but not using ByteBuffer.
     */
    BASIC_NON_BYTE_BUFFER,
  }

  private TestRunner() {} // prevent instantiation

  /** Directory for the test generated files. */
  public static final String TEST_DIRECTORY_NAME = "default_tests_files";

  /**
   * Console program that validates the configuration.
   *
   * @param args array of arguments given by the user's input from the terminal
   */
  public static void main(String[] args) throws Exception {
    TestRunner runner = new TestRunner();
    JCommander jCommander = new JCommander(runner);
    jCommander.setProgramName("TestRunner");
    jCommander.parse(args);
    if (runner.mHelp) {
      jCommander.usage();
      return;
    }

    int ret = runner.runTests();
    System.exit(ret);
  }

  /**
   * Runs combinations of tests of operation, read and write type.
   *
   * @return the number of failed tests
   */
  private int runTests() throws Exception {
    if (mWorkerAddresses != null) {
      Configuration.set(PropertyKey.USER_FILE_PASSIVE_CACHE_ENABLED, false);
    }

    mDirectory = PathUtils.concatPath(mDirectory, TEST_DIRECTORY_NAME);

    AlluxioURI testDir = new AlluxioURI(mDirectory);
    FileSystemContext fsContext = FileSystemContext.create();
    FileSystem fs = FileSystem.Factory.create(fsContext);
    if (fs.exists(testDir)) {
      fs.delete(testDir, DeletePOptions.newBuilder().setRecursive(true).setUnchecked(true).build());
    }

    int failed = 0;

    List<ReadType> readTypes =
        mReadType == null ? READ_TYPES : Lists.newArrayList(ReadType.valueOf(mReadType));
    List<WriteType> writeTypes =
        mWriteType == null ? WRITE_TYPES : Lists.newArrayList(WriteType.valueOf(mWriteType));
    List<OperationType> operations = mOperation == null ? Lists.newArrayList(OperationType.values())
        : Lists.newArrayList(OperationType.valueOf(mOperation));

    if (mWorkerAddresses == null) {
      for (ReadType readType : readTypes) {
        for (WriteType writeType : writeTypes) {
          for (OperationType opType : operations) {
            System.out.println(String.format("runTest --operation %s --readType %s --writeType %s",
                opType, readType, writeType));
            failed += runTest(opType, readType, writeType, fsContext, null);
          }
        }
      }
      if (failed > 0) {
        System.out.println("Number of failed tests: " + failed);
      }
    } else {
      // If workers are specified, the test will iterate through all workers and
      // for each worker:
      // 1. Create a file
      // 2. Write the blocks of test files into a specific worker using SpecificHostPolicy
      // 3. Open the file to read
      // 4. If blocks are cached on the worker (MUST_CACHE/CACHE_THROUGH/ASYNC_THROUGH),
      //    then data will be read from that specific worker as there is only one copy of data.
      //    If blocks are not cached on the worker (THROUGH),
      //    then data will be loaded from UFS into that specific worker, by setting the
      //    ufsReadWorkerLocation field InStreamOptions.
      // In this way, we made sure that a worker works normally on both reads and writes.
      HashMap<WorkerNetAddress, Integer> failedTestWorkers = new HashMap<>();
      boolean hasFailedWorkers = false;
      for (WorkerNetAddress workerNetAddress : mWorkerAddresses) {
        System.out.println("Running test for worker:" + getWorkerAddressString(workerNetAddress));
        for (ReadType readType : readTypes) {
          for (WriteType writeType : writeTypes) {
            for (OperationType opType : operations) {
              System.out.printf("[%s] runTest --operation %s --readType %s --writeType %s%n",
                  getWorkerAddressString(workerNetAddress), opType, readType, writeType);
              failed += runTest(opType, readType, writeType, fsContext, workerNetAddress);
              failedTestWorkers.put(
                  workerNetAddress, failedTestWorkers.getOrDefault(workerNetAddress, 0) + failed);
              if (failed != 0) {
                hasFailedWorkers = true;
              }
            }
          }
        }
      }
      if (!hasFailedWorkers) {
        System.out.println(
            Constants.ANSI_GREEN + "All workers passed tests!" + Constants.ANSI_RESET);
      } else {
        System.out.println(
            Constants.ANSI_RED + "Some workers failed tests!" + Constants.ANSI_RESET);
        failedTestWorkers.forEach((k, v) -> {
          if (v > 0) {
            System.out.printf(
                "%sWorker %s failed %s tests %s%n",
                Constants.ANSI_RED, getWorkerAddressString(k), 4, Constants.ANSI_RESET);
          }
        });
      }
    }
    return failed;
  }

    /**
     * Runs a single test given operation, read and write type.
     *
     * @param opType operation type
     * @param readType read type
     * @param writeType write type
     * @return 0 on success, 1 on failure
     */
  private int runTest(
      OperationType opType, ReadType readType, WriteType writeType,
      FileSystemContext fsContext, @Nullable WorkerNetAddress workerAddress) {
    final AlluxioURI filePath;
    if (workerAddress == null) {
      filePath = new AlluxioURI(
          String.format("%s/%s_%s_%s", mDirectory, opType, readType, writeType));
    } else {
      String workerAddressString = getWorkerAddressString(workerAddress);
      filePath = new AlluxioURI(
          String.format("%s/%s/%s_%s_%s", mDirectory, workerAddressString, opType, readType,
              writeType));
    }

    boolean result = true;
    switch (opType) {
      case BASIC:
        result = RunTestUtils.runExample(
          new BasicOperations(filePath, readType, writeType, fsContext, workerAddress));
        break;
      case BASIC_NON_BYTE_BUFFER:
        result = RunTestUtils.runExample(
          new BasicNonByteBufferOperations(
              filePath, readType, writeType, true, 20, fsContext, workerAddress));
        break;
      default:
        System.out.println("Unrecognized operation type " + opType);
    }
    return result ? 0 : 1;
  }

  /**
   * Parses worker address param.
   */
  public static class WorkerAddressConverter implements IStringConverter<WorkerNetAddress> {
    @Override
    public WorkerNetAddress convert(String s) {
      if (s.contains(":")) {
        String[] components = s.split(":");
        Preconditions.checkState(components.length == 2);
        return WorkerNetAddress.newBuilder().setHost(components[0])
            .setRpcPort(Integer.parseInt(components[1])).build();
      } else {
        return WorkerNetAddress.newBuilder().setHost(s).build();
      }
    }
  }

  private String getWorkerAddressString(WorkerNetAddress workerAddress) {
    return workerAddress.getRpcPort() == 0 ? workerAddress.getHost() :
        workerAddress.getHost() + "_" + workerAddress.getRpcPort();
  }
}
