From 950bccb130997b7d0e4b49c0344a2313421abb9c Mon Sep 17 00:00:00 2001 From: Vitaly Baranov Date: Sat, 7 Nov 2020 14:42:29 +0300 Subject: [PATCH] Add test for authentication. --- tests/integration/test_grpc_protocol/test.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/integration/test_grpc_protocol/test.py b/tests/integration/test_grpc_protocol/test.py index fcd214509d6..6589e2336d7 100644 --- a/tests/integration/test_grpc_protocol/test.py +++ b/tests/integration/test_grpc_protocol/test.py @@ -41,7 +41,8 @@ def create_channel(): main_channel = channel return channel -def query_common(query_text, settings={}, input_data=[], input_data_delimiter='', output_format='TabSeparated', external_tables=[], query_id='123', session_id='', stream_output=False, channel=None): +def query_common(query_text, settings={}, input_data=[], input_data_delimiter='', output_format='TabSeparated', external_tables=[], + user_name='', password='', query_id='123', session_id='', stream_output=False, channel=None): if type(input_data) == str: input_data = [input_data] if not channel: @@ -50,8 +51,8 @@ def query_common(query_text, settings={}, input_data=[], input_data_delimiter='' def query_info(): input_data_part = input_data.pop(0) if input_data else '' return clickhouse_grpc_pb2.QueryInfo(query=query_text, settings=settings, input_data=input_data_part, input_data_delimiter=input_data_delimiter, - output_format=output_format, external_tables=external_tables, query_id=query_id, session_id=session_id, - next_query_info=bool(input_data)) + output_format=output_format, external_tables=external_tables, user_name=user_name, password=password, + query_id=query_id, session_id=session_id, next_query_info=bool(input_data)) def send_query_info(): yield query_info() while input_data: @@ -201,6 +202,10 @@ def test_errors_handling(): e = query_and_get_error("CREATE TABLE t (a UInt8) ENGINE = Memory") assert "Table default.t already exists" in e.display_text +def test_authentication(): + query("CREATE USER john IDENTIFIED BY 'qwe123'") + assert query("SELECT currentUser()", user_name="john", password="qwe123") == "john\n" + def test_logs(): logs = query_and_get_logs("SELECT 1", settings={'send_logs_level':'debug'}) assert "SELECT 1" in logs