package test;

import com.airbnb.di.hive.batchreplication.hivecopy.MetastoreReplicationJob;
import com.airbnb.di.hive.common.HiveObjectSpec;
import com.airbnb.di.hive.replication.ReplicationUtils;
import com.airbnb.di.utils.ReplicationTestUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.util.ToolRunner;
import org.junit.BeforeClass;
import org.junit.Test;

import java.io.IOException;
import java.sql.SQLException;

import static org.junit.Assert.assertTrue;

/**
 * Unit Test for MetastoreReplicationJob
 */
public class BatchMetastoreCopyTest extends MockClusterTest {
    @BeforeClass
    public static void setupClass() throws IOException, SQLException {
        MockClusterTest.setupClass();
   }


    @Test
    public void testCopyNewTables() throws Exception {
        // Create an unpartitioned table in the source
        HiveObjectSpec spec = new HiveObjectSpec("test", "test_table");
        Table srcTable = ReplicationTestUtils.createUnpartitionedTable(conf,
                srcMetastore,
                spec,
                TableType.MANAGED_TABLE,
                srcWarehouseRoot);

        // Create a partitioned table in the source
        HiveObjectSpec tableSpec = new HiveObjectSpec("test", "partitioned_table");
        Table srcTable2 = ReplicationTestUtils.createPartitionedTable(conf,
                srcMetastore,
                tableSpec,
                TableType.MANAGED_TABLE,
                srcWarehouseRoot);

        // Create several partitions in the source table
        HiveObjectSpec partitionSpec1 = new HiveObjectSpec("test",
                "partitioned_table", "ds=1/hr=1");
        HiveObjectSpec partitionSpec2 = new HiveObjectSpec("test",
                "partitioned_table", "ds=1/hr=2");
        HiveObjectSpec partitionSpec3 = new HiveObjectSpec("test",
                "partitioned_table", "ds=1/hr=3");

        Partition srcPartition1 = ReplicationTestUtils.createPartition(conf,
                srcMetastore, partitionSpec1);
        Partition srcPartition2 = ReplicationTestUtils.createPartition(conf,
                srcMetastore, partitionSpec2);
        Partition srcPartition3 = ReplicationTestUtils.createPartition(conf,
                srcMetastore, partitionSpec3);

        JobConf jobConf = new JobConf(conf);

        String[] args = {};
        jobConf.set("airbnb.reair.clusters.batch.output.dir",
                new Path(destCluster.getFsRoot(), "test_output").toString());
        jobConf.set("airbnb.reair.clusters.batch.test.injection.class", "test.MockClusterTest");

        ToolRunner.run(jobConf, new MetastoreReplicationJob(), args);

        assertTrue(ReplicationUtils.exists(destMetastore, spec));

        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec1));
        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec2));
        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec3));

        ReplicationTestUtils.dropTable(srcMetastore, spec);
        ReplicationTestUtils.dropPartition(srcMetastore, partitionSpec2);

        ToolRunner.run(jobConf, new MetastoreReplicationJob(), args);

        assertTrue(!ReplicationUtils.exists(destMetastore, partitionSpec2));
    }
}

@@ -781,16 +781,17 @@
 rTest %7B%0A
+%0A
     @Bef
@@ -914,17 +914,16 @@
 ;%0A   %7D%0A%0A
-%0A
     @Tes
@@ -2733,14 +2733,8 @@
 s%22, 
-%22test.
 Mock
@@ -2744,17 +2744,32 @@
 sterTest
-%22
+.class.getName()
 );%0A%0A    
@@ -2914,224 +2914,1498 @@
     
-assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec1));%0A        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec2));%0A        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec3
+Table dstTable = destMetastore.getTable(spec.getDbName(), spec.getTableName());%0A        assertTrue(directoryCopier.equalDirs(new Path(srcTable.getSd().getLocation()),%0A                new Path(dstTable.getSd().getLocation())));%0A%0A        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec1));%0A        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec2));%0A        assertTrue(ReplicationUtils.exists(destMetastore, partitionSpec3));%0A%0A        Partition dstPartition1 = destMetastore.getPartition(partitionSpec1.getDbName(),%0A                partitionSpec1.getTableName(),%0A                partitionSpec1.getPartitionName());%0A        assertTrue(directoryCopier.equalDirs(new Path(srcPartition1.getSd().getLocation()),%0A                new Path(dstPartition1.getSd().getLocation())));%0A%0A        Partition dstPartition2 = destMetastore.getPartition(partitionSpec2.getDbName(),%0A                partitionSpec2.getTableName(),%0A                partitionSpec2.getPartitionName());%0A        assertTrue(directoryCopier.equalDirs(new Path(srcPartition2.getSd().getLocation()),%0A                new Path(dstPartition2.getSd().getLocation())));%0A%0A        Partition dstPartition3 = destMetastore.getPartition(partitionSpec3.getDbName(),%0A                partitionSpec3.getTableName(),%0A                partitionSpec3.getPartitionName());%0A        assertTrue(directoryCopier.equalDirs(new Path(srcPartition3.getSd().getLocation()),%0A                new Path(dstPartition3.getSd().getLocation())
 ));%0A

fix minor comment