ClickHouse/tests/integration/test_kerberos_auth/test.py
Max Kainov 9f3c625ab5 fix mysql client tests to use mysql8 image
#ci_set_integration
 #ci_set_arm
2024-03-20 17:35:11 +00:00

92 lines
2.2 KiB
Python

import pytest
from helpers.cluster import ClickHouseCluster, is_arm
if is_arm():
pytestmark = pytest.mark.skip
cluster = ClickHouseCluster(__file__)
instance1 = cluster.add_instance(
"instance1",
main_configs=["configs/kerberos_with_keytab.xml"],
user_configs=["configs/users.xml"],
with_kerberos_kdc=True,
)
instance2 = cluster.add_instance(
"instance2",
main_configs=["configs/kerberos_without_keytab.xml"],
user_configs=["configs/users.xml"],
with_kerberos_kdc=True,
)
instance3 = cluster.add_instance(
"instance3",
main_configs=["configs/kerberos_bad_path_to_keytab.xml"],
user_configs=["configs/users.xml"],
with_kerberos_kdc=True,
)
client = cluster.add_instance(
"client",
main_configs=["configs/kerberos_without_keytab.xml"],
user_configs=["configs/users.xml"],
with_kerberos_kdc=True,
)
# Fixtures
@pytest.fixture(scope="module")
def kerberos_cluster():
try:
cluster.start()
yield cluster
finally:
cluster.shutdown()
# Tests
def make_auth(instance):
instance_ip = cluster.get_instance_ip(instance.name)
client.exec_in_container(
["bash", "-c", f"echo '{instance_ip} {instance.hostname}' >> /etc/hosts"]
)
client.exec_in_container(
["bash", "-c", "kinit -k -t /tmp/keytab/kuser.keytab kuser"]
)
return client.exec_in_container(
[
"bash",
"-c",
f"echo 'select currentUser()' | curl --negotiate -u : http://{instance.hostname}:8123/ --data-binary @-",
]
)
def test_kerberos_auth_with_keytab(kerberos_cluster):
assert make_auth(instance1) == "kuser\n"
def test_kerberos_auth_without_keytab(kerberos_cluster):
assert (
"DB::Exception: : Authentication failed: password is incorrect, or there is no user with such name."
in make_auth(instance2)
)
def test_bad_path_to_keytab(kerberos_cluster):
assert (
"DB::Exception: : Authentication failed: password is incorrect, or there is no user with such name."
in make_auth(instance3)
)
assert instance3.contains_in_log("Keytab file not found")
if __name__ == "__main__":
cluster.start()
input("Cluster created, press any key to destroy...")
cluster.shutdown()