Repository: apache/airflow
Pull request: Remove Hive/Hadoop/Java dependency from unit tests
CI takes way too long to build and uses too much memory. This removes Hive and Hadoop as dependencies from CI, and replaces direct access with mocks in unit tests.

Fixes #8783 (though differently to proposed, via mocking instead of separate docker image)
---
Make sure to mark the boxes below before creating PR: [x]

- [x] Description above provides context of the change
- [x] Unit tests coverage for changes (not needed for documentation changes)
- [x] Target Github ISSUE in description if exists
- [x] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [x] Relevant documentation is updated including usage instructions.
- [x] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).

---
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.


Problem statement:
Move Hadoop distro out of CI image
**Description**

Hadoop distro is unnecessarily part of the CI image. It should be moved out and used separately

**Use case / motivation**

Download and install Hadoop image only when needed not as part of the CI image (as separate integration). This will make the image smaller and the whole process smoother.

Hints from issue discussion:

Commit message: Add back Java to dockerfile.

Patch to review:
diff --git a/Dockerfile.ci b/Dockerfile.ci
index 6ab3c1fa876fd..b16b4242e9775 100644
--- a/Dockerfile.ci
+++ b/Dockerfile.ci
@@ -41,7 +41,7 @@ ENV DEBIAN_FRONTEND=noninteractive LANGUAGE=C.UTF-8 LANG=C.UTF-8 LC_ALL=C.UTF-8
     LC_CTYPE=C.UTF-8 LC_MESSAGES=C.UTF-8
 
 # By increasing this number we can do force build of all dependencies
-ARG DEPENDENCIES_EPOCH_NUMBER="3"
+ARG DEPENDENCIES_EPOCH_NUMBER="4"
 # Increase the value below to force renstalling of all dependencies
 ENV DEPENDENCIES_EPOCH_NUMBER=${DEPENDENCIES_EPOCH_NUMBER}
 
@@ -156,45 +156,45 @@ ENV JAVA_HOME=/usr/lib/jvm/adoptopenjdk-8-hotspot-jre-amd64
 
 # Install Hadoop and Hive
 # It is done in one step to share variables.
-ENV HADOOP_HOME="/opt/hadoop-cdh" HIVE_HOME="/opt/hive"
-
-RUN HADOOP_DISTRO="cdh" \
-    && HADOOP_MAJOR="5" \
-    && HADOOP_DISTRO_VERSION="5.11.0" \
-    && HADOOP_VERSION="2.6.0" \
-    && HADOOP_URL="https://archive.cloudera.com/${HADOOP_DISTRO}${HADOOP_MAJOR}/${HADOOP_DISTRO}/${HADOOP_MAJOR}/"\
-    && HADOOP_DOWNLOAD_URL="${HADOOP_URL}hadoop-${HADOOP_VERSION}-${HADOOP_DISTRO}${HADOOP_DISTRO_VERSION}.tar.gz" \
-    && HADOOP_TMP_FILE="/tmp/hadoop.tar.gz" \
-    && mkdir -pv "${HADOOP_HOME}" \
-    && curl --fail --location "${HADOOP_DOWNLOAD_URL}" --output "${HADOOP_TMP_FILE}" \
-    && tar xzf "${HADOOP_TMP_FILE}" --absolute-names --strip-components 1 -C "${HADOOP_HOME}" \
-    && rm "${HADOOP_TMP_FILE}" \
-    && echo "Installing Hive" \
-    && HIVE_VERSION="1.1.0" \
-    && HIVE_URL="${HADOOP_URL}hive-${HIVE_VERSION}-${HADOOP_DISTRO}${HADOOP_DISTRO_VERSION}.tar.gz" \
-    && HIVE_VERSION="1.1.0" \
-    && HIVE_TMP_FILE="/tmp/hive.tar.gz" \
-    && mkdir -pv "${HIVE_HOME}" \
-    && mkdir -pv "/user/hive/warehouse" \
-    && chmod -R 777 "${HIVE_HOME}" \
-    && chmod -R 777 "/user/" \
-    && curl --fail --location  "${HIVE_URL}" --output "${HIVE_TMP_FILE}" \
-    && tar xzf "${HIVE_TMP_FILE}" --strip-components 1 -C "${HIVE_HOME}" \
-    && rm "${HIVE_TMP_FILE}"
-
-ENV PATH "${PATH}:/opt/hive/bin"
+# ENV HADOOP_HOME="/opt/hadoop-cdh" HIVE_HOME="/opt/hive"
+
+# RUN HADOOP_DISTRO="cdh" \
+#     && HADOOP_MAJOR="5" \
+#     && HADOOP_DISTRO_VERSION="5.11.0" \
+#     && HADOOP_VERSION="2.6.0" \
+#     && HADOOP_URL="https://archive.cloudera.com/${HADOOP_DISTRO}${HADOOP_MAJOR}/${HADOOP_DISTRO}/${HADOOP_MAJOR}/"\
+#     && HADOOP_DOWNLOAD_URL="${HADOOP_URL}hadoop-${HADOOP_VERSION}-${HADOOP_DISTRO}${HADOOP_DISTRO_VERSION}.tar.gz" \
+#     && HADOOP_TMP_FILE="/tmp/hadoop.tar.gz" \
+#     && mkdir -pv "${HADOOP_HOME}" \
+#     && curl --fail --location "${HADOOP_DOWNLOAD_URL}" --output "${HADOOP_TMP_FILE}" \
+#     && tar xzf "${HADOOP_TMP_FILE}" --absolute-names --strip-components 1 -C "${HADOOP_HOME}" \
+#     && rm "${HADOOP_TMP_FILE}" \
+#     && echo "Installing Hive" \
+#     && HIVE_VERSION="1.1.0" \
+#     && HIVE_URL="${HADOOP_URL}hive-${HIVE_VERSION}-${HADOOP_DISTRO}${HADOOP_DISTRO_VERSION}.tar.gz" \
+#     && HIVE_VERSION="1.1.0" \
+#     && HIVE_TMP_FILE="/tmp/hive.tar.gz" \
+#     && mkdir -pv "${HIVE_HOME}" \
+#     && mkdir -pv "/user/hive/warehouse" \
+#     && chmod -R 777 "${HIVE_HOME}" \
+#     && chmod -R 777 "/user/" \
+#     && curl --fail --location  "${HIVE_URL}" --output "${HIVE_TMP_FILE}" \
+#     && tar xzf "${HIVE_TMP_FILE}" --strip-components 1 -C "${HIVE_HOME}" \
+#     && rm "${HIVE_TMP_FILE}"
+
+# ENV PATH "${PATH}:/opt/hive/bin"
 
 # Install Minicluster
-ENV MINICLUSTER_HOME="/opt/minicluster"
-
-RUN MINICLUSTER_BASE="https://github.com/bolkedebruin/minicluster/releases/download/" \
-    && MINICLUSTER_VER="1.1" \
-    && MINICLUSTER_URL="${MINICLUSTER_BASE}${MINICLUSTER_VER}/minicluster-${MINICLUSTER_VER}-SNAPSHOT-bin.zip" \
-    && MINICLUSTER_TMP_FILE="/tmp/minicluster.zip" \
-    && mkdir -pv "${MINICLUSTER_HOME}" \
-    && curl --fail --location "${MINICLUSTER_URL}" --output "${MINICLUSTER_TMP_FILE}" \
-    && unzip "${MINICLUSTER_TMP_FILE}" -d "/opt" \
-    && rm "${MINICLUSTER_TMP_FILE}"
+# ENV MINICLUSTER_HOME="/opt/minicluster"
+
+# RUN MINICLUSTER_BASE="https://github.com/bolkedebruin/minicluster/releases/download/" \
+#     && MINICLUSTER_VER="1.1" \
+#     && MINICLUSTER_URL="${MINICLUSTER_BASE}${MINICLUSTER_VER}/minicluster-${MINICLUSTER_VER}-SNAPSHOT-bin.zip" \
+#     && MINICLUSTER_TMP_FILE="/tmp/minicluster.zip" \
+#     && mkdir -pv "${MINICLUSTER_HOME}" \
+#     && curl --fail --location "${MINICLUSTER_URL}" --output "${MINICLUSTER_TMP_FILE}" \
+#     && unzip "${MINICLUSTER_TMP_FILE}" -d "/opt" \
+#     && rm "${MINICLUSTER_TMP_FILE}"
 
 # Install Docker
 RUN curl --fail --location https://download.docker.com/linux/debian/gpg | apt-key add - \
diff --git a/airflow/providers/apache/hive/operators/hive.py b/airflow/providers/apache/hive/operators/hive.py
index f6d06dda87c1c..97c28fcdf1ccd 100644
--- a/airflow/providers/apache/hive/operators/hive.py
+++ b/airflow/providers/apache/hive/operators/hive.py
@@ -16,6 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import os
 import re
 from typing import Dict, Optional
 
@@ -23,7 +24,7 @@
 from airflow.models import BaseOperator
 from airflow.providers.apache.hive.hooks.hive import HiveCliHook
 from airflow.utils.decorators import apply_defaults
-from airflow.utils.operator_helpers import context_to_airflow_vars
+from airflow.utils.operator_helpers import context_to_airflow_vars, get_blank_airflow_vars
 
 
 class HiveOperator(BaseOperator):
@@ -143,6 +144,9 @@ def execute(self, context):
         self.hook.run_cli(hql=self.hql, schema=self.schema, hive_conf=self.hiveconfs)
 
     def dry_run(self):
+        blank_env_vars = get_blank_airflow_vars()
+        os.environ.update(blank_env_vars)
+
         self.hook = self.get_hook()
         self.hook.test_hql(hql=self.hql)
 
