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

import static org.apache.hcatalog.HcatTestUtils.perm300;
import static org.apache.hcatalog.HcatTestUtils.perm500;
import static org.apache.hcatalog.HcatTestUtils.perm555;
import static org.apache.hcatalog.HcatTestUtils.perm700;
import static org.apache.hcatalog.HcatTestUtils.perm755;

import java.io.IOException;
import java.util.Random;

import junit.framework.Assert;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.cli.CliSessionState;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.Warehouse;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.shims.ShimLoader;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hcatalog.HcatTestUtils;
import org.apache.hcatalog.cli.HCatDriver;
import org.apache.hcatalog.cli.SemanticAnalysis.HCatSemanticAnalyzer;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

public class TestHdfsAuthorizationProvider {
  
  protected HCatDriver hcatDriver;
  protected HiveMetaStoreClient msc;
  protected HiveConf conf;
  protected String whDir;
  protected Path whPath;
  protected FileSystem whFs;
  protected Warehouse wh;
  protected Hive hive;

  @Before
  public void setUp() throws Exception {
    
    conf = new HiveConf(this.getClass());
    conf.set(ConfVars.PREEXECHOOKS.varname, "");
    conf.set(ConfVars.POSTEXECHOOKS.varname, "");
    conf.set(ConfVars.HIVE_SUPPORT_CONCURRENCY.varname, "false");
  
    conf.set("hive.metastore.local", "true");
    conf.set(ConfVars.SEMANTIC_ANALYZER_HOOK.varname, HCatSemanticAnalyzer.class.getName());
    conf.setBoolVar(HiveConf.ConfVars.HIVE_AUTHORIZATION_ENABLED, true);
    conf.setVar(HiveConf.ConfVars.HIVE_AUTHORIZATION_MANAGER, 
        StorageDelegationAuthorizationProvider.class.getCanonicalName());
    conf.set("fs.pfile.impl", "org.apache.hadoop.fs.ProxyLocalFileSystem");
    
    whDir = System.getProperty("test.warehouse.dir", "/tmp/testhdfsauthorization_wh");
    conf.setVar(HiveConf.ConfVars.METASTOREWAREHOUSE, whDir);
    
    UserGroupInformation ugi = ShimLoader.getHadoopShims().getUGIForConf(conf);
    String username = ShimLoader.getHadoopShims().getShortUserName(ugi); 
    
    whPath = new Path(whDir);
    whFs = whPath.getFileSystem(conf);
    
    wh = new Warehouse(conf);
    hive = Hive.get(conf);
    
    //clean up mess in HMS 
    HcatTestUtils.cleanupHMS(hive, wh, perm700);
    
    whFs.delete(whPath, true);
    whFs.mkdirs(whPath, perm755);
    
    SessionState.start(new CliSessionState(conf));
    hcatDriver = new HCatDriver();
  }

  @After
  public void tearDown() throws IOException {
    whFs.close();
    hcatDriver.close();
    Hive.closeCurrent();
  }

  public Path getDbPath(String dbName) throws MetaException, HiveException {
    return HcatTestUtils.getDbPath(hive, wh, dbName); 
  }
  
  public Path getTablePath(String dbName, String tableName) throws HiveException {
    Table table = hive.getTable(dbName, tableName);
    return table.getPath();
  }

  public Path getPartPath(String partName, String dbName, String tableName) throws HiveException {
    return new Path(getTablePath(dbName, tableName), partName);
  }

  /** Execute the query expecting success*/
  public void exec(String format, Object ... args) throws Exception {
    String command = String.format(format, args);
    CommandProcessorResponse resp = hcatDriver.run(command);
    Assert.assertEquals(resp.getErrorMessage(), 0, resp.getResponseCode());
    Assert.assertEquals(resp.getErrorMessage(), null, resp.getErrorMessage());
  }

