wehub-resource-sync
498b235461
Build and test / Build and test AMD64 Ubuntu 22.04 (push) Failing after 0s
Publish Builder / amazonlinux2023 (push) Failing after 1s
Build and test / UT for Go (push) Has been skipped
Publish KRTE Images / KRTE (push) Failing after 1s
Build and test / Integration Test (push) Has been skipped
Build and test / Upload Code Coverage (push) Has been skipped
Publish Builder / rockylinux9 (push) Failing after 1s
Publish Builder / ubuntu22.04 (push) Failing after 0s
Publish Builder / ubuntu24.04 (push) Failing after 0s
Publish Gpu Builder / publish-gpu-builder (push) Failing after 1s
Publish Test Images / PyTest (push) Failing after 0s
Build and test / UT for Cpp (push) Has been cancelled
467 lines
17 KiB
Go
467 lines
17 KiB
Go
package manager
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/mock"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/resolver"
|
|
|
|
"github.com/milvus-io/milvus/internal/mocks/util/streamingutil/service/mock_lazygrpc"
|
|
"github.com/milvus-io/milvus/internal/mocks/util/streamingutil/service/mock_resolver"
|
|
kvfactory "github.com/milvus-io/milvus/internal/util/dependency/kv"
|
|
"github.com/milvus-io/milvus/internal/util/sessionutil"
|
|
"github.com/milvus-io/milvus/internal/util/streamingutil/service/attributes"
|
|
"github.com/milvus-io/milvus/internal/util/streamingutil/service/contextutil"
|
|
"github.com/milvus-io/milvus/internal/util/streamingutil/service/discoverer"
|
|
"github.com/milvus-io/milvus/pkg/v3/mocks/proto/mock_streamingpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/proto/streamingpb"
|
|
"github.com/milvus-io/milvus/pkg/v3/streaming/util/types"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/paramtable"
|
|
"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
|
|
)
|
|
|
|
func TestManager(t *testing.T) {
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
managerService := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeManagerServiceClient](t)
|
|
m := &managerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
stopped: make(chan struct{}),
|
|
rb: rb,
|
|
service: managerService,
|
|
}
|
|
r := mock_resolver.NewMockResolver(t)
|
|
rb.EXPECT().Resolver().Return(r)
|
|
r.EXPECT().Watch(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, f func(discoverer.VersionedState) error) error {
|
|
f(discoverer.VersionedState{})
|
|
return context.Canceled
|
|
})
|
|
|
|
// Test WatchNodeChanged.
|
|
resultCh, err := m.WatchNodeChanged(context.Background())
|
|
assert.NoError(t, err)
|
|
_, ok := <-resultCh
|
|
assert.True(t, ok)
|
|
_, ok = <-resultCh
|
|
assert.False(t, ok)
|
|
|
|
// Test CollectAllStatus
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
return discoverer.VersionedState{
|
|
Version: typeutil.VersionInt64(1),
|
|
State: resolver.State{},
|
|
}, nil
|
|
})
|
|
// Not address here.
|
|
nodes, err := m.CollectAllStatus(context.Background(), "")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 0)
|
|
|
|
// Has address.
|
|
managerServiceClient := mock_streamingpb.NewMockStreamingNodeManagerServiceClient(t)
|
|
managerService.EXPECT().GetService(mock.Anything).RunAndReturn(func(ctx context.Context) (streamingpb.StreamingNodeManagerServiceClient, error) {
|
|
return managerServiceClient, nil
|
|
})
|
|
managerServiceClient.EXPECT().CollectStatus(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, snmcsr *streamingpb.StreamingNodeManagerCollectStatusRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerCollectStatusResponse, error) {
|
|
return &streamingpb.StreamingNodeManagerCollectStatusResponse{}, nil
|
|
})
|
|
|
|
i := 0
|
|
states := []map[uint64]bool{
|
|
{1: false, 2: false, 3: true},
|
|
{1: true, 2: false},
|
|
{1: true, 2: false},
|
|
{1: true, 2: false},
|
|
}
|
|
r.EXPECT().GetLatestState(mock.Anything).Unset()
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
s := newVersionedState(int64(i), states[i])
|
|
i++
|
|
return s, nil
|
|
})
|
|
|
|
nodes, err = m.CollectAllStatus(context.Background(), "")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 3)
|
|
assert.ErrorIs(t, nodes[3].Err, types.ErrNotAlive)
|
|
assert.ErrorIs(t, nodes[1].Err, types.ErrStopping)
|
|
|
|
nodeInfos, err := m.GetAllStreamingNodes(context.Background())
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodeInfos, 2)
|
|
|
|
// Test Assign
|
|
serverID := int64(2)
|
|
managerServiceClient.EXPECT().Assign(mock.Anything, mock.Anything).RunAndReturn(
|
|
func(ctx context.Context, snmcsr *streamingpb.StreamingNodeManagerAssignRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerAssignResponse, error) {
|
|
pickedServerID, ok := contextutil.GetPickServerID(ctx)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, serverID, pickedServerID)
|
|
return nil, nil
|
|
})
|
|
err = m.Assign(context.Background(), types.PChannelInfoAssigned{
|
|
Channel: types.PChannelInfo{Name: "p", Term: 1},
|
|
Node: types.StreamingNodeInfo{ServerID: serverID},
|
|
})
|
|
assert.NoError(t, err)
|
|
|
|
// Test Remove
|
|
serverID = int64(2)
|
|
managerServiceClient.EXPECT().Remove(mock.Anything, mock.Anything).RunAndReturn(
|
|
func(ctx context.Context, snmrr *streamingpb.StreamingNodeManagerRemoveRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerRemoveResponse, error) {
|
|
pickedServerID, ok := contextutil.GetPickServerID(ctx)
|
|
assert.True(t, ok)
|
|
assert.Equal(t, serverID, pickedServerID)
|
|
return nil, nil
|
|
})
|
|
err = m.Remove(context.Background(), types.PChannelInfoAssigned{
|
|
Channel: types.PChannelInfo{Name: "p", Term: 1},
|
|
Node: types.StreamingNodeInfo{ServerID: serverID},
|
|
})
|
|
assert.NoError(t, err)
|
|
|
|
// Test Close
|
|
managerService.EXPECT().Close().Return()
|
|
rb.EXPECT().Close().Return()
|
|
m.Close()
|
|
|
|
nodeInfos, err = m.GetAllStreamingNodes(context.Background())
|
|
assert.Nil(t, nodeInfos)
|
|
assert.Error(t, err)
|
|
nodes, err = m.CollectAllStatus(context.Background(), "")
|
|
assert.Nil(t, nodes)
|
|
assert.Error(t, err)
|
|
err = m.Assign(context.Background(), types.PChannelInfoAssigned{})
|
|
assert.Error(t, err)
|
|
err = m.Remove(context.Background(), types.PChannelInfoAssigned{})
|
|
assert.Error(t, err)
|
|
resultCh, err = m.WatchNodeChanged(context.Background())
|
|
assert.Nil(t, resultCh)
|
|
assert.Error(t, err)
|
|
}
|
|
|
|
type serverInfo struct {
|
|
stopping bool
|
|
resourceGroup string
|
|
}
|
|
|
|
func newVersionedState(version int64, serverIDs map[uint64]bool) discoverer.VersionedState {
|
|
infos := make(map[uint64]serverInfo, len(serverIDs))
|
|
for id, stopping := range serverIDs {
|
|
infos[id] = serverInfo{stopping: stopping}
|
|
}
|
|
return newVersionedStateWithRG(version, infos)
|
|
}
|
|
|
|
func newVersionedStateWithRG(version int64, servers map[uint64]serverInfo) discoverer.VersionedState {
|
|
state := discoverer.VersionedState{
|
|
Version: typeutil.VersionInt64(version),
|
|
State: resolver.State{
|
|
Addresses: make([]resolver.Address, 0, len(servers)),
|
|
},
|
|
}
|
|
|
|
for serverID, info := range servers {
|
|
session := &sessionutil.SessionRaw{
|
|
ServerID: int64(serverID),
|
|
Stopping: info.stopping,
|
|
}
|
|
if info.resourceGroup != "" {
|
|
session.ServerLabels = map[string]string{
|
|
sessionutil.LabelResourceGroup: info.resourceGroup,
|
|
}
|
|
}
|
|
state.State.Addresses = append(state.State.Addresses, resolver.Address{
|
|
Addr: fmt.Sprintf("localhost:%d", serverID),
|
|
BalancerAttributes: attributes.WithSession(new(attributes.Attributes), session),
|
|
})
|
|
}
|
|
return state
|
|
}
|
|
|
|
func TestGetAllStreamingNodesWithResourceGroup(t *testing.T) {
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
managerService := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeManagerServiceClient](t)
|
|
m := &managerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
stopped: make(chan struct{}),
|
|
rb: rb,
|
|
service: managerService,
|
|
}
|
|
r := mock_resolver.NewMockResolver(t)
|
|
rb.EXPECT().Resolver().Return(r)
|
|
|
|
// Return sessions with resource group labels
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
return newVersionedStateWithRG(1, map[uint64]serverInfo{
|
|
1: {resourceGroup: "rg_a"},
|
|
2: {resourceGroup: "rg_b"},
|
|
3: {resourceGroup: ""}, // empty RG should default to __default_resource_group
|
|
}), nil
|
|
})
|
|
|
|
nodes, err := m.GetAllStreamingNodes(context.Background())
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 3)
|
|
assert.Equal(t, "rg_a", nodes[1].ResourceGroup)
|
|
assert.Equal(t, "rg_b", nodes[2].ResourceGroup)
|
|
assert.Equal(t, "__default_resource_group", nodes[3].ResourceGroup)
|
|
|
|
managerService.EXPECT().Close().Return()
|
|
rb.EXPECT().Close().Return()
|
|
m.Close()
|
|
}
|
|
|
|
func TestCollectAllStatusWithResourceGroupFilter(t *testing.T) {
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
managerService := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeManagerServiceClient](t)
|
|
m := &managerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
stopped: make(chan struct{}),
|
|
rb: rb,
|
|
service: managerService,
|
|
}
|
|
r := mock_resolver.NewMockResolver(t)
|
|
rb.EXPECT().Resolver().Return(r)
|
|
|
|
managerServiceClient := mock_streamingpb.NewMockStreamingNodeManagerServiceClient(t)
|
|
managerService.EXPECT().GetService(mock.Anything).RunAndReturn(func(ctx context.Context) (streamingpb.StreamingNodeManagerServiceClient, error) {
|
|
return managerServiceClient, nil
|
|
})
|
|
managerServiceClient.EXPECT().CollectStatus(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, snmcsr *streamingpb.StreamingNodeManagerCollectStatusRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerCollectStatusResponse, error) {
|
|
return &streamingpb.StreamingNodeManagerCollectStatusResponse{}, nil
|
|
})
|
|
|
|
// Two calls to GetLatestState: first for initial state, second for re-check
|
|
callCount := 0
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
callCount++
|
|
return newVersionedStateWithRG(int64(callCount), map[uint64]serverInfo{
|
|
1: {resourceGroup: "rg_a"},
|
|
2: {resourceGroup: "rg_a"},
|
|
3: {resourceGroup: "rg_b"},
|
|
4: {resourceGroup: ""}, // defaults to __default_resource_group
|
|
}), nil
|
|
})
|
|
|
|
// Hint rg_a - should only return status from servers 1 and 2 while rg_a is available.
|
|
nodes, err := m.CollectAllStatus(context.Background(), "rg_a")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 2)
|
|
assert.Contains(t, nodes, int64(1))
|
|
assert.Contains(t, nodes, int64(2))
|
|
assert.NotContains(t, nodes, int64(3))
|
|
assert.NotContains(t, nodes, int64(4))
|
|
|
|
// Empty resource group falls back to __default_resource_group when unlabeled legacy nodes exist.
|
|
callCount = 0
|
|
nodes, err = m.CollectAllStatus(context.Background(), "")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 1)
|
|
assert.Contains(t, nodes, int64(4))
|
|
|
|
// Hint __default_resource_group - should only return status from server 4.
|
|
callCount = 0
|
|
nodes, err = m.CollectAllStatus(context.Background(), "__default_resource_group")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 1)
|
|
assert.Contains(t, nodes, int64(4))
|
|
|
|
// Unavailable hint falls back to the RG with the most available nodes.
|
|
callCount = 0
|
|
nodes, err = m.CollectAllStatus(context.Background(), "non_existent_rg")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 2)
|
|
assert.Contains(t, nodes, int64(1))
|
|
assert.Contains(t, nodes, int64(2))
|
|
|
|
managerService.EXPECT().Close().Return()
|
|
rb.EXPECT().Close().Return()
|
|
m.Close()
|
|
}
|
|
|
|
func TestCollectAllStatusFallbackWhenHintUnavailable(t *testing.T) {
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
managerService := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeManagerServiceClient](t)
|
|
m := &managerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
stopped: make(chan struct{}),
|
|
rb: rb,
|
|
service: managerService,
|
|
}
|
|
r := mock_resolver.NewMockResolver(t)
|
|
rb.EXPECT().Resolver().Return(r)
|
|
|
|
managerServiceClient := mock_streamingpb.NewMockStreamingNodeManagerServiceClient(t)
|
|
managerService.EXPECT().GetService(mock.Anything).RunAndReturn(func(ctx context.Context) (streamingpb.StreamingNodeManagerServiceClient, error) {
|
|
return managerServiceClient, nil
|
|
})
|
|
var mu sync.Mutex
|
|
pickedServerIDs := make([]int64, 0, 2)
|
|
managerServiceClient.EXPECT().CollectStatus(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, snmcsr *streamingpb.StreamingNodeManagerCollectStatusRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerCollectStatusResponse, error) {
|
|
serverID, ok := contextutil.GetPickServerID(ctx)
|
|
assert.True(t, ok)
|
|
mu.Lock()
|
|
pickedServerIDs = append(pickedServerIDs, serverID)
|
|
mu.Unlock()
|
|
return &streamingpb.StreamingNodeManagerCollectStatusResponse{}, nil
|
|
})
|
|
|
|
callCount := 0
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
callCount++
|
|
return newVersionedStateWithRG(int64(callCount), map[uint64]serverInfo{
|
|
2: {resourceGroup: "rg_a"},
|
|
3: {resourceGroup: "rg_a"},
|
|
4: {resourceGroup: "rg_b"},
|
|
}), nil
|
|
})
|
|
|
|
nodes, err := m.CollectAllStatus(context.Background(), "rg_primary")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 2)
|
|
assert.Contains(t, nodes, int64(2))
|
|
assert.Contains(t, nodes, int64(3))
|
|
assert.ElementsMatch(t, []int64{2, 3}, pickedServerIDs)
|
|
|
|
managerService.EXPECT().Close().Return()
|
|
rb.EXPECT().Close().Return()
|
|
m.Close()
|
|
}
|
|
|
|
func TestCollectAllStatusFallbackTieBreakByMaxServerID(t *testing.T) {
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
managerService := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeManagerServiceClient](t)
|
|
m := &managerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
stopped: make(chan struct{}),
|
|
rb: rb,
|
|
service: managerService,
|
|
}
|
|
r := mock_resolver.NewMockResolver(t)
|
|
rb.EXPECT().Resolver().Return(r)
|
|
|
|
managerServiceClient := mock_streamingpb.NewMockStreamingNodeManagerServiceClient(t)
|
|
managerService.EXPECT().GetService(mock.Anything).RunAndReturn(func(ctx context.Context) (streamingpb.StreamingNodeManagerServiceClient, error) {
|
|
return managerServiceClient, nil
|
|
})
|
|
managerServiceClient.EXPECT().CollectStatus(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, snmcsr *streamingpb.StreamingNodeManagerCollectStatusRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerCollectStatusResponse, error) {
|
|
return &streamingpb.StreamingNodeManagerCollectStatusResponse{}, nil
|
|
})
|
|
|
|
callCount := 0
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
callCount++
|
|
return newVersionedStateWithRG(int64(callCount), map[uint64]serverInfo{
|
|
1: {resourceGroup: "rg_a"},
|
|
5: {resourceGroup: "rg_b"},
|
|
}), nil
|
|
})
|
|
|
|
nodes, err := m.CollectAllStatus(context.Background(), "non_existent_rg")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 1)
|
|
assert.Contains(t, nodes, int64(5))
|
|
|
|
managerService.EXPECT().Close().Return()
|
|
rb.EXPECT().Close().Return()
|
|
m.Close()
|
|
}
|
|
|
|
func TestCollectAllStatusWithEmptyResourceGroupAndExplicitDefaultRG(t *testing.T) {
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
managerService := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeManagerServiceClient](t)
|
|
m := &managerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
stopped: make(chan struct{}),
|
|
rb: rb,
|
|
service: managerService,
|
|
}
|
|
r := mock_resolver.NewMockResolver(t)
|
|
rb.EXPECT().Resolver().Return(r)
|
|
|
|
managerServiceClient := mock_streamingpb.NewMockStreamingNodeManagerServiceClient(t)
|
|
managerService.EXPECT().GetService(mock.Anything).RunAndReturn(func(ctx context.Context) (streamingpb.StreamingNodeManagerServiceClient, error) {
|
|
return managerServiceClient, nil
|
|
})
|
|
managerServiceClient.EXPECT().CollectStatus(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, snmcsr *streamingpb.StreamingNodeManagerCollectStatusRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerCollectStatusResponse, error) {
|
|
return &streamingpb.StreamingNodeManagerCollectStatusResponse{}, nil
|
|
})
|
|
|
|
callCount := 0
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
callCount++
|
|
return newVersionedStateWithRG(int64(callCount), map[uint64]serverInfo{
|
|
1: {resourceGroup: "__default_resource_group"},
|
|
2: {resourceGroup: "rg_a"},
|
|
}), nil
|
|
})
|
|
|
|
nodes, err := m.CollectAllStatus(context.Background(), "")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 1)
|
|
assert.Contains(t, nodes, int64(1))
|
|
assert.NotContains(t, nodes, int64(2))
|
|
|
|
managerService.EXPECT().Close().Return()
|
|
rb.EXPECT().Close().Return()
|
|
m.Close()
|
|
}
|
|
|
|
func TestCollectAllStatusWithEmptyResourceGroupAndDefaultRGAbsent(t *testing.T) {
|
|
rb := mock_resolver.NewMockBuilder(t)
|
|
managerService := mock_lazygrpc.NewMockService[streamingpb.StreamingNodeManagerServiceClient](t)
|
|
m := &managerClientImpl{
|
|
lifetime: typeutil.NewLifetime(),
|
|
stopped: make(chan struct{}),
|
|
rb: rb,
|
|
service: managerService,
|
|
}
|
|
r := mock_resolver.NewMockResolver(t)
|
|
rb.EXPECT().Resolver().Return(r)
|
|
|
|
managerServiceClient := mock_streamingpb.NewMockStreamingNodeManagerServiceClient(t)
|
|
managerService.EXPECT().GetService(mock.Anything).RunAndReturn(func(ctx context.Context) (streamingpb.StreamingNodeManagerServiceClient, error) {
|
|
return managerServiceClient, nil
|
|
})
|
|
managerServiceClient.EXPECT().CollectStatus(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, snmcsr *streamingpb.StreamingNodeManagerCollectStatusRequest, co ...grpc.CallOption) (*streamingpb.StreamingNodeManagerCollectStatusResponse, error) {
|
|
return &streamingpb.StreamingNodeManagerCollectStatusResponse{}, nil
|
|
})
|
|
|
|
callCount := 0
|
|
r.EXPECT().GetLatestState(mock.Anything).RunAndReturn(func(ctx context.Context) (discoverer.VersionedState, error) {
|
|
callCount++
|
|
return newVersionedStateWithRG(int64(callCount), map[uint64]serverInfo{
|
|
1: {resourceGroup: "rg_a"},
|
|
2: {resourceGroup: "rg_b"},
|
|
}), nil
|
|
})
|
|
|
|
nodes, err := m.CollectAllStatus(context.Background(), "")
|
|
assert.NoError(t, err)
|
|
assert.Len(t, nodes, 2)
|
|
assert.Contains(t, nodes, int64(1))
|
|
assert.Contains(t, nodes, int64(2))
|
|
|
|
managerService.EXPECT().Close().Return()
|
|
rb.EXPECT().Close().Return()
|
|
m.Close()
|
|
}
|
|
|
|
func TestDial(t *testing.T) {
|
|
paramtable.Init()
|
|
|
|
c, _ := kvfactory.GetEtcdAndPath()
|
|
assert.NotNil(t, c)
|
|
|
|
client := NewManagerClient(c)
|
|
assert.NotNil(t, client)
|
|
time.Sleep(100 * time.Millisecond)
|
|
client.Close()
|
|
}
|