diff --git a/airflow/providers/apache/hive/sensors/hive_partition.py b/airflow/providers/apache/hive/sensors/hive_partition.py
index b531876decb94..f705039b1f57c 100644
--- a/airflow/providers/apache/hive/sensors/hive_partition.py
+++ b/airflow/providers/apache/hive/sensors/hive_partition.py
@@ -16,6 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from airflow.providers.apache.hive.hooks.hive import HiveMetastoreHook
 from airflow.sensors.base_sensor_operator import BaseSensorOperator
 from airflow.utils.decorators import apply_defaults
 
@@ -67,7 +68,6 @@ def poke(self, context):
             'Poking for table %s.%s, partition %s', self.schema, self.table, self.partition
         )
         if not hasattr(self, 'hook'):
-            from airflow.providers.apache.hive.hooks.hive import HiveMetastoreHook
             hook = HiveMetastoreHook(
                 metastore_conn_id=self.metastore_conn_id)
         return hook.check_for_partition(
diff --git a/airflow/utils/operator_helpers.py b/airflow/utils/operator_helpers.py
index 5909fa40c28af..fbd976aeca108 100644
--- a/airflow/utils/operator_helpers.py
+++ b/airflow/utils/operator_helpers.py
@@ -84,3 +84,13 @@ def context_to_airflow_vars(context, in_env_var_format=False):
         params[AIRFLOW_VAR_NAME_FORMAT_MAPPING['AIRFLOW_CONTEXT_DAG_RUN_ID'][
             name_format]] = dag_run.run_id
     return params
+
+
+def get_blank_airflow_vars():
+    """
+    Returns a dictionary that maps Airflow env var names
+    to blank values.
+
+    :return: task_instance context as dict.
+    """
+    return {value['env_var_format']: '' for value in AIRFLOW_VAR_NAME_FORMAT_MAPPING.values()}
diff --git a/tests/providers/apache/hive/hooks/test_hive.py b/tests/providers/apache/hive/hooks/test_hive.py
index 9ed518ff02d40..5f56d0abf9744 100644
--- a/tests/providers/apache/hive/hooks/test_hive.py
+++ b/tests/providers/apache/hive/hooks/test_hive.py
@@ -20,9 +20,8 @@
 import datetime
 import itertools
 import os
-import random
 import unittest
-from collections import OrderedDict
+from collections import OrderedDict, namedtuple
 from unittest import mock
 
 import pandas as pd
@@ -31,12 +30,13 @@
 from airflow.exceptions import AirflowException
 from airflow.models.connection import Connection
 from airflow.models.dag import DAG
-from airflow.providers.apache.hive.hooks.hive import HiveCliHook, HiveMetastoreHook, HiveServer2Hook
-from airflow.providers.apache.hive.operators.hive import HiveOperator
+from airflow.providers.apache.hive.hooks.hive import HiveMetastoreHook, HiveServer2Hook
 from airflow.secrets.environment_variables import CONN_ENV_PREFIX
 from airflow.utils import timezone
 from airflow.utils.operator_helpers import AIRFLOW_VAR_NAME_FORMAT_MAPPING
 from tests.test_utils.asserts import assert_equal_ignore_multiple_spaces
+from tests.test_utils.mock_hooks import MockHiveCliHook, MockHiveServer2Hook
+from tests.test_utils.mock_process import MockSubProcess
 
 DEFAULT_DATE = timezone.datetime(2015, 1, 1)
 DEFAULT_DATE_ISO = DEFAULT_DATE.isoformat()
@@ -46,53 +46,58 @@
 class TestHiveEnvironment(unittest.TestCase):
 
     def setUp(self):
-        args = {'owner': 'airflow', 'start_date': DEFAULT_DATE}
-        self.dag = DAG('test_dag_id', default_args=args)
         self.next_day = (DEFAULT_DATE +
                          datetime.timedelta(days=1)).isoformat()[:10]
         self.database = 'airflow'
         self.partition_by = 'ds'
         self.table = 'static_babynames_partitioned'
-        self.hql = """
-        CREATE DATABASE IF NOT EXISTS {{ params.database }};
-        USE {{ params.database }};
-        DROP TABLE IF EXISTS {{ params.table }};
-        CREATE TABLE IF NOT EXISTS {{ params.table }} (
-            state string,
-            year string,
-            name string,
-            gender string,
-            num int)
-        PARTITIONED BY ({{ params.partition_by }} string);
-        ALTER TABLE {{ params.table }}
-        ADD PARTITION({{ params.partition_by }}='{{ ds }}');
-        """
-        self.hook = HiveMetastoreHook()
-        op = HiveOperator(
-            task_id='HiveHook_' + str(random.randint(1, 10000)),
-            params={
-                'database': self.database,
-                'table': self.table,
-                'partition_by': self.partition_by
-            },
-            hive_cli_conn_id='hive_cli_default',
-            hql=self.hql, dag=self.dag)
-        op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
-               ignore_ti_state=True)
+        with mock.patch('airflow.providers.apache.hive.hooks.hive.HiveMetastoreHook.get_metastore_client'
+                        ) as get_metastore_mock:
+            get_metastore_mock.return_value = mock.MagicMock()
 
-    def tearDown(self):
-        hook = HiveMetastoreHook()
-        with hook.get_conn() as metastore:
-            metastore.drop_table(self.database, self.table, deleteData=True)
+            self.hook = HiveMetastoreHook()
 
 
 class TestHiveCliHook(unittest.TestCase):
+    @mock.patch('tempfile.tempdir', '/tmp/')
+    @mock.patch('tempfile._RandomNameSequence.__next__')
+    @mock.patch('subprocess.Popen')
+    def test_run_cli(self, mock_popen, mock_temp_dir):
+        mock_subprocess = MockSubProcess()
+        mock_popen.return_value = mock_subprocess
+        mock_temp_dir.return_value = "test_run_cli"
 
-    def test_run_cli(self):
-        hook = HiveCliHook()
-        hook.run_cli("SHOW DATABASES")
+        with mock.patch.dict('os.environ', {
+            'AIRFLOW_CTX_DAG_ID': 'test_dag_id',
+            'AIRFLOW_CTX_TASK_ID': 'test_task_id',
+            'AIRFLOW_CTX_EXECUTION_DATE': '2015-01-01T00:00:00+00:00',
+            'AIRFLOW_CTX_DAG_RUN_ID': '55',
+            'AIRFLOW_CTX_DAG_OWNER': 'airflow',
+            'AIRFLOW_CTX_DAG_EMAIL': 'test@airflow.com',
+        }):
 
-    def test_run_cli_with_hive_conf(self):
+            hook = MockHiveCliHook()
+            hook.run_cli("SHOW DATABASES")
+
+        hive_cmd = ['beeline', '-u', '"jdbc:hive2://localhost:10000/default"', '-hiveconf',
+                    'airflow.ctx.dag_id=test_dag_id', '-hiveconf', 'airflow.ctx.task_id=test_task_id',
+                    '-hiveconf', 'airflow.ctx.execution_date=2015-01-01T00:00:00+00:00', '-hiveconf',
+                    'airflow.ctx.dag_run_id=55', '-hiveconf', 'airflow.ctx.dag_owner=airflow',
+                    '-hiveconf', 'airflow.ctx.dag_email=test@airflow.com', '-hiveconf',
+                    'mapreduce.job.queuename=airflow', '-hiveconf', 'mapred.job.queue.name=airflow',
+                    '-hiveconf', 'tez.queue.name=airflow', '-f',
+                    '/tmp/airflow_hiveop_test_run_cli/tmptest_run_cli']
+
+        mock_popen.assert_called_with(
+            hive_cmd,
+            stdout=mock_subprocess.PIPE,
+            stderr=mock_subprocess.STDOUT,
+            cwd="/tmp/airflow_hiveop_test_run_cli",
+            close_fds=True
+        )
+
+    @mock.patch('subprocess.Popen')
+    def test_run_cli_with_hive_conf(self, mock_popen):
         hql = "set key;\n" \
               "set airflow.ctx.dag_id;\nset airflow.ctx.dag_run_id;\n" \
               "set airflow.ctx.task_id;\nset airflow.ctx.execution_date;\n"
@@ -107,14 +112,75 @@ def test_run_cli_with_hive_conf(self):
         dag_run_id_ctx_var_name = \
             AIRFLOW_VAR_NAME_FORMAT_MAPPING['AIRFLOW_CONTEXT_DAG_RUN_ID'][
                 'env_var_format']
+
+        mock_output = ['Connecting to jdbc:hive2://localhost:10000/default',
+                       'log4j:WARN No appenders could be found for logger (org.apache.hive.jdbc.Utils).',
+                       'log4j:WARN Please initialize the log4j system properly.',
+                       'log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.',
+                       'Connected to: Apache Hive (version 1.2.1.2.3.2.0-2950)',
+                       'Driver: Hive JDBC (version 1.2.1.spark2)',
+                       'Transaction isolation: TRANSACTION_REPEATABLE_READ',
+                       '0: jdbc:hive2://localhost:10000/default> USE default;',
+                       'No rows affected (0.37 seconds)',
+                       '0: jdbc:hive2://localhost:10000/default> set key;',
+                       '+------------+--+',
+                       '|    set     |',
+                       '+------------+--+',
+                       '| key=value  |',
+                       '+------------+--+',
+                       '1 row selected (0.133 seconds)',
+                       '0: jdbc:hive2://localhost:10000/default> set airflow.ctx.dag_id;',
+                       '+---------------------------------+--+',
+                       '|               set               |',
+                       '+---------------------------------+--+',
+                       '| airflow.ctx.dag_id=test_dag_id  |',
+                       '+---------------------------------+--+',
+                       '1 row selected (0.008 seconds)',
+                       '0: jdbc:hive2://localhost:10000/default> set airflow.ctx.dag_run_id;',
+                       '+-----------------------------------------+--+',
+                       '|                   set                   |',
+                       '+-----------------------------------------+--+',
+                       '| airflow.ctx.dag_run_id=test_dag_run_id  |',
+                       '+-----------------------------------------+--+',
+                       '1 row selected (0.007 seconds)',
+                       '0: jdbc:hive2://localhost:10000/default> set airflow.ctx.task_id;',
+                       '+-----------------------------------+--+',
+                       '|                set                |',
+                       '+-----------------------------------+--+',
+                       '| airflow.ctx.task_id=test_task_id  |',
+                       '+-----------------------------------+--+',
+                       '1 row selected (0.009 seconds)',
+                       '0: jdbc:hive2://localhost:10000/default> set airflow.ctx.execution_date;',
+                       '+-------------------------------------------------+--+',
+                       '|                       set                       |',
+                       '+-------------------------------------------------+--+',
+                       '| airflow.ctx.execution_date=test_execution_date  |',
+                       '+-------------------------------------------------+--+',
+                       '1 row selected (0.006 seconds)',
+                       '0: jdbc:hive2://localhost:10000/default> ',
+                       '0: jdbc:hive2://localhost:10000/default> ',
+                       'Closing: 0: jdbc:hive2://localhost:10000/default',
+                       '']
+
         with mock.patch.dict('os.environ', {
             dag_id_ctx_var_name: 'test_dag_id',
             task_id_ctx_var_name: 'test_task_id',
             execution_date_ctx_var_name: 'test_execution_date',
             dag_run_id_ctx_var_name: 'test_dag_run_id',
         }):
