planner, executor: add unit test

executor/show.go

@@ -20,8 +20,10 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"testing"
 	"time"
 
+	"github.com/coreos/etcd/integration"
 	"github.com/cznic/mathutil"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/ast"
@@ -50,6 +52,8 @@ import (
 
 var etcdDialTimeout = 5 * time.Second
 
+const binlogTest = "/tmp/tidb"
+
 // ShowExec represents a show executor.
 type ShowExec struct {
 	baseExecutor
@@ -1031,6 +1035,12 @@ func (e *ShowExec) fetchShowPumpOrDrainerStatus(kind string) error {
 
 // createRegistry returns an ectd registry
 func createRegistry(urls string) (*node.EtcdRegistry, error) {
+	if urls == binlogTest {
+		var t *testing.T
+		testEtcdCluster := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+		etcdclient := etcd.NewClient(testEtcdCluster.RandClient(), node.DefaultRootPath)
+		return node.NewEtcdRegistry(etcdclient, time.Duration(5)*time.Second), nil
+	}
 	ectdEndpoints, err := utils.ParseHostPortAddr(urls)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -1041,6 +1051,7 @@ func createRegistry(urls string) (*node.EtcdRegistry, error) {
 	}
 
 	return node.NewEtcdRegistry(cli, etcdDialTimeout), nil
+
 }
 
 func (e *ShowExec) getTable() (table.Table, error) {

executor/show_test.go

@@ -16,12 +16,18 @@ package executor_test
 import (
 	"context"
 	"fmt"
+	"path"
+	"testing"
+	"time"
 
+	"github.com/coreos/etcd/integration"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/auth"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
+	"github.com/pingcap/tidb-tools/pkg/etcd"
+	"github.com/pingcap/tidb-tools/tidb-binlog/node"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/executor"
 	plannercore "github.com/pingcap/tidb/planner/core"
@@ -404,3 +410,29 @@ func (s *testSuite2) TestShowEscape(c *C) {
 	tk.MustExec("rename table \"t`abl\"\"e\" to t")
 	tk.MustExec("set sql_mode=@old_sql_mode")
 }
+
+func (s *testSuite2) TestShowPumpOrDrainerStatus(c *C) {
+	var t *testing.T
+	testEtcdCluster := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+
+	etcdclient := etcd.NewClient(testEtcdCluster.RandClient(), node.DefaultRootPath)
+	r := node.NewEtcdRegistry(etcdclient, time.Duration(5)*time.Second)
+	defer func() {
+		testEtcdCluster.Terminate(t)
+		r.Close()
+	}()
+
+	ns := &node.Status{
+		NodeID:  "pump1",
+		Addr:    "127.0.0.1:8249",
+		State:   node.Online,
+		IsAlive: true,
+	}
+	nodePrefix := path.Join(node.DefaultRootPath, node.NodePrefix[node.PumpNode])
+	err := r.UpdateNode(context.Background(), nodePrefix, ns)
+	c.Assert(err, IsNil)
+
+	tk := testkit.NewTestKit(c, s.store)
+	result := tk.MustQuery("SHOW PUMP STATUS")
+	c.Check(result.Rows(), HasLen, 0)
+}