milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add more testcase to improve code coverage for data service (#5041) 

* fix TestChannel run error

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* comment validateAllocRequest

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* add test APIs

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* fix static-check

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>
pull/5046/head
Cai Yudong 2021-04-26 09:45:54 +08:00 committed by GitHub
parent 0d68c3f2fd
commit 74f457bd14
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 66 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
internal/dataservice/server.go
View File

@ -624,20 +624,20 @@ func (s *Server) AssignSegmentID(ctx context.Context, req *datapb.AssignSegmentI
}, nil
}
func (s *Server) validateAllocRequest(collID UniqueID, partID UniqueID, channelName string) error {
if !s.meta.HasCollection(collID) {
return fmt.Errorf("can not find collection %d", collID)
}
if !s.meta.HasPartition(collID, partID) {
return fmt.Errorf("can not find partition %d", partID)
}
for _, name := range s.insertChannels {
if name == channelName {
return nil
}
}
return fmt.Errorf("can not find channel %s", channelName)
}
//func (s *Server) validateAllocRequest(collID UniqueID, partID UniqueID, channelName string) error {
// if !s.meta.HasCollection(collID) {
// return fmt.Errorf("can not find collection %d", collID)
// }
// if !s.meta.HasPartition(collID, partID) {
// return fmt.Errorf("can not find partition %d", partID)
// }
// for _, name := range s.insertChannels {
// if name == channelName {
// return nil
// }
// }
// return fmt.Errorf("can not find channel %s", channelName)
//}
func (s *Server) loadCollectionFromMaster(ctx context.Context, collectionID int64) error {
resp, err := s.masterClient.DescribeCollection(ctx, &milvuspb.DescribeCollectionRequest{

58
internal/dataservice/server_test.go
View File

@ -347,6 +347,47 @@ func TestGetSegmentStates(t *testing.T) {
}
}
func TestGetInsertBinlogPaths(t *testing.T) {
svr := newTestServer(t)
defer closeTestServer(t, svr)
req := &datapb.GetInsertBinlogPathsRequest{
SegmentID: 0,
}
resp, err := svr.GetInsertBinlogPaths(svr.ctx, req)
assert.Nil(t, err)
assert.EqualValues(t, commonpb.ErrorCode_Success, resp.Status.ErrorCode)
}
func TestGetCollectionStatistics(t *testing.T) {
svr := newTestServer(t)
defer closeTestServer(t, svr)
req := &datapb.GetCollectionStatisticsRequest{
CollectionID: 0,
}
resp, err := svr.GetCollectionStatistics(svr.ctx, req)
assert.Nil(t, err)
assert.EqualValues(t, commonpb.ErrorCode_Success, resp.Status.ErrorCode)
}
func TestGetSegmentInfo(t *testing.T) {
svr := newTestServer(t)
defer closeTestServer(t, svr)
segInfo := &datapb.SegmentInfo{
ID: 0,
}
svr.meta.AddSegment(segInfo)
req := &datapb.GetSegmentInfoRequest{
SegmentIDs: []int64{0},
}
resp, err := svr.GetSegmentInfo(svr.ctx, req)
assert.Nil(t, err)
assert.EqualValues(t, commonpb.ErrorCode_Success, resp.Status.ErrorCode)
}
func TestChannel(t *testing.T) {
svr := newTestServer(t)
defer closeTestServer(t, svr)
@ -358,8 +399,7 @@ func TestChannel(t *testing.T) {
segInfo := &datapb.SegmentInfo{
ID: segID,
}
err := svr.meta.AddSegment(segInfo)
assert.Nil(t, err)
svr.meta.AddSegment(segInfo)
stats := &internalpb.SegmentStatisticsUpdates{
SegmentID: segID,
@ -384,12 +424,14 @@ func TestChannel(t *testing.T) {
statsStream, _ := svr.msFactory.NewMsgStream(svr.ctx)
statsStream.AsProducer([]string{Params.StatisticsChannelName})
statsStream.Start()
defer statsStream.Close()
msgPack := msgstream.MsgPack{}
msgPack.Msgs = append(msgPack.Msgs, genMsg(commonpb.MsgType_SegmentStatistics, 123))
msgPack.Msgs = append(msgPack.Msgs, genMsg(commonpb.MsgType_SegmentInfo, 234))
msgPack.Msgs = append(msgPack.Msgs, genMsg(commonpb.MsgType_SegmentStatistics, 345))
err = statsStream.Produce(&msgPack)
err := statsStream.Produce(&msgPack)
assert.Nil(t, err)
time.Sleep(time.Second)
@ -416,14 +458,16 @@ func TestChannel(t *testing.T) {
}
}
statsStream, _ := svr.msFactory.NewMsgStream(svr.ctx)
statsStream.AsProducer([]string{Params.SegmentInfoChannelName})
segInfoStream, _ := svr.msFactory.NewMsgStream(svr.ctx)
segInfoStream.AsProducer([]string{Params.SegmentInfoChannelName})
segInfoStream.Start()
defer segInfoStream.Close()
msgPack := msgstream.MsgPack{}
msgPack.Msgs = append(msgPack.Msgs, genMsg(commonpb.MsgType_SegmentFlushDone, 123))
msgPack.Msgs = append(msgPack.Msgs, genMsg(commonpb.MsgType_SegmentInfo, 234))
msgPack.Msgs = append(msgPack.Msgs, genMsg(commonpb.MsgType_SegmentFlushDone, 345))
err := statsStream.Produce(&msgPack)
err := segInfoStream.Produce(&msgPack)
assert.Nil(t, err)
time.Sleep(time.Second)
})
@ -447,6 +491,8 @@ func TestChannel(t *testing.T) {
timeTickStream, _ := svr.msFactory.NewMsgStream(svr.ctx)
timeTickStream.AsProducer([]string{Params.ProxyTimeTickChannelName})
timeTickStream.Start()
defer timeTickStream.Close()
msgPack := msgstream.MsgPack{}
msgPack.Msgs = append(msgPack.Msgs, genMsg(commonpb.MsgType_TimeTick, 123))