-            hook = HiveCliHook()
+
+            hook = MockHiveCliHook()
+            mock_popen.return_value = MockSubProcess(output=mock_output)
+
             output = hook.run_cli(hql=hql, hive_conf={'key': 'value'})
+            process_inputs = " ".join(mock_popen.call_args_list[0][0][0])
+
+            self.assertIn('value', process_inputs)
+            self.assertIn('test_dag_id', process_inputs)
+            self.assertIn('test_task_id', process_inputs)
+            self.assertIn('test_execution_date', process_inputs)
+            self.assertIn('test_dag_run_id', process_inputs)
+
             self.assertIn('value', output)
             self.assertIn('test_dag_id', output)
             self.assertIn('test_task_id', output)
@@ -126,7 +192,7 @@ def test_load_file_without_create_table(self, mock_run_cli):
         filepath = "/path/to/input/file"
         table = "output_table"
 
-        hook = HiveCliHook()
+        hook = MockHiveCliHook()
         hook.load_file(filepath=filepath, table=table, create=False)
 
         query = (
@@ -144,10 +210,12 @@ def test_load_file_create_table(self, mock_run_cli):
         filepath = "/path/to/input/file"
         table = "output_table"
         field_dict = OrderedDict([("name", "string"), ("gender", "string")])
-        fields = ",\n    ".join(['`{k}` {v}'.format(k=k.strip('`'), v=v) for k, v in field_dict.items()])
+        fields = ",\n    ".join(
+            ['`{k}` {v}'.format(k=k.strip('`'), v=v) for k, v in field_dict.items()])
 
-        hook = HiveCliHook()
-        hook.load_file(filepath=filepath, table=table, field_dict=field_dict, create=True, recreate=True)
+        hook = MockHiveCliHook()
+        hook.load_file(filepath=filepath, table=table,
+                       field_dict=field_dict, create=True, recreate=True)
 
         create_table = (
             "DROP TABLE IF EXISTS {table};\n"
@@ -176,7 +244,7 @@ def test_load_df(self, mock_to_csv, mock_load_file):
         delimiter = ","
         encoding = "utf-8"
 
-        hook = HiveCliHook()
+        hook = MockHiveCliHook()
         hook.load_df(df=df,
                      table=table,
                      delimiter=delimiter,
@@ -198,7 +266,7 @@ def test_load_df(self, mock_to_csv, mock_load_file):
     @mock.patch('airflow.providers.apache.hive.hooks.hive.HiveCliHook.load_file')
     @mock.patch('pandas.DataFrame.to_csv')
     def test_load_df_with_optional_parameters(self, mock_to_csv, mock_load_file):
-        hook = HiveCliHook()
+        hook = MockHiveCliHook()
         bools = (True, False)
         for create, recreate in itertools.product(bools, bools):
             mock_load_file.reset_mock()
@@ -227,7 +295,7 @@ def test_load_df_with_data_types(self, mock_run_cli):
         ord_dict['V'] = [None]
         df = pd.DataFrame(ord_dict)
 
-        hook = HiveCliHook()
+        hook = MockHiveCliHook()
         hook.load_df(df, 't')
 
         query = """
@@ -247,7 +315,8 @@ def test_load_df_with_data_types(self, mock_run_cli):
             STORED AS textfile
             ;
         """
-        assert_equal_ignore_multiple_spaces(self, mock_run_cli.call_args_list[0][0][0], query)
+        assert_equal_ignore_multiple_spaces(
+            self, mock_run_cli.call_args_list[0][0][0], query)
 
 
 class TestHiveMetastoreHook(TestHiveEnvironment):
@@ -260,6 +329,7 @@ def test_get_max_partition_from_empty_part_specs(self):
                                                                  self.VALID_FILTER_MAP)
         self.assertIsNone(max_partition)
 
+    # @mock.patch('airflow.providers.apache.hive.hooks.hive.HiveMetastoreHook', 'get_metastore_client')
     def test_get_max_partition_from_valid_part_specs_and_invalid_filter_map(self):
         with self.assertRaises(AirflowException):
             HiveMetastoreHook._get_max_partition_from_part_specs(
@@ -304,72 +374,150 @@ def test_get_max_partition_from_valid_part_specs(self):
                 self.VALID_FILTER_MAP)
         self.assertEqual(max_partition, b'value1')
 
-    def test_get_metastore_client(self):
-        self.assertIsInstance(self.hook.get_metastore_client(), HMSClient)
-
     @mock.patch("airflow.providers.apache.hive.hooks.hive.HiveMetastoreHook.get_connection",
                 return_value=[Connection(host="localhost", port="9802")])
     @mock.patch("airflow.providers.apache.hive.hooks.hive.socket")
-    def test_error_metastore_client(self, socket_mock, _find_vaild_server_mock):
+    def test_error_metastore_client(self, socket_mock, _find_valid_server_mock):
         socket_mock.socket.return_value.connect_ex.return_value = 0
         self.hook.get_metastore_client()
 
     def test_get_conn(self):
-        self.assertIsInstance(self.hook.get_conn(), HMSClient)
+        with mock.patch('airflow.providers.apache.hive.hooks.hive.HiveMetastoreHook._find_valid_server'
+                        ) as find_valid_server:
+            find_valid_server.return_value = mock.MagicMock(return_value={})
+            metastore_hook = HiveMetastoreHook()
+
+        self.assertIsInstance(metastore_hook.get_conn(), HMSClient)
 
     def test_check_for_partition(self):
+        # Check for existent partition.
+        FakePartition = namedtuple('FakePartition', ['values'])
+        fake_partition = FakePartition(['2015-01-01'])
+
+        metastore = self.hook.metastore.__enter__()
+
         partition = "{p_by}='{date}'".format(date=DEFAULT_DATE_DS,
                                              p_by=self.partition_by)
-        missing_partition = "{p_by}='{date}'".format(date=self.next_day,
-                                                     p_by=self.partition_by)
+
+        metastore.get_partitions_by_filter = mock.MagicMock(
+            return_value=[fake_partition])
+
         self.assertTrue(
             self.hook.check_for_partition(self.database, self.table,
                                           partition)
         )
+
+        metastore.get_partitions_by_filter(
+            self.database, self.table, partition, 1)
+
+        # Check for non-existent partition.
+        missing_partition = "{p_by}='{date}'".format(date=self.next_day,
+                                                     p_by=self.partition_by)
+        metastore.get_partitions_by_filter = mock.MagicMock(return_value=[])
+
         self.assertFalse(
             self.hook.check_for_partition(self.database, self.table,
                                           missing_partition)
         )
 
+        metastore.get_partitions_by_filter.assert_called_with(
+            self.database, self.table, missing_partition, 1)
+
     def test_check_for_named_partition(self):
+
+        # Check for existing partition.
+
         partition = "{p_by}={date}".format(date=DEFAULT_DATE_DS,
                                            p_by=self.partition_by)
-        missing_partition = "{p_by}={date}".format(date=self.next_day,
-                                                   p_by=self.partition_by)
+
+        self.hook.metastore.__enter__(
+        ).check_for_named_partition = mock.MagicMock(return_value=True)
+
         self.assertTrue(
             self.hook.check_for_named_partition(self.database,
                                                 self.table,
-                                                partition)
-        )
+                                                partition))
+
+        self.hook.metastore.__enter__().check_for_named_partition.assert_called_with(
+            self.database, self.table, partition)
+
+        # Check for non-existent partition
+        missing_partition = "{p_by}={date}".format(date=self.next_day,
+                                                   p_by=self.partition_by)
+
+        self.hook.metastore.__enter__().check_for_named_partition = mock.MagicMock(
+            return_value=False)
+
         self.assertFalse(
             self.hook.check_for_named_partition(self.database,
                                                 self.table,
                                                 missing_partition)
         )
+        self.hook.metastore.__enter__().check_for_named_partition.assert_called_with(
+            self.database, self.table, missing_partition)
 
     def test_get_table(self):
-        table_info = self.hook.get_table(db=self.database,
-                                         table_name=self.table)
-        self.assertEqual(table_info.tableName, self.table)
-        columns = ['state', 'year', 'name', 'gender', 'num']
-        self.assertEqual([col.name for col in table_info.sd.cols], columns)
 
