fix the lint error and unit test

Signed-off-by: SimFG <bang.fu@zilliz.com>
zilliztech · Jul 22, 2024 · 1cbf104 · 1cbf104
1 parent 69eabf1
commit 1cbf104
Show file tree

Hide file tree

Showing 8 changed files with 78 additions and 24 deletions.
diff --git a/core/api/replicate_manager_test.go b/core/api/replicate_manager_test.go
@@ -24,7 +24,6 @@ import (
 	"testing"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/msgpb"
-	"github.com/milvus-io/milvus/pkg/mq/msgstream"
 
 	"github.com/zilliztech/milvus-cdc/core/model"
 	"github.com/zilliztech/milvus-cdc/core/pb"
@@ -104,7 +103,7 @@ func TestDefaultChannelManager_GetMsgChan(t *testing.T) {
 	tests := []struct {
 		name string
 		args args
-		want <-chan *msgstream.MsgPack
+		want <-chan *ReplicateMsg
 	}{
 		{
 			name: "TestDefaultChannelManager_GetMsgChan",

diff --git a/core/mocks/channel_manager.go b/core/mocks/channel_manager.go
diff --git a/core/reader/replicate_channel_manager_test.go b/core/reader/replicate_channel_manager_test.go
@@ -620,7 +620,8 @@ func TestReplicateChannelHandler(t *testing.T) {
 			defer close(done)
 			{
 				// timetick pack
-				pack := <-handler.msgPackChan
+				replicateMsg := <-handler.msgPackChan
+				pack := replicateMsg.MsgPack
 				// assert pack
 				assert.NotNil(t, pack)
 				assert.EqualValues(t, 1, pack.BeginTs)
@@ -631,7 +632,8 @@ func TestReplicateChannelHandler(t *testing.T) {
 			}
 			{
 				// insert msg
-				pack := <-handler.msgPackChan
+				replicateMsg := <-handler.msgPackChan
+				pack := replicateMsg.MsgPack
 				assert.Len(t, pack.Msgs, 1)
 				insertMsg := pack.Msgs[0].(*msgstream.InsertMsg)
 				assert.EqualValues(t, 100, insertMsg.CollectionID)
@@ -641,7 +643,8 @@ func TestReplicateChannelHandler(t *testing.T) {
 
 			{
 				// delete msg
-				pack := <-handler.msgPackChan
+				replicateMsg := <-handler.msgPackChan
+				pack := replicateMsg.MsgPack
 				assert.Len(t, pack.Msgs, 2)
 				{
 					deleteMsg := pack.Msgs[0].(*msgstream.DeleteMsg)
@@ -659,7 +662,8 @@ func TestReplicateChannelHandler(t *testing.T) {
 
 			{
 				// drop partition msg
-				pack := <-handler.msgPackChan
+				replicateMsg := <-handler.msgPackChan
+				pack := replicateMsg.MsgPack
 				assert.Len(t, pack.Msgs, 1)
 				dropMsg := pack.Msgs[0].(*msgstream.DropPartitionMsg)
 				assert.EqualValues(t, 100, dropMsg.CollectionID)
@@ -669,7 +673,8 @@ func TestReplicateChannelHandler(t *testing.T) {
 
 			{
 				// drop collection msg
-				pack := <-handler.msgPackChan
+				replicateMsg := <-handler.msgPackChan
+				pack := replicateMsg.MsgPack
 				assert.Len(t, pack.Msgs, 2)
 				dropMsg := pack.Msgs[0].(*msgstream.DropCollectionMsg)
 				assert.EqualValues(t, 100, dropMsg.CollectionID)

diff --git a/core/util/msgpack.go b/core/util/msgpack.go
@@ -25,9 +25,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/mq/msgstream"
 )
 
-var (
-	SuffixSnapshotTombstone = []byte{0xE2, 0x9B, 0xBC} // base64 value: "4pu8"
-)
+var SuffixSnapshotTombstone = []byte{0xE2, 0x9B, 0xBC} // base64 value: "4pu8"
 
 type OnceWriteChan[T any] struct {
 	once sync.Once

diff --git a/core/util/msgpack_test.go b/core/util/msgpack_test.go
@@ -43,7 +43,7 @@ func TestOnceChan(t *testing.T) {
 
 func TestGetCollectionNameFromMsgPack(t *testing.T) {
 	t.Run("empty pack", func(t *testing.T) {
-		assert.Equal(t, "", GetCollectionNameFromMsgPack(EmptyMsgPack))
+		assert.Equal(t, "", GetCollectionNameFromMsgPack(&msgstream.MsgPack{}))
 	})
 
 	t.Run("success", func(t *testing.T) {

diff --git a/server/cdc_impl_test.go b/server/cdc_impl_test.go
@@ -458,7 +458,7 @@ func TestCreateRequest(t *testing.T) {
 					"ch1_v0": {
 						Time: 1,
 						DataPair: &commonpb.KeyDataPair{
-							Key:  "ch1_v0",
+							Key:  "rootcoord-dml-channel_1_123v0",
 							Data: []byte("ch1-position"),
 						},
 					},
@@ -498,9 +498,9 @@ func TestCreateRequest(t *testing.T) {
 				},
 				CollectionInfos: []model.CollectionInfo{
 					{
-						Name: "*",
+						Name: "hello_milvus",
 						Positions: map[string]string{
-							"ch1_v0": base64.StdEncoding.EncodeToString([]byte("ch1-position")),
+							"rootcoord-dml-channel_1_123v0": base64.StdEncoding.EncodeToString([]byte("ch1-position")),
 						},
 					},
 				},

diff --git a/server/mocks/cdc_service.go b/server/mocks/cdc_service.go
diff --git a/server/tool/msg_count/main.go b/server/tool/msg_count/main.go
@@ -109,12 +109,12 @@ func main() {
 	GlobalConfig = positionConfig
 
 	if GlobalConfig.EnableCSV {
-		InsertCSVFile, err = os.OpenFile("insert.csv", os.O_CREATE|os.O_WRONLY, 0644)
+		InsertCSVFile, err = os.OpenFile("insert.csv", os.O_CREATE|os.O_WRONLY, 0o644)
 		if err != nil {
 			panic(err)
 		}
 		InsertCSVFile.Write([]byte("row_id,timestamp,pk\n"))
-		DeleteCSVFile, err = os.OpenFile("delete.csv", os.O_CREATE|os.O_WRONLY, 0644)
+		DeleteCSVFile, err = os.OpenFile("delete.csv", os.O_CREATE|os.O_WRONLY, 0o644)
 		if err != nil {
 			panic(err)
 		}