import random import string import time import logging import pymysql import json import uuid import semver import exrex from . import schemas logger = logging.getLogger(__name__) def test_get_instances_list(cli, unauthorised_cli): get_instance_list(cli) assert get_instance_list(unauthorised_cli).json()["code"] == 401 def test_post_instance(cli, unauthorised_cli): # test clould create instance normal. instance = create_instance(cli, schemas.new_cluster_info()) assert_success(cli, instance) # TODO should delete the instance after test. # test should be failed when replicationFactor does't match replicas instance = create_instance(cli, schemas.failure_info(), replicationFactor=7) assert instance['status'] == 'Failure' # test should be failed when no azs. instance = create_instance(cli, schemas.failure_info(), azs=None) assert instance['status'] == 'Failure' # test should be failed when no azs not in plat. azs = ['t1', 't2', 't3'] instance = create_instance(cli, schemas.failure_info(), azs=azs) assert instance['status'] == 'Failure' # test should be failed when no azs > 3. azs = ['t1', 't2', 't3', 't4'] instance = create_instance(cli, schemas.failure_info(), azs=azs) assert instance['status'] == 'Failure' # TODO Need more cases and result assert. unauthorised_instance = create_instance(unauthorised_cli) assert unauthorised_instance['code'] == 401 def test_get_instance_by_id(cli, unauthorised_cli): instance_id = get_instance_id(cli) path = "/v1/tidbclusters/" + instance_id cli.get(path, schemas.cluster_info()) assert unauthorised_cli.get(path).json()["code"] == 401 def test_get_tikv_by_id(cli, unauthorised_cli): instance_id = get_instance_id(cli) path = "/v1/namespaces/" + instance_id + "/tikv" cli.get(path, schemas.tikv_list_info(10)) assert unauthorised_cli.get(path).json()["code"] == 401 def test_get_pd_by_id(cli, unauthorised_cli): instance_id = get_instance_id(cli) path = "/v1/namespaces/" + instance_id + "/pd" cli.get(path, schemas.pd_list_info(10)) assert unauthorised_cli.get(path).json()["code"] == 401 def test_get_tidb_by_id(cli, unauthorised_cli): instance_id = get_instance_id(cli) path = "/v1/namespaces/" + instance_id + "/tidb" cli.get(path, schemas.tidb_list_info(10)) assert unauthorised_cli.get(path).json()["code"] == 401 def test_delete_instance_by_id(cli, unauthorised_cli): instance_id = get_instance_id(cli) path = "/v1/tidbclusters/" + instance_id assert cli.delete(path).json()["code"] == 200 assert cli.get(path).json()["code"] == 404 assert unauthorised_cli.delete(path).json()["code"] == 401 def test_update_instance_by_id(cli, unauthorised_cli): instance_id = get_instance_id(cli) path = "/v1/tidbclusters/" + instance_id instance = cli.get(path).json() clusterName = exrex.getone('^[a-zA-Z][a-zA-Z0-9-]{7,31}$') tags = [str(uuid.uuid1())] version = "v3.0.5" tidb_replicas = instance["tidb"]["replicas"] + 2 tikv_replicas = instance["tikv"]["replicas"] + 2 update_body = { "clusterName": clusterName, "tags": tags, "tidb": { "replicas": tidb_replicas }, "tikv": { "replicas": tikv_replicas }, "version": version } res = cli.patch(path, json=update_body, headers={"Content-Type": "application/strategic-merge-patch+json"}).json() logger.info("patch result: {}".format(res)) if semver.compare(version.lstrip("v"), instance["version"].lstrip("v")) >= 1: assert res["clusterName"] == clusterName and \ res["tags"] == tags and \ res["tidb"]["replicas"] == tidb_replicas and \ res["tikv"]["replicas"] == tikv_replicas and \ res["version"] == version else: assert res["code"] == 400 assert unauthorised_cli.patch(path, json=update_body, headers={"Content-Type": "application/strategic-merge-patch+json"}).json()[ "code"] == 401 def test_get_failed_instances(cli, unauthorised_cli): path = "/v1/failed-tidbclusters" res = cli.get(path, schemas.failed_cluster_info(10)) assert unauthorised_cli.get(path).json()["code"] == 401 def get_instance_id(cli): instance_list = get_instance_list(cli).json() if len(instance_list['items']) == 0: instance = create_instance(cli, schemas.new_cluster_info()) assert_success(cli, instance) instance_id = instance['metadata']['name'] else: instance_id = instance_list['items'][0]['metadata']['name'] return instance_id def get_instance_list(cli): path = '/v1/tidbclusters' schema = schemas.clusters_list(10) return cli.get(path, schema=schema) def assert_success(cli, instance): instance_id, password = instance['metadata']['name'], instance['password'] # endpoints maybe null before it's running, instance = wait_instance_final_status(cli, instance_id) assert instance['status'] == "running" host, port = instance['endpoints'][0], instance['port'] assert connect_tidb(host, port, password) def create_instance(cli, schema=None, **kwargs): cluster_name = "CI" + "".join(random.choices(string.ascii_letters + string.digits, k=random.randint(6, 30))) path = '/v1/tidbclusters' default_instance = { "replicationFactor": 3, "clusterName": cluster_name, "azs": ["cn-east-1", "cn-west-1", "cn-north-1"], "pd": { "replicas": 3, "resourceTemplate": { "cpu": 1, "memory": 1 }, "storageClassName": "rook-ceph-block" }, "tidb": { "replicas": 1, "resourceTemplate": { "cpu": 1, "memory": 1 } }, "tikv": { "replicas": 3, "resourceTemplate": { "cpu": 1, "memory": 1 }, "storageClassName": "rook-ceph-block" }, "monitor": { "storageClassName": "rook-ceph-block" }, "password": "Pa$$word123", "tags": ["fromCI"], "version": "v3.0.4" } default_instance.update(kwargs) logger.info("start to create instance with Parameters: {}".format(default_instance)) if schema: instance = cli.post(path, json=default_instance, schema=schema).json() else: instance = cli.post(path, json=default_instance).json() return instance def wait_instance_final_status(cli, instance_id): """ get instance the finally status """ path = "/v1/tidbclusters/" + instance_id # wait for 15 minutes or consider it is failed. for i in range(90): instance = cli.get(path, schemas.cluster_info()).json() status = instance['status'] roll_status = ["creating", "upgrading", "scaling", "recovering"] if status not in roll_status: return instance else: logger.info("the instance {} status is {}".format(instance_id, status)) time.sleep(10) logger.error("get instance final status timeout.") return False def connect_tidb(host, port, password): result = None connection = pymysql.connect(host=host, user="root", port=port, password=password, db="test") try: with connection.cursor() as cursor: # Create table cursor.execute("DROP TABLE IF EXISTS `users`;") create_table_sql = """ CREATE TABLE `users` ( `id` int(11) NOT NULL AUTO_INCREMENT, `email` varchar(255) COLLATE utf8_bin NOT NULL, `password` varchar(255) COLLATE utf8_bin NOT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin AUTO_INCREMENT=1 ; """ cursor.execute(create_table_sql) # Create a new record sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)" cursor.execute(sql, ('tep@pingcap.com', 'very-secret')) connection.commit() with connection.cursor() as cursor: # Read a single record sql = "SELECT `id`, `password` FROM `users` WHERE `email`=%s" cursor.execute(sql, ('tep@pingcap.com',)) result = cursor.fetchone() finally: connection.close() return result == (1, 'very-secret') # def test_get_instance_detail(cli, unauthorised_cli): # path = "/v1/tidbclusters/" + instance_id # assert unauthorised_cli.get(path).json()['code'] == 401 # cli.get(path, schemas.cluster_info()).json()