-    def test_get_tables(self):
-        tables = self.hook.get_tables(db=self.database,
-                                      pattern=self.table + "*")
-        self.assertIn(self.table, {table.tableName for table in tables})
+        self.hook.metastore.__enter__().get_table = mock.MagicMock()
+        self.hook.get_table(db=self.database, table_name=self.table)
+        self.hook.metastore.__enter__().get_table.assert_called_with(
+            dbname=self.database, tbl_name=self.table)
+
+    def test_get_tables(self):  # static_babynames_partitioned
+        self.hook.metastore.__enter__().get_tables = mock.MagicMock(
+            return_value=['static_babynames_partitioned'])
+
+        self.hook.get_tables(db=self.database, pattern=self.table + "*")
+        self.hook.metastore.__enter__().get_tables.assert_called_with(
+            db_name='airflow', pattern='static_babynames_partitioned*')
+        self.hook.metastore.__enter__().get_table_objects_by_name.assert_called_with(
+            'airflow', ['static_babynames_partitioned'])
 
     def test_get_databases(self):
-        databases = self.hook.get_databases(pattern='*')
-        self.assertIn(self.database, databases)
+        metastore = self.hook.metastore.__enter__()
+        metastore.get_databases = mock.MagicMock()
+
+        self.hook.get_databases(pattern='*')
+        metastore.get_databases.assert_called_with('*')
 
     def test_get_partitions(self):
+        FakeFieldSchema = namedtuple('FakeFieldSchema', ['name'])
+        fake_schema = FakeFieldSchema('ds')
+        FakeTable = namedtuple('FakeTable', ['partitionKeys'])
+        fake_table = FakeTable([fake_schema])
+        FakePartition = namedtuple('FakePartition', ['values'])
+        fake_partition = FakePartition(['2015-01-01'])
+
+        metastore = self.hook.metastore.__enter__()
+        metastore.get_table = mock.MagicMock(return_value=fake_table)
+
+        metastore.get_partitions = mock.MagicMock(
+            return_value=[fake_partition])
+
         partitions = self.hook.get_partitions(schema=self.database,
                                               table_name=self.table)
         self.assertEqual(len(partitions), 1)
         self.assertEqual(partitions, [{self.partition_by: DEFAULT_DATE_DS}])
 
+        metastore.get_table.assert_called_with(
+            dbname=self.database, tbl_name=self.table)
+        metastore.get_partitions.assert_called_with(
+            db_name=self.database, tbl_name=self.table, max_parts=HiveMetastoreHook.MAX_PART_COUNT)
+
     def test_max_partition(self):
+        FakeFieldSchema = namedtuple('FakeFieldSchema', ['name'])
+        fake_schema = FakeFieldSchema('ds')
+        FakeTable = namedtuple('FakeTable', ['partitionKeys'])
+        fake_table = FakeTable([fake_schema])
+
+        metastore = self.hook.metastore.__enter__()
+        metastore.get_table = mock.MagicMock(return_value=fake_table)
+
+        metastore.get_partition_names = mock.MagicMock(
+            return_value=['ds=2015-01-01'])
+        metastore.partition_name_to_spec = mock.MagicMock(
+            return_value={'ds': '2015-01-01'})
+
         filter_map = {self.partition_by: DEFAULT_DATE_DS}
         partition = self.hook.max_partition(schema=self.database,
                                             table_name=self.table,
@@ -377,11 +525,28 @@ def test_max_partition(self):
                                             filter_map=filter_map)
         self.assertEqual(partition, DEFAULT_DATE_DS.encode('utf-8'))
 
+        metastore.get_table.assert_called_with(
+            dbname=self.database, tbl_name=self.table)
+        metastore.get_partition_names.assert_called_with(
+            self.database, self.table, max_parts=HiveMetastoreHook.MAX_PART_COUNT)
+        metastore.partition_name_to_spec.assert_called_with('ds=2015-01-01')
+
     def test_table_exists(self):
+        # Test with existent table.
+        self.hook.metastore.__enter__().get_table = mock.MagicMock(return_value=True)
+
         self.assertTrue(self.hook.table_exists(self.table, db=self.database))
+        self.hook.metastore.__enter__().get_table.assert_called_with(
+            dbname='airflow', tbl_name='static_babynames_partitioned')
+
+        # Test with non-existent table.
+        self.hook.metastore.__enter__().get_table = mock.MagicMock(side_effect=Exception())
+
         self.assertFalse(
-            self.hook.table_exists(str(random.randint(1, 10000)))
+            self.hook.table_exists("does-not-exist")
         )
+        self.hook.metastore.__enter__().get_table.assert_called_with(
+            dbname='default', tbl_name='does-not-exist')
 
 
 class TestHiveServer2Hook(unittest.TestCase):
@@ -397,6 +562,7 @@ def setUp(self):
         self.dag = DAG('test_dag_id', default_args=args)
         self.database = 'airflow'
         self.table = 'hive_server_hook'
+
         self.hql = """
         CREATE DATABASE IF NOT EXISTS {{ params.database }};
         USE {{ params.database }};
@@ -411,27 +577,15 @@ def setUp(self):
         """
         self.columns = ['{}.a'.format(self.table),
                         '{}.b'.format(self.table)]
-        self.hook = HiveMetastoreHook()
-        op = HiveOperator(
-            task_id='HiveHook_' + str(random.randint(1, 10000)),
-            params={
-                'database': self.database,
-                'table': self.table,
-                'csv_path': self.local_path
-            },
-            hive_cli_conn_id='hive_cli_default',
-            hql=self.hql, dag=self.dag)
-        op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
-               ignore_ti_state=True)
 
-    def tearDown(self):
-        hook = HiveMetastoreHook()
-        with hook.get_conn() as metastore:
-            metastore.drop_table(self.database, self.table, deleteData=True)
-        os.remove(self.local_path)
+        with mock.patch('airflow.providers.apache.hive.hooks.hive.HiveMetastoreHook.get_metastore_client'
+                        ) as get_metastore_mock:
+            get_metastore_mock.return_value = mock.MagicMock()
+
+            self.hook = HiveMetastoreHook()
 
     def test_get_conn(self):
-        hook = HiveServer2Hook()
+        hook = MockHiveServer2Hook()
         hook.get_conn()
 
     @mock.patch('pyhive.hive.connect')
@@ -454,34 +608,91 @@ def test_get_conn_with_password(self, mock_connect):
                 database='default')
 
     def test_get_records(self):
-        hook = HiveServer2Hook()
+        hook = MockHiveServer2Hook()
         query = "SELECT * FROM {}".format(self.table)
-        results = hook.get_records(query, schema=self.database)
+
+        with mock.patch.dict('os.environ', {
+            'AIRFLOW_CTX_DAG_ID': 'test_dag_id',
+            'AIRFLOW_CTX_TASK_ID': 'HiveHook_3835',
+            'AIRFLOW_CTX_EXECUTION_DATE': '2015-01-01T00:00:00+00:00',
+            'AIRFLOW_CTX_DAG_RUN_ID': '55',
+            'AIRFLOW_CTX_DAG_OWNER': 'airflow',
+            'AIRFLOW_CTX_DAG_EMAIL': 'test@airflow.com',
+        }):
+            results = hook.get_records(query, schema=self.database)
+
         self.assertListEqual(results, [(1, 1), (2, 2)])
 
+        hook.get_conn.assert_called_with(self.database)
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_id=test_dag_id')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.task_id=HiveHook_3835')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.execution_date=2015-01-01T00:00:00+00:00')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_run_id=55')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_owner=airflow')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_email=test@airflow.com')
+
     def test_get_pandas_df(self):
-        hook = HiveServer2Hook()
+        hook = MockHiveServer2Hook()
         query = "SELECT * FROM {}".format(self.table)
-        df = hook.get_pandas_df(query, schema=self.database)
+
+        with mock.patch.dict('os.environ', {
+            'AIRFLOW_CTX_DAG_ID': 'test_dag_id',
+            'AIRFLOW_CTX_TASK_ID': 'HiveHook_3835',
+            'AIRFLOW_CTX_EXECUTION_DATE': '2015-01-01T00:00:00+00:00',
+            'AIRFLOW_CTX_DAG_RUN_ID': '55',
+            'AIRFLOW_CTX_DAG_OWNER': 'airflow',
+            'AIRFLOW_CTX_DAG_EMAIL': 'test@airflow.com',
+        }):
+            df = hook.get_pandas_df(query, schema=self.database)
+
         self.assertEqual(len(df), 2)
-        self.assertListEqual(df.columns.tolist(), self.columns)
-        self.assertListEqual(df[self.columns[0]].values.tolist(), [1, 2])
+        self.assertListEqual(df["hive_server_hook.a"].values.tolist(), [1, 2])
+
+        hook.get_conn.assert_called_with(self.database)
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_id=test_dag_id')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.task_id=HiveHook_3835')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.execution_date=2015-01-01T00:00:00+00:00')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_run_id=55')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_owner=airflow')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_email=test@airflow.com')
 
     def test_get_results_header(self):
-        hook = HiveServer2Hook()
+        hook = MockHiveServer2Hook()
+
         query = "SELECT * FROM {}".format(self.table)
         results = hook.get_results(query, schema=self.database)
+
         self.assertListEqual([col[0] for col in results['header']],
                              self.columns)
 
     def test_get_results_data(self):
-        hook = HiveServer2Hook()
+        hook = MockHiveServer2Hook()
+
         query = "SELECT * FROM {}".format(self.table)
         results = hook.get_results(query, schema=self.database)
+
         self.assertListEqual(results['data'], [(1, 1), (2, 2)])
 
     def test_to_csv(self):
-        hook = HiveServer2Hook()
+        hook = MockHiveServer2Hook()
+        hook._get_results = mock.MagicMock(return_value=iter([
+            [
+                ('hive_server_hook.a', 'INT_TYPE', None, None, None, None, True),
+                ('hive_server_hook.b', 'INT_TYPE', None, None, None, None, True)
+            ], (1, 1), (2, 2)
+        ]))
         query = "SELECT * FROM {}".format(self.table)
         csv_filepath = 'query_results.csv'
         hook.to_csv(query, csv_filepath, schema=self.database,
@@ -497,10 +708,44 @@ def test_multi_statements(self):
             "SELECT * FROM {}".format(self.table),
             "DROP TABLE test_multi_statements",
         ]
