package console

import (
	"context"
	"encoding/json"
	"testing"

	aws_unit "github.com/aws/aws-sdk-go/awstesting/unit"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	feature_gate_impl "github.com/tidbcloud/dataflow-service/app/internal/feature_gate/impl"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_mock "github.com/tidbcloud/dataflow-service/app/models/ticdc/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db/mock_db"
	"go.uber.org/mock/gomock"
)

type bizChangefeedControllerTestSuite struct {
	suite.Suite
	mockBizChangefeedModel *biz_changefeed_mock.BizChangefeedModel
	mockCDCClusterModel    *ticdc_mock.ClusterModel
	mockCoreCli            *grpcli_mock.CoreGrpcClient
	controller             *bizChangefeedControllerImpl
}

func (s *bizChangefeedControllerTestSuite) SetupSuite() {
	trans := mock_db.NewMockTransProvider(gomock.NewController(s.T()))
	s.mockBizChangefeedModel = biz_changefeed_mock.NewBizChangefeedModel(s.T())
	s.mockCDCClusterModel = ticdc_mock.NewClusterModel(s.T())
	s.mockCoreCli = grpcli_mock.NewCoreGrpcClient(s.T())
	downstreamFactory := NewBizChangefeedDownstreamAdaptorFactory(nil, aws_unit.Session.Copy())
	s.controller = &bizChangefeedControllerImpl{
		BaseController:    NewBaseController(trans),
		changefeedModel:   s.mockBizChangefeedModel,
		cdcClusterModel:   s.mockCDCClusterModel,
		downstreamFactory: downstreamFactory,
		coreCli:           s.mockCoreCli,
		featureGate:       feature_gate_impl.NewFeatureGate(),
	}
}

func (s *bizChangefeedControllerTestSuite) TestListConnectors() {
	mockReq := &dataflow.ConnectorListReq{
		ClusterId:     111,
		Page:          1,
		PerPage:       5,
		ConnectorType: dataflow.ConnectorType_MYSQL,
	}
	listReqMatcher := mock.MatchedBy(func(req *domain.ListBizChangefeedsReq) bool {
		return req.ClusterID == mockReq.ClusterId &&
			req.Page == mockReq.Page &&
			req.PerPage == mockReq.PerPage &&
			*req.DownstreamType == entity.TiCDCDownStreamTypeMySQL
	})

	s.Run("list failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedModel.
			On("ListBizChangefeeds", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.ListConnectors(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	mockListResp := &domain.ListBizChangefeedsResp{
		Total: 10,
		BizChangefeeds: []*domain.BizChangefeed{
			{
				BizChangefeedID:     1,
				Name:                "a",
				DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
				DownstreamParameter: s.mustGenDownstreamParams(entity.TiCDCDownStreamTypeMySQL),
			},
			{
				BizChangefeedID:     2,
				Name:                "b",
				DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
				DownstreamParameter: s.mustGenDownstreamParams(entity.TiCDCDownStreamTypeMySQL),
			},
			{
				BizChangefeedID:     3,
				Name:                "c",
				DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
				DownstreamParameter: s.mustGenDownstreamParams(entity.TiCDCDownStreamTypeMySQL),
			},
			{
				BizChangefeedID:     4,
				Name:                "d",
				DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
				DownstreamParameter: s.mustGenDownstreamParams(entity.TiCDCDownStreamTypeMySQL),
			},
			{
				BizChangefeedID:     5,
				Name:                "e",
				DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
				DownstreamParameter: s.mustGenDownstreamParams(entity.TiCDCDownStreamTypeMySQL),
			},
		},
	}
	s.Run("list successfully", func() {
		call := s.mockBizChangefeedModel.
			On("ListBizChangefeeds", mock.Anything, mock.Anything).
			Return(mockListResp, nil)
		defer call.Unset()
		resp, err := s.controller.ListConnectors(context.Background(), mockReq)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(mockListResp.Total, resp.Total)
		s.Equal(len(mockListResp.BizChangefeeds), len(resp.Items))
		for i := range resp.Items {
			s.Equal(mockListResp.BizChangefeeds[i].BizChangefeedID, resp.Items[i].ConnectorId)
			s.Equal(mockListResp.BizChangefeeds[i].Name, resp.Items[i].Name)
			s.NotNil(resp.Items[i].Request)
		}
		s.mockBizChangefeedModel.AssertCalled(s.T(), "ListBizChangefeeds", mock.Anything, listReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) mustGenDownstreamParams(
	downstreamType entity.TiCDCChangefeedDownStreamType,
) string {
	var downstreamParams interface{}
	switch downstreamType {
	case entity.TiCDCDownStreamTypeMySQL:
		downstreamParams = &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "a.b:4000",
				Username: "u",
				Password: "p",
			},
		}
	default:
		s.FailNow("gen downstream params failed", "not supported downstream type %s", downstreamType)
	}

	paramsBytes, err := json.Marshal(downstreamParams)
	if err != nil {
		s.FailNow("gen downstream params failed", "marshal downstream params error: %s", err.Error())
	}
	return string(paramsBytes)
}

func (s *bizChangefeedControllerTestSuite) TestDeleteConnector() {
	mockReq := &dataflow.ConnectorID{
		ConnectorId: 1111,
	}
	deleteReqMatcher := mock.MatchedBy(func(req *domain.DeleteBizChangefeedReq) bool {
		return req.BizChangefeedID == mockReq.ConnectorId
	})
	s.Run("delete failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedModel.
			On("DeleteBizChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.DeleteConnector(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("delete successfully", func() {
		mockDeleteResp := &domain.DeleteBizChangefeedResp{
			WorkflowID: "a",
		}
		call := s.mockBizChangefeedModel.
			On("DeleteBizChangefeed", mock.Anything, mock.Anything).
			Return(mockDeleteResp, nil)
		defer call.Unset()
		_, err := s.controller.DeleteConnector(context.Background(), mockReq)
		s.NoError(err)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "DeleteBizChangefeed", mock.Anything, deleteReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestScaleChangefeed() {
	mockReq := &dataflow.ChangefeedScaleReq{
		ClusterId:              111,
		ChangefeedId:           1111,
		ChangefeedInstanceType: "abc",
	}
	scaleReqMatcher := mock.MatchedBy(func(req *domain.ScaleBizChangefeedReq) bool {
		return req.BizChangefeedID == mockReq.ChangefeedId && req.SpecificationCode == mockReq.ChangefeedInstanceType
	})
	s.Run("scale failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedModel.
			On("ScaleBizChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.ScaleChangefeed(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("scale successfully", func() {
		mockDeleteResp := &domain.ScaleBizChangefeedResp{
			WorkflowID: "a",
		}
		call := s.mockBizChangefeedModel.
			On("ScaleBizChangefeed", mock.Anything, mock.Anything).
			Return(mockDeleteResp, nil)
		defer call.Unset()
		_, err := s.controller.ScaleChangefeed(context.Background(), mockReq)
		s.NoError(err)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "ScaleBizChangefeed", mock.Anything, scaleReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestCalChangefeedFee() {
	mockReq := &dataflow.CalChangefeedFeeReq{
		OrgId:                  1,
		ProjectId:              11,
		RegionName:             "us-west-2",
		CloudProviderName:      "aws",
		ChangefeedInstanceType: "abc",
		Duration:               60,
	}
	calReqMatcher := mock.MatchedBy(func(req *domain.CalBizChangefeedFeeReq) bool {
		return req.TenantID == mockReq.OrgId &&
			req.ProjectID == mockReq.ProjectId &&
			req.RegionName == mockReq.RegionName &&
			string(req.CloudProvider) == mockReq.CloudProviderName &&
			req.SpecificationCode == mockReq.ChangefeedInstanceType &&
			req.Duration == mockReq.Duration
	})
	s.Run("cal failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockBizChangefeedModel.
			On("CalBizChangefeedFee", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.CalChangefeedFee(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("cal successfully", func() {
		mockCalResp := &domain.CalBizChangefeedFeeResp{
			Fee:    10.0,
			RawFee: 20.0,
		}
		call := s.mockBizChangefeedModel.
			On("CalBizChangefeedFee", mock.Anything, mock.Anything).
			Return(mockCalResp, nil)
		defer call.Unset()
		resp, err := s.controller.CalChangefeedFee(context.Background(), mockReq)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(mockCalResp.RawFee, resp.RawFee)
		s.Equal(mockCalResp.Fee, resp.Fee)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "CalBizChangefeedFee", mock.Anything, calReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestStartConnector() {
	mockReq := &dataflow.ConnectorID{
		ClusterId:   111,
		ConnectorId: 1111,
	}
	resumeReqMatcher := mock.MatchedBy(func(req *domain.ResumeBizChangefeedReq) bool {
		return req.BizChangefeedID == mockReq.ConnectorId
	})
	s.Run("start failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockBizChangefeedModel.
			On("ResumeBizChangefeed", mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		_, err := s.controller.StartConnector(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("start successfully", func() {
		call := s.mockBizChangefeedModel.
			On("ResumeBizChangefeed", mock.Anything, mock.Anything).
			Return(nil)
		defer call.Unset()
		_, err := s.controller.StartConnector(context.Background(), mockReq)
		s.NoError(err)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "ResumeBizChangefeed", mock.Anything, resumeReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestStopConnector() {
	mockReq := &dataflow.ConnectorID{
		ClusterId:   111,
		ConnectorId: 1111,
	}
	pauseReqMatcher := mock.MatchedBy(func(req *domain.PauseBizChangefeedReq) bool {
		return req.BizChangefeedID == mockReq.ConnectorId
	})
	s.Run("stop failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockBizChangefeedModel.
			On("PauseBizChangefeed", mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		_, err := s.controller.StopConnector(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("stop successfully", func() {
		call := s.mockBizChangefeedModel.
			On("PauseBizChangefeed", mock.Anything, mock.Anything).
			Return(nil)
		defer call.Unset()
		_, err := s.controller.StopConnector(context.Background(), mockReq)
		s.NoError(err)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "PauseBizChangefeed", mock.Anything, pauseReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestGetConnector() {
	mockReq := &dataflow.ConnectorID{
		ClusterId:   111,
		ConnectorId: 1111,
	}
	getReqMatcher := mock.MatchedBy(func(req *domain.GetBizChangefeedReq) bool {
		return req.BizChangefeedID == mockReq.ConnectorId
	})
	s.Run("get failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeRecordNotFound, "not found")
		call := s.mockBizChangefeedModel.
			On("GetBizChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.GetConnector(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	mockGetResp := &domain.GetBizChangefeedResp{
		BizChangefeed: &domain.BizChangefeed{
			BizChangefeedID:     1,
			Name:                "a",
			DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
			DownstreamParameter: s.mustGenDownstreamParams(entity.TiCDCDownStreamTypeMySQL),
		},
	}
	s.Run("get successfully", func() {
		call := s.mockBizChangefeedModel.
			On("GetBizChangefeed", mock.Anything, mock.Anything).
			Return(mockGetResp, nil)
		defer call.Unset()
		resp, err := s.controller.GetConnector(context.Background(), mockReq)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(mockGetResp.BizChangefeed.BizChangefeedID, resp.ConnectorId)
		s.Equal(mockGetResp.BizChangefeed.Name, resp.Name)
		s.NotNil(resp.Request)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "GetBizChangefeed", mock.Anything, getReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestGetClusterTCU() {
	mockReq := &dataflow.GetClusterTCUReq{
		ClusterId: 111,
	}
	rcuReqMatcher := mock.MatchedBy(func(req *domain.GetClusterBizChangefeedRCUReq) bool {
		return req.ClusterID == mockReq.ClusterId
	})
	s.Run("get tcu failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedModel.
			On("GetClusterBizChangefeedRCU", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.GetClusterTCU(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("get tcu successfully", func() {
		mockRCUResp := &domain.GetClusterBizChangefeedRCUResp{
			RCUNumber: 8,
		}
		call := s.mockBizChangefeedModel.
			On("GetClusterBizChangefeedRCU", mock.Anything, mock.Anything).
			Return(mockRCUResp, nil)
		defer call.Unset()
		resp, err := s.controller.GetClusterTCU(context.Background(), mockReq)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(uint64(mockRCUResp.RCUNumber), resp.TcuNumber)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "GetClusterBizChangefeedRCU", mock.Anything, rcuReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestCreateConnector() {
	mockReq := &dataflow.ConnectorCreateReq{
		ClusterId:              111,
		Name:                   "a",
		ChangefeedInstanceType: "abc",
	}

	s.Run("get downstream adaptor failed", func() {
		mockReq.ConnectorType = dataflow.ConnectorType(10)
		_, err := s.controller.CreateConnector(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "invalid connector type")
	})

	mockReq.ConnectorType = dataflow.ConnectorType_MYSQL
	s.Run("get cluster failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeGRPC, "network error")
		call := s.mockCoreCli.
			On("GetAvailableClusterByID", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.CreateConnector(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "get cluster")
	})

	mockCluster := &core.Cluster{
		Id:      111,
		Version: "v7.5.0",
	}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, uint64(111)).
		Return(mockCluster, nil)
	defer clusterCall.Unset()

	mockTiCDCCluster := entity.TiCDCCluster{
		DeployType: entity.TiCDCClusterDeployTypeShared,
	}
	mockTiCDCCluster.ID = 1111
	mockListCDCResp := &domain.ListTiCDCClustersResp{
		Items: []entity.TiCDCCluster{mockTiCDCCluster},
	}
	listCDCCall := s.mockCDCClusterModel.On("List", mock.Anything, mock.Anything).Return(mockListCDCResp, nil)
	defer listCDCCall.Unset()

	mockVersionResp := &domain.GetExistingTiCDCClusterVersionResp{
		Version: "v7.5.0",
		NewArch: false,
	}
	versionCall := s.mockCDCClusterModel.
		On("GetExistingClusterVersion", mock.Anything, mock.Anything).
		Return(mockVersionResp, nil)
	defer versionCall.Unset()

	accessPointsCall := s.mockCoreCli.
		On("GetClusterAccessPoints", mock.Anything, mock.Anything).
		Return([]*core.AccessPoint{}, nil)
	defer accessPointsCall.Unset()

	mockReq.MysqlInfo = &dataflow.MySQLInfo{}
	s.Run("validate downstream params failed", func() {
		_, err := s.controller.CreateConnector(context.Background(), mockReq)
		s.Error(err)
	})

	mockReq.MysqlInfo = &dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: "a.b:4000",
			Username: "u",
			Password: "p",
		},
		TableConfig: &dataflow.TableConfig{
			FilterRule:      []string{},
			EventFilterRule: []*dataflow.EventFilterRule{},
		},
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_NOW,
		},
	}
	createReqMatcher := mock.MatchedBy(func(req *domain.CreateBizChangefeedReq) bool {
		return req.ClusterID == mockReq.ClusterId &&
			req.Name == mockReq.Name &&
			req.DownstreamType == mockReq.ConnectorType &&
			req.SpecificationCode == mockReq.ChangefeedInstanceType &&
			s.Equal(mockReq.MysqlInfo, req.DownstreamParameter)
	})
	s.Run("create failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedModel.
			On("CreateBizChangefeed", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.controller.CreateConnector(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	mockCreateResp := &domain.CreateBizChangefeedResp{
		BizChangefeedID: 1111,
		WorkflowID:      "123",
	}
	s.Run("create successfully", func() {
		call := s.mockBizChangefeedModel.
			On("CreateBizChangefeed", mock.Anything, mock.Anything).
			Return(mockCreateResp, nil)
		defer call.Unset()
		resp, err := s.controller.CreateConnector(context.Background(), mockReq)
		s.NoError(err)
		s.NotNil(resp)
		s.Equal(mockCreateResp.BizChangefeedID, resp.ConnectorId)
		s.Equal(mockReq.ClusterId, resp.ClusterId)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "CreateBizChangefeed", mock.Anything, createReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestEditConnector() {
	mockReq := &dataflow.ConnectorEditReq{
		ConnectorId: 1111,
		ClusterId:   111,
		Name:        "a",
	}
	s.Run("get downstream adaptor failed", func() {
		mockReq.ConnectorType = dataflow.ConnectorType(10)
		_, err := s.controller.EditConnector(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "invalid connector type")
	})

	mockReq.ConnectorType = dataflow.ConnectorType_MYSQL
	mockChangefeedResp := &domain.GetBizChangefeedResp{
		BizChangefeed: &domain.BizChangefeed{
			BizChangefeedID: 11111,
			TiCDCClusterID:  1111,
		},
	}
	getChangefeedCall := s.mockBizChangefeedModel.
		On("GetBizChangefeed", mock.Anything, mock.Anything).
		Return(mockChangefeedResp, nil)
	defer getChangefeedCall.Unset()

	mockCDCVersionResp := &domain.GetExistingTiCDCClusterVersionResp{
		Version: "v7.5.0",
		NewArch: false,
	}
	getCDCClusterCall := s.mockCDCClusterModel.
		On("GetExistingClusterVersion", mock.Anything, mock.Anything).
		Return(mockCDCVersionResp, nil)
	defer getCDCClusterCall.Unset()

	accessPointsCall := s.mockCoreCli.
		On("GetClusterAccessPoints", mock.Anything, mock.Anything).
		Return([]*core.AccessPoint{}, nil)
	defer accessPointsCall.Unset()

	mockReq.MysqlInfo = &dataflow.MySQLInfo{}
	s.Run("validate downstream params failed", func() {
		_, err := s.controller.EditConnector(context.Background(), mockReq)
		s.Error(err)
	})

	mockReq.MysqlInfo = &dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: "a.b:4000",
			Username: "u",
			Password: "p",
		},
		TableConfig: &dataflow.TableConfig{
			FilterRule:      []string{},
			EventFilterRule: []*dataflow.EventFilterRule{},
		},
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_NOW,
		},
	}
	editReqMatcher := mock.MatchedBy(func(req *domain.EditBizChangefeedReq) bool {
		return req.BizChangefeedID == mockReq.ConnectorId &&
			req.Name == mockReq.Name &&
			req.DownstreamType == mockReq.ConnectorType &&
			s.Equal(mockReq.MysqlInfo, req.DownstreamParameter)
	})
	s.Run("edit failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedModel.
			On("EditBizChangefeed", mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		_, err := s.controller.EditConnector(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("edit successfully", func() {
		call := s.mockBizChangefeedModel.
			On("EditBizChangefeed", mock.Anything, mock.Anything).
			Return(nil)
		defer call.Unset()
		_, err := s.controller.EditConnector(context.Background(), mockReq)
		s.NoError(err)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "EditBizChangefeed", mock.Anything, editReqMatcher)
	})
}

func (s *bizChangefeedControllerTestSuite) TestTestConnector() {
	mockReq := &dataflow.ConnectorCreateReq{
		ClusterId: 111,
		Name:      "a",
	}
	s.Run("get downstream adaptor failed", func() {
		mockReq.ConnectorType = dataflow.ConnectorType(10)
		_, err := s.controller.TestConnector(context.Background(), mockReq)
		s.Error(err)
		s.Contains(err.Error(), "invalid connector type")
	})

	mockReq.ConnectorType = dataflow.ConnectorType_MYSQL
	mockReq.MysqlInfo = &dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: "a.b:4000",
			Username: "u",
			Password: "p",
		},
	}
	testReqMatcher := mock.MatchedBy(func(req *domain.TestBizChangefeedReq) bool {
		return req.ClusterID == mockReq.ClusterId &&
			req.DownstreamType == mockReq.ConnectorType &&
			s.Equal(mockReq.MysqlInfo, req.DownstreamParameter)
	})

	mockCluster := &core.Cluster{
		Id:      111,
		Version: "v7.5.0",
	}
	clusterCall := s.mockCoreCli.
		On("GetAvailableClusterByID", mock.Anything, uint64(111)).
		Return(mockCluster, nil)
	defer clusterCall.Unset()

	accessPointsCall := s.mockCoreCli.
		On("GetClusterAccessPoints", mock.Anything, mock.Anything).
		Return([]*core.AccessPoint{}, nil)
	defer accessPointsCall.Unset()

	s.Run("test failed", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockBizChangefeedModel.
			On("TestBizChangefeed", mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		_, err := s.controller.TestConnector(context.Background(), mockReq)
		s.Error(err)
		s.Equal(mockErr, err)
	})

	s.Run("test successfully", func() {
		call := s.mockBizChangefeedModel.
			On("TestBizChangefeed", mock.Anything, mock.Anything).
			Return(nil)
		defer call.Unset()
		_, err := s.controller.TestConnector(context.Background(), mockReq)
		s.NoError(err)
		s.mockBizChangefeedModel.AssertCalled(s.T(), "TestBizChangefeed", mock.Anything, testReqMatcher)
	})
}

func TestBizChangefeedControllerTestSuite(t *testing.T) {
	suite.Run(t, new(bizChangefeedControllerTestSuite))
}