added mysql protocol test

dbms/programs/server/MySQLHandler.cpp

@@ -36,7 +36,7 @@ void MySQLHandler::run()
 
     try
     {
-        Handshake handshake(connection_id, VERSION_FULL);
+        Handshake handshake(connection_id, VERSION_STRING);
         packet_sender.sendPacket<Handshake>(handshake, true);
 
         LOG_TRACE(log, "Sent handshake");
@@ -82,10 +82,24 @@ void MySQLHandler::run()
             packet.readPayload(std::move(handshake_response.auth_response));
             password = packet.value;
         }
-        connection_context.setUser(handshake_response.username, password, socket().address(), "");
-        connection_context.setCurrentDatabase(handshake_response.database);
-        connection_context.setCurrentQueryId("");
-        LOG_ERROR(log, "Authentication for user " << handshake_response.username << " succeeded.");
+        LOG_TRACE(log, "password: " << password);
+        try
+        {
+            connection_context.setUser(handshake_response.username, password, socket().address(), "");
+            connection_context.setCurrentDatabase(handshake_response.database);
+            connection_context.setCurrentQueryId("");
+            LOG_ERROR(log, "Authentication for user " << handshake_response.username << " succeeded.");
+        }
+        catch (const NetException &)
+        {
+            throw;
+        }
+        catch (const Exception & exc)
+        {
+            LOG_ERROR(log, "Authentication for user " << handshake_response.username << " failed.");
+            packet_sender.sendPacket(ERR_Packet(exc.code(), "00000", exc.message()), true);
+            throw;
+        }
 
         OK_Packet ok_packet(0, handshake_response.capability_flags, 0, 0, 0, 0, "");
         packet_sender.sendPacket(ok_packet, true);
@@ -187,6 +201,10 @@ void MySQLHandler::comQuery(String payload)
 
         LOG_TRACE(log, "Sent columns definitions.");
 