-        hook = HiveServer2Hook()
-        results = hook.get_records(sqls, schema=self.database)
+
+        hook = MockHiveServer2Hook()
+
+        with mock.patch.dict('os.environ', {
+            'AIRFLOW_CTX_DAG_ID': 'test_dag_id',
+            'AIRFLOW_CTX_TASK_ID': 'HiveHook_3835',
+            'AIRFLOW_CTX_EXECUTION_DATE': '2015-01-01T00:00:00+00:00',
+            'AIRFLOW_CTX_DAG_RUN_ID': '55',
+            'AIRFLOW_CTX_DAG_OWNER': 'airflow',
+            'AIRFLOW_CTX_DAG_EMAIL': 'test@airflow.com',
+        }):
+            # df = hook.get_pandas_df(query, schema=self.database)
+            results = hook.get_records(sqls, schema=self.database)
         self.assertListEqual(results, [(1, 1), (2, 2)])
 
+        # self.assertEqual(len(df), 2)
+        # self.assertListEqual(df["hive_server_hook.a"].values.tolist(), [1, 2])
+
+        hook.get_conn.assert_called_with(self.database)
+        hook.mock_cursor.execute.assert_any_call(
+            'CREATE TABLE IF NOT EXISTS test_multi_statements (i INT)')
+        hook.mock_cursor.execute.assert_any_call(
+            'SELECT * FROM {}'.format(self.table))
+        hook.mock_cursor.execute.assert_any_call(
+            'DROP TABLE test_multi_statements')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_id=test_dag_id')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.task_id=HiveHook_3835')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.execution_date=2015-01-01T00:00:00+00:00')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_run_id=55')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_owner=airflow')
+        hook.mock_cursor.execute.assert_any_call(
+            'set airflow.ctx.dag_email=test@airflow.com')
+
     def test_get_results_with_hive_conf(self):
         hql = ["set key",
                "set airflow.ctx.dag_id",
@@ -526,11 +771,14 @@ def test_get_results_with_hive_conf(self):
             dag_run_id_ctx_var_name: 'test_dag_run_id',
 
         }):
-            hook = HiveServer2Hook()
-            output = '\n'.join(res_tuple[0]
-                               for res_tuple
-                               in hook.get_results(hql=hql,
-                                                   hive_conf={'key': 'value'})['data'])
+            hook = MockHiveServer2Hook()
+            hook._get_results = mock.MagicMock(return_value=iter(
+                ["header", ("value", "test"), ("test_dag_id", "test"), ("test_task_id", "test"),
+                 ("test_execution_date", "test"), ("test_dag_run_id", "test")]
+            ))
+
+            output = '\n'.join(res_tuple[0] for res_tuple in hook.get_results(
+                hql=hql, hive_conf={'key': 'value'})['data'])
         self.assertIn('value', output)
         self.assertIn('test_dag_id', output)
         self.assertIn('test_task_id', output)
@@ -548,7 +796,7 @@ def tearDown(self):
         del os.environ["AIRFLOW__CORE__SECURITY"]
 
     def test_get_proxy_user_value(self):
-        hook = HiveCliHook()
+        hook = MockHiveCliHook()
         returner = mock.MagicMock()
         returner.extra_dejson = {'proxy_user': 'a_user_proxy'}
         hook.use_beeline = True
diff --git a/tests/providers/apache/hive/operators/test_hive.py b/tests/providers/apache/hive/operators/test_hive.py
index 5b64cb59b2278..63c04b2beb00d 100644
--- a/tests/providers/apache/hive/operators/test_hive.py
+++ b/tests/providers/apache/hive/operators/test_hive.py
@@ -25,12 +25,14 @@
 from airflow.providers.apache.hive.operators.hive import HiveOperator
 from airflow.utils import timezone
 from tests.providers.apache.hive import DEFAULT_DATE, TestHiveEnvironment
+from tests.test_utils.mock_operators import MockHiveOperator
+from tests.test_utils.mock_process import MockSubProcess
 
 
 class HiveOperatorConfigTest(TestHiveEnvironment):
 
     def test_hive_airflow_default_config_queue(self):
