/**
 * 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
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * 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.hcatalog.common;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.metastore.api.SerDeInfo;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.serde.Constants;
import org.apache.hcatalog.data.schema.HCatFieldSchema;
import org.apache.hcatalog.data.schema.HCatSchema;
import org.junit.Assert;
import org.junit.Test;

public class TestHCatUtil {

  @Test
  public void testFsPermissionOperation(){

    HashMap<String,Integer> permsCode = new HashMap<String,Integer>();

    for (int i = 0; i < 8; i++){
      for (int j = 0; j < 8; j++){
        for (int k = 0; k < 8; k++){
          StringBuilder sb = new StringBuilder();
          sb.append("0");
          sb.append(i);
          sb.append(j);
          sb.append(k);
          Integer code = (((i*8)+j)*8)+k;
          String perms = (new FsPermission(Short.decode(sb.toString()))).toString();
          if (permsCode.containsKey(perms)){
            Assert.assertEquals("permissions(" + perms + ") mapped to multiple codes", code, permsCode.get(perms));
          }
          permsCode.put(perms, code);
          assertFsPermissionTransformationIsGood(perms);
        }
      }
    }
  }

  private void assertFsPermissionTransformationIsGood(String perms) {
    Assert.assertEquals(perms, FsPermission.valueOf("-" + perms).toString());
  }

  @Test
  public void testValidateMorePermissive(){
    assertConsistentFsPermissionBehaviour(FsAction.ALL,true,true,true,true,true,true,true,true);
    assertConsistentFsPermissionBehaviour(FsAction.READ,false,true,false,true,false,false,false,false);
    assertConsistentFsPermissionBehaviour(FsAction.WRITE,false,true,false,false,true,false,false,false);
    assertConsistentFsPermissionBehaviour(FsAction.EXECUTE,false,true,true,false,false,false,false,false);
    assertConsistentFsPermissionBehaviour(FsAction.READ_EXECUTE,false,true,true,true,false,true,false,false);
    assertConsistentFsPermissionBehaviour(FsAction.READ_WRITE,false,true,false,true,true,false,true,false);
    assertConsistentFsPermissionBehaviour(FsAction.WRITE_EXECUTE,false,true,true,false,true,false,false,true);
    assertConsistentFsPermissionBehaviour(FsAction.NONE,false,true,false,false,false,false,false,false);
  }


  private void assertConsistentFsPermissionBehaviour(
      FsAction base, boolean versusAll, boolean versusNone,
      boolean versusX, boolean versusR, boolean versusW,
      boolean versusRX, boolean versusRW,  boolean versusWX){

    Assert.assertTrue(versusAll == HCatUtil.validateMorePermissive(base, FsAction.ALL));
    Assert.assertTrue(versusX == HCatUtil.validateMorePermissive(base, FsAction.EXECUTE));
    Assert.assertTrue(versusNone == HCatUtil.validateMorePermissive(base, FsAction.NONE));
    Assert.assertTrue(versusR == HCatUtil.validateMorePermissive(base, FsAction.READ));
    Assert.assertTrue(versusRX == HCatUtil.validateMorePermissive(base, FsAction.READ_EXECUTE));
    Assert.assertTrue(versusRW == HCatUtil.validateMorePermissive(base, FsAction.READ_WRITE));
    Assert.assertTrue(versusW == HCatUtil.validateMorePermissive(base, FsAction.WRITE));
    Assert.assertTrue(versusWX == HCatUtil.validateMorePermissive(base, FsAction.WRITE_EXECUTE));
  }

  @Test
  public void testExecutePermissionsCheck(){
    Assert.assertTrue(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.ALL));
    Assert.assertTrue(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.NONE));
    Assert.assertTrue(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.EXECUTE));
    Assert.assertTrue(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.READ_EXECUTE));
    Assert.assertTrue(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.WRITE_EXECUTE));

    Assert.assertFalse(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.READ));
    Assert.assertFalse(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.WRITE));
    Assert.assertFalse(HCatUtil.validateExecuteBitPresentIfReadOrWrite(FsAction.READ_WRITE));

  }

  @Test
  public void testGetTableSchemaWithPtnColsApi() throws IOException {
    // Check the schema of a table with one field & no partition keys.
    StorageDescriptor sd = new StorageDescriptor(
        Lists.newArrayList(new FieldSchema("username", Constants.STRING_TYPE_NAME, null)),
        "location", "org.apache.hadoop.mapred.TextInputFormat",
        "org.apache.hadoop.mapred.TextOutputFormat", false, -1, new SerDeInfo(),
        new ArrayList<String>(), new ArrayList<Order>(), new HashMap<String, String>());
    org.apache.hadoop.hive.metastore.api.Table apiTable =
        new org.apache.hadoop.hive.metastore.api.Table("test_tblname", "test_dbname", "test_owner",
            0, 0, 0, sd, new ArrayList<FieldSchema>(), new HashMap<String, String>(),
            "viewOriginalText", "viewExpandedText", TableType.EXTERNAL_TABLE.name());
    Table table = new Table(apiTable);

    List<HCatFieldSchema> expectedHCatSchema =
        Lists.newArrayList(new HCatFieldSchema("username", HCatFieldSchema.Type.STRING, null));

    Assert.assertEquals(new HCatSchema(expectedHCatSchema),
        HCatUtil.getTableSchemaWithPtnCols(table));

    // Add a partition key & ensure its reflected in the schema.
    List<FieldSchema> partitionKeys =
        Lists.newArrayList(new FieldSchema("dt", Constants.STRING_TYPE_NAME, null));
    table.getTTable().setPartitionKeys(partitionKeys);
    expectedHCatSchema.add(new HCatFieldSchema("dt", HCatFieldSchema.Type.STRING, null));
    Assert.assertEquals(new HCatSchema(expectedHCatSchema),
        HCatUtil.getTableSchemaWithPtnCols(table));
  }

  /**
   * Hive represents tables in two ways:
   * <ul>
   *   <li>org.apache.hadoop.hive.metastore.api.Table - exactly whats stored in the metastore</li>
   *   <li>org.apache.hadoop.hive.ql.metadata.Table - adds business logic over api.Table</li>
   * </ul>
   * Here we check SerDe-reported fields are included in the table schema.
   */
  @Test
  public void testGetTableSchemaWithPtnColsSerDeReportedFields() throws IOException {
    Map<String, String> parameters = Maps.newHashMap();
    parameters.put(Constants.SERIALIZATION_CLASS,
        "org.apache.hadoop.hive.serde2.thrift.test.IntString");
    parameters.put(Constants.SERIALIZATION_FORMAT, "org.apache.thrift.protocol.TBinaryProtocol");

    SerDeInfo serDeInfo = new SerDeInfo(null,
        "org.apache.hadoop.hive.serde2.thrift.ThriftDeserializer", parameters);

    // StorageDescriptor has an empty list of fields - SerDe will report them.
    StorageDescriptor sd = new StorageDescriptor(new ArrayList<FieldSchema>(), "location",
        "org.apache.hadoop.mapred.TextInputFormat", "org.apache.hadoop.mapred.TextOutputFormat",
        false, -1, serDeInfo, new ArrayList<String>(), new ArrayList<Order>(),
        new HashMap<String, String>());

    org.apache.hadoop.hive.metastore.api.Table apiTable =
        new org.apache.hadoop.hive.metastore.api.Table("test_tblname", "test_dbname", "test_owner",
            0, 0, 0, sd, new ArrayList<FieldSchema>(), new HashMap<String, String>(),
            "viewOriginalText", "viewExpandedText", TableType.EXTERNAL_TABLE.name());
    Table table = new Table(apiTable);

    List<HCatFieldSchema> expectedHCatSchema = Lists.newArrayList(
        new HCatFieldSchema("myint", HCatFieldSchema.Type.INT, null),
        new HCatFieldSchema("mystring", HCatFieldSchema.Type.STRING, null),
        new HCatFieldSchema("underscore_int", HCatFieldSchema.Type.INT, null));

    Assert.assertEquals(new HCatSchema(expectedHCatSchema),
        HCatUtil.getTableSchemaWithPtnCols(table));
  }
}