AMBARI-17460. Hive JDBC Add top level unit tests. (Ashwin Rajeev via dipayanb)

contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/actor/DeathWatch.java

@@ -33,7 +33,7 @@ public class DeathWatch extends HiveActor {
             LoggerFactory.getLogger(DeathWatch.class);
 
     @Override
-    void handleMessage(HiveMessage hiveMessage) {
+    public void handleMessage(HiveMessage hiveMessage) {
         Object message = hiveMessage.getMessage();
         if(message instanceof RegisterActor){
             RegisterActor registerActor = (RegisterActor) message;

contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/actor/HiveActor.java

@@ -43,7 +43,7 @@ public abstract class HiveActor extends UntypedActor {
         }
     }
 
-    abstract void handleMessage(HiveMessage hiveMessage);
+    public abstract void handleMessage(HiveMessage hiveMessage);
 
 
 

contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/actor/ResultSetIterator.java

@@ -72,7 +72,7 @@ public class ResultSetIterator extends HiveActor {
   }
 
   @Override
-  void handleMessage(HiveMessage hiveMessage) {
+  public void handleMessage(HiveMessage hiveMessage) {
     sendKeepAlive();
     Object message = hiveMessage.getMessage();
     if (message instanceof Next) {

contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/actor/StatementExecutor.java

@@ -65,7 +65,7 @@ public class StatementExecutor extends HiveActor {
   }
 
   @Override
-  void handleMessage(HiveMessage hiveMessage) {
+  public void handleMessage(HiveMessage hiveMessage) {
     Object message = hiveMessage.getMessage();
     if (message instanceof RunStatement) {
       runStatement((RunStatement) message);

contrib/views/hive-next/src/main/java/org/apache/ambari/view/hive2/actor/YarnAtsGUIDFetcher.java

@@ -28,6 +28,7 @@ import scala.concurrent.duration.Duration;
 
 import java.util.concurrent.TimeUnit;
 
+
 /**
  * Queries YARN/ATS time to time to fetch the status of the ExecuteJob and updates database
  */

contrib/views/hive-next/src/test/java/org/apache/ambari/view/hive2/AsyncJobRunnerImplTest.java

@@ -0,0 +1,140 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.ambari.view.hive2;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import com.google.common.base.Optional;
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.hive2.actor.HiveActor;
+import org.apache.ambari.view.hive2.actor.ResultSetIterator;
+import org.apache.ambari.view.hive2.actor.message.Connect;
+import org.apache.ambari.view.hive2.actor.message.ExecuteJob;
+import org.apache.ambari.view.hive2.actor.message.FetchError;
+import org.apache.ambari.view.hive2.actor.message.FetchResult;
+import org.apache.ambari.view.hive2.actor.message.HiveMessage;
+import org.apache.ambari.view.hive2.actor.message.SQLStatementJob;
+import org.apache.ambari.view.hive2.actor.message.job.CancelJob;
+import org.apache.ambari.view.hive2.actor.message.job.Failure;
+import org.apache.ambari.view.hive2.client.AsyncJobRunnerImpl;
+import org.apache.ambari.view.hive2.client.ConnectionConfig;
+import org.apache.ambari.view.hive2.client.NonPersistentCursor;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.Job;
+import org.apache.hive.jdbc.HiveQueryResultSet;
+import org.apache.tools.ant.taskdefs.Execute;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+
+import static org.easymock.EasyMock.*;
+import static org.junit.Assert.*;
+
+public class AsyncJobRunnerImplTest {
+
+  private ActorSystem actorSystem;
+
+  @Before
+  public void setUp() throws Exception {
+    actorSystem = ActorSystem.create("TestingActorSystem");
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    JavaTestKit.shutdownActorSystem(actorSystem);
+  }
+
+
+  @Test
+  public void testSubmitJob() throws Exception {
+    ConnectionConfig connectionConfig = createNiceMock(ConnectionConfig.class);
+    SQLStatementJob sqlStatementJob = createNiceMock(SQLStatementJob.class);
+    Job job = createNiceMock(Job.class);
+    Connect connect = createNiceMock(Connect.class);
+    ViewContext viewContext = createNiceMock(ViewContext.class);
+    ActorRef controller = actorSystem.actorOf(
+            Props.create(TestParent.class));
+    AsyncJobRunnerImpl runner = new AsyncJobRunnerImpl(viewContext, controller, actorSystem);
+    expect(job.getId()).andReturn("1");
+    expect(connect.getJdbcUrl()).andReturn("testjdbc");
+    expect(connectionConfig.createConnectMessage("1")).andReturn(connect);
+    replay(job, connectionConfig);
+    runner.submitJob(connectionConfig, sqlStatementJob, job);
+    verify(job, connectionConfig);
+  }
+
+  @Test
+  public void testCancelJob() throws Exception {
+    ViewContext viewContext = createNiceMock(ViewContext.class);
+    ActorRef controller = actorSystem.actorOf(
+            Props.create(TestParent.class));
+    AsyncJobRunnerImpl runner = new AsyncJobRunnerImpl(viewContext, controller, actorSystem);
+    runner.cancelJob("1", "test");
+  }
+
+  @Test
+  public void testGetCursor() throws Exception {
+    ViewContext viewContext = createNiceMock(ViewContext.class);
+    ActorRef controller = actorSystem.actorOf(
+            Props.create(TestParent.class));
+    AsyncJobRunnerImpl runner = new AsyncJobRunnerImpl(viewContext, controller, actorSystem);
+    Optional<NonPersistentCursor> cursor = runner.getCursor("1", "test");
+    assertTrue(cursor.isPresent());
+  }
+
+
+  @Test
+  public void testGetError() throws Exception {
+    ViewContext viewContext = createNiceMock(ViewContext.class);
+    ActorRef controller = actorSystem.actorOf(
+            Props.create(TestParent.class));
+    AsyncJobRunnerImpl runner = new AsyncJobRunnerImpl(viewContext, controller, actorSystem);
+    Optional<Failure> failure = runner.getError("1", "test");
+    assertTrue(failure.isPresent());
+    assertEquals("failure", failure.get().getMessage());
+  }
+
+  private static class TestParent extends HiveActor {
+
+    @Override
+    public void handleMessage(HiveMessage hiveMessage) {
+      if (hiveMessage.getMessage() instanceof ExecuteJob) {
+        ExecuteJob executeJob = (ExecuteJob) hiveMessage.getMessage();
+        assertEquals(executeJob.getConnect().getJdbcUrl(), "testjdbc");
+      }
+      if (hiveMessage.getMessage() instanceof CancelJob) {
+        CancelJob cancelJob = (CancelJob) hiveMessage.getMessage();
+        assertEquals("1", cancelJob.getJobId());
+        assertEquals("test", cancelJob.getUsername());
+      }
+      if (hiveMessage.getMessage() instanceof FetchError) {
+        sender().tell(Optional.of(new Failure("failure", new NullPointerException())), self());
+      }
+      if (hiveMessage.getMessage() instanceof FetchResult) {
+        ResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+        ActorRef rsi = context().actorOf(
+                Props.create(ResultSetIterator.class, self(), resultSet));
+        sender().tell(Optional.of(rsi), self());
+      }
+    }
+  }
+}

contrib/views/hive-next/src/test/java/org/apache/ambari/view/hive2/ConnectionFailuresTest.java

@@ -0,0 +1,154 @@
+/*
+ * 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.ambari.view.hive2;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import com.beust.jcommander.internal.Lists;
+import com.google.common.base.Optional;
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.hive2.actor.DeathWatch;
+import org.apache.ambari.view.hive2.actor.OperationController;
+import org.apache.ambari.view.hive2.actor.message.Connect;
+import org.apache.ambari.view.hive2.actor.message.ExecuteJob;
+import org.apache.ambari.view.hive2.actor.message.HiveJob;
+import org.apache.ambari.view.hive2.actor.message.SQLStatementJob;
+import org.apache.ambari.view.hive2.internal.ConnectionSupplier;
+import org.apache.ambari.view.hive2.internal.DataStorageSupplier;
+import org.apache.ambari.view.hive2.internal.HdfsApiSupplier;
+import org.apache.ambari.view.hive2.internal.HiveConnectionWrapper;
+import org.apache.ambari.view.hive2.persistence.Storage;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.Job;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.JobImpl;
+import org.apache.ambari.view.utils.hdfs.HdfsApi;
+import org.apache.hive.jdbc.HiveConnection;
+import org.apache.hive.jdbc.HiveQueryResultSet;
+import org.apache.hive.jdbc.HiveStatement;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+import static org.easymock.EasyMock.*;
+
+public class ConnectionFailuresTest {
+
+  private ActorSystem actorSystem;
+
+  @Before
+  public void setUp() throws Exception {
+    actorSystem = ActorSystem.create("TestingActorSystem");
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    JavaTestKit.shutdownActorSystem(actorSystem);
+  }
+
+  @Test
+  public void testConnectionFailure() throws Exception {
+    ViewContext viewContext = createNiceMock(ViewContext.class);
+    ConnectionSupplier connectionSupplier = createNiceMock(ConnectionSupplier.class);
+    DataStorageSupplier dataStorageSupplier = createNiceMock(DataStorageSupplier.class);
+    HdfsApi hdfsApi = createNiceMock(HdfsApi.class);
+    HdfsApiSupplier hdfsApiSupplier = createNiceMock(HdfsApiSupplier.class);
+    Connect connect = createNiceMock(Connect.class);
+    Storage storage = createNiceMock(Storage.class);
+    JobImpl jobImpl = createNiceMock(JobImpl.class);
+    ResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+    HiveStatement statement = createNiceMock(HiveStatement.class);
+    ConnectionDelegate delegate = createNiceMock(ConnectionDelegate.class);
+    HiveConnectionWrapper connectionWrapper = createNiceMock(HiveConnectionWrapper.class);
+    HiveConnection hiveConnection = createNiceMock(HiveConnection.class);
+    HiveJob test = new SQLStatementJob(HiveJob.Type.ASYNC, new String[]{"select * from test"}, "test", "1", "test.log");
+    ExecuteJob executeJob = new ExecuteJob(connect, test);
+    ActorRef deathwatch = actorSystem.actorOf(Props.create(DeathWatch.class));
+    ActorRef operationControl = actorSystem.actorOf(
+            Props.create(OperationController.class, actorSystem, deathwatch, viewContext, connectionSupplier, dataStorageSupplier, hdfsApiSupplier), "operationController-test");
+    expect(hdfsApiSupplier.get(viewContext)).andReturn(Optional.of(hdfsApi));
+    expect(viewContext.getInstanceName()).andReturn("test").anyTimes();
+    expect(connect.getConnectable()).andReturn(connectionWrapper);
+    expect(connectionWrapper.isOpen()).andReturn(false).anyTimes();
+    expect(connectionWrapper.getConnection()).andReturn(Optional.<HiveConnection>absent()).anyTimes();
+    expect(dataStorageSupplier.get(viewContext)).andReturn(storage);
+    expect(connectionSupplier.get(viewContext)).andReturn(delegate);
+    expect(storage.load(JobImpl.class, "1")).andReturn(jobImpl).anyTimes();
+    connectionWrapper.connect();
+    jobImpl.setStatus(Job.JOB_STATE_ERROR);
+    storage.store(JobImpl.class, jobImpl);
+    replay(viewContext, connect, hdfsApiSupplier, dataStorageSupplier, connectionWrapper,
+            storage, jobImpl, connectionSupplier, delegate, statement, resultSet);
+
+    operationControl.tell(executeJob, ActorRef.noSender());
+    Thread.sleep(5000);
+    verify(connect, hdfsApiSupplier, dataStorageSupplier, connectionWrapper,
+            storage, jobImpl, connectionSupplier, delegate, statement, resultSet);
+
+  }
+
+  @Test
+  public void testExecutionFailure() throws Exception {
+    ViewContext viewContext = createNiceMock(ViewContext.class);
+    ConnectionSupplier connectionSupplier = createNiceMock(ConnectionSupplier.class);
+    DataStorageSupplier dataStorageSupplier = createNiceMock(DataStorageSupplier.class);
+    HdfsApi hdfsApi = createNiceMock(HdfsApi.class);
+    HdfsApiSupplier hdfsApiSupplier = createNiceMock(HdfsApiSupplier.class);
+    Connect connect = createNiceMock(Connect.class);
+    Storage storage = createNiceMock(Storage.class);
+    JobImpl jobImpl = createNiceMock(JobImpl.class);
+    ResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+    HiveStatement statement = createNiceMock(HiveStatement.class);
+    ConnectionDelegate delegate = createNiceMock(ConnectionDelegate.class);
+    HiveConnectionWrapper connectionWrapper = createNiceMock(HiveConnectionWrapper.class);
+    HiveConnection hiveConnection = createNiceMock(HiveConnection.class);
+    HiveJob test = new SQLStatementJob(HiveJob.Type.ASYNC, new String[]{"select * from test"}, "test", "1", "test.log");
+    ExecuteJob executeJob = new ExecuteJob(connect, test);
+    ActorRef deathwatch = actorSystem.actorOf(Props.create(DeathWatch.class));
+    ActorRef operationControl = actorSystem.actorOf(
+            Props.create(OperationController.class, actorSystem, deathwatch, viewContext, connectionSupplier, dataStorageSupplier, hdfsApiSupplier), "operationController-test");
+    expect(hdfsApiSupplier.get(viewContext)).andReturn(Optional.of(hdfsApi));
+    expect(connect.getConnectable()).andReturn(connectionWrapper);
+    expect(connectionWrapper.isOpen()).andReturn(false);
+    expect(connectionWrapper.getConnection()).andReturn(Optional.of(hiveConnection)).anyTimes();
+    expect(dataStorageSupplier.get(viewContext)).andReturn(storage);
+    expect(connectionSupplier.get(viewContext)).andReturn(delegate);
+    expect(storage.load(JobImpl.class, "1")).andReturn(jobImpl).anyTimes();
+    expect(delegate.createStatement(hiveConnection)).andReturn(statement);
+    expect(delegate.execute("select * from test")).andThrow(new SQLException("Syntax error"));
+    expect(statement.getQueryLog()).andReturn(Lists.<String>newArrayList());
+    jobImpl.setStatus(Job.JOB_STATE_RUNNING);
+    storage.store(JobImpl.class, jobImpl);
+    connectionWrapper.connect();
+    jobImpl.setStatus(Job.JOB_STATE_ERROR);
+    storage.store(JobImpl.class, jobImpl);
+    replay(connect, hdfsApiSupplier, dataStorageSupplier, connectionWrapper,
+            storage, jobImpl, connectionSupplier, delegate, statement, resultSet);
+
+    operationControl.tell(executeJob, ActorRef.noSender());
+    Thread.sleep(5000);
+    verify(connect, hdfsApiSupplier, dataStorageSupplier, connectionWrapper,
+            storage, jobImpl, connectionSupplier, delegate, statement, resultSet);
+  }
+
+
+}

contrib/views/hive-next/src/test/java/org/apache/ambari/view/hive2/HiveJdbcConnectionDelegateTest.java

@@ -0,0 +1,105 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.ambari.view.hive2;
+
+import com.google.common.base.Optional;
+import org.apache.ambari.view.hive2.actor.message.GetColumnMetadataJob;
+import org.apache.hive.jdbc.HiveConnection;
+import org.apache.hive.jdbc.HiveQueryResultSet;
+import org.apache.hive.jdbc.HiveStatement;
+import org.junit.Test;
+
+import java.sql.DatabaseMetaData;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+
+import static org.easymock.EasyMock.*;
+import static org.junit.Assert.assertEquals;
+
+public class HiveJdbcConnectionDelegateTest {
+
+
+  @Test
+  public void testCreateStatement() throws SQLException {
+    HiveConnection hiveConnection = createNiceMock(HiveConnection.class);
+    HiveStatement hiveStatement = createNiceMock(HiveStatement.class);
+    expect(hiveConnection.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY)).andReturn(hiveStatement);
+    replay(hiveConnection);
+    HiveStatement statement = new HiveJdbcConnectionDelegate().createStatement(hiveConnection);
+    assertEquals(hiveStatement, statement);
+
+  }
+
+
+  @Test
+  public void testExecute() throws SQLException {
+    HiveConnection hiveConnection = createNiceMock(HiveConnection.class);
+    HiveStatement hiveStatement = createNiceMock(HiveStatement.class);
+    HiveQueryResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+    expect(hiveConnection.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY)).andReturn(hiveStatement);
+    String query = "select * from test";
+    expect(hiveStatement.execute(query)).andReturn(true);
+    expect(hiveStatement.getResultSet()).andReturn(resultSet);
+    replay(hiveConnection, hiveStatement, resultSet);
+    HiveJdbcConnectionDelegate hiveJdbcConnectionDelegate = new HiveJdbcConnectionDelegate();
+    Optional<ResultSet> execute = hiveJdbcConnectionDelegate.execute(hiveConnection, query);
+    assertEquals(execute.get(), resultSet);
+    verify(hiveConnection, hiveStatement, resultSet);
+
+  }
+
+
+  @Test
+  public void testGetColumnMetaData() throws SQLException {
+
+    HiveConnection hiveConnection = createNiceMock(HiveConnection.class);
+    DatabaseMetaData metadata = createNiceMock(DatabaseMetaData.class);
+    expect(hiveConnection.getMetaData()).andReturn(metadata);
+    ResultSet resultSet = createNiceMock(ResultSet.class);
+    expect(metadata.getColumns(anyString(), anyString(), anyString(), anyString())).andReturn(resultSet);
+    replay(hiveConnection, metadata, resultSet);
+    HiveJdbcConnectionDelegate hiveJdbcConnectionDelegate = new HiveJdbcConnectionDelegate();
+    ResultSet columnMetadata = hiveJdbcConnectionDelegate.getColumnMetadata(hiveConnection, new GetColumnMetadataJob("", "", "", ""));
+    assertEquals(resultSet, columnMetadata);
+    verify(hiveConnection, metadata, resultSet);
+  }
+
+
+  @Test
+  public void testCancel() throws SQLException {
+    HiveConnection hiveConnection = createNiceMock(HiveConnection.class);
+    HiveStatement hiveStatement = createNiceMock(HiveStatement.class);
+    HiveQueryResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+    expect(hiveConnection.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY)).andReturn(hiveStatement);
+    String query = "select * from test";
+    expect(hiveStatement.execute(query)).andReturn(true);
+    expect(hiveStatement.getResultSet()).andReturn(resultSet);
+    hiveStatement.cancel();
+    resultSet.close();
+    replay(hiveConnection, hiveStatement, resultSet);
+    HiveJdbcConnectionDelegate hiveJdbcConnectionDelegate = new HiveJdbcConnectionDelegate();
+    hiveJdbcConnectionDelegate.execute(hiveConnection, query);
+    hiveJdbcConnectionDelegate.cancel();
+    hiveJdbcConnectionDelegate.closeResultSet();
+    hiveJdbcConnectionDelegate.closeStatement();
+    verify(hiveConnection, hiveStatement, resultSet);
+  }
+
+}

contrib/views/hive-next/src/test/java/org/apache/ambari/view/hive2/JobExecutionTest.java

@@ -0,0 +1,115 @@
+/*
+ * 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.ambari.view.hive2;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import com.beust.jcommander.internal.Lists;
+import com.google.common.base.Optional;
+import org.apache.ambari.view.ViewContext;
+import org.apache.ambari.view.hive2.ConnectionDelegate;
+import org.apache.ambari.view.hive2.actor.DeathWatch;
+import org.apache.ambari.view.hive2.actor.OperationController;
+import org.apache.ambari.view.hive2.actor.message.Connect;
+import org.apache.ambari.view.hive2.actor.message.ExecuteJob;
+import org.apache.ambari.view.hive2.actor.message.HiveJob;
+import org.apache.ambari.view.hive2.actor.message.SQLStatementJob;
+import org.apache.ambari.view.hive2.internal.ConnectionSupplier;
+import org.apache.ambari.view.hive2.internal.DataStorageSupplier;
+import org.apache.ambari.view.hive2.internal.HdfsApiSupplier;
+import org.apache.ambari.view.hive2.internal.HiveConnectionWrapper;
+import org.apache.ambari.view.hive2.persistence.Storage;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.Job;
+import org.apache.ambari.view.hive2.resources.jobs.viewJobs.JobImpl;
+import org.apache.ambari.view.utils.hdfs.HdfsApi;
+import org.apache.hive.jdbc.HiveConnection;
+import org.apache.hive.jdbc.HiveQueryResultSet;
+import org.apache.hive.jdbc.HiveStatement;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.ResultSet;
+
+import static org.easymock.EasyMock.*;
+import static org.junit.Assert.*;
+
+public class JobExecutionTest {
+
+  private ActorSystem actorSystem;
+
+  @Before
+  public void setUp() throws Exception {
+    actorSystem = ActorSystem.create("TestingActorSystem");
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    JavaTestKit.shutdownActorSystem(actorSystem);
+  }
+
+  @Test
+  public void testExecuteJob() throws Exception {
+    ViewContext viewContext = createNiceMock(ViewContext.class);
+    ConnectionSupplier connectionSupplier = createNiceMock(ConnectionSupplier.class);
+    DataStorageSupplier dataStorageSupplier = createNiceMock(DataStorageSupplier.class);
+    HdfsApi hdfsApi = createNiceMock(HdfsApi.class);
+    HdfsApiSupplier hdfsApiSupplier = createNiceMock(HdfsApiSupplier.class);
+    Connect connect = createNiceMock(Connect.class);
+    Storage storage = createNiceMock(Storage.class);
+    JobImpl jobImpl = createNiceMock(JobImpl.class);
+    ResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+    HiveStatement statement = createNiceMock(HiveStatement.class);
+    ConnectionDelegate delegate = createNiceMock(ConnectionDelegate.class);
+    HiveConnectionWrapper connectionWrapper = createNiceMock(HiveConnectionWrapper.class);
+    HiveConnection hiveConnection = createNiceMock(HiveConnection.class);
+    HiveJob test = new SQLStatementJob(HiveJob.Type.ASYNC, new String[]{"select * from test"}, "test", "1", "test.log");
+    ExecuteJob executeJob = new ExecuteJob(connect, test);
+    ActorRef deathwatch = actorSystem.actorOf(Props.create(DeathWatch.class));
+    ActorRef operationControl = actorSystem.actorOf(
+            Props.create(OperationController.class, actorSystem, deathwatch, viewContext, connectionSupplier, dataStorageSupplier, hdfsApiSupplier), "operationController-test");
+    expect(hdfsApiSupplier.get(viewContext)).andReturn(Optional.of(hdfsApi));
+    expect(connect.getConnectable()).andReturn(connectionWrapper);
+    expect(connectionWrapper.isOpen()).andReturn(false);
+    expect(connectionWrapper.getConnection()).andReturn(Optional.of(hiveConnection)).anyTimes();
+    expect(dataStorageSupplier.get(viewContext)).andReturn(storage);
+    expect(connectionSupplier.get(viewContext)).andReturn(delegate);
+    expect(storage.load(JobImpl.class, "1")).andReturn(jobImpl).anyTimes();
+    expect(delegate.createStatement(hiveConnection)).andReturn(statement);
+    expect(delegate.execute("select * from test")).andReturn(Optional.of(resultSet));
+    expect(statement.getQueryLog()).andReturn(Lists.<String>newArrayList());
+    jobImpl.setStatus(Job.JOB_STATE_RUNNING);
+    storage.store(JobImpl.class, jobImpl);
+    connectionWrapper.connect();
+    jobImpl.setStatus(Job.JOB_STATE_FINISHED);
+    storage.store(JobImpl.class, jobImpl);
+    replay(connect, hdfsApiSupplier, dataStorageSupplier, connectionWrapper,
+            storage, jobImpl, connectionSupplier, delegate, statement, resultSet);
+
+    operationControl.tell(executeJob, ActorRef.noSender());
+    Thread.sleep(5000);
+    verify(connect, hdfsApiSupplier, dataStorageSupplier, connectionWrapper,
+            storage, jobImpl, connectionSupplier, delegate, statement, resultSet);
+
+  }
+
+
+}

contrib/views/hive-next/src/test/java/org/apache/ambari/view/hive2/PropertyValidatorTest.java

@@ -0,0 +1,113 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.ambari.view.hive2;
+
+import org.apache.ambari.view.ViewInstanceDefinition;
+import org.apache.ambari.view.validation.Validator;
+import org.easymock.EasyMock;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.replay;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class PropertyValidatorTest {
+
+  @Test
+  public void testValidatePropertyWebHDFSCom() throws Exception {
+    PropertyValidator validator = new PropertyValidator();
+    ViewInstanceDefinition definition = getViewInstanceDefinition();
+
+    definition.getPropertyMap().put(PropertyValidator.WEBHDFS_URL, "hdfs://hostname.com:8020");
+
+    assertTrue(validator.validateProperty(PropertyValidator.WEBHDFS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+
+    definition.getPropertyMap().put(PropertyValidator.WEBHDFS_URL, "webhdfs://hostname.com:50070");
+
+    assertTrue(validator.validateProperty(PropertyValidator.WEBHDFS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+
+    definition.getPropertyMap().put(PropertyValidator.WEBHDFS_URL, "http://hostname.com:50070");
+
+    assertFalse(validator.validateProperty(PropertyValidator.WEBHDFS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+  }
+
+  @Test
+  public void testValidatePropertyWebHDFSInternal() throws Exception {
+    PropertyValidator validator = new PropertyValidator();
+    ViewInstanceDefinition definition = getViewInstanceDefinition();
+
+    definition.getPropertyMap().put(PropertyValidator.WEBHDFS_URL, "hdfs://hostname.internal:8020");
+
+    assertTrue(validator.validateProperty(PropertyValidator.WEBHDFS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+
+    definition.getPropertyMap().put(PropertyValidator.WEBHDFS_URL, "webhdfs://hostname.internal:50070");
+
+    assertTrue(validator.validateProperty(PropertyValidator.WEBHDFS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+
+    definition.getPropertyMap().put(PropertyValidator.WEBHDFS_URL, "swebhdfs://hostname.internal:50070");
+
+    assertTrue(validator.validateProperty(PropertyValidator.WEBHDFS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+
+    definition.getPropertyMap().put(PropertyValidator.WEBHDFS_URL, "http://hostname.internal:50070");
+
+    assertFalse(validator.validateProperty(PropertyValidator.WEBHDFS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+  }
+
+  @Test
+  public void testValidatePropertyATSCom() throws Exception {
+    PropertyValidator validator = new PropertyValidator();
+    ViewInstanceDefinition definition = getViewInstanceDefinition();
+
+    definition.getPropertyMap().put(PropertyValidator.YARN_ATS_URL, "http://hostname.com:8088");
+
+    assertTrue(validator.validateProperty(PropertyValidator.YARN_ATS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+  }
+
+  @Test
+  public void testValidatePropertyATSInternal() throws Exception {
+    PropertyValidator validator = new PropertyValidator();
+    ViewInstanceDefinition definition = getViewInstanceDefinition();
+
+    definition.getPropertyMap().put(PropertyValidator.YARN_ATS_URL, "http://hostname.internal:8088");
+
+    assertTrue(validator.validateProperty(PropertyValidator.YARN_ATS_URL,
+            definition, Validator.ValidationContext.PRE_CREATE).isValid());
+  }
+
+  private ViewInstanceDefinition getViewInstanceDefinition() {
+    ViewInstanceDefinition definition = EasyMock.createNiceMock(ViewInstanceDefinition.class);
+    expect(definition.getClusterHandle()).andReturn(null).anyTimes();
+    Map<String, String> properties = new HashMap<String, String>();
+    expect(definition.getPropertyMap()).andReturn(properties).anyTimes();
+    replay(definition);
+    return definition;
+  }
+}

contrib/views/hive-next/src/test/java/org/apache/ambari/view/hive2/ResultSetIteratorTest.java

@@ -0,0 +1,100 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.ambari.view.hive2;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.testkit.JavaTestKit;
+import org.apache.ambari.view.hive2.actor.HiveActor;
+import org.apache.ambari.view.hive2.actor.ResultSetIterator;
+import org.apache.ambari.view.hive2.actor.message.HiveMessage;
+import org.apache.ambari.view.hive2.actor.message.ResetCursor;
+import org.apache.ambari.view.hive2.actor.message.job.Next;
+import org.apache.hive.jdbc.HiveQueryResultSet;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+
+import static org.easymock.EasyMock.*;
+
+
+public class ResultSetIteratorTest {
+
+  private ActorSystem actorSystem;
+
+  @Before
+  public void setUp() throws Exception {
+    actorSystem = ActorSystem.create("TestingActorSystem");
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    JavaTestKit.shutdownActorSystem(actorSystem);
+  }
+
+  @Test
+  public void testGetNext() throws Exception {
+    ResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+    ResultSetMetaData resultSetMetaData = createNiceMock(ResultSetMetaData.class);
+
+    ActorRef parent = actorSystem.actorOf(
+            Props.create(TestParent.class));
+    ActorRef rsi = actorSystem.actorOf(
+            Props.create(ResultSetIterator.class, parent, resultSet));
+    expect(resultSet.getMetaData()).andReturn(resultSetMetaData);
+    expect(resultSetMetaData.getColumnCount()).andReturn(1);
+    expect(resultSetMetaData.getColumnName(1)).andReturn("test");
+    expect(resultSetMetaData.getColumnTypeName(1)).andReturn("string");
+    replay(resultSet, resultSetMetaData);
+    rsi.tell(new Next(), parent);
+    Thread.sleep(2000);
+    verify(resultSet, resultSetMetaData);
+
+  }
+
+  @Test
+  public void testResetCursor() throws Exception {
+    ResultSet resultSet = createNiceMock(HiveQueryResultSet.class);
+
+    ActorRef parent = actorSystem.actorOf(
+            Props.create(TestParent.class));
+    ActorRef rsi = actorSystem.actorOf(
+            Props.create(ResultSetIterator.class, parent, resultSet));
+    resultSet.beforeFirst();
+    replay(resultSet);
+    rsi.tell(new ResetCursor(), parent);
+    Thread.sleep(2000);
+    verify(resultSet);
+
+  }
+
+
+  private static class TestParent extends HiveActor {
+
+    @Override
+    public void handleMessage(HiveMessage hiveMessage) {
+
+    }
+  }
+
+
+}