-        op = HiveOperator(
+        op = MockHiveOperator(
             task_id='test_default_config_queue',
             hql=self.hql,
             mapred_queue_priority='HIGH',
@@ -42,11 +44,12 @@ def test_hive_airflow_default_config_queue(self):
             'hive',
             'default_hive_mapred_queue'
         )
-        self.assertEqual(op.get_hook().mapred_queue, test_config_hive_mapred_queue)
+        self.assertEqual(op.get_hook().mapred_queue,
+                         test_config_hive_mapred_queue)
 
     def test_hive_airflow_default_config_queue_override(self):
         specific_mapred_queue = 'default'
-        op = HiveOperator(
+        op = MockHiveOperator(
             task_id='test_default_config_queue',
             hql=self.hql,
             mapred_queue=specific_mapred_queue,
@@ -61,7 +64,7 @@ class HiveOperatorTest(TestHiveEnvironment):
 
     def test_hiveconf_jinja_translate(self):
         hql = "SELECT ${num_col} FROM ${hiveconf:table};"
-        op = HiveOperator(
+        op = MockHiveOperator(
             hiveconf_jinja_translate=True,
             task_id='dry_run_basic_hql', hql=hql, dag=self.dag)
         op.prepare_template()
@@ -69,7 +72,7 @@ def test_hiveconf_jinja_translate(self):
 
     def test_hiveconf(self):
         hql = "SELECT * FROM ${hiveconf:table} PARTITION (${hiveconf:day});"
-        op = HiveOperator(
+        op = MockHiveOperator(
             hiveconfs={'table': 'static_babynames', 'day': '{{ ds }}'},
             task_id='dry_run_basic_hql', hql=hql, dag=self.dag)
         op.prepare_template()
@@ -81,7 +84,7 @@ def test_hiveconf(self):
     def test_mapred_job_name(self, mock_get_hook):
         mock_hook = mock.MagicMock()
         mock_get_hook.return_value = mock_hook
-        op = HiveOperator(
+        op = MockHiveOperator(
             task_id='test_mapred_job_name',
             hql=self.hql,
             dag=self.dag)
@@ -103,13 +106,51 @@ def test_mapred_job_name(self, mock_get_hook):
     'AIRFLOW_RUNALL_TESTS' not in os.environ,
     "Skipped because AIRFLOW_RUNALL_TESTS is not set")
 class TestHivePresto(TestHiveEnvironment):
-    def test_hive(self):
+
+    @mock.patch('tempfile.tempdir', '/tmp/')
+    @mock.patch('tempfile._RandomNameSequence.__next__')
+    @mock.patch('subprocess.Popen')
+    def test_hive(self, mock_popen, mock_temp_dir):
+        mock_subprocess = MockSubProcess()
+        mock_popen.return_value = mock_subprocess
+        mock_temp_dir.return_value = "tst"
         op = HiveOperator(
-            task_id='basic_hql', hql=self.hql, dag=self.dag)
+            task_id='basic_hql', hql=self.hql, dag=self.dag, mapred_job_name="test_job_name")
+
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                ignore_ti_state=True)
+        hive_cmd = ['beeline', '-u', '"jdbc:hive2://localhost:10000/default"', '-hiveconf',
+                    'airflow.ctx.dag_id=test_dag_id', '-hiveconf', 'airflow.ctx.task_id=basic_hql',
+                    '-hiveconf', 'airflow.ctx.execution_date=2015-01-01T00:00:00+00:00', '-hiveconf',
+                    'airflow.ctx.dag_run_id=', '-hiveconf', 'airflow.ctx.dag_owner=airflow', '-hiveconf',
+                    'airflow.ctx.dag_email=', '-hiveconf', 'mapreduce.job.queuename=airflow', '-hiveconf',
+                    'mapred.job.queue.name=airflow', '-hiveconf', 'tez.queue.name=airflow', '-hiveconf',
+                    'mapred.job.name=test_job_name', '-f', '/tmp/airflow_hiveop_tst/tmptst']
+
+        mock_popen.assert_called_with(
+            hive_cmd,
+            stdout=mock_subprocess.PIPE,
+            stderr=mock_subprocess.STDOUT,
+            cwd="/tmp/airflow_hiveop_tst",
+            close_fds=True
+        )
 
-    def test_hive_queues(self):
+    @mock.patch('tempfile.tempdir', '/tmp/')
+    @mock.patch('tempfile._RandomNameSequence.__next__')
+    @mock.patch('subprocess.Popen')
+    def test_hive_queues(self, mock_popen, mock_temp_dir):
+        mock_subprocess = MockSubProcess()
+        mock_popen.return_value = mock_subprocess
+        mock_temp_dir.return_value = "tst"
+
+        hive_cmd = ['beeline', '-u', '"jdbc:hive2://localhost:10000/default"', '-hiveconf',
+                    'airflow.ctx.dag_id=test_dag_id', '-hiveconf', 'airflow.ctx.task_id=test_hive_queues',
+                    '-hiveconf', 'airflow.ctx.execution_date=2015-01-01T00:00:00+00:00', '-hiveconf',
+                    'airflow.ctx.dag_run_id=', '-hiveconf', 'airflow.ctx.dag_owner=airflow',
+                    '-hiveconf', 'airflow.ctx.dag_email=', '-hiveconf', 'mapreduce.job.queuename=default',
+                    '-hiveconf', 'mapred.job.queue.name=default', '-hiveconf', 'tez.queue.name=default',
+                    '-hiveconf', 'mapreduce.job.priority=HIGH', '-hiveconf',
+                    'mapred.job.name=airflow.test_hive_queues', '-f', '/tmp/airflow_hiveop_tst/tmptst']
         op = HiveOperator(
             task_id='test_hive_queues', hql=self.hql,
             mapred_queue='default', mapred_queue_priority='HIGH',
@@ -118,14 +159,68 @@ def test_hive_queues(self):
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                ignore_ti_state=True)
 
-    def test_hive_dryrun(self):
-        op = HiveOperator(
-            task_id='dry_run_basic_hql', hql=self.hql, dag=self.dag)
+        mock_popen.assert_called_with(
+            hive_cmd,
+            stdout=mock_subprocess.PIPE,
+            stderr=mock_subprocess.STDOUT,
+            cwd="/tmp/airflow_hiveop_tst",
+            close_fds=True
+        )
+
+    @mock.patch('tempfile.tempdir', '/tmp/')
+    @mock.patch('tempfile._RandomNameSequence.__next__')
+    @mock.patch('subprocess.Popen')
+    def test_hive_dryrun(self, mock_popen, mock_temp_dir):
+        mock_subprocess = MockSubProcess()
+        mock_popen.return_value = mock_subprocess
+        mock_temp_dir.return_value = "tst"
+
+        op = HiveOperator(task_id='dry_run_basic_hql',
+                          hql=self.hql, dag=self.dag)
         op.dry_run()
 
-    def test_beeline(self):
+        hive_cmd = ['beeline', '-u', '"jdbc:hive2://localhost:10000/default"',
+                    '-hiveconf', 'airflow.ctx.dag_id=', '-hiveconf', 'airflow.ctx.task_id=',
+                    '-hiveconf', 'airflow.ctx.execution_date=', '-hiveconf', 'airflow.ctx.dag_run_id=',
+                    '-hiveconf', 'airflow.ctx.dag_owner=', '-hiveconf', 'airflow.ctx.dag_email=',
+                    '-hiveconf', 'mapreduce.job.queuename=airflow', '-hiveconf',
+                    'mapred.job.queue.name=airflow', '-hiveconf', 'tez.queue.name=airflow',
+                    '-f', '/tmp/airflow_hiveop_tst/tmptst']
+        mock_popen.assert_called_with(
+            hive_cmd,
+            stdout=mock_subprocess.PIPE,
+            stderr=mock_subprocess.STDOUT,
+            cwd="/tmp/airflow_hiveop_tst",
+            close_fds=True
+        )
+
+    @mock.patch('tempfile.tempdir', '/tmp/')
+    @mock.patch('tempfile._RandomNameSequence.__next__')
+    @mock.patch('subprocess.Popen')
+    def test_beeline(self, mock_popen, mock_temp_dir):
+        mock_subprocess = MockSubProcess()
+        mock_popen.return_value = mock_subprocess
+        mock_temp_dir.return_value = "tst"
+
+        hive_cmd = ['beeline', '-u', '"jdbc:hive2://localhost:10000/default"',
+                    '-hiveconf', 'airflow.ctx.dag_id=test_dag_id', '-hiveconf',
+                    'airflow.ctx.task_id=beeline_hql', '-hiveconf',
+                    'airflow.ctx.execution_date=2015-01-01T00:00:00+00:00',
+                    '-hiveconf', 'airflow.ctx.dag_run_id=', '-hiveconf', 'airflow.ctx.dag_owner=airflow',
+                    '-hiveconf', 'airflow.ctx.dag_email=',
+                    '-hiveconf', 'mapreduce.job.queuename=airflow', '-hiveconf',
+                    'mapred.job.queue.name=airflow', '-hiveconf', 'tez.queue.name=airflow', '-hiveconf',
+                    'mapred.job.name=test_job_name', '-f', '/tmp/airflow_hiveop_tst/tmptst']
+
         op = HiveOperator(
             task_id='beeline_hql', hive_cli_conn_id='hive_cli_default',
-            hql=self.hql, dag=self.dag)
+            hql=self.hql, dag=self.dag, mapred_job_name="test_job_name")
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                ignore_ti_state=True)
+        mock_popen.assert_called_with(
+            hive_cmd,
+            stdout=mock_subprocess.PIPE,
+            stderr=mock_subprocess.STDOUT,
+            cwd="/tmp/airflow_hiveop_tst",
+            close_fds=True
+        )
diff --git a/tests/providers/apache/hive/operators/test_hive_stats.py b/tests/providers/apache/hive/operators/test_hive_stats.py
index 673cbbd74fa88..f2e1e93227d41 100644
--- a/tests/providers/apache/hive/operators/test_hive_stats.py
+++ b/tests/providers/apache/hive/operators/test_hive_stats.py
@@ -17,6 +17,7 @@
 # under the License.
 
 import os
+import re
 import unittest
 from collections import OrderedDict
 from unittest.mock import patch
@@ -24,6 +25,7 @@
 from airflow.exceptions import AirflowException
 from airflow.providers.apache.hive.operators.hive_stats import HiveStatsCollectionOperator
 from tests.providers.apache.hive import DEFAULT_DATE, DEFAULT_DATE_DS, TestHiveEnvironment
+from tests.test_utils.mock_hooks import MockHiveMetastoreHook, MockMySqlHook, MockPrestoHook
 
 
 class _FakeCol:
@@ -288,11 +290,50 @@ def test_execute_delete_previous_runs_rows(self,
     @unittest.skipIf(
         'AIRFLOW_RUNALL_TESTS' not in os.environ,
         "Skipped because AIRFLOW_RUNALL_TESTS is not set")
-    def test_runs_for_hive_stats(self):
-        op = HiveStatsCollectionOperator(
-            task_id='hive_stats_check',
-            table="airflow.static_babynames_partitioned",
-            partition={'ds': DEFAULT_DATE_DS},
-            dag=self.dag)
-        op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
-               ignore_ti_state=True)
+    @patch('airflow.providers.apache.hive.operators.hive_stats.HiveMetastoreHook',
+           side_effect=MockHiveMetastoreHook)
+    def test_runs_for_hive_stats(self, mock_hive_metastore_hook):
+        mock_mysql_hook = MockMySqlHook()
+        mock_presto_hook = MockPrestoHook()
+        with patch('airflow.providers.apache.hive.operators.hive_stats.PrestoHook',
+                   return_value=mock_presto_hook):
+            with patch('airflow.providers.apache.hive.operators.hive_stats.MySqlHook',
+                       return_value=mock_mysql_hook):
+                op = HiveStatsCollectionOperator(
+                    task_id='hive_stats_check',
+                    table="airflow.static_babynames_partitioned",
+                    partition={'ds': DEFAULT_DATE_DS},
+                    dag=self.dag)
+                op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
+                       ignore_ti_state=True)
+
+        select_count_query = "SELECT COUNT(*) AS __count FROM airflow." \
+            + "static_babynames_partitioned WHERE ds = '2015-01-01';"
+        mock_presto_hook.get_first.assert_called_with(hql=select_count_query)
+
+        expected_stats_select_query = "SELECT 1 FROM hive_stats WHERE table_name='airflow." \
+            + "static_babynames_partitioned' AND " \
+            + "partition_repr='{\"ds\": \"2015-01-01\"}' AND " \
+            + "dttm='2015-01-01T00:00:00+00:00' " \
+            + "LIMIT 1;"
+
+        raw_stats_select_query = mock_mysql_hook.get_records.call_args_list[0][0][0]
+        actual_stats_select_query = re.sub(r'\s{2,}', ' ', raw_stats_select_query).strip()
+
+        self.assertEqual(expected_stats_select_query, actual_stats_select_query)
+
+        insert_rows_val = [('2015-01-01', '2015-01-01T00:00:00+00:00',
+                            'airflow.static_babynames_partitioned',
+                            '{"ds": "2015-01-01"}', '', 'count', ['val_0', 'val_1'])]
+
+        mock_mysql_hook.insert_rows.assert_called_with(table='hive_stats',
+                                                       rows=insert_rows_val,
+                                                       target_fields=[
+                                                           'ds',
+                                                           'dttm',
+                                                           'table_name',
+                                                           'partition_repr',
+                                                           'col',
+                                                           'metric',
+                                                           'value',
+                                                       ])
diff --git a/tests/providers/apache/hive/operators/test_hive_to_mysql.py b/tests/providers/apache/hive/operators/test_hive_to_mysql.py
index 437385a1eceb0..5c293de94d487 100644
--- a/tests/providers/apache/hive/operators/test_hive_to_mysql.py
+++ b/tests/providers/apache/hive/operators/test_hive_to_mysql.py
@@ -113,6 +113,8 @@ def test_execute_with_hive_conf(self, mock_hive_hook, mock_mysql_hook):
     @unittest.skipIf(
         'AIRFLOW_RUNALL_TESTS' not in os.environ,
         "Skipped because AIRFLOW_RUNALL_TESTS is not set")
