tests: add more unit tests in cdc/model package (#1081) (#1105)

Signed-off-by: ti-srebot <ti-srebot@pingcap.com>
Co-authored-by: amyangfei <amyangfei@gmail.com>

cdc/model/capture_test.go

@@ -0,0 +1,37 @@
+// Copyright 2020 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 model
+
+import (
+	"github.com/pingcap/check"
+)
+
+type captureSuite struct{}
+
+var _ = check.Suite(&captureSuite{})
+
+func (s *captureSuite) TestMarshalUnmarshal(c *check.C) {
+	info := &CaptureInfo{
+		ID:            "9ff52aca-aea6-4022-8ec4-fbee3f2c7890",
+		AdvertiseAddr: "127.0.0.1:8300",
+	}
+	expected := []byte(`{"id":"9ff52aca-aea6-4022-8ec4-fbee3f2c7890","address":"127.0.0.1:8300"}`)
+	data, err := info.Marshal()
+	c.Assert(err, check.IsNil)
+	c.Assert(data, check.DeepEquals, expected)
+	decodedInfo := &CaptureInfo{}
+	err = decodedInfo.Unmarshal(data)
+	c.Assert(err, check.IsNil)
+	c.Assert(decodedInfo, check.DeepEquals, info)
+}

cdc/model/changefeed_test.go

@@ -14,13 +14,16 @@
 package model
 
 import (
+	"math"
 	"time"
 
 	"github.com/pingcap/check"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/ticdc/pkg/config"
+	cerror "github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/util/testleak"
 	filter "github.com/pingcap/tidb-tools/pkg/table-filter"
+	"github.com/pingcap/tidb/store/tikv/oracle"
 )
 
 type configSuite struct{}
@@ -95,7 +98,7 @@ func (s *configSuite) TestFillV1(c *check.C) {
             ]
         },
         "cyclic-replication":{
-            "enable":false,
+            "enable":true,
             "replica-id":1,
             "filter-replica-ids":[
                 2,
@@ -112,7 +115,9 @@ func (s *configSuite) TestFillV1(c *check.C) {
 	c.Assert(err, check.IsNil)
 	c.Assert(cfg, check.DeepEquals, &ChangeFeedInfo{
 		SinkURI: "blackhole://",
-		Opts:    map[string]string{},
+		Opts: map[string]string{
+			"_cyclic_relax_sql_mode": `{"enable":true,"replica-id":1,"filter-replica-ids":[2,3],"id-buckets":4,"sync-ddl":true}`,
+		},
 		StartTs: 417136892416622595,
 		Engine:  "memory",
 		SortDir: ".",
@@ -150,6 +155,7 @@ func (s *configSuite) TestFillV1(c *check.C) {
 				},
 			},
 			Cyclic: &config.CyclicConfig{
+				Enable:          true,
 				ReplicaID:       1,
 				FilterReplicaID: []uint64{2, 3},
 				IDBuckets:       4,
@@ -223,3 +229,49 @@ func (s *changefeedSuite) TestChangefeedInfoStringer(c *check.C) {
 	str := info.String()
 	c.Check(str, check.Matches, ".*sink-uri\":\"\\*\\*\\*\".*")
 }
+
+func (s *changefeedSuite) TestValidateChangefeedID(c *check.C) {
+	validIDs := []string{
+		"test",
+		"1",
+		"9ff52aca-aea6-4022-8ec4-fbee3f2c7890",
+	}
+	for _, id := range validIDs {
+		err := ValidateChangefeedID(id)
+		c.Assert(err, check.IsNil)
+	}
+
+	invalidIDs := []string{
+		"",
+		"test_task",
+		"job$",
+	}
+	for _, id := range invalidIDs {
+		err := ValidateChangefeedID(id)
+		c.Assert(cerror.ErrInvalidChangefeedID.Equal(err), check.IsTrue)
+	}
+}
+
+func (s *changefeedSuite) TestGetTs(c *check.C) {
+	var (
+		startTs      uint64 = 418881574869139457
+		targetTs     uint64 = 420891571239139085
+		checkpointTs uint64 = 420874357546418177
+		createTime          = time.Now()
+		info                = &ChangeFeedInfo{
+			SinkURI:    "blackhole://",
+			CreateTime: createTime,
+		}
+	)
+	c.Assert(info.GetStartTs(), check.Equals, oracle.EncodeTSO(createTime.Unix()*1000))
+	info.StartTs = startTs
+	c.Assert(info.GetStartTs(), check.Equals, startTs)
+
+	c.Assert(info.GetTargetTs(), check.Equals, uint64(math.MaxUint64))
+	info.TargetTs = targetTs
+	c.Assert(info.GetTargetTs(), check.Equals, targetTs)
+
+	c.Assert(info.GetCheckpointTs(nil), check.Equals, startTs)
+	status := &ChangeFeedStatus{CheckpointTs: checkpointTs}
+	c.Assert(info.GetCheckpointTs(status), check.Equals, checkpointTs)
+}

cdc/model/kv_test.go

@@ -0,0 +1,56 @@
+// Copyright 2020 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 model
+
+import (
+	"github.com/pingcap/check"
+	"github.com/pingcap/ticdc/pkg/regionspan"
+)
+
+type kvSuite struct{}
+
+var _ = check.Suite(&kvSuite{})
+
+func (s *kvSuite) TestRegionFeedEvent(c *check.C) {
+	raw := &RawKVEntry{
+		CRTs:   1,
+		OpType: OpTypePut,
+	}
+	resolved := &ResolvedSpan{
+		Span:       regionspan.ComparableSpan{},
+		ResolvedTs: 111,
+	}
+
+	ev := &RegionFeedEvent{}
+	c.Assert(ev.GetValue(), check.IsNil)
+
+	ev = &RegionFeedEvent{Val: raw}
+	c.Assert(ev.GetValue(), check.DeepEquals, raw)
+
+	ev = &RegionFeedEvent{Resolved: resolved}
+	c.Assert(ev.GetValue(), check.DeepEquals, resolved)
+}
+
+func (s *kvSuite) TestRawKVEntry(c *check.C) {
+	raw := &RawKVEntry{
+		StartTs: 100,
+		CRTs:    101,
+		OpType:  OpTypePut,
+		Key:     []byte("123"),
+		Value:   []byte("345"),
+	}
+
+	c.Assert(raw.String(), check.Equals, "OpType: 1, Key: 123, Value: 345, StartTs: 100, CRTs: 101, RegionID: 0")
+	c.Assert(raw.ApproximateSize(), check.Equals, int64(6))
+}

cdc/model/mounter_test.go

@@ -0,0 +1,73 @@
+// Copyright 2020 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 model
+
+import (
+	"context"
+	"sync"
+
+	"github.com/pingcap/check"
+)
+
+type mounterSuite struct{}
+
+var _ = check.Suite(&mounterSuite{})
+
+func (s *mounterSuite) TestPolymorphicEvent(c *check.C) {
+	raw := &RawKVEntry{
+		StartTs:  99,
+		CRTs:     100,
+		OpType:   OpTypePut,
+		RegionID: 2,
+	}
+	resolved := &RawKVEntry{
+		OpType: OpTypeResolved,
+		CRTs:   101,
+	}
+
+	polyEvent := NewPolymorphicEvent(raw)
+	c.Assert(polyEvent.RawKV, check.DeepEquals, raw)
+	c.Assert(polyEvent.CRTs, check.Equals, raw.CRTs)
+	c.Assert(polyEvent.StartTs, check.Equals, raw.StartTs)
+	c.Assert(polyEvent.RegionID(), check.Equals, raw.RegionID)
+
+	rawResolved := &RawKVEntry{CRTs: resolved.CRTs, OpType: OpTypeResolved}
+	polyEvent = NewPolymorphicEvent(resolved)
+	c.Assert(polyEvent.RawKV, check.DeepEquals, rawResolved)
+	c.Assert(polyEvent.CRTs, check.Equals, resolved.CRTs)
+	c.Assert(polyEvent.StartTs, check.Equals, uint64(0))
+}
+
+func (s *mounterSuite) TestPolymorphicEventPrepare(c *check.C) {
+	ctx := context.Background()
+	polyEvent := NewPolymorphicEvent(&RawKVEntry{OpType: OpTypeResolved})
+	c.Assert(polyEvent.WaitPrepare(ctx), check.IsNil)
+
+	polyEvent = NewPolymorphicEvent(&RawKVEntry{OpType: OpTypePut})
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err := polyEvent.WaitPrepare(ctx)
+		c.Assert(err, check.IsNil)
+	}()
+	polyEvent.PrepareFinished()
+	wg.Wait()
+
+	cctx, cancel := context.WithCancel(ctx)
+	polyEvent = NewPolymorphicEvent(&RawKVEntry{OpType: OpTypePut})
+	cancel()
+	err := polyEvent.WaitPrepare(cctx)
+	c.Assert(err, check.Equals, context.Canceled)
+}