# 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.
#
# Client tests for SQL statement authorization

import os
import pytest
import tempfile
import grp
import re
import random
import sys
import subprocess
import urllib

from getpass import getuser
from ImpalaService import ImpalaHiveServer2Service
from TCLIService import TCLIService
from thrift.transport.TSocket import TSocket
from thrift.transport.TTransport import TBufferedTransport
from thrift.protocol import TBinaryProtocol
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
from tests.common.file_utils import assert_file_in_dir_contains,\
    assert_no_files_in_dir_contain

SENTRY_CONFIG_DIR = os.getenv('IMPALA_HOME') + '/fe/src/test/resources/'
SENTRY_BASE_LOG_DIR = os.getenv('IMPALA_CLUSTER_LOGS_DIR') + "/sentry"
SENTRY_CONFIG_FILE = SENTRY_CONFIG_DIR + 'sentry-site.xml'
SENTRY_CONFIG_FILE_OO = SENTRY_CONFIG_DIR + 'sentry-site_oo.xml'
PRIVILEGES = ['all', 'alter', 'drop', 'insert', 'refresh', 'select']
ADMIN = "admin"

class TestAuthorization(CustomClusterTestSuite):
  def setup(self):
    host, port = (self.cluster.impalads[0].service.hostname,
                  self.cluster.impalads[0].service.hs2_port)
    self.socket = TSocket(host, port)
    self.transport = TBufferedTransport(self.socket)
    self.transport.open()
    self.protocol = TBinaryProtocol.TBinaryProtocol(self.transport)
    self.hs2_client = ImpalaHiveServer2Service.Client(self.protocol)

  def teardown(self):
    if self.socket:
      self.socket.close()

  def __execute_hs2_stmt(self, statement, verify=True):
    """
    Executes an hs2 statement

    :param statement: the statement to execute
    :param verify: If set to true, will thrown an exception on a failed hs2 execution
    :return: the result of execution
    """
    from tests.hs2.test_hs2 import TestHS2
    execute_statement_req = TCLIService.TExecuteStatementReq()
    execute_statement_req.sessionHandle = self.session_handle
    execute_statement_req.statement = statement
    result = self.hs2_client.ExecuteStatement(execute_statement_req)
    if verify:
      TestHS2.check_response(result)
    return result

  def __open_hs2(self, user, configuration, verify=True):
    """
    Open a session with hs2

    :param user: the user to open the session
    :param configuration: the configuration for the session
    :param verify: If set to true, will thrown an exception on failed session open
    :return: the result of opening the session
    """
    from tests.hs2.test_hs2 import TestHS2
    open_session_req = TCLIService.TOpenSessionReq()
    open_session_req.username = user
    open_session_req.configuration = configuration
    resp = self.hs2_client.OpenSession(open_session_req)
    if verify:
      TestHS2.check_response(resp)
    return resp

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
      impalad_args="--server_name=server1 "
                   "--sentry_config={0} "
                   "--authorization_policy_provider_class="
                   "org.apache.impala.testutil.TestSentryResourceAuthorizationProvider"
                   .format(SENTRY_CONFIG_FILE),
      catalogd_args="--sentry_config={0} "
                    "--authorization_policy_provider_class="
                    "org.apache.impala.testutil.TestSentryResourceAuthorizationProvider"
                    .format(SENTRY_CONFIG_FILE),
      sentry_config=SENTRY_CONFIG_FILE)
  def test_custom_authorization_provider(self, unique_role):
    try:
      self.session_handle = self.__open_hs2(getuser(), dict()).sessionHandle
      self.__execute_hs2_stmt("create role {0}".format(unique_role))
      self.__execute_hs2_stmt("grant role {0} to group {1}"
                              .format(unique_role, grp.getgrnam(getuser()).gr_name))
      self.__execute_hs2_stmt("grant select on table tpch.lineitem to role {0}"
                              .format(unique_role))

      bad_resp = self.__execute_hs2_stmt("describe tpch_seq.lineitem", False)
      assert 'User \'%s\' does not have privileges to access' % getuser() in \
             str(bad_resp)
      self.__execute_hs2_stmt("describe tpch.lineitem")
    finally:
      self.__execute_hs2_stmt("drop role {0}".format(unique_role))

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
      impalad_args="--server_name=server1 "
                   "--authorized_proxy_user_config=hue={0}".format(getuser()),
      catalogd_args="--sentry_config=" + SENTRY_CONFIG_FILE,
      sentry_config=SENTRY_CONFIG_FILE)
  def test_access_runtime_profile(self, unique_role, unique_name):
    unique_db = unique_name + "_db"

    try:
      self.session_handle = self.__open_hs2(getuser(), dict()).sessionHandle
      self.__execute_hs2_stmt("create role {0}".format(unique_role))
      self.__execute_hs2_stmt("grant create on server to role {0}".format(unique_role))
      self.__execute_hs2_stmt("grant all on database tpch to role {0}"
                              .format(unique_role))
      self.__execute_hs2_stmt("grant select on table functional.complex_view to role {0}"
                              .format(unique_role))
      self.__execute_hs2_stmt("grant role {0} to group {1}"
                              .format(unique_role, grp.getgrnam(getuser()).gr_name))
      # Create db with permissions
      self.__execute_hs2_stmt("create database {0}".format(unique_db))
      self.__execute_hs2_stmt("grant all on database {0} to role {1}"
                              .format(unique_db, unique_role))

      # Current user can't access view's underlying tables
      bad_resp = self.__execute_hs2_stmt("explain select * from functional.complex_view",
                                         False)
      assert 'User \'%s\' does not have privileges to EXPLAIN' % getuser() in \
             str(bad_resp)
      # User should not have access to the runtime profile
      self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
                                                False, False)
      self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
                                                False, True)

      # Repeat as a delegated user
      self.session_handle = \
          self.__open_hs2('hue', {'impala.doas.user': getuser()}).sessionHandle
      # User should not have access to the runtime profile
      self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
                                                False, False)
      self.__run_stmt_and_verify_profile_access("select * from functional.complex_view",
                                                False, True)

      # Create a view for which the user has access to the underlying tables.
      self.session_handle = self.__open_hs2(getuser(), dict()).sessionHandle
      self.__execute_hs2_stmt(
          "create view if not exists {0}.customer_view as select * from tpch.customer "
          "limit 1".format(unique_db))

      # User should be able to run EXPLAIN
      self.__execute_hs2_stmt("explain select * from {0}.customer_view"
                              .format(unique_db))

      # User should have access to the runtime profile and exec summary
      self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
                                                .format(unique_db), True, False)
      self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
                                                .format(unique_db), True, True)

      # Repeat as a delegated user
      self.session_handle = \
          self.__open_hs2('hue', {'impala.doas.user': getuser()}).sessionHandle
      # Delegated user is the current user
      self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
                                                .format(unique_db), True, False)
      self.__run_stmt_and_verify_profile_access("select * from {0}.customer_view"
                                                .format(unique_db), True, True)
    finally:
      self.__execute_hs2_stmt("grant all on server to role {0}".format(unique_role))
      self.__execute_hs2_stmt("drop view if exists {0}.customer_view".format(unique_db))
      self.__execute_hs2_stmt("drop table if exists {0}.customer".format(unique_db))
      self.__execute_hs2_stmt("drop database if exists {0}".format(unique_db))
      self.__execute_hs2_stmt("drop role {0}".format(unique_role))

  def __run_stmt_and_verify_profile_access(self, stmt, has_access, close_operation):
    """Runs 'stmt' and retrieves the runtime profile and exec summary. If
      'has_access' is true, it verifies that no runtime profile or exec summary are
      returned. If 'close_operation' is true, make sure the operation is closed before
      retrieving the profile and exec summary."""
    from tests.hs2.test_hs2 import TestHS2
    execute_statement_resp = self.__execute_hs2_stmt(stmt, False)

    if close_operation:
      close_operation_req = TCLIService.TCloseOperationReq()
      close_operation_req.operationHandle = execute_statement_resp.operationHandle
      TestHS2.check_response(self.hs2_client.CloseOperation(close_operation_req))

    get_profile_req = ImpalaHiveServer2Service.TGetRuntimeProfileReq()
    get_profile_req.operationHandle = execute_statement_resp.operationHandle
    get_profile_req.sessionHandle = self.session_handle
    get_profile_resp = self.hs2_client.GetRuntimeProfile(get_profile_req)

    if has_access:
      TestHS2.check_response(get_profile_resp)
      assert "Plan: " in get_profile_resp.profile
    else:
      assert "User %s is not authorized to access the runtime profile or "\
          "execution summary." % (getuser()) in str(get_profile_resp)

    exec_summary_req = ImpalaHiveServer2Service.TGetExecSummaryReq()
    exec_summary_req.operationHandle = execute_statement_resp.operationHandle
    exec_summary_req.sessionHandle = self.session_handle
    exec_summary_resp = self.hs2_client.GetExecSummary(exec_summary_req)

    if has_access:
      TestHS2.check_response(exec_summary_resp)
    else:
      assert "User %s is not authorized to access the runtime profile or "\
          "execution summary." % (getuser()) in str(exec_summary_resp)

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
      impalad_args="--server_name=server1 --sentry_config=" + SENTRY_CONFIG_FILE,
      catalogd_args="--sentry_config=" + SENTRY_CONFIG_FILE,
      impala_log_dir=tempfile.mkdtemp(prefix="test_deprecated_none_",
      dir=os.getenv("LOG_DIR")))
  def test_deprecated_flag_doesnt_show(self):
    assert_no_files_in_dir_contain(self.impala_log_dir, "Ignoring removed flag "
                                                        "authorization_policy_file")

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args("--server_name=server1\
      --authorization_policy_file=ignored_file",
      impala_log_dir=tempfile.mkdtemp(prefix="test_deprecated_",
      dir=os.getenv("LOG_DIR")))
  def test_deprecated_flags(self):
    assert_file_in_dir_contains(self.impala_log_dir, "Ignoring removed flag "
                                                     "authorization_policy_file")

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
    catalogd_args="--sentry_config=%s" % SENTRY_CONFIG_FILE,
    impala_log_dir=tempfile.mkdtemp(prefix="test_catalog_restart_",
                                    dir=os.getenv("LOG_DIR")))
  def test_catalog_restart(self, unique_role):
    """IMPALA-7713: Tests that a catalogd restart when authorization is enabled should
    reset the previous privileges stored in impalad's catalog to avoid stale privilege
    data in the impalad's catalog."""
    def assert_privileges():
      result = self.client.execute("show grant role %s_foo" % unique_role)
      TestAuthorization._check_privileges(result, [["database", "functional",
                                                    "", "", "", "all", "false"]])

      result = self.client.execute("show grant role %s_bar" % unique_role)
      TestAuthorization._check_privileges(result, [["database", "functional_kudu",
                                                    "", "", "", "all", "false"]])

      result = self.client.execute("show grant role %s_baz" % unique_role)
      TestAuthorization._check_privileges(result, [["database", "functional_avro",
                                                    "", "", "", "all", "false"]])

    self.role_cleanup(unique_role)
    try:
      self.client.execute("create role %s_foo" % unique_role)
      self.client.execute("create role %s_bar" % unique_role)
      self.client.execute("create role %s_baz" % unique_role)
      self.client.execute("grant all on database functional to role %s_foo" %
                          unique_role)
      self.client.execute("grant all on database functional_kudu to role %s_bar" %
                          unique_role)
      self.client.execute("grant all on database functional_avro to role %s_baz" %
                          unique_role)

      assert_privileges()
      self._start_impala_cluster(["--catalogd_args=--sentry_config=%s" %
                                  SENTRY_CONFIG_FILE, "--restart_catalogd_only"])
      assert_privileges()
    finally:
      self.role_cleanup(unique_role)

  def role_cleanup(self, role_name_match):
    """Cleans up any roles that match the given role name."""
    for role_name in self.client.execute("show roles").data:
      if role_name_match in role_name:
        self.client.execute("drop role %s" % role_name)

  @staticmethod
  def _check_privileges(result, expected):
    def columns(row):
      cols = row.split("\t")
      return cols[0:len(cols) - 1]
    assert map(columns, result.data) == expected

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
    catalogd_args="--sentry_config=%s" % SENTRY_CONFIG_FILE,
    impala_log_dir=tempfile.mkdtemp(prefix="test_catalog_restart_",
                                    dir=os.getenv("LOG_DIR")))
  def test_catalog_object(self, unique_role):
    """IMPALA-7721: Tests /catalog_object web API for principal and privilege"""
    self.role_cleanup(unique_role)
    try:
      self.client.execute("create role %s" % unique_role)
      self.client.execute("grant select on database functional to role %s" % unique_role)
      for service in [self.cluster.catalogd.service,
                      self.cluster.get_first_impalad().service]:
        obj_dump = service.get_catalog_object_dump("PRINCIPAL", "%s.ROLE" % unique_role)
        assert "catalog_version" in obj_dump

        # Get the privilege associated with that principal ID.
        principal_id = re.search(r"principal_id \(i32\) = (\d+)", obj_dump)
        assert principal_id is not None
        obj_dump = service.get_catalog_object_dump("PRIVILEGE", urllib.quote(
            "server=server1->db=functional->action=select->grantoption=false.%s.ROLE" %
            principal_id.group(1)))
        assert "catalog_version" in obj_dump

        # Get the principal that does not exist.
        obj_dump = service.get_catalog_object_dump("PRINCIPAL", "doesnotexist.ROLE")
        assert "CatalogException" in obj_dump

        # Get the privilege that does not exist.
        obj_dump = service.get_catalog_object_dump("PRIVILEGE", urllib.quote(
            "server=server1->db=doesntexist->action=select->grantoption=false.%s.ROLE" %
            principal_id.group(1)))
        assert "CatalogException" in obj_dump
    finally:
      self.role_cleanup(unique_role)

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
    catalogd_args="--sentry_config=%s --sentry_catalog_polling_frequency_s=3600" %
                  SENTRY_CONFIG_FILE,
    impala_log_dir=tempfile.mkdtemp(prefix="test_invalidate_metadata_sentry_unavailable_",
                                    dir=os.getenv("LOG_DIR")))
  def test_invalidate_metadata_sentry_unavailable(self, unique_role):
    """IMPALA-7824: Tests that running INVALIDATE METADATA when Sentry is unavailable
    should not cause Impala to hang."""
    self.role_cleanup(unique_role)
    try:
      group_name = grp.getgrnam(getuser()).gr_name
      self.client.execute("create role %s" % unique_role)
      self.client.execute("grant all on server to role %s" % unique_role)
      self.client.execute("grant role %s to group `%s`" % (unique_role, group_name))

      self._stop_sentry_service()
      # Calling INVALIDATE METADATA when Sentry is unavailable should return an error.
      result = self.execute_query_expect_failure(self.client, "invalidate metadata")
      result_str = str(result)
      assert "MESSAGE: CatalogException: Error refreshing authorization policy:" \
             in result_str
      assert "CAUSED BY: ImpalaRuntimeException: Error refreshing authorization policy." \
             " Sentry is unavailable. Ensure Sentry is up:" in result_str

      self._start_sentry_service(SENTRY_CONFIG_FILE)
      # Calling INVALIDATE METADATA after Sentry is up should not return an error.
      self.execute_query_expect_success(self.client, "invalidate metadata")
    finally:
      self.role_cleanup(unique_role)

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
      impalad_args="--server_name=server1 --sentry_config=%s" % SENTRY_CONFIG_FILE,
      catalogd_args="--sentry_config=%s --sentry_catalog_polling_frequency_s=3600 " %
                    SENTRY_CONFIG_FILE,
      impala_log_dir=tempfile.mkdtemp(prefix="test_refresh_authorization_",
                                      dir=os.getenv("LOG_DIR")))
  def test_refresh_authorization(self, unique_role):
    """Tests refresh authorization statement by adding and removing roles and privileges
       externally. The long Sentry polling is used so that any authorization metadata
       updated externally does not get polled by Impala in order to test an an explicit
       call to refresh authorization statement."""
    group_name = grp.getgrnam(getuser()).gr_name
    self.role_cleanup(unique_role)
    for sync_ddl in [1, 0]:
      query_options = {'sync_ddl': sync_ddl}
      clients = []
      if sync_ddl:
        # When sync_ddl is True, we want to ensure the changes are propagated to all
        # coordinators.
        for impalad in self.cluster.impalads:
          clients.append(impalad.service.create_beeswax_client())
      else:
        clients.append(self.client)
      try:
        self.client.execute("create role %s" % unique_role)
        self.client.execute("grant role %s to group `%s`" % (unique_role, group_name))
        self.client.execute("grant refresh on server to %s" % unique_role)

        self.validate_refresh_authorization_roles(unique_role, query_options, clients)
        self.validate_refresh_authorization_privileges(unique_role, query_options,
                                                       clients)
      finally:
        self.role_cleanup(unique_role)

  def validate_refresh_authorization_roles(self, unique_role, query_options, clients):
    """This method tests refresh authorization statement by adding and removing
       roles externally."""
    try:
      # Create two roles inside Impala.
      self.client.execute("create role %s_internal1" % unique_role)
      self.client.execute("create role %s_internal2" % unique_role)
      # Drop an existing role (_internal1) outside Impala.
      role = "%s_internal1" % unique_role
      subprocess.check_call(
        ["/bin/bash", "-c",
         "%s/bin/sentryShell --conf %s/sentry-site.xml -dr -r %s" %
         (os.getenv("SENTRY_HOME"), os.getenv("SENTRY_CONF_DIR"), role)],
        stdout=sys.stdout, stderr=sys.stderr)

      result = self.execute_query_expect_success(self.client, "show roles")
      assert any(role in x for x in result.data)
      self.execute_query_expect_success(self.client, "refresh authorization",
                                        query_options=query_options)
      for client in clients:
        result = self.execute_query_expect_success(client, "show roles")
        assert not any(role in x for x in result.data)

      # Add a new role outside Impala.
      role = "%s_external" % unique_role
      subprocess.check_call(
          ["/bin/bash", "-c",
           "%s/bin/sentryShell --conf %s/sentry-site.xml -cr -r %s" %
           (os.getenv("SENTRY_HOME"), os.getenv("SENTRY_CONF_DIR"), role)],
          stdout=sys.stdout, stderr=sys.stderr)

      result = self.execute_query_expect_success(self.client, "show roles")
      assert not any(role in x for x in result.data)
      self.execute_query_expect_success(self.client, "refresh authorization",
                                        query_options=query_options)
      for client in clients:
        result = self.execute_query_expect_success(client, "show roles")
        assert any(role in x for x in result.data)
    finally:
      for suffix in ["internal1", "internal2", "external"]:
        self.role_cleanup("%s_%s" % (unique_role, suffix))

  def validate_refresh_authorization_privileges(self, unique_role, query_options,
                                                clients):
    """This method tests refresh authorization statement by adding and removing
       privileges externally."""
    # Grant select privilege outside Impala.
    subprocess.check_call(
        ["/bin/bash", "-c",
         "%s/bin/sentryShell --conf %s/sentry-site.xml -gpr -p "
         "'server=server1->db=functional->table=alltypes->action=select' -r %s" %
         (os.getenv("SENTRY_HOME"), os.getenv("SENTRY_CONF_DIR"), unique_role)],
        stdout=sys.stdout, stderr=sys.stderr)

    # Before refresh authorization, there should only be one refresh privilege.
    result = self.execute_query_expect_success(self.client, "show grant role %s" %
                                               unique_role)
    assert len(result.data) == 1
    assert any("refresh" in x for x in result.data)

    for client in clients:
      self.execute_query_expect_failure(client,
                                        "select * from functional.alltypes limit 1")

    self.execute_query_expect_success(self.client, "refresh authorization",
                                      query_options=query_options)

    for client in clients:
      # Ensure select privilege was granted after refresh authorization.
      result = self.execute_query_expect_success(client, "show grant role %s" %
                                                 unique_role)
      assert len(result.data) == 2
      assert any("select" in x for x in result.data)
      assert any("refresh" in x for x in result.data)
      self.execute_query_expect_success(client,
                                        "select * from functional.alltypes limit 1")

  @staticmethod
  def _verify_show_dbs(result, unique_name, visibility_privileges=PRIVILEGES):
    """ Helper function for verifying the results of SHOW DATABASES below.
    Only show databases with privileges implying any of the visibility_privileges.
    """
    for priv in PRIVILEGES:
      # Result lines are in the format of "db_name\tdb_comment"
      db_name = 'db_%s_%s\t' % (unique_name, priv)
      if priv != 'all' and priv not in visibility_privileges:
        assert db_name not in result.data
      else:
        assert db_name in result.data

  def _test_sentry_show_stmts_helper(self, unique_role, unique_name,
                                     visibility_privileges):
    unique_db = unique_name + "_db"
    # TODO: can we create and use a temp username instead of using root?
    another_user = 'root'
    another_user_grp = 'root'
    self.role_cleanup(unique_role)
    try:
      self.client.execute("create role %s" % unique_role)
      self.client.execute("grant create on server to role %s" % unique_role)
      self.client.execute("grant drop on server to role %s" % unique_role)
      self.client.execute("grant role %s to group %s" %
                          (unique_role, grp.getgrnam(getuser()).gr_name))

      self.client.execute("drop database if exists %s cascade" % unique_db)
      self.client.execute("create database %s" % unique_db)
      for priv in PRIVILEGES:
        self.client.execute("create database db_%s_%s" % (unique_name, priv))
        self.client.execute("grant {0} on database db_{1}_{2} to role {3}"
                            .format(priv, unique_name, priv, unique_role))
        self.client.execute("create table %s.tbl_%s (i int)" % (unique_db, priv))
        self.client.execute("grant {0} on table {1}.tbl_{2} to role {3}"
                            .format(priv, unique_db, priv, unique_role))
      self.client.execute("grant role %s to group %s" %
                          (unique_role, another_user_grp))

      # Owner (current user) can still see all the owned databases and tables
      result = self.client.execute("show databases")
      TestAuthorization._verify_show_dbs(result, unique_name)
      result = self.client.execute("show tables in %s" % unique_db)
      assert result.data == ["tbl_%s" % p for p in PRIVILEGES]

      # Check SHOW DATABASES and SHOW TABLES using another username
      # Create another client so we can user another username
      root_impalad_client = self.create_impala_client()
      result = self.execute_query_expect_success(
          root_impalad_client, "show databases", user=another_user)
      TestAuthorization._verify_show_dbs(result, unique_name, visibility_privileges)
      result = self.execute_query_expect_success(
          root_impalad_client, "show tables in %s" % unique_db, user=another_user)
      # Only show tables with privileges implying any of the visibility privileges
      assert 'tbl_all' in result.data   # ALL can imply to any privilege
      for p in visibility_privileges:
        assert 'tbl_%s' % p in result.data
    finally:
      self.client.execute("drop database if exists %s cascade" % unique_db)
      for priv in PRIVILEGES:
        self.client.execute(
            "drop database if exists db_%s_%s cascade" % (unique_name, priv))
      self.role_cleanup(unique_role)

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server_name=server1 --sentry_config=%s "
                 "--authorized_proxy_user_config=%s=* "
                 "--min_privilege_set_for_show_stmts=select" %
                 (SENTRY_CONFIG_FILE, getuser()),
    catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
    sentry_config=SENTRY_CONFIG_FILE_OO,  # Enable Sentry Object Ownership
    sentry_log_dir="{0}/test_sentry_show_stmts_with_select".format(SENTRY_BASE_LOG_DIR))
  def test_sentry_show_stmts_with_select(self, unique_role, unique_name):
    self._test_sentry_show_stmts_helper(unique_role, unique_name, ['select'])

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server_name=server1 --sentry_config=%s "
                 "--authorized_proxy_user_config=%s=* "
                 "--min_privilege_set_for_show_stmts=select,insert" %
                 (SENTRY_CONFIG_FILE, getuser()),
    catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
    sentry_config=SENTRY_CONFIG_FILE_OO,  # Enable Sentry Object Ownership
    sentry_log_dir="{0}/test_sentry_show_stmts_with_select_insert"
                   .format(SENTRY_BASE_LOG_DIR))
  def test_sentry_show_stmts_with_select_insert(self, unique_role, unique_name):
    self._test_sentry_show_stmts_helper(unique_role, unique_name,
                                        ['select', 'insert'])

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server_name=server1 --sentry_config=%s "
                 "--authorized_proxy_user_config=%s=* "
                 "--min_privilege_set_for_show_stmts=any" %
                 (SENTRY_CONFIG_FILE, getuser()),
    catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
    sentry_config=SENTRY_CONFIG_FILE_OO,  # Enable Sentry Object Ownership
    sentry_log_dir="{0}/test_sentry_show_stmts_with_any".format(SENTRY_BASE_LOG_DIR))
  def test_sentry_show_stmts_with_any(self, unique_role, unique_name):
    self._test_sentry_show_stmts_helper(unique_role, unique_name, PRIVILEGES)

  def _test_ranger_show_stmts_helper(self, unique_name, visibility_privileges):
    unique_db = unique_name + "_db"
    admin_client = self.create_impala_client()
    try:
      admin_client.execute("drop database if exists %s cascade" % unique_db, user=ADMIN)
      admin_client.execute("create database %s" % unique_db, user=ADMIN)
      for priv in PRIVILEGES:
        admin_client.execute("create database db_%s_%s" % (unique_name, priv))
        admin_client.execute("grant {0} on database db_{1}_{2} to user {3}"
                             .format(priv, unique_name, priv, getuser()))
        admin_client.execute("create table %s.tbl_%s (i int)" % (unique_db, priv))
        admin_client.execute("grant {0} on table {1}.tbl_{2} to user {3}"
                             .format(priv, unique_db, priv, getuser()))

      # Admin can still see all the databases and tables
      result = admin_client.execute("show databases")
      TestAuthorization._verify_show_dbs(result, unique_name)
      result = admin_client.execute("show tables in %s" % unique_db)
      assert result.data == ["tbl_%s" % p for p in PRIVILEGES]

      # Check SHOW DATABASES and SHOW TABLES using another username
      result = self.client.execute("show databases")
      TestAuthorization._verify_show_dbs(result, unique_name, visibility_privileges)
      result = self.client.execute("show tables in %s" % unique_db)
      # Only show tables with privileges implying any of the visibility privileges
      assert 'tbl_all' in result.data   # ALL can imply to any privilege
      for p in visibility_privileges:
        assert 'tbl_%s' % p in result.data
    finally:
      admin_client.execute("drop database if exists %s cascade" % unique_db)
      for priv in PRIVILEGES:
        admin_client.execute(
            "drop database if exists db_%s_%s cascade" % (unique_name, priv))

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server-name=server1 --ranger_service_type=hive "
                 "--ranger_app_id=impala --authorization_provider=ranger "
                 "--min_privilege_set_for_show_stmts=select",
    catalogd_args="--server-name=server1 --ranger_service_type=hive "
                  "--ranger_app_id=impala --authorization_provider=ranger")
  def test_ranger_show_stmts_with_select(self, unique_name):
    self._test_ranger_show_stmts_helper(unique_name, ['select'])

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server-name=server1 --ranger_service_type=hive "
                 "--ranger_app_id=impala --authorization_provider=ranger "
                 "--min_privilege_set_for_show_stmts=select,insert",
    catalogd_args="--server-name=server1 --ranger_service_type=hive "
                  "--ranger_app_id=impala --authorization_provider=ranger")
  def test_ranger_show_stmts_with_select_insert(self, unique_name):
    self._test_ranger_show_stmts_helper(unique_name, ['select', 'insert'])

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server-name=server1 --ranger_service_type=hive "
                 "--ranger_app_id=impala --authorization_provider=ranger "
                 "--min_privilege_set_for_show_stmts=any",
    catalogd_args="--server-name=server1 --ranger_service_type=hive "
                  "--ranger_app_id=impala --authorization_provider=ranger")
  def test_ranger_show_stmts_with_any(self, unique_name):
    self._test_ranger_show_stmts_helper(unique_name, PRIVILEGES)

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server-name=server1 --ranger_service_type=hive "
                 "--ranger_app_id=impala --authorization_provider=ranger "
                 "--num_check_authorization_threads=%d" % (random.randint(2, 128)),
    catalogd_args="--server-name=server1 --ranger_service_type=hive "
                  "--ranger_app_id=impala --authorization_provider=ranger")
  def test_num_check_authorization_threads_with_ranger(self, unique_name):
    self._test_ranger_show_stmts_helper(unique_name, PRIVILEGES)

  @pytest.mark.execute_serially
  @CustomClusterTestSuite.with_args(
    impalad_args="--server_name=server1 --sentry_config=%s "
                 "--authorized_proxy_user_config=%s=* "
                 "--num_check_authorization_threads=%d" %
                 (SENTRY_CONFIG_FILE, getuser(), random.randint(2, 128)),
    catalogd_args="--sentry_config={0}".format(SENTRY_CONFIG_FILE),
    sentry_config=SENTRY_CONFIG_FILE_OO,  # Enable Sentry Object Ownership
    sentry_log_dir="{0}/test_num_check_authorization_threads_with_sentry"
                   .format(SENTRY_BASE_LOG_DIR))
  def test_num_check_authorization_threads_with_sentry(self, unique_role, unique_name):
    self._test_sentry_show_stmts_helper(unique_role, unique_name, PRIVILEGES)