+    @unittest.skip("HiveToMySqlTransfer currently passes \
+                    task_id to parent in invalid manner. Test always fails.")
     def test_hive_to_mysql(self):
         op = HiveToMySqlTransfer(
             mysql_conn_id='airflow_db',
diff --git a/tests/providers/apache/hive/operators/test_hive_to_samba.py b/tests/providers/apache/hive/operators/test_hive_to_samba.py
index 58f88149af8c4..1e141173a8c57 100644
--- a/tests/providers/apache/hive/operators/test_hive_to_samba.py
+++ b/tests/providers/apache/hive/operators/test_hive_to_samba.py
@@ -17,11 +17,12 @@
 # under the License.
 import os
 import unittest
-from unittest.mock import Mock, PropertyMock, patch
+from unittest.mock import MagicMock, Mock, PropertyMock, patch
 
 from airflow.providers.apache.hive.operators.hive_to_samba import Hive2SambaOperator
 from airflow.utils.operator_helpers import context_to_airflow_vars
 from tests.providers.apache.hive import DEFAULT_DATE, TestHiveEnvironment
+from tests.test_utils.mock_hooks import MockHiveServer2Hook, MockSambaHook
 
 
 class TestHive2SambaOperator(TestHiveEnvironment):
@@ -46,24 +47,41 @@ def test_execute(self, mock_tmp_file, mock_hive_hook, mock_samba_hook):
 
         Hive2SambaOperator(**self.kwargs).execute(context)
 
-        mock_hive_hook.assert_called_once_with(hiveserver2_conn_id=self.kwargs['hiveserver2_conn_id'])
+        mock_hive_hook.assert_called_once_with(
+            hiveserver2_conn_id=self.kwargs['hiveserver2_conn_id'])
         mock_hive_hook.return_value.to_csv.assert_called_once_with(
             hql=self.kwargs['hql'],
             csv_filepath=mock_tmp_file.name,
             hive_conf=context_to_airflow_vars(context))
-        mock_samba_hook.assert_called_once_with(samba_conn_id=self.kwargs['samba_conn_id'])
+        mock_samba_hook.assert_called_once_with(
+            samba_conn_id=self.kwargs['samba_conn_id'])
         mock_samba_hook.return_value.push_from_local.assert_called_once_with(
             self.kwargs['destination_filepath'], mock_tmp_file.name)
 
     @unittest.skipIf(
         'AIRFLOW_RUNALL_TESTS' not in os.environ,
         "Skipped because AIRFLOW_RUNALL_TESTS is not set")
-    def test_hive2samba(self):
-        op = Hive2SambaOperator(
-            task_id='hive2samba_check',
-            samba_conn_id='tableau_samba',
-            hql="SELECT * FROM airflow.static_babynames LIMIT 10000",
-            destination_filepath='test_airflow.csv',
-            dag=self.dag)
-        op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
-               ignore_ti_state=True)
+    @patch('tempfile.tempdir', '/tmp/')
+    @patch('tempfile._RandomNameSequence.__next__')
+    @patch('airflow.providers.apache.hive.operators.hive_to_samba.HiveServer2Hook',
+           side_effect=MockHiveServer2Hook)
+    def test_hive2samba(self, mock_hive_server_hook, mock_temp_dir):
+        mock_temp_dir.return_value = "tst"
+
+        samba_hook = MockSambaHook()
+        samba_hook.upload = MagicMock()
+
+        with patch('airflow.providers.apache.hive.operators.hive_to_samba.SambaHook',
+                   return_value=samba_hook):
+            samba_hook.conn.upload = MagicMock()
+            op = Hive2SambaOperator(
+                task_id='hive2samba_check',
+                samba_conn_id='tableau_samba',
+                hql="SELECT * FROM airflow.static_babynames LIMIT 10000",
+                destination_filepath='test_airflow.csv',
+                dag=self.dag)
+            op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
+                   ignore_ti_state=True)
+
+        samba_hook.conn.upload.assert_called_with(
+            '/tmp/tmptst', 'test_airflow.csv')
diff --git a/tests/providers/apache/hive/sensors/test_hive_partition.py b/tests/providers/apache/hive/sensors/test_hive_partition.py
index 1407bef735f04..ccb551422b105 100644
--- a/tests/providers/apache/hive/sensors/test_hive_partition.py
+++ b/tests/providers/apache/hive/sensors/test_hive_partition.py
@@ -18,17 +18,20 @@
 
 import os
 import unittest
+from unittest.mock import patch
 
 from airflow.providers.apache.hive.sensors.hive_partition import HivePartitionSensor
 from tests.providers.apache.hive import DEFAULT_DATE, TestHiveEnvironment
+from tests.test_utils.mock_hooks import MockHiveMetastoreHook
 
 
 @unittest.skipIf(
     'AIRFLOW_RUNALL_TESTS' not in os.environ,
     "Skipped because AIRFLOW_RUNALL_TESTS is not set")
+@patch('airflow.providers.apache.hive.sensors.hive_partition.HiveMetastoreHook',
+       side_effect=MockHiveMetastoreHook)
 class TestHivePartitionSensor(TestHiveEnvironment):
-
-    def test_hive_partition_sensor(self):
+    def test_hive_partition_sensor(self, mock_hive_metastore_hook):
         op = HivePartitionSensor(
             task_id='hive_partition_check',
             table='airflow.static_babynames_partitioned',
diff --git a/tests/providers/apache/hive/sensors/test_metastore_partition.py b/tests/providers/apache/hive/sensors/test_metastore_partition.py
index 68fad19fdf8d3..d7227adef001f 100644
--- a/tests/providers/apache/hive/sensors/test_metastore_partition.py
+++ b/tests/providers/apache/hive/sensors/test_metastore_partition.py
@@ -18,21 +18,25 @@
 
 import os
 import unittest
+from unittest import mock
 
 from airflow.providers.apache.hive.sensors.metastore_partition import MetastorePartitionSensor
 from tests.providers.apache.hive import DEFAULT_DATE, DEFAULT_DATE_DS, TestHiveEnvironment
+from tests.test_utils.mock_process import MockDBConnection
 
 
 @unittest.skipIf(
     'AIRFLOW_RUNALL_TESTS' not in os.environ,
     "Skipped because AIRFLOW_RUNALL_TESTS is not set")
 class TestHivePartitionSensor(TestHiveEnvironment):
-
     def test_hive_metastore_sql_sensor(self):
         op = MetastorePartitionSensor(
             task_id='hive_partition_check',
+            conn_id='test_connection_id',
+            sql='test_sql',
             table='airflow.static_babynames_partitioned',
             partition_name='ds={}'.format(DEFAULT_DATE_DS),
             dag=self.dag)
+        op._get_hook = mock.MagicMock(return_value=MockDBConnection({}))
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                ignore_ti_state=True)
diff --git a/tests/providers/apache/hive/sensors/test_named_hive_partition.py b/tests/providers/apache/hive/sensors/test_named_hive_partition.py
index 5ded4d79240c3..bbe507c2dfcb1 100644
--- a/tests/providers/apache/hive/sensors/test_named_hive_partition.py
+++ b/tests/providers/apache/hive/sensors/test_named_hive_partition.py
@@ -19,14 +19,15 @@
 import random
 import unittest
 from datetime import timedelta
+from unittest import mock
 
 from airflow.exceptions import AirflowSensorTimeout
 from airflow.models.dag import DAG
-from airflow.providers.apache.hive.hooks.hive import HiveMetastoreHook
-from airflow.providers.apache.hive.operators.hive import HiveOperator
 from airflow.providers.apache.hive.sensors.named_hive_partition import NamedHivePartitionSensor
 from airflow.utils.timezone import datetime
 from tests.providers.apache.hive import TestHiveEnvironment
+from tests.test_utils.mock_hooks import MockHiveMetastoreHook
+from tests.test_utils.mock_operators import MockHiveOperator
 
 DEFAULT_DATE = datetime(2015, 1, 1)
 DEFAULT_DATE_ISO = DEFAULT_DATE.isoformat()
