Chore: Remove context.TODO() from services (#42555)
* Remove context.TODO() from services * Fix live test
This commit is contained in:
@@ -25,7 +25,7 @@ type ChannelLocalPublisher interface {
|
||||
}
|
||||
|
||||
type PluginContextGetter interface {
|
||||
GetPluginContext(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error)
|
||||
GetPluginContext(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error)
|
||||
}
|
||||
|
||||
type NumLocalSubscribersGetter interface {
|
||||
@@ -182,7 +182,7 @@ func (s *Manager) watchStream(ctx context.Context, cancelFn func(), sr streamReq
|
||||
case <-datasourceTicker.C:
|
||||
if sr.PluginContext.DataSourceInstanceSettings != nil {
|
||||
dsUID := sr.PluginContext.DataSourceInstanceSettings.UID
|
||||
pCtx, ok, err := s.pluginContextGetter.GetPluginContext(sr.user, sr.PluginContext.PluginID, dsUID, false)
|
||||
pCtx, ok, err := s.pluginContextGetter.GetPluginContext(ctx, sr.user, sr.PluginContext.PluginID, dsUID, false)
|
||||
if err != nil {
|
||||
logger.Error("Error getting datasource context", "channel", sr.Channel, "path", sr.Path, "error", err)
|
||||
continue
|
||||
@@ -283,7 +283,7 @@ func (s *Manager) runStream(ctx context.Context, cancelFn func(), sr streamReque
|
||||
if pluginCtx.DataSourceInstanceSettings != nil {
|
||||
datasourceUID = pluginCtx.DataSourceInstanceSettings.UID
|
||||
}
|
||||
newPluginCtx, ok, err := s.pluginContextGetter.GetPluginContext(sr.user, pluginCtx.PluginID, datasourceUID, false)
|
||||
newPluginCtx, ok, err := s.pluginContextGetter.GetPluginContext(ctx, sr.user, pluginCtx.PluginID, datasourceUID, false)
|
||||
if err != nil {
|
||||
logger.Error("Error getting plugin context", "path", sr.Path, "error", err)
|
||||
isReconnect = true
|
||||
@@ -407,7 +407,7 @@ func (s *Manager) SubmitStream(ctx context.Context, user *models.SignedInUser, c
|
||||
if pCtx.DataSourceInstanceSettings != nil {
|
||||
datasourceUID = pCtx.DataSourceInstanceSettings.UID
|
||||
}
|
||||
newPluginCtx, ok, err := s.pluginContextGetter.GetPluginContext(user, pCtx.PluginID, datasourceUID, false)
|
||||
newPluginCtx, ok, err := s.pluginContextGetter.GetPluginContext(ctx, user, pCtx.PluginID, datasourceUID, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -71,7 +71,7 @@ func TestStreamManager_SubmitStream_Send(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
mockContextGetter.EXPECT().GetPluginContext(context.Background(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
require.Equal(t, int64(2), user.UserId)
|
||||
require.Equal(t, int64(1), user.OrgId)
|
||||
require.Equal(t, testPluginContext.PluginID, pluginID)
|
||||
@@ -83,7 +83,7 @@ func TestStreamManager_SubmitStream_Send(t *testing.T) {
|
||||
|
||||
mockStreamRunner := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner.EXPECT().RunStream(
|
||||
gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
require.Equal(t, "test", req.Path)
|
||||
close(startedCh)
|
||||
@@ -133,13 +133,13 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
|
||||
mockPacketSender.EXPECT().PublishLocal("1/test", gomock.Any()).Times(1)
|
||||
mockPacketSender.EXPECT().PublishLocal("2/test", gomock.Any()).Times(1)
|
||||
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
mockContextGetter.EXPECT().GetPluginContext(context.Background(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
return backend.PluginContext{}, true, nil
|
||||
}).Times(0)
|
||||
|
||||
mockStreamRunner1 := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner1.EXPECT().RunStream(
|
||||
gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
require.Equal(t, "test", req.Path)
|
||||
close(startedCh1)
|
||||
@@ -152,7 +152,7 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
|
||||
|
||||
mockStreamRunner2 := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner2.EXPECT().RunStream(
|
||||
gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
require.Equal(t, "test", req.Path)
|
||||
close(startedCh2)
|
||||
@@ -205,14 +205,14 @@ func TestStreamManager_SubmitStream_CloseNoSubscribers(t *testing.T) {
|
||||
startedCh := make(chan struct{})
|
||||
doneCh := make(chan struct{})
|
||||
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
mockContextGetter.EXPECT().GetPluginContext(context.Background(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
return backend.PluginContext{}, true, nil
|
||||
}).Times(0)
|
||||
|
||||
mockNumSubscribersGetter.EXPECT().GetNumLocalSubscribers("1/test").Return(0, nil).Times(3)
|
||||
|
||||
mockStreamRunner := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner.EXPECT().RunStream(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
mockStreamRunner.EXPECT().RunStream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
close(startedCh)
|
||||
<-ctx.Done()
|
||||
close(doneCh)
|
||||
@@ -254,7 +254,7 @@ func TestStreamManager_SubmitStream_ErrorRestartsRunStream(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
require.Equal(t, int64(2), user.UserId)
|
||||
require.Equal(t, int64(1), user.OrgId)
|
||||
require.Equal(t, testPluginContext.PluginID, pluginID)
|
||||
@@ -264,7 +264,7 @@ func TestStreamManager_SubmitStream_ErrorRestartsRunStream(t *testing.T) {
|
||||
|
||||
mockStreamRunner := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner.EXPECT().RunStream(
|
||||
gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
if currentErrors >= numErrors {
|
||||
return nil
|
||||
@@ -296,13 +296,13 @@ func TestStreamManager_SubmitStream_NilErrorStopsRunStream(t *testing.T) {
|
||||
_ = manager.Run(ctx)
|
||||
}()
|
||||
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
mockContextGetter.EXPECT().GetPluginContext(context.Background(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
return backend.PluginContext{}, true, nil
|
||||
}).Times(0)
|
||||
|
||||
mockStreamRunner := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner.EXPECT().RunStream(
|
||||
gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
return nil
|
||||
}).Times(1)
|
||||
@@ -337,7 +337,7 @@ func TestStreamManager_HandleDatasourceUpdate(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
require.Equal(t, int64(2), user.UserId)
|
||||
require.Equal(t, int64(1), user.OrgId)
|
||||
require.Equal(t, testPluginContext.PluginID, pluginID)
|
||||
@@ -352,7 +352,7 @@ func TestStreamManager_HandleDatasourceUpdate(t *testing.T) {
|
||||
|
||||
mockStreamRunner := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner.EXPECT().RunStream(
|
||||
gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
if isFirstCall {
|
||||
// first RunStream will wait till context done.
|
||||
@@ -403,7 +403,7 @@ func TestStreamManager_HandleDatasourceDelete(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
mockContextGetter.EXPECT().GetPluginContext(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
mockContextGetter.EXPECT().GetPluginContext(context.Background(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, user *models.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
|
||||
require.Equal(t, int64(2), user.UserId)
|
||||
require.Equal(t, int64(1), user.OrgId)
|
||||
require.Equal(t, testPluginContext.PluginID, pluginID)
|
||||
@@ -415,7 +415,7 @@ func TestStreamManager_HandleDatasourceDelete(t *testing.T) {
|
||||
|
||||
mockStreamRunner := NewMockStreamRunner(mockCtrl)
|
||||
mockStreamRunner.EXPECT().RunStream(
|
||||
gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
|
||||
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
close(doneCh)
|
||||
<-ctx.Done()
|
||||
|
||||
@@ -120,7 +120,7 @@ func (m *MockStreamRunner) RunStream(arg0 context.Context, arg1 *backend.RunStre
|
||||
}
|
||||
|
||||
// RunStream indicates an expected call of RunStream.
|
||||
func (mr *MockStreamRunnerMockRecorder) RunStream(arg0, arg1, arg2 interface{}) *gomock.Call {
|
||||
func (mr *MockStreamRunnerMockRecorder) RunStream(ctx, arg0, arg1, arg2 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RunStream", reflect.TypeOf((*MockStreamRunner)(nil).RunStream), arg0, arg1, arg2)
|
||||
}
|
||||
@@ -149,9 +149,9 @@ func (m *MockPluginContextGetter) EXPECT() *MockPluginContextGetterMockRecorder
|
||||
}
|
||||
|
||||
// GetPluginContext mocks base method.
|
||||
func (m *MockPluginContextGetter) GetPluginContext(arg0 *models.SignedInUser, arg1, arg2 string, arg3 bool) (backend.PluginContext, bool, error) {
|
||||
func (m *MockPluginContextGetter) GetPluginContext(ctx context.Context, arg0 *models.SignedInUser, arg1, arg2 string, arg3 bool) (backend.PluginContext, bool, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "GetPluginContext", arg0, arg1, arg2, arg3)
|
||||
ret := m.ctrl.Call(m, "GetPluginContext", ctx, arg0, arg1, arg2, arg3)
|
||||
ret0, _ := ret[0].(backend.PluginContext)
|
||||
ret1, _ := ret[1].(bool)
|
||||
ret2, _ := ret[2].(error)
|
||||
@@ -159,7 +159,7 @@ func (m *MockPluginContextGetter) GetPluginContext(arg0 *models.SignedInUser, ar
|
||||
}
|
||||
|
||||
// GetPluginContext indicates an expected call of GetPluginContext.
|
||||
func (mr *MockPluginContextGetterMockRecorder) GetPluginContext(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
|
||||
func (mr *MockPluginContextGetterMockRecorder) GetPluginContext(ctx, arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPluginContext", reflect.TypeOf((*MockPluginContextGetter)(nil).GetPluginContext), arg0, arg1, arg2, arg3)
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPluginContext", reflect.TypeOf((*MockPluginContextGetter)(nil).GetPluginContext), ctx, arg0, arg1, arg2, arg3)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user