mirror of
https://gitee.com/milvus-io/milvus.git
synced 2024-12-02 11:59:00 +08:00
Add unittests for getChannelName in DataNode (#6346)
Signed-off-by: yangxuan <xuan.yang@zilliz.com>
This commit is contained in:
parent
d972a0361d
commit
f0a68704e2
@ -40,6 +40,7 @@ func TestMain(t *testing.M) {
|
||||
func TestDataNode(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
node := newIDLEDataNodeMock(ctx)
|
||||
node.Init()
|
||||
node.Start()
|
||||
node.Register()
|
||||
|
||||
@ -228,6 +229,52 @@ func TestDataNode(t *testing.T) {
|
||||
|
||||
})
|
||||
|
||||
t.Run("Test GetChannelName", func(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
node := newIDLEDataNodeMock(ctx)
|
||||
|
||||
testCollIDs := []UniqueID{0, 1, 2, 1}
|
||||
testSegIDs := []UniqueID{10, 11, 12, 13}
|
||||
testchanNames := []string{"a", "b", "c", "d"}
|
||||
|
||||
node.chanMut.Lock()
|
||||
for i, name := range testchanNames {
|
||||
replica := &SegmentReplica{
|
||||
collectionID: testCollIDs[i],
|
||||
newSegments: make(map[UniqueID]*Segment),
|
||||
}
|
||||
|
||||
replica.addNewSegment(testSegIDs[i], testCollIDs[i], 0, name, &internalpb.MsgPosition{}, nil)
|
||||
node.vchan2SyncService[name] = &dataSyncService{collectionID: testCollIDs[i], replica: replica}
|
||||
}
|
||||
node.chanMut.Unlock()
|
||||
|
||||
type Test struct {
|
||||
inCollID UniqueID
|
||||
expectedChannels []string
|
||||
|
||||
inSegID UniqueID
|
||||
expectedChannel string
|
||||
}
|
||||
tests := []Test{
|
||||
{0, []string{"a"}, 10, "a"},
|
||||
{1, []string{"b", "d"}, 11, "b"},
|
||||
{2, []string{"c"}, 12, "c"},
|
||||
{3, []string{}, 13, "d"},
|
||||
{3, []string{}, 100, ""},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
actualChannels := node.getChannelNamesbyCollectionID(test.inCollID)
|
||||
assert.ElementsMatch(t, test.expectedChannels, actualChannels)
|
||||
|
||||
actualChannel := node.getChannelNamebySegmentID(test.inSegID)
|
||||
assert.Equal(t, test.expectedChannel, actualChannel)
|
||||
}
|
||||
|
||||
cancel()
|
||||
})
|
||||
|
||||
t.Run("Test BackGroundGC", func(t *testing.T) {
|
||||
t.Skipf("Skip for data race")
|
||||
collIDCh := make(chan UniqueID)
|
||||
|
@ -122,7 +122,6 @@ func (dsService *dataSyncService) initNodes(vchanInfo *datapb.VchannelInfo) erro
|
||||
zap.Int64("SegmentID", fu.segID),
|
||||
zap.Int64("CollectionID", fu.collID),
|
||||
zap.Int("Length of Field2BinlogPaths", len(id2path)),
|
||||
zap.Any("Start Positions", fu.startPositions),
|
||||
)
|
||||
|
||||
req := &datapb.SaveBinlogPathsRequest{
|
||||
|
@ -57,6 +57,7 @@ func (ddn *ddNode) Operate(in []flowgraph.Msg) []flowgraph.Msg {
|
||||
msMsg, ok := in[0].(*MsgStreamMsg)
|
||||
if !ok {
|
||||
log.Error("type assertion failed for MsgStreamMsg")
|
||||
return []flowgraph.Msg{}
|
||||
// TODO: add error handling
|
||||
}
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user