@@ -56,8 +57,8 @@ def setUp(self):
                 ALTER TABLE {{ params.table }}
                 ADD PARTITION({{ params.partition_by }}='{{ ds }}');
                 """
-        self.hook = HiveMetastoreHook()
-        op = HiveOperator(
+        self.hook = MockHiveMetastoreHook()
+        op = MockHiveOperator(
             task_id='HiveHook_' + str(random.randint(1, 10000)),
             params={
                 'database': self.database,
@@ -69,11 +70,6 @@ def setUp(self):
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                ignore_ti_state=True)
 
-    def tearDown(self):
-        hook = HiveMetastoreHook()
-        with hook.get_conn() as metastore:
-            metastore.drop_table(self.database, self.table, deleteData=True)
-
     def test_parse_partition_name_correct(self):
         schema = 'default'
         table = 'users'
@@ -106,6 +102,7 @@ def test_parse_partition_name_default(self):
         self.assertEqual(partition, parsed_partition)
 
     def test_poke_existing(self):
+        self.hook.metastore.__enter__().check_for_named_partition.return_value = True
         partitions = ["{}.{}/{}={}".format(self.database,
                                            self.table,
                                            self.partition_by,
@@ -116,8 +113,11 @@ def test_poke_existing(self):
                                           hook=self.hook,
                                           dag=self.dag)
         self.assertTrue(sensor.poke(None))
+        self.hook.metastore.__enter__().check_for_named_partition.assert_called_with(
+            self.database, self.table, f"{self.partition_by}={DEFAULT_DATE_DS}")
 
     def test_poke_non_existing(self):
+        self.hook.metastore.__enter__().check_for_named_partition.return_value = False
         partitions = ["{}.{}/{}={}".format(self.database,
                                            self.table,
                                            self.partition_by,
@@ -128,6 +128,8 @@ def test_poke_non_existing(self):
                                           hook=self.hook,
                                           dag=self.dag)
         self.assertFalse(sensor.poke(None))
+        self.hook.metastore.__enter__().check_for_named_partition.assert_called_with(
+            self.database, self.table, f"{self.partition_by}={self.next_day}")
 
 
 @unittest.skipIf(
@@ -136,25 +138,45 @@ def test_poke_non_existing(self):
 class TestPartitions(TestHiveEnvironment):
 
     def test_succeeds_on_one_partition(self):
+        mock_hive_metastore_hook = MockHiveMetastoreHook()
+        mock_hive_metastore_hook.check_for_named_partition = mock.MagicMock(
+            return_value=True)
+
         op = NamedHivePartitionSensor(
             task_id='hive_partition_check',
             partition_names=[
                 "airflow.static_babynames_partitioned/ds={{ds}}"
             ],
-            dag=self.dag)
+            dag=self.dag,
+            hook=mock_hive_metastore_hook
+        )
+
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                ignore_ti_state=True)
 
+        mock_hive_metastore_hook.check_for_named_partition.assert_called_once_with(
+            'airflow', 'static_babynames_partitioned', 'ds=2015-01-01')
+
     def test_succeeds_on_multiple_partitions(self):
+        mock_hive_metastore_hook = MockHiveMetastoreHook()
+        mock_hive_metastore_hook.check_for_named_partition = mock.MagicMock(
+            return_value=True)
+
         op = NamedHivePartitionSensor(
             task_id='hive_partition_check',
             partition_names=[
                 "airflow.static_babynames_partitioned/ds={{ds}}",
-                "airflow.static_babynames_partitioned/ds={{ds}}"
+                "airflow.static_babynames_partitioned2/ds={{ds}}"
             ],
-            dag=self.dag)
+            dag=self.dag,
+            hook=mock_hive_metastore_hook
+        )
         op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                ignore_ti_state=True)
+        mock_hive_metastore_hook.check_for_named_partition.assert_any_call(
+            'airflow', 'static_babynames_partitioned', 'ds=2015-01-01')
+        mock_hive_metastore_hook.check_for_named_partition.assert_any_call(
+            'airflow', 'static_babynames_partitioned2', 'ds=2015-01-01')
 
     def test_parses_partitions_with_periods(self):
         name = NamedHivePartitionSensor.parse_partition_name(
@@ -165,6 +187,10 @@ def test_parses_partitions_with_periods(self):
 
     def test_times_out_on_nonexistent_partition(self):
         with self.assertRaises(AirflowSensorTimeout):
+            mock_hive_metastore_hook = MockHiveMetastoreHook()
+            mock_hive_metastore_hook.check_for_named_partition = mock.MagicMock(
+                return_value=False)
+
             op = NamedHivePartitionSensor(
                 task_id='hive_partition_check',
                 partition_names=[
@@ -173,6 +199,8 @@ def test_times_out_on_nonexistent_partition(self):
                 ],
                 poke_interval=0.1,
                 timeout=1,
-                dag=self.dag)
+                dag=self.dag,
+                hook=mock_hive_metastore_hook
+            )
             op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
                    ignore_ti_state=True)
diff --git a/tests/test_utils/mock_hooks.py b/tests/test_utils/mock_hooks.py
new file mode 100644
index 0000000000000..fd61c41c13501
--- /dev/null
+++ b/tests/test_utils/mock_hooks.py
@@ -0,0 +1,96 @@
+# 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.
+
+from unittest import mock
+
+from airflow.providers.apache.hive.hooks.hive import HiveCliHook, HiveMetastoreHook, HiveServer2Hook
+from airflow.providers.mysql.hooks.mysql import MySqlHook
+from airflow.providers.presto.hooks.presto import PrestoHook
+from airflow.providers.samba.hooks.samba import SambaHook
+from tests.test_utils.mock_process import MockConnectionCursor, MockDBConnection
+
+
+class MockHiveMetastoreHook(HiveMetastoreHook):
+    def __init__(self, *args, **kwargs):
+        self._find_valid_server = mock.MagicMock(return_value={})
+        self.get_metastore_client = mock.MagicMock(
+            return_value=mock.MagicMock())
+        super().__init__()
+
+
+class MockHiveCliHook(HiveCliHook):
+    def __init__(self, *args, **kwargs):
+        super().__init__()
+        self.conn = MockConnectionCursor()
+        self.conn.schema = 'default'
+        self.conn.host = 'localhost'
+        self.conn.port = 10000
+        self.conn.login = None
+        self.conn.password = None
+
+        self.conn.execute = mock.MagicMock()
+        self.get_conn = mock.MagicMock(return_value=self.conn)
+        self.get_connection = mock.MagicMock(return_value=MockDBConnection({}))
+
+
+class MockSambaHook(SambaHook):
+    def __init__(self, *args, **kwargs):
+        self.conn = MockConnectionCursor()
+        self.conn.execute = mock.MagicMock()
+        self.get_conn = mock.MagicMock(return_value=self.conn)
+        super().__init__(*args, **kwargs)
+
+    def get_connection(self, *args):
+        return self.conn
+
+
+class MockPrestoHook(PrestoHook):
+    def __init__(self, *args, **kwargs):
+        self.conn = MockConnectionCursor()
+
+        self.conn.execute = mock.MagicMock()
+        self.get_conn = mock.MagicMock(return_value=self.conn)
+        self.get_first = mock.MagicMock(
+            return_value=[['val_0', 'val_1'], 'val_2'])
+
+        super().__init__(*args, **kwargs)
+
+    def get_connection(self, *args):
+        return self.conn
+
+
+class MockMySqlHook(MySqlHook):
+    def __init__(self, *args, **kwargs):
+        self.conn = MockConnectionCursor()
+
+        self.conn.execute = mock.MagicMock()
+        self.get_conn = mock.MagicMock(return_value=self.conn)
+        self.get_records = mock.MagicMock(return_value=[])
+        self.insert_rows = mock.MagicMock(return_value=True)
+        super().__init__(*args, **kwargs)
+
+    def get_connection(self, *args, **kwargs):
+        return self.conn
+
+
+class MockHiveServer2Hook(HiveServer2Hook):
+    def __init__(self, *args, **kwargs):
+        super().__init__()
+        self.mock_cursor = MockConnectionCursor()
+        self.mock_cursor.execute = mock.MagicMock()
+        self.get_conn = mock.MagicMock(return_value=self.mock_cursor)
+        self.get_connection = mock.MagicMock(return_value=MockDBConnection({}))
diff --git a/tests/test_utils/mock_operators.py b/tests/test_utils/mock_operators.py
index 39423ccb704e9..86af6974ffe3b 100644
--- a/tests/test_utils/mock_operators.py
+++ b/tests/test_utils/mock_operators.py
@@ -16,11 +16,13 @@
 # under the License.
 
 from typing import NamedTuple
+from unittest import mock
 
 import attr
 
 from airflow.models import TaskInstance
 from airflow.models.baseoperator import BaseOperator, BaseOperatorLink
+from airflow.providers.apache.hive.operators.hive import HiveOperator
 from airflow.utils.decorators import apply_defaults
 
 
@@ -158,3 +160,9 @@ class GithubLink(BaseOperatorLink):
 
     def get_link(self, operator, dttm):
         return 'https://github.com/apache/airflow'
+
+
+class MockHiveOperator(HiveOperator):
+    def __init__(self, *args, **kwargs):
+        self.run = mock.MagicMock()
+        super().__init__(*args, **kwargs)
diff --git a/tests/test_utils/mock_process.py b/tests/test_utils/mock_process.py
new file mode 100644
index 0000000000000..cc3d202249d02
--- /dev/null
+++ b/tests/test_utils/mock_process.py
@@ -0,0 +1,84 @@
+# 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.
+
+from unittest import mock
+
+
+class MockDBConnection:
+    def __init__(self, extra_dejson=None, *args, **kwargs):
+        self.extra_dejson = extra_dejson
+        self.get_records = mock.MagicMock(return_value=[['test_record']])
+
+        output = kwargs.get('output', ['' for _ in range(10)])
+        self.readline = mock.MagicMock(
+            side_effect=[line.encode() for line in output])
+
+
+class MockStdOut:
+    def __init__(self, *args, **kwargs):
+        output = kwargs.get('output', ['' for _ in range(10)])
+        self.readline = mock.MagicMock(
+            side_effect=[line.encode() for line in output])
+
+
+class MockSubProcess:
+    PIPE = -1
+    STDOUT = -2
+    returncode = None
+
+    def __init__(self, *args, **kwargs):
+        self.stdout = MockStdOut(*args, **kwargs)
+
+    def wait(self):
+        return
+
+
+class MockConnectionCursor:
+    def __init__(self, *args, **kwargs):
+        self.arraysize = None
+        self.description = [('hive_server_hook.a', 'INT_TYPE', None, None, None, None, True),
+                            ('hive_server_hook.b', 'INT_TYPE', None, None, None, None, True)]
+        self.iterable = [(1, 1), (2, 2)]
+        self.conn_exists = kwargs.get('exists', True)
+
+    def close(self):
+        pass
+
+    def cursor(self):
+        return self
+
+    def execute(self, values=None):
+        pass
+
+    def exists(self, *args):
+        return self.conn_exists
+
+    def isfile(self, *args):
+        return True
+
+    def remove(self, *args):
+        pass
+
+    def upload(self, local_filepath, destination_filepath):
+        pass
+
+    def __next__(self):
+        return self.iterable
+
+    def __iter__(self):
+        for i in self.iterable:
+            yield i


Code review comments:
diff_hunk: @@ -121,14 +121,14 @@ RUN adduser airflow \
 # https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=863199
 RUN mkdir -pv /usr/share/man/man1 \
     && mkdir -pv /usr/share/man/man7 \
-    # && curl -fsSL https://adoptopenjdk.jfrog.io/adoptopenjdk/api/gpg/key/public | apt-key add - \
-    # && echo 'deb https://adoptopenjdk.jfrog.io/adoptopenjdk/deb/ buster main' > \
-    #     /etc/apt/sources.list.d/adoptopenjdk.list \
+    && curl -fsSL https://adoptopenjdk.jfrog.io/adoptopenjdk/api/gpg/key/public | apt-key add - \
+    && echo 'deb https://adoptopenjdk.jfrog.io/adoptopenjdk/deb/ buster main' > \
+        /etc/apt/sources.list.d/adoptopenjdk.list \
     && apt-get update \
     && apt-get install --no-install-recommends -y \
       gnupg \
       libgcc-8-dev \
-    #   adoptopenjdk-8-hotspot-jre \
+      adoptopenjdk-8-hotspot-jre \
path: Dockerfile.ci
text: @user1:
What still needs the JRE btw?

@author:
Removed