  /** Execute the query expecting it to fail with AuthorizationException */
  public void execFail(String format, Object ... args) throws Exception {
    String command = String.format(format, args);
    CommandProcessorResponse resp = hcatDriver.run(command);
    Assert.assertNotSame(resp.getErrorMessage(), 0, resp.getResponseCode());
    Assert.assertTrue((resp.getResponseCode() == 40000) || (resp.getResponseCode() == 403));
    if(resp.getErrorMessage() != null){
     Assert.assertTrue(resp.getErrorMessage().contains("org.apache.hadoop.security.AccessControlException"));
    }
  }

  
  /** 
   * Tests whether the warehouse directory is writable by the current user (as defined by Hadoop)
   */
  @Test
  public void testWarehouseIsWritable() throws Exception {
    Path top = new Path(whPath, "_foobarbaz12_");
    try {
      whFs.mkdirs(top);
    } finally {
      whFs.delete(top, true);
    }
  }
  
  @Test
  public void testShowDatabases() throws Exception {
    exec("CREATE DATABASE doo");
    exec("SHOW DATABASES");
    
    whFs.setPermission(whPath, perm300); //revoke r
    execFail("SHOW DATABASES");
  }
  
  @Test
  public void testDatabaseOps() throws Exception {
    exec("SHOW TABLES");
    exec("SHOW TABLE EXTENDED LIKE foo1");
    
    whFs.setPermission(whPath, perm700);
    exec("CREATE DATABASE doo");
    exec("DESCRIBE DATABASE doo");
    exec("USE doo");
    exec("SHOW TABLES");
    exec("SHOW TABLE EXTENDED LIKE foo1");
    exec("DROP DATABASE doo");
    
    //custom location
    Path dbPath = new Path(whPath, new Random().nextInt() + "/mydb");
    whFs.mkdirs(dbPath, perm700);
    exec("CREATE DATABASE doo2 LOCATION '%s'", dbPath.toUri());
    exec("DESCRIBE DATABASE doo2", dbPath.toUri());
    exec("USE doo2");
    exec("SHOW TABLES");
    exec("SHOW TABLE EXTENDED LIKE foo1");
    exec("DROP DATABASE doo2", dbPath.toUri());
    
    //custom non-existing location
    exec("CREATE DATABASE doo3 LOCATION '%s/subpath'", dbPath.toUri());
  }
  
  @Test
  public void testCreateDatabaseFail1() throws Exception {
    whFs.setPermission(whPath, perm500);
    execFail("CREATE DATABASE doo"); //in the default location
    
    whFs.setPermission(whPath, perm555);
    execFail("CREATE DATABASE doo2");
  }

  @Test
  public void testCreateDatabaseFail2() throws Exception {
    //custom location
    Path dbPath = new Path(whPath, new Random().nextInt() + "/mydb");
    
    whFs.mkdirs(dbPath, perm700);
    whFs.setPermission(dbPath, perm500);
    execFail("CREATE DATABASE doo2 LOCATION '%s'", dbPath.toUri());
  }
  
  @Test
  public void testDropDatabaseFail1() throws Exception {
    whFs.setPermission(whPath, perm700);
    exec("CREATE DATABASE doo"); //in the default location
    
    whFs.setPermission(getDbPath("doo"), perm500); //revoke write
    execFail("DROP DATABASE doo");
  }
  
  @Test
  public void testDropDatabaseFail2() throws Exception {
    //custom location
    Path dbPath = new Path(whPath, new Random().nextInt() + "/mydb");
    
    whFs.mkdirs(dbPath, perm700);
    exec("CREATE DATABASE doo2 LOCATION '%s'", dbPath.toUri());
    
    whFs.setPermission(dbPath, perm500);
    execFail("DROP DATABASE doo2");
  }
  
  @Test
  public void testDescSwitchDatabaseFail() throws Exception {
    whFs.setPermission(whPath, perm700);
    exec("CREATE DATABASE doo");
    whFs.setPermission(getDbPath("doo"), perm300); //revoke read
    execFail("DESCRIBE DATABASE doo");
    execFail("USE doo");
    
    //custom location
    Path dbPath = new Path(whPath, new Random().nextInt() + "/mydb");
    whFs.mkdirs(dbPath, perm700);
    exec("CREATE DATABASE doo2 LOCATION '%s'", dbPath.toUri());
    whFs.mkdirs(dbPath, perm300); //revoke read
    execFail("DESCRIBE DATABASE doo2", dbPath.toUri());
    execFail("USE doo2");
  }
  
  @Test 
  public void testShowTablesFail() throws Exception {
    whFs.setPermission(whPath, perm700);
    exec("CREATE DATABASE doo");
    exec("USE doo");
    whFs.setPermission(getDbPath("doo"), perm300); //revoke read
    execFail("SHOW TABLES");
    execFail("SHOW TABLE EXTENDED LIKE foo1");
  }
  
