// Copyright 2017 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.package spec

package e2e

import (
	"database/sql"
	"errors"
	"fmt"
	"os/exec"
	"strings"
	"time"

	"github.com/ngaut/log"
	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"

	"github.com/pingcap/tidb-cloud-manager/pkg/manager"
	tiapi "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/api"
	tictl "github.com/pingcap/tidb-cloud-manager/pkg/tidbinstance/controller"
	"github.com/pingcap/tidb-cloud-manager/test/e2e/framework"
	"github.com/pingcap/private-tidb-operator/pkg/util/label"
	"github.com/pingcap/private-tidb-operator/pkg/volumeprovisioner"
	"k8s.io/api/core/v1"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"
)

const clusterName = "tidb-cloud-manager-e2e"

var _ = Describe("ManageTidbCluster", func() {

	newPassword := "pingcape2e"
	tcName := "tidb-cluster"
	password := "pingcap123456"
	pushgatewayAnnKey := "pushgateway-image"

	kafkaAddr := fmt.Sprintf("kafka-0.kafka-svc.%s.svc.cluster.local:9092", clusterName)
	binlogImage := "localhost:5000/pingcap/tidb-binlog:latest"
	upgradeBinlogImage := "uhub.ucloud.cn/pingcap/tidb-binlog:latest"
	pushgatewayAnnValue := "localhost:5000/pingcap/pushgateway:v0.3.1"

	It("start functional testing for tidb-cloud-manager", func() {
		cm, err := framework.NewClusterManger(clusterName)
		Expect(err).NotTo(HaveOccurred())
		tc, err := framework.NewTidbCluster(cm.ClusterName)
		Expect(err).NotTo(HaveOccurred())

		By("When TidbCluster is created")
		err = wait.Poll(10*time.Second, 10*time.Minute, func() (bool, error) {
			err = cm.CreateTidbCluster(tc)
			if err != nil {
				log.Error(err)
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("Then TidbCluster exists")
		err = wait.Poll(5*time.Second, 30*time.Minute, func() (bool, error) {
			clusters, err := cm.ListTidbClusters()
			if err != nil {
				return false, nil
			}
			if clusters == nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("And TidbCluster's retentionDuration is 6m")
		tcobj, err := cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
		Expect(err).NotTo(HaveOccurred())

		By(fmt.Sprintf("And make sure annotation %s has existed in TidbCluster CRD", pushgatewayAnnKey))
		Expect(tcobj.Annotations[pushgatewayAnnKey]).To(Equal(pushgatewayAnnValue))

		By("And make sure TidbCluster has initialized")
		rtc := &manager.Cluster{}
		err = wait.Poll(10*time.Second, 30*time.Minute, func() (bool, error) {
			rtc, err = cm.GetTidbCluster()
			if err != nil {
				return false, nil
			}
			err = cm.CheckClusterStatus(tc, rtc)
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By(fmt.Sprintf("When clear annotation %s from TidbCluster CRD", pushgatewayAnnKey))
		err = wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
			tcobj, err = cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			delete(tcobj.Annotations, pushgatewayAnnKey)
			tcobj, err = cm.Cli.PingcapV1().TidbClusters(clusterName).Update(tcobj)
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By(fmt.Sprintf("Then make sure annotation %s will be sync to TidbCluster CRD again", pushgatewayAnnKey))
		err = wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
			tcobj, err = cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
			if err != nil {
				framework.Errorf("get TidbCluster CRD failed: %v", err)
				return false, nil
			}

			if tcobj.Annotations[pushgatewayAnnKey] != pushgatewayAnnValue {
				return false, nil
			}

			return true, nil

		})
		Expect(err).NotTo(HaveOccurred())

		By(fmt.Sprintf("When add data to tidb"))
		dsn := fmt.Sprintf("root:%s@(%s-tidb.%s:4000)/test?charset=utf8", password, tcName, clusterName)
		err = wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
			db, err := sql.Open("mysql", dsn)
			if err != nil {
				framework.Errorf("can't open connection to mysql: %v", err)
				return false, nil
			}
			defer db.Close()

			_, err = db.Exec("CREATE TABLE tidb_cluster (clusterName VARCHAR(64))")
			if err != nil {
				framework.Errorf("can't create table to mysql: %v", err)
				return false, nil
			}

			_, err = db.Exec("INSERT INTO tidb_cluster VALUES (?)", clusterName)
			if err != nil {
				framework.Errorf("can't insert data to mysql: %v", err)
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When create backup for database")
		nodeSelector := map[string]string{
			"kubernetes.io/hostname": "kube-node-1",
		}
		backupName, err := cm.CreateBackup("test", []string{"tidb_cluster"}, nodeSelector)
		Expect(err).NotTo(HaveOccurred())

		By("Then backup job should be completed")
		var backupStatus *manager.BackupStatus
		err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
			backupStatus, err = cm.GetBackup(backupName)
			if err != nil || backupStatus.Status != manager.JobComplete {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When create a temporary tidb for restoring data")
		cmd := exec.Command("/bin/sh", "-c", fmt.Sprintf("/usr/local/bin/kubectl apply -n %s -f /tmp/data/standalone-tidb.yaml", clusterName))
		data, err := cmd.CombinedOutput()
		if err != nil {
			framework.Errorf(string(data))
		}
		Expect(err).NotTo(HaveOccurred())

		By("Then temporary tidb should be created successfully")
		err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
			pod, err := cm.Cli.CoreV1().Pods(clusterName).Get("standalone-tidb", metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			if pod.Status.Phase != v1.PodRunning {
				return false, nil
			}
			_, err = cm.Cli.CoreV1().Services(clusterName).Get("standalone-tidb", metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("And make sure getVolumes interface work as expected")
		err = wait.Poll(10*time.Second, 30*time.Minute, func() (bool, error) {
			vols, err := cm.GetVolumes(clusterName, "", false)
			if err != nil {
				return false, nil
			}
			opts := metav1.ListOptions{
				LabelSelector: labels.SelectorFromSet(
					label.New().Namespace(clusterName).Cluster(tcName).Labels(),
				).String(),
			}
			pvList, err := cm.Cli.CoreV1().PersistentVolumes().List(opts)
			if err != nil || len(pvList.Items) != len(vols) {
				return false, nil
			}
			for _, vol := range vols {
				if len(vol.Used) == 0 {
					return false, nil
				}
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("And when restore data from backup")
		Expect(cm.RestoreData(clusterName, "/data", backupStatus.Path, nodeSelector)).NotTo(HaveOccurred())
		By("Then restore data job finished")
		err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
			job, err := cm.Cli.BatchV1().Jobs(clusterName).Get(framework.RestoreJobName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			if job.Status.Succeeded == 1 {
				return true, nil
			}
			return false, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("And data should be restored")
		dsn = fmt.Sprintf("root@(standalone-tidb.%s:4000)/test?charset=utf8", clusterName)
		ensureDataIsRight(dsn, clusterName)

		By("When reset Tidb's password")
		jobName, err := cm.ResetTidbPassword(newPassword)
		Expect(err).NotTo(HaveOccurred())

		By("Then validate password reset works as expected")
		err = wait.Poll(5*time.Second, 5*time.Minute, func() (bool, error) {
			status, err := cm.GetJobStatus(jobName)
			if status == manager.JobFailed {
				//fast fail when job has failed
				return false, framework.JobFailed
			}
			if status == manager.JobRunning {
				return false, nil
			}
			//password reset job has completed
			err = cm.CheckTidbConn(newPassword)
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		jobName = fmt.Sprintf("%s-set-tidb-password", tcName)
		By(fmt.Sprintf("When delete job %s", jobName))
		err = wait.Poll(5*time.Second, 2*time.Minute, func() (bool, error) {
			err = cm.Cli.BatchV1().Jobs(clusterName).Delete(jobName, nil)
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())
		By("Then cluster initialized is still true")
		err = wait.Poll(5*time.Second, 2*time.Minute, func() (bool, error) {
			rtc, err = cm.GetTidbCluster()
			if err != nil {
				return false, nil
			}
			if !rtc.Initialized {
				return false, framework.InitNotCompleted
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When resize pd and tidb's replicas to 3 and tikv's replicas to 5")
		tc.Pd.Size = 3
		tc.Tidb.Size = 3
		tc.Tikv.Size = 5
		//update tc's password for subsequent validation
		tc.RootPassword = newPassword
		err = cm.UpdateTidbCluster(tc)
		Expect(err).NotTo(HaveOccurred())

		By("Then make sure scale up TidbCluster has prepared")
		err = wait.Poll(10*time.Second, 30*time.Minute, func() (bool, error) {
			rtc, err = cm.GetTidbCluster()
			if err != nil {
				return false, nil
			}
			err = cm.CheckClusterStatus(tc, rtc)
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When resize Tikv's replicas to 3")
		tc.Tikv.Size = 3
		err = wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
			err = cm.UpdateTidbCluster(tc)
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("Then make sure scale in TidbCluster has prepared")
		var shouldDeletedPVName string
		err = wait.Poll(10*time.Second, 20*time.Minute, func() (bool, error) {
			rtc, err = cm.GetTidbCluster()
			if err != nil {
				return false, nil
			}
			err = cm.CheckClusterStatus(tc, rtc)
			if err != nil {
				return false, nil
			}

			podList, err := cm.Cli.CoreV1().Pods(clusterName).List(metav1.ListOptions{
				LabelSelector: labels.SelectorFromSet(
					label.New().Cluster(tcName).App("tikv").Labels(),
				).String(),
			})
			if err != nil {
				return false, nil
			}

			if len(podList.Items) != 3 {
				return false, nil
			}

			pvList, err := cm.Cli.CoreV1().PersistentVolumes().List(label.ClusterListOptions(tcName))
			if err != nil || len(pvList.Items) != 8 {
				return false, nil
			}

			// there must 2 tombstoned pv
			tombstonedCount := 0
			for _, pv := range pvList.Items {
				if pv.Annotations[volumeprovisioner.AnnPVTombstoned] != "" {
					tombstonedCount++
				}
			}
			if tombstonedCount != 2 {
				framework.Errorf("tombstoned PV count must be 2, now: [%s]", tombstonedCount)
				return false, nil
			}
			framework.Infof("tombstoned PV count is: %d", tombstonedCount)

			vols, err := cm.GetVolumes(clusterName, "", true)
			if err != nil {
				return false, nil
			}
			if len(vols) != 2 {
				framework.Errorf("getVolumes onlyTombstoned PV count must be 2, now: [%s]", len(vols))
				return false, nil
			}
			shouldDeletedPVName = vols[0].Name
			framework.Infof("selected a should deleted pv: %s", shouldDeletedPVName)

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When delete a tombstoned PV manually")
		err = cm.DeleteVolume(shouldDeletedPVName)
		Expect(err).NotTo(HaveOccurred())

		By(fmt.Sprintf("Then it annotations[%s] must be set", volumeprovisioner.AnnPVManualDeleted))
		tmpPV, err := cm.Cli.CoreV1().PersistentVolumes().Get(shouldDeletedPVName, metav1.GetOptions{})
		Expect(err).NotTo(HaveOccurred())
		Expect(tmpPV.Annotations[volumeprovisioner.AnnPVManualDeleted]).NotTo(Equal(""))

		By("Then it must be deleted")
		err = wait.Poll(5*time.Second, 20*time.Minute, func() (bool, error) {
			_, err := cm.Cli.CoreV1().PersistentVolumes().Get(shouldDeletedPVName, metav1.GetOptions{})
			if err != nil && apierrs.IsNotFound(err) {
				return true, nil
			}

			return false, nil
		})
		Expect(err).NotTo(HaveOccurred())

		// wait all pods are really running
		// TODO use tidb-operator's style
		time.Sleep(2 * time.Minute)

		By("When upgrade pd, tikv and tidb")
		framework.SetVersionByTiPodSpec(tictl.PDContainerType, tc.Pd, "master")
		framework.SetVersionByTiPodSpec(tictl.TiDBContainerType, tc.Tidb, "master")
		framework.SetVersionByTiPodSpec(tictl.TiKVContainerType, tc.Tikv, "master")
		err = cm.UpdateTidbCluster(tc)
		Expect(err).NotTo(HaveOccurred())

		By("Then the upgrade order is pd -> tikv -> tidb")
		err = wait.Poll(2*time.Second, 20*time.Minute, func() (bool, error) {
			ti, err := cm.Cli.PingcapV1().TidbInstances(clusterName).Get(clusterName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}

			if ti.PDUpgrading() {
				if ti.TiKVUpgrading() {
					return true, errors.New("pd and tikv are upgrading at the same time")
				}

				if ti.TiDBUpgrading() {
					return true, errors.New("pd and tidb are upgrading at the same time")
				}
			}

			if ti.TiKVUpgrading() {
				if ti.TiDBUpgrading() {
					return true, errors.New("tikv and tidb are upgrading at the same time")
				}
			}

			if ti.Status.PDStatus.UpgradeState == tiapi.Upgraded &&
				ti.Status.TiKVStatus.UpgradeState == tiapi.Upgraded &&
				ti.Status.TiDBStatus.UpgradeState == tiapi.Upgraded {
				return true, nil
			}

			return false, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When delete a PD pod")
		podName := rtc.PdStatus[0].Name
		err = wait.Poll(3*time.Second, 30*time.Second, func() (bool, error) {
			err = cm.DeletePod(podName)
			if err != nil {
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("Then PD pod is deleted and PD replica still equals PD size of TidbCluster ")
		err = wait.Poll(3*time.Second, 10*time.Minute, func() (bool, error) {
			rtc, err = cm.GetTidbCluster()
			if err != nil {
				return false, nil
			}
			//verify pd size still equals tc.Pd.Size
			if len(rtc.PdStatus) != tc.Pd.Size {
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When delete a TiKV pod")
		podName = rtc.TikvStatus[0].Name
		err = wait.Poll(3*time.Second, 30*time.Second, func() (bool, error) {
			err = cm.DeletePod(podName)
			if err != nil {
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("Then TiKV pod is deleted and TiKV replica still equals TiKV size of TidbCluster ")
		err = wait.Poll(3*time.Second, 30*time.Minute, func() (bool, error) {
			rtc, err = cm.GetTidbCluster()
			if err != nil {
				return false, nil
			}
			//verify tikv size still equals tc.Tikv.Size
			if len(rtc.TikvStatus) != tc.Tikv.Size {
				return false, nil
			}
			for _, tikv := range rtc.TikvStatus {
				if tikv.Name == podName {
					return false, nil
				}
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When delete a TiDB pod")
		podName = rtc.TidbStatus[0].Name
		err = wait.Poll(3*time.Second, 30*time.Second, func() (bool, error) {
			err = cm.DeletePod(podName)
			if err != nil {
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("Then TiDB pod is deleted and TiDB replica still equals TiDB size of TidbCluster ")
		err = wait.Poll(3*time.Second, 30*time.Minute, func() (bool, error) {
			rtc, err = cm.GetTidbCluster()
			if err != nil {
				return false, nil
			}
			//verify tikv size still equals tc.Tikv.Size
			if len(rtc.TidbStatus) != tc.Tidb.Size {
				return false, nil
			}
			for _, tidb := range rtc.TidbStatus {
				if tidb.Name == podName {
					return false, nil
				}
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("Ensure kafka is ready")
		Expect(framework.EnsureKafkaIsReady(bootstrap, clusterName)).To(Succeed())

		By("Ensure TiDB has not enabled binlog")
		tidbPL, err := cm.GetAllTiDBPods(clusterName, tcName)
		Expect(err).NotTo(HaveOccurred())
		Expect(framework.HasAllBinlogEnabled(tidbPL)).NotTo(Equal(true))

		By("When upgrade TiDB to support binlog")
		framework.AddOrUpdateBinlog(tc.Tidb, binlogImage)
		tc.KafkaAddr = kafkaAddr
		err = cm.UpdateTidbCluster(tc)
		Expect(err).NotTo(HaveOccurred())

		By("Then all TiDB pods should upgrade to contain binlog container")
		err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
			tidbPL, err = cm.GetAllTiDBPods(clusterName, tcName)
			if err != nil || len(tidbPL.Items) != tc.Tidb.Size {
				return false, nil
			}

			if !framework.HasAllBinlogEnabled(tidbPL) {
				return false, nil
			}

			for _, pod := range tidbPL.Items {
				running, err := cm.IsPodRunning(&pod)
				if err != nil || !running {
					return false, nil
				}

				blImage := manager.GetImageByK8sPodSpec(tictl.BinlogContainerType, &pod.Spec)
				if binlogImage != blImage {
					return false, nil
				}
			}

			_, err := cm.Cli.CoreV1().Secrets(clusterName).Get(fmt.Sprintf("%s-secret", clusterName), metav1.GetOptions{})
			if err == nil || !apierrs.IsNotFound(err) {
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By(fmt.Sprintf("When upgrade Binlog image to %s", upgradeBinlogImage))
		framework.AddOrUpdateBinlog(tc.Tidb, upgradeBinlogImage)
		err = cm.UpdateTidbCluster(tc)
		Expect(err).NotTo(HaveOccurred())

		By(fmt.Sprintf("Then all Binlog containers along with TiDB pods are upgraded to %s", upgradeBinlogImage))
		err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
			tidbPL, err = cm.GetAllTiDBPods(clusterName, tcName)
			if err != nil || len(tidbPL.Items) != tc.Tidb.Size {
				return false, nil
			}

			for _, pod := range tidbPL.Items {
				running, err := cm.IsPodRunning(&pod)
				if err != nil || !running {
					return false, nil
				}

				blImage := manager.GetImageByK8sPodSpec(tictl.BinlogContainerType, &pod.Spec)
				if upgradeBinlogImage != blImage {
					return false, nil
				}
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When set TidbInstance.Spec.Deleted to true")
		err = cm.DeleteTidbCluster()
		Expect(err).NotTo(HaveOccurred())

		By("Then getCluster would return empty result but listCluster still have one cluster")
		err = wait.Poll(3*time.Second, 1*time.Minute, func() (bool, error) {
			_, err = cm.GetTidbCluster()
			if err != framework.ClusterNotFound {
				return false, nil
			}
			clusters, err := cm.ListTidbClusters()
			if err != nil || len(clusters) != 1 {
				return false, nil
			}
			if !clusters[0].Deleted {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("Then all resources will been deleted")
		err = wait.Poll(3*time.Second, 10*time.Minute, func() (bool, error) {
			priDeployName := fmt.Sprintf("%s-privileged-tidb", clusterName)
			priSvcName := priDeployName
			_, err := cm.Cli.AppsV1beta1().Deployments(clusterName).Get(priDeployName, metav1.GetOptions{})
			if err == nil || !apierrs.IsNotFound(err) {
				return false, nil
			}

			_, err = cm.Cli.CoreV1().Services(clusterName).Get(priSvcName, metav1.GetOptions{})
			if err == nil || !apierrs.IsNotFound(err) {
				return false, nil
			}

			svcNames := []string{
				"prometheus",
				"grafana",
			}

			for _, svcName := range svcNames {
				name := fmt.Sprintf("%s-%s", clusterName, svcName)
				_, err := cm.Cli.CoreV1().Services(clusterName).Get(name, metav1.GetOptions{})
				if err == nil || !apierrs.IsNotFound(err) {
					return false, nil
				}
			}

			monitorDeployName := fmt.Sprintf("%s-monitor", clusterName)
			jobName := monitorDeployName
			_, err = cm.Cli.AppsV1beta1().Deployments(clusterName).Get(monitorDeployName, metav1.GetOptions{})
			if err == nil || !apierrs.IsNotFound(err) {
				return false, nil
			}

			_, err = cm.Cli.BatchV1().Jobs(clusterName).Get(jobName, metav1.GetOptions{})
			if err == nil || !apierrs.IsNotFound(err) {
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("And TidbCluster will be graceful deleted")
		err = wait.Poll(3*time.Second, 10*time.Minute, func() (bool, error) {
			tc, err := cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}

			if !tc.ReallyGracefulDeleted() {
				return false, nil
			}

			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		time.Sleep(2 * time.Minute)

		By("When set TidbInstance.Spec.Deleted to false to restore the cluster")
		err = cm.RestoreTidbCluster()
		Expect(err).NotTo(HaveOccurred())

		By("Then TidbCluster.State will be StateRestore")
		err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
			tc, err := cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}

			if tc.ReallyRestoring() {
				return true, nil
			}

			return false, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("And TidbCluster will be restored, its state will be StateNormal")
		err = wait.Poll(5*time.Second, 15*time.Minute, func() (bool, error) {
			tc, err := cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}

			if tc.ReallyNormal() {
				return true, nil
			}

			return false, nil
		})
		Expect(err).NotTo(HaveOccurred())

		By("When set TidbInstance.Spec.Deleted to true again")
		err = wait.Poll(2*time.Second, 2*time.Minute, func() (bool, error) {
			tidbCluster, err := cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
			if err != nil {
				return false, nil
			}
			_, err = cm.Cli.PingcapV1().TidbClusters(clusterName).Update(tidbCluster)
			if err != nil {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())

		_, err = cm.Cli.PingcapV1().TidbClusters(clusterName).Get(tcName, metav1.GetOptions{})
		err = cm.DeleteTidbCluster()
		Expect(err).NotTo(HaveOccurred())

		By("Then TidbInstance will not be deleted")
		err = wait.Poll(5*time.Second, 10*time.Minute, func() (bool, error) {
			_, err := cm.Cli.PingcapV1().TidbInstances(clusterName).Get(clusterName, metav1.GetOptions{})
			if err != nil && apierrs.IsNotFound(err) {
				return true, fmt.Errorf("TidbInstance [%s/%s] had been deleted", clusterName, clusterName)
			}

			return false, nil
		})
		if err != nil {
			if !strings.Contains(err.Error(), wait.ErrWaitTimeout.Error()) {
				Expect(err).NotTo(HaveOccurred())
			}
		}

		By("Finally delete TidbCluster")
		err = cm.Cli.CoreV1().Namespaces().Delete(clusterName, nil)
		if err != nil {
			if apierrs.IsNotFound(err) || apierrs.IsConflict(err) {
				err = nil
			}
		}
		Expect(err).NotTo(HaveOccurred())
		err = wait.Poll(3*time.Second, 10*time.Minute, func() (bool, error) {
			//verify cluster has been deleted
			clusters, err := cm.ListTidbClusters()
			if err != nil || len(clusters) != 0 {
				return false, nil
			}
			return true, nil
		})
		Expect(err).NotTo(HaveOccurred())
	})
})

func ensureDataIsRight(dsn, clusterName string) {
	By("Ensure saved data is right")
	err := wait.Poll(10*time.Second, 5*time.Minute, func() (bool, error) {
		db, err := sql.Open("mysql", dsn)
		if err != nil {
			return false, nil
		}

		rows, err := db.Query("SELECT * FROM tidb_cluster")
		if err != nil {
			framework.Errorf(err.Error())
			return false, nil
		}

		for rows.Next() {
			var val string
			err := rows.Scan(&val)
			if err != nil {
				framework.Errorf(err.Error())
			}

			if val == clusterName {
				return true, nil
			}

			return true, fmt.Errorf("val should equal: %s", clusterName)
		}

		return false, nil
	})
	Expect(err).NotTo(HaveOccurred())
}