+        if (!(capabilities & Capability::CLIENT_DEPRECATE_EOF)) {
+            packet_sender.sendPacket(EOF_Packet(0, 0));
+        }
+
         while (Block block = res.in->read())
         {
             size_t rows = block.rows();

dbms/programs/server/MySQLHandler.h

@@ -34,9 +34,6 @@ private:
     Poco::Logger * log;
     Context connection_context;
 
-    std::shared_ptr<ReadBuffer> in;
-    std::shared_ptr<WriteBuffer> out;
-
     MySQLProtocol::PacketSender packet_sender;
 
     uint32_t connection_id = 0;

dbms/src/Core/MySQLProtocol.h

@@ -365,7 +365,7 @@ public:
     {
         String result;
         result.append(1, 0xfe);
-        result.append(plugin_name);
+        writeNulTerminatedString(result, plugin_name);
         result.append(auth_plugin_data);
         return result;
     }

dbms/tests/integration/helpers/cluster.py

@@ -102,7 +102,7 @@ class ClickHouseCluster:
         self.with_hdfs = False
         self.with_mongo = False
 
-        self.docker_client = None
+        self.docker_client = None  # type: docker.DockerClient
         self.is_up = False
 
 

dbms/tests/integration/test_mysql_protocol/configs/config.xml

+<?xml version="1.0"?>
+<yandex>
+    <logger>
+        <level>trace</level>
+        <log>/var/log/clickhouse-server/clickhouse-server.log</log>
+        <errorlog>/var/log/clickhouse-server/clickhouse-server.err.log</errorlog>
+        <size>1000M</size>
+        <count>10</count>
+    </logger>
+
+    <tcp_port>9000</tcp_port>
+    <mysql_port>9001</mysql_port>
+    <listen_host>127.0.0.1</listen_host>
+
+    <max_concurrent_queries>500</max_concurrent_queries>
+    <mark_cache_size>5368709120</mark_cache_size>
+    <path>./clickhouse/</path>
+    <users_config>users.xml</users_config>
+
+</yandex>

dbms/tests/integration/test_mysql_protocol/configs/users.xml

+<?xml version="1.0"?>
+<yandex>
+    <profiles>
+        <default>
+        </default>
+    </profiles>
+
+    <users>
+        <default>
+            <password>123</password>
+            <networks incl="networks" replace="replace">
+                <ip>::/0</ip>
+            </networks>
+            <profile>default</profile>
+            <quota>default</quota>
+        </default>
+    </users>
+
+    <quotas>
+        <default>
+        </default>
+    </quotas>
+</yandex>

dbms/tests/integration/test_mysql_protocol/test.py

+# coding: utf-8
+
+import os
+import pytest
+import pymysql.connections
+
+from docker.models.containers import Container
+
+from helpers.cluster import ClickHouseCluster
+
+
+SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
+
+cluster = ClickHouseCluster(__file__, base_configs_dir=os.path.join(SCRIPT_DIR, './configs'))
+
+node = cluster.add_instance('node', with_mysql=True)
+
+server_port = 9001
+
+
+@pytest.fixture(scope="module")
+def started_cluster():
+    cluster.start()
+    try:
+        yield cluster
+    finally:
+        cluster.shutdown()
+
+
+@pytest.fixture(scope='module')
+def mysql_container(started_cluster):
+    # type: (ClickHouseCluster) -> Container
+    yield started_cluster.docker_client.containers.get(started_cluster.get_instance_docker_id('mysql1'))
+
+
+@pytest.fixture(scope='module')
+def server_address(started_cluster):
+    yield started_cluster.get_instance_ip('node')
+
+
+def test_select(mysql_container, server_address):
+    # type: (Container, str) -> None
+    code, (stdout, stderr) = mysql_container.exec_run('''
+        mysql --protocol tcp -h {host} -P {port} default -u default --enable-cleartext-plugin --password=123
+        -e "select 1 as a;"
+        -e "select 'тест' as b;"
+    '''.format(host=server_address, port=server_port), demux=True)
+
+    assert stdout == 'a\n1\nb\nтест\n'
+
+
+def test_authentication(mysql_container, server_address):
+    # type: (Container, str) -> None
+
+    code, (stdout, stderr) = mysql_container.exec_run('''
+        mysql --protocol tcp -h {host} -P {port} default -u default --enable-cleartext-plugin --password=abc -e "select 1 as a;"
+    '''.format(host=server_address, port=server_port), demux=True)
+
+    assert stderr == 'mysql: [Warning] Using a password on the command line interface can be insecure.\n' \
+                     'ERROR 193 (00000): Wrong password for user default\n'
+
+
+def test_change_database(mysql_container, server_address):
+    # type: (Container, str) -> None
+
+    code, (stdout, stderr) = mysql_container.exec_run('''
+        mysql --protocol tcp -h {host} -P {port} default -u default --enable-cleartext-plugin --password=123
+        -e "use system;"
+        -e "select count(*) from (select name from tables limit 1);"
+        -e "use system2;"
+    '''.format(host=server_address, port=server_port), demux=True)
+
+    assert stdout == 'count()\n1\n'
+    assert stderr == "mysql: [Warning] Using a password on the command line interface can be insecure.\n" \
+                     "ERROR 81 (00000) at line 1: Database system2 doesn't exist\n"
+
+
+def test_py_client(server_address):
+    with pytest.raises(pymysql.InternalError) as exc_info:
+        pymysql.connections.Connection(host=server_address, user='default', password='abacab', database='default', port=server_port)
+
+    assert exc_info.value.args == (193, 'Wrong password for user default')
+
+    client = pymysql.connections.Connection(host=server_address, user='default', password='123', database='default', port=server_port)
+
+    with pytest.raises(pymysql.InternalError) as exc_info:
+        client.query('select name from tables')
+
+    assert exc_info.value.args == (60, "Table default.tables doesn't exist.")
+
+    cursor = client.cursor(pymysql.cursors.DictCursor)
+    cursor.execute("select 1 as a, 'тест' as b")
+    assert cursor.fetchall() == [{'a': '1', 'b': 'тест'}]
+
+    client.select_db('system')
+
+    with pytest.raises(pymysql.InternalError) as exc_info:
+        client.select_db('system2')
+
+    assert exc_info.value.args == (81, "Database system2 doesn't exist")