  @Test
  public void testTableOps() throws Exception {
    //default db
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    exec("DESCRIBE foo1");
    exec("DROP TABLE foo1");
    
    //default db custom location
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm700);
    exec("CREATE EXTERNAL TABLE foo2 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    exec("DESCRIBE foo2");
    exec("DROP TABLE foo2");
    
    //default db custom non existing location
    exec("CREATE EXTERNAL TABLE foo3 (foo INT) STORED AS RCFILE LOCATION '%s/subpath'", tablePath);
    exec("DESCRIBE foo3");
    exec("DROP TABLE foo3");
    
    //non default db
    exec("CREATE DATABASE doo");
    exec("USE doo");
    
    exec("CREATE TABLE foo4 (foo INT) STORED AS RCFILE");
    exec("DESCRIBE foo4");
    exec("DROP TABLE foo4");
    
    //non-default db custom location
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm700);
    exec("CREATE EXTERNAL TABLE foo5 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    exec("DESCRIBE foo5");
    exec("DROP TABLE foo5");
    
    //non-default db custom non existing location
    exec("CREATE EXTERNAL TABLE foo6 (foo INT) STORED AS RCFILE LOCATION '%s/subpath'", tablePath);
    exec("DESCRIBE foo6");
    exec("DROP TABLE foo6");
    
    exec("DROP TABLE IF EXISTS foo_non_exists");
    
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    exec("DESCRIBE EXTENDED foo1");
    exec("DESCRIBE FORMATTED foo1");
    exec("DESCRIBE foo1.foo");
    
    //deep non-existing path for the table
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm700);
    exec("CREATE EXTERNAL TABLE foo2 (foo INT) STORED AS RCFILE LOCATION '%s/a/a/a/'", tablePath);
  }
  
  @Test
  public void testCreateTableFail1() throws Exception {
    //default db
    whFs.mkdirs(whPath, perm500); //revoke w
    execFail("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
  }
  
  @Test
  public void testCreateTableFail2() throws Exception {
    //default db custom location
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm500);
    execFail("CREATE EXTERNAL TABLE foo2 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    
    //default db custom non existing location
    execFail("CREATE EXTERNAL TABLE foo3 (foo INT) STORED AS RCFILE LOCATION '%s/subpath'", tablePath);
  }
  
  @Test
  public void testCreateTableFail3() throws Exception {
    //non default db
    exec("CREATE DATABASE doo");
    whFs.setPermission(getDbPath("doo"), perm500);

    execFail("CREATE TABLE doo.foo4 (foo INT) STORED AS RCFILE");
    
    //non-default db custom location, permission to write to tablePath, but not on db path
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm700);
    exec("USE doo");
    execFail("CREATE EXTERNAL TABLE foo5 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
  }

  @Test
  public void testCreateTableFail4() throws Exception {
    //non default db
    exec("CREATE DATABASE doo");

    //non-default db custom location
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm500);
    execFail("CREATE EXTERNAL TABLE doo.foo5 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    
    //non-default db custom non existing location
    execFail("CREATE EXTERNAL TABLE doo.foo6 (foo INT) STORED AS RCFILE LOCATION '%s/a/a/a/'", tablePath);
  }
  
  @Test
  public void testDropTableFail1() throws Exception {
    //default db
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    whFs.mkdirs(getTablePath("default", "foo1"), perm500); //revoke w
    execFail("DROP TABLE foo1");
  }
  
  @Test
  public void testDropTableFail2() throws Exception {
    //default db custom location
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    exec("CREATE EXTERNAL TABLE foo2 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    whFs.mkdirs(tablePath, perm500);
    execFail("DROP TABLE foo2");
  }

  @Test
  public void testDropTableFail4() throws Exception {
    //non default db
    exec("CREATE DATABASE doo");

    //non-default db custom location
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    
    exec("CREATE EXTERNAL TABLE doo.foo5 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    whFs.mkdirs(tablePath, perm500);
    exec("USE doo"); //There is no DROP TABLE doo.foo5 support in Hive
    execFail("DROP TABLE foo5");
  }
  
  @Test
  public void testDescTableFail() throws Exception {
    //default db
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    whFs.mkdirs(getTablePath("default", "foo1"), perm300); //revoke read
    execFail("DESCRIBE foo1");
    
    //default db custom location
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm700);
    exec("CREATE EXTERNAL TABLE foo2 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    whFs.mkdirs(tablePath, perm300); //revoke read
    execFail("DESCRIBE foo2");
  }
  
  @Test
  public void testAlterTableRename() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    exec("ALTER TABLE foo1 RENAME TO foo2");
    
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    exec("CREATE EXTERNAL TABLE foo3 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    exec("ALTER TABLE foo3 RENAME TO foo4");
  }
  
  @Test
  public void testAlterTableRenameFail() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    whFs.mkdirs(getTablePath("default", "foo1"), perm500); //revoke write
    execFail("ALTER TABLE foo1 RENAME TO foo2");
    
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    exec("CREATE EXTERNAL TABLE foo3 (foo INT) STORED AS RCFILE LOCATION '%s'", tablePath);
    whFs.mkdirs(tablePath, perm500); //revoke write 
    execFail("ALTER TABLE foo3 RENAME TO foo4");
  }
  
  @Test
  public void testAlterTableRelocate() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    exec("ALTER TABLE foo1 SET LOCATION '%s'", tablePath.makeQualified(whFs));
    
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable2");
    exec("CREATE EXTERNAL TABLE foo3 (foo INT) STORED AS RCFILE LOCATION '%s'", 
        tablePath.makeQualified(whFs));
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable2");
    exec("ALTER TABLE foo3 SET LOCATION '%s'", tablePath.makeQualified(whFs));
  }
  
  @Test
  public void testAlterTableRelocateFail() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) STORED AS RCFILE");
    Path tablePath = new Path(whPath, new Random().nextInt() + "/mytable");
    whFs.mkdirs(tablePath, perm500); //revoke write
    execFail("ALTER TABLE foo1 SET LOCATION '%s'", tablePath.makeQualified(whFs));
    
    //dont have access to new table loc
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable2");
    exec("CREATE EXTERNAL TABLE foo3 (foo INT) STORED AS RCFILE LOCATION '%s'", 
        tablePath.makeQualified(whFs));
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable2");
    whFs.mkdirs(tablePath, perm500); //revoke write
    execFail("ALTER TABLE foo3 SET LOCATION '%s'", tablePath.makeQualified(whFs));
    
    //have access to new table loc, but not old table loc 
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable3");
    exec("CREATE EXTERNAL TABLE foo4 (foo INT) STORED AS RCFILE LOCATION '%s'", 
        tablePath.makeQualified(whFs));
    whFs.mkdirs(tablePath, perm500); //revoke write
    tablePath = new Path(whPath, new Random().nextInt() + "/mytable3");
    execFail("ALTER TABLE foo4 SET LOCATION '%s'", tablePath.makeQualified(whFs));
  }
  
  @Test
  public void testAlterTable() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS TEXTFILE");
    exec("ALTER TABLE foo1 SET TBLPROPERTIES ('foo'='bar')");
    exec("ALTER TABLE foo1 SET SERDEPROPERTIES ('foo'='bar')");
    exec("ALTER TABLE foo1 ADD COLUMNS (foo2 INT)");
  }
  
  @Test
  public void testAddDropPartition() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS TEXTFILE");
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-10')");
    exec("ALTER TABLE foo1 ADD IF NOT EXISTS PARTITION (b='2010-10-10')");
    String relPath = new Random().nextInt() + "/mypart";
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-11') LOCATION '%s'", relPath);
    
    exec("ALTER TABLE foo1 PARTITION (b='2010-10-10') SET FILEFORMAT RCFILE");
    
    exec("ALTER TABLE foo1 PARTITION (b='2010-10-10') SET FILEFORMAT INPUTFORMAT "
        + "'org.apache.hadoop.hive.ql.io.RCFileInputFormat' OUTPUTFORMAT "
        + "'org.apache.hadoop.hive.ql.io.RCFileOutputFormat' inputdriver "
        + "'mydriver' outputdriver 'yourdriver'");    
    
    exec("ALTER TABLE foo1 DROP PARTITION (b='2010-10-10')");
    exec("ALTER TABLE foo1 DROP PARTITION (b='2010-10-11')");
  }
  
  @Test
  public void testAddPartitionFail1() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS TEXTFILE");
    whFs.mkdirs(getTablePath("default", "foo1"), perm500);
    execFail("ALTER TABLE foo1 ADD PARTITION (b='2010-10-10')");
  }
  
  @Test
  public void testAddPartitionFail2() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS TEXTFILE");
    String relPath = new Random().nextInt() + "/mypart";
    Path partPath = new Path(getTablePath("default", "foo1"), relPath);
    whFs.mkdirs(partPath, perm500);
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-10') LOCATION '%s'", partPath);
  }
  
  @Test
  public void testDropPartitionFail1() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS TEXTFILE");
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-10')");
    whFs.mkdirs(getPartPath("b=2010-10-10", "default", "foo1"), perm500);
    execFail("ALTER TABLE foo1 DROP PARTITION (b='2010-10-10')");
  }

  @Test
  public void testDropPartitionFail2() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS TEXTFILE");
    String relPath = new Random().nextInt() + "/mypart";
    Path partPath = new Path(getTablePath("default", "foo1"), relPath);
    whFs.mkdirs(partPath, perm700);
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-10') LOCATION '%s'", partPath);
    whFs.mkdirs(partPath, perm500); //revoke write
    execFail("ALTER TABLE foo1 DROP PARTITION (b='2010-10-10')");
  }
  
  @Test
  public void testAlterTableFail() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (boo STRING) STORED AS TEXTFILE");
    whFs.mkdirs(getTablePath("default", "foo1"), perm500); //revoke write
    execFail("ALTER TABLE foo1 SET TBLPROPERTIES ('foo'='bar')");
    execFail("ALTER TABLE foo1 SET SERDEPROPERTIES ('foo'='bar')");
    execFail("ALTER TABLE foo1 ADD COLUMNS (foo2 INT)");
  }
  
  @Test
  public void testShowTables() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (boo STRING) STORED AS TEXTFILE");
    exec("SHOW PARTITIONS foo1");
    
    whFs.mkdirs(getTablePath("default", "foo1"), perm300); //revoke read
    execFail("SHOW PARTITIONS foo1");
  }
  
  @Test
  public void testAlterTablePartRename() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS RCFILE");
    Path loc = new Path(whPath, new Random().nextInt() + "/mypart");
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-16') LOCATION '%s'", loc);
    exec("ALTER TABLE foo1 PARTITION (b='2010-10-16') RENAME TO PARTITION (b='2010-10-17')");
  }
  
  @Test
  public void testAlterTablePartRenameFail() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS RCFILE");
    Path loc = new Path(whPath, new Random().nextInt() + "/mypart");
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-16') LOCATION '%s'", loc);
    whFs.setPermission(loc, perm500); //revoke w
    execFail("ALTER TABLE foo1 PARTITION (b='2010-10-16') RENAME TO PARTITION (b='2010-10-17')");
  }
  
  @Test
  public void testAlterTablePartRelocate() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS RCFILE");
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-16')");
    Path partPath = new Path(whPath, new Random().nextInt() + "/mypart");
    exec("ALTER TABLE foo1 PARTITION (b='2010-10-16') SET LOCATION '%s'", partPath.makeQualified(whFs));
  }

  @Test
  public void testAlterTablePartRelocateFail() throws Exception {
    exec("CREATE TABLE foo1 (foo INT) PARTITIONED BY (b STRING) STORED AS RCFILE");
    
    Path oldLoc = new Path(whPath, new Random().nextInt() + "/mypart");
    Path newLoc = new Path(whPath, new Random().nextInt() + "/mypart2");
    
    exec("ALTER TABLE foo1 ADD PARTITION (b='2010-10-16') LOCATION '%s'", oldLoc);
    whFs.mkdirs(oldLoc, perm500);
    execFail("ALTER TABLE foo1 PARTITION (b='2010-10-16') SET LOCATION '%s'", newLoc.makeQualified(whFs));
    whFs.mkdirs(oldLoc, perm700);
    whFs.mkdirs(newLoc, perm500);
    execFail("ALTER TABLE foo1 PARTITION (b='2010-10-16') SET LOCATION '%s'", newLoc.makeQualified(whFs));
  }
  
}