thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-21 15:12:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

add some except error test

Browse Source
This commit is contained in:
taichong 2020-11-10 17:08:12 +08:00
parent 5c3d4b2245
commit 3287f691ee
2 changed files with 29 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests/integration/test_materialize_mysql_database/materialize_with_ddl.py
View File

@ -5,6 +5,8 @@ import pymysql.cursors
import pytest
from helpers.client import QueryRuntimeException
from helpers.network import PartitionManager
import pytest
from helpers.client import QueryRuntimeException
def check_query(clickhouse_node, query, result_set, retry_count=60, interval_seconds=3):
lastest_result = ''
@ -557,28 +559,42 @@ def err_sync_user_privs_with_materialize_mysql_database(clickhouse_node, mysql_n
mysql_node.query("DROP DATABASE priv_err_db;")
mysql_node.grant_min_priv_for_user("test")
def restore_instance_mysql_connections(clickhouse_node, pm, action='DROP'):
pm._check_instance(clickhouse_node)
pm._delete_rule({'source': clickhouse_node.ip_address, 'destination_port': 3306, 'action': action})
pm._delete_rule({'destination': clickhouse_node.ip_address, 'source_port': 3306, 'action': action})
time.sleep(5)
def drop_instance_mysql_connections(clickhouse_node, pm, action='DROP'):
pm._check_instance(clickhouse_node)
pm._add_rule({'source': clickhouse_node.ip_address, 'destination_port': 3306, 'action': action})
pm._add_rule({'destination': clickhouse_node.ip_address, 'source_port': 3306, 'action': action})
time.sleep(5)
def network_partition_test(clickhouse_node, mysql_node, service_name):
mysql_node.query("CREATE DATABASE test_database;")
mysql_node.query("CREATE DATABASE test;")
mysql_node.query("CREATE TABLE test_database.test_table ( `id` int(11) NOT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB;")
with PartitionManager() as pm:
clickhouse_node.query(
"CREATE DATABASE test_database ENGINE = MaterializeMySQL('{}:3306', 'test_database', 'root', 'clickhouse')".format(service_name))
pm._check_instance(clickhouse_node)
pm._add_rule({'source': clickhouse_node.ip_address, 'destination_port': 3306, 'action': 'DROP'})
pm._add_rule({'destination': clickhouse_node.ip_address, 'source_port': 3306, 'action': 'DROP'})
time.sleep(5)
drop_instance_mysql_connections(clickhouse_node, pm)
mysql_node.query('INSERT INTO test_database.test_table VALUES(1)')
check_query(clickhouse_node, "SHOW TABLES FROM test_database FORMAT TSV", "test_table\n")
check_query(clickhouse_node, "SELECT * FROM test_database.test_table FORMAT TSV", '')
check_query(clickhouse_node, "SELECT * FROM test_database.test_table", '')
with pytest.raises(QueryRuntimeException) as exception:
clickhouse_node.query(
"CREATE DATABASE test ENGINE = MaterializeMySQL('{}:3306', 'test', 'root', 'clickhouse')".format(service_name))
assert "Can't connect to MySQL server" in str(exception.value)
clickhouse_node.query("DETACH DATABASE test_database")
pm._delete_rule({'source': clickhouse_node.ip_address, 'destination_port': 3306, 'action': 'DROP'})
pm._delete_rule({'destination': clickhouse_node.ip_address, 'source_port': 3306, 'action': 'DROP'})
time.sleep(5)
restore_instance_mysql_connections(clickhouse_node, pm)
clickhouse_node.query("ATTACH DATABASE test_database")
check_query(clickhouse_node, "SELECT * FROM test_database.test_table FORMAT TSV", '1\n')

12
tests/integration/test_materialize_mysql_database/test.py
View File

@ -20,24 +20,22 @@ def started_cluster():
cluster.start()
yield cluster
finally:
cluster.shutdown()
#cluster.shutdown()
pass
class MySQLNodeInstance:
def __init__(self, user='root', password='clickhouse', hostname='127.0.0.1', port=3308):
def __init__(self, user='root', password='clickhouse', ip_address='127.0.0.1', port=3308):
self.user = user
self.port = port
self.hostname = hostname
self.ip_address = ip_address
self.password = password
self.mysql_connection = None # lazy init
def getPort(self):
if self.port is not None:
return self.port
def alloc_connection(self):
if self.mysql_connection is None:
self.mysql_connection = pymysql.connect(user=self.user, password=self.password, host=self.hostname,
self.mysql_connection = pymysql.connect(user=self.user, password=self.password, host=self.ip_address,
port=self.port, autocommit=True)
return self.mysql_connection