package console

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"testing"
	"time"

	"github.com/Masterminds/semver/v3"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	aws_unit "github.com/aws/aws-sdk-go/awstesting/unit"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/app/config"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	sink_endpoint_mock "github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/pkgv2/bizerr"
	v12 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"k8s.io/apimachinery/pkg/util/rand"
)

var letterRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")

func TestBizChangefeedDownstreamAdaptorFactory(t *testing.T) {
	testcases := []struct {
		connectorType dataflow.ChangefeedType
		hasInstance   bool
	}{
		{
			connectorType: dataflow.ChangefeedType_KAFKA,
			hasInstance:   true,
		},
		{
			connectorType: dataflow.ChangefeedType_MYSQL,
			hasInstance:   true,
		},
		{
			connectorType: dataflow.ChangefeedType_SERVERLESS_TIDB,
			hasInstance:   true,
		},
		{
			connectorType: dataflow.ChangefeedType_S3,
			hasInstance:   true,
		},
		{
			connectorType: dataflow.ChangefeedType_GCS,
			hasInstance:   true,
		},
		{
			connectorType: dataflow.ChangefeedType_PULSAR,
			hasInstance:   true,
		},
	}

	fac := newBizChangefeedDownstreamAdaptorFactory(aws_unit.Session.Copy(), nil)
	for _, tc := range testcases {
		_, ok := fac.Get(tc.connectorType)
		if tc.hasInstance {
			assert.True(t, ok)
		} else {
			assert.False(t, ok)
		}
	}
}

func TestMySQLBizChangefeedDownstreamAdaptor_GetDownstreamParamsFromCreateReq(t *testing.T) {
	reqs := []*dataflow.ChangefeedCreateReq{
		{
			MysqlInfo: nil,
		},
		{
			MysqlInfo: &dataflow.MySQLInfo{},
		},
		{
			MysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a",
				},
			},
		},
	}
	adaptor := newMySQLBizChangefeedDownstreamAdaptor(nil)
	ctx := context.Background()
	for _, req := range reqs {
		mysqlInfo, err := adaptor.ParseCreateParams(ctx, req, nil)
		require.NoError(t, err)
		assert.Equal(t, req.MysqlInfo, mysqlInfo)
	}
}

func TestMySQLBizChangefeedDownstreamAdaptor_PrivateLinkUsesClusterScopedSinkEndpoint(t *testing.T) {
	sinkEndpointModel := sink_endpoint_mock.NewSinkEndpointModel(t)
	adaptor := newMySQLBizChangefeedDownstreamAdaptor(sinkEndpointModel)
	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId: 123,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &v12.Cluster{},
		},
	})
	endpointID := uint64(456)
	req := &dataflow.ChangefeedCreateReq{
		NetworkInfo: &dataflow.NetworkInfo{
			NetworkType:    dataflow.NetworkType_NetworkType_Private_Link,
			SinkEndpointId: &endpointID,
			Ports:          []uint32{3306},
		},
		MysqlInfo: &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{},
		},
	}
	queryMatcher := mock.MatchedBy(func(req *domain.QuerySinkEndpointReq) bool {
		return req.ID == endpointID && req.ClusterID == cluster.GetClusterID()
	})
	sinkEndpointModel.
		On("QuerySinkEndpoint", mock.Anything, queryMatcher).
		Return(&domain.QuerySinkEndpointResp{
			SinkEndpoint: &entity.BizSinkEndpoint{
				Status:          entity.BizPLEndpointStatusAvailable,
				EndpointAddress: "endpoint.local",
			},
		}, nil).
		Once()

	params, err := adaptor.ParseCreateParams(context.Background(), req, cluster)
	require.NoError(t, err)
	require.Equal(t, req.MysqlInfo, params)
	require.Equal(t, "endpoint.local:3306", req.MysqlInfo.Connection.Endpoint)
}

func TestMySQLBizChangefeedDownstreamAdaptor_GetDownstreamParamsFromEditReq(t *testing.T) {
	reqs := []*dataflow.ChangefeedEditReq{
		{
			MysqlInfo: nil,
		},
		{
			MysqlInfo: &dataflow.MySQLInfo{},
		},
		{
			MysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a",
				},
			},
		},
	}
	adaptor := newMySQLBizChangefeedDownstreamAdaptor(nil)
	for _, req := range reqs {
		mysqlInfo := adaptor.ParseEditParams(req)
		assert.Equal(t, req.MysqlInfo, mysqlInfo)
	}
}

func TestMySQLBizChangefeedDownstreamAdaptor_ValidateCreateDownstreamParams(t *testing.T) {
	testcases := []struct {
		name    string
		params  interface{}
		cluster *utils.Cluster
		err     bool
	}{
		{
			name:   "downstream params nil",
			params: nil,
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name:   "downstream params not mysql info type",
			params: &dataflow.KafkaInfo{},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name:   "empty downstream params",
			params: &dataflow.MySQLInfo{},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "downstream params connection nil",
			params: &dataflow.MySQLInfo{
				TableConfig:   &dataflow.TableConfig{},
				StartPosition: &dataflow.StartPosition{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "downstream params table config nil",
			params: &dataflow.MySQLInfo{
				Connection:    &dataflow.Connection{},
				StartPosition: &dataflow.StartPosition{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "downstream params connection nil",
			params: &dataflow.MySQLInfo{
				Connection:  &dataflow.Connection{},
				TableConfig: &dataflow.TableConfig{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "check connection failed",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "check table config failed",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:1000",
					Username: "u",
					Password: "p",
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "check start position failed",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:1000",
					Username: "u",
					Password: "p",
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_TSO,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "valid downstream params",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:1000",
					Username: "u",
					Password: "p",
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: false,
		},
	}

	adaptor := newMySQLBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateCreateParams(tc.params, tc.cluster)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestMySQLBizChangefeedDownstreamAdaptor_ValidateEditDownstreamParams(t *testing.T) {
	testcases := []struct {
		name           string
		params         interface{}
		clusterVersion string
		err            bool
	}{
		{
			name:           "downstream params nil",
			params:         nil,
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name:           "downstream params not mysql info type",
			params:         &dataflow.KafkaInfo{},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name:           "empty downstream params",
			params:         &dataflow.MySQLInfo{},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "downstream params connection nil",
			params: &dataflow.MySQLInfo{
				TableConfig:   &dataflow.TableConfig{},
				StartPosition: &dataflow.StartPosition{},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "downstream params table config nil",
			params: &dataflow.MySQLInfo{
				Connection:    &dataflow.Connection{},
				StartPosition: &dataflow.StartPosition{},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "downstream params connection nil",
			params: &dataflow.MySQLInfo{
				Connection:  &dataflow.Connection{},
				TableConfig: &dataflow.TableConfig{},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "check connection failed",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "check table config failed",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:1000",
					Username: "u",
					Password: "p",
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "start position is nil",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:1000",
					Username: "u",
					Password: "p",
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
			},
			clusterVersion: "v6.5.9",
			err:            false,
		},
		{
			name: "valid downstream params",
			params: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Username: "u",
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
			},
			clusterVersion: "v6.5.9",
			err:            false,
		},
	}

	adaptor := newMySQLBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateEditParams(tc.params, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestMySQLBizChangefeedDownstreamAdaptor_CheckClusterVersion(t *testing.T) {
	testcases := []struct {
		name           string
		params         *dataflow.MySQLInfo
		clusterVersion string
		err            bool
	}{
		{
			name: "event filter version check failed",
			params: &dataflow.MySQLInfo{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v6.5.2",
			err:            true,
		},
		{
			name:           "cluster version check failed",
			params:         &dataflow.MySQLInfo{},
			clusterVersion: "v6.0.0",
			err:            true,
		},
		{
			name: "check successful",
			params: &dataflow.MySQLInfo{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v6.5.9",
			err:            false,
		},
	}

	adaptor := &mysqlBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.checkClusterVersion(tc.params, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestMySQLBizChangefeedDownstreamAdaptor_AssembleCreateConnectorReq(t *testing.T) {
	testcases := []struct {
		downstreamParams interface{}
		assembledParams  *dataflow.MySQLInfo
		err              bool
	}{
		{
			downstreamParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "a.b.c:9092",
				},
			},
			assembledParams: nil,
			err:             true,
		},
		{
			downstreamParams: &dataflow.MySQLInfo{},
			assembledParams:  &dataflow.MySQLInfo{},
			err:              false,
		},
		{
			downstreamParams: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:4000",
					Username: "u",
					Password: "p",
				},
			},
			assembledParams: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:4000",
					Username: "u",
					Password: "",
				},
			},
			err: false,
		},
	}

	adaptor := newMySQLBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		paramsBytes, err := json.Marshal(tc.downstreamParams)
		require.NoError(t, err)
		bizChangefeed := &domain.BizChangefeed{
			DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
			DownstreamParameter: string(paramsBytes),
		}
		req, err := adaptor.AssembleGetResponse(bizChangefeed)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
			assert.Equal(t, tc.assembledParams, req.MysqlInfo)
		}
	}
}

func TestValidateEndpoint(t *testing.T) {
	testcases := []struct {
		endpoint string
		valid    bool
	}{
		{
			endpoint: "",
			valid:    false,
		},
		{
			endpoint: "a.b.c",
			valid:    false,
		},
		{
			endpoint: "a.b.c:100:100",
			valid:    false,
		},
		{
			endpoint: "a.b.c:",
			valid:    false,
		},
		{
			endpoint: "a.b.c:70000",
			valid:    false,
		},
		{
			endpoint: "a.b.c:65536",
			valid:    false,
		},
		{
			endpoint: "a.b.c:0",
			valid:    true,
		},
		{
			endpoint: "a.b.c:9999",
			valid:    true,
		},
		{
			endpoint: "a.b.c:59999",
			valid:    true,
		},
		{
			endpoint: "a.b.c:64999",
			valid:    true,
		},
		{
			endpoint: "a.b.c:65499",
			valid:    true,
		},
		{
			endpoint: "a.b.c:65529",
			valid:    true,
		},
		{
			endpoint: "a.b.c:65535",
			valid:    true,
		},
	}

	for _, tc := range testcases {
		result := validateEndpoint(tc.endpoint)
		if tc.valid {
			assert.True(t, result)
		} else {
			assert.False(t, result)
		}
	}
}

func TestCheckMySQLConnection(t *testing.T) {
	testcases := []struct {
		connection *dataflow.Connection
		err        bool
	}{
		{
			connection: nil,
			err:        true,
		},
		{
			connection: &dataflow.Connection{},
			err:        true,
		},
		{
			connection: &dataflow.Connection{
				Endpoint: "a.b.c:30000",
			},
			err: true,
		},
		{
			connection: &dataflow.Connection{
				Endpoint: "a.b.c:30000",
				Username: "u",
			},
			err: true,
		},
		{
			connection: &dataflow.Connection{
				Endpoint: "a.b.c:30000",
				Username: "u",
				Password: "p",
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkMySQLConnection(tc.connection)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckStartPosition(t *testing.T) {
	testcases := []struct {
		startPosition *dataflow.StartPosition
		err           bool
	}{
		{
			startPosition: nil,
			err:           true,
		},
		{
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode(10),
			},
			err: true,
		},
		{
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_NOW,
			},
			err: false,
		},
		{
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
			},
			err: true,
		},
		{
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  100,
			},
			err: false,
		},
		{
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
			},
			err: true,
		},
		{
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.DateTime),
			},
			err: true,
		},
		{
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC3339),
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkStartPosition(tc.startPosition)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckTableFilter(t *testing.T) {
	rand.Seed(time.Now().UnixNano())
	testcases := []struct {
		filters []string
		err     bool
	}{
		{
			filters: nil,
			err:     false,
		},
		{
			filters: []string{},
			err:     false,
		},
		{
			filters: genStrList(maxFilterRuleNum+1, 1),
			err:     true,
		},
		{
			filters: genStrList(maxFilterRuleNum, maxTotalTableFilterRuleLength/maxFilterRuleNum+1),
			err:     true,
		},
		{
			filters: []string{"<abc>"},
			err:     true,
		},
		{
			filters: []string{"a.b"},
			err:     false,
		},
	}

	for _, tc := range testcases {
		err := checkTableFilter(tc.filters)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckEventFilter(t *testing.T) {
	testcases := []struct {
		eventRules []*dataflow.EventFilterRule
		err        bool
	}{
		{
			eventRules: nil,
			err:        false,
		},
		{
			eventRules: []*dataflow.EventFilterRule{},
			err:        false,
		},
		{
			eventRules: genEmptyEventFilters(maxEventFilterRuleNum + 1),
			err:        true,
		},
		{
			eventRules: []*dataflow.EventFilterRule{
				{
					Matcher: genStrList(maxFilterRuleLength+1, 1),
				},
			},
			err: true,
		},
		{
			eventRules: genEmptyEventFilters(1),
			err:        false,
		},
	}

	filters := []string{"*.*"}
	for _, tc := range testcases {
		err := checkEventFilter(tc.eventRules, filters)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_GetDownstreamParamsFromCreateReq(t *testing.T) {
	reqs := []*dataflow.ChangefeedCreateReq{
		{
			KafkaInfo: nil,
		},
		{
			KafkaInfo: &dataflow.KafkaInfo{},
		},
		{
			KafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "a",
				},
			},
		},
	}
	adaptor := newKafkaBizChangefeedDownstreamAdaptor(nil)
	ctx := context.Background()
	for _, req := range reqs {
		kafkaInfo, err := adaptor.ParseCreateParams(ctx, req, nil)
		require.NoError(t, err)
		assert.Equal(t, req.KafkaInfo, kafkaInfo)
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_GetDownstreamParamsFromEditReq(t *testing.T) {
	reqs := []*dataflow.ChangefeedEditReq{
		{
			KafkaInfo: nil,
		},
		{
			KafkaInfo: &dataflow.KafkaInfo{},
		},
		{
			KafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "a",
				},
			},
		},
	}
	adaptor := newKafkaBizChangefeedDownstreamAdaptor(nil)
	for _, req := range reqs {
		kafkaInfo := adaptor.ParseEditParams(req)
		assert.Equal(t, req.KafkaInfo, kafkaInfo)
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_ValidateCreateDownstreamParams(t *testing.T) {
	testcases := []struct {
		name        string
		kafkaParams interface{}
		cluster     *utils.Cluster
		err         bool
		errKeyword  string
	}{
		{
			name:        "kafka params nil",
			kafkaParams: nil,
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "kafka_connection_info",
		},
		{
			name:        "kafka params type error",
			kafkaParams: &dataflow.MySQLInfo{},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "invalid kafka connection info",
		},
		{
			name:        "kafka params empty",
			kafkaParams: &dataflow.KafkaInfo{},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "kafka_connection_info",
		},
		{
			name: "invalid broker",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion: dataflow.KafkaVersion(5),
				},
				Authentication:       &dataflow.Authentication{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
				StartPosition:        &dataflow.StartPosition{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "check kafka broker",
		},
		{
			name: "invalid authentication",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType(5),
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
				StartPosition:        &dataflow.StartPosition{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "check kafka authentication",
		},
		{
			name: "invalid topic partition config",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType: dataflow.DispatchType(5),
				},
				DataFormat:    &dataflow.DataFormat{},
				TableConfig:   &dataflow.TableConfig{},
				StartPosition: &dataflow.StartPosition{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "check kafka topic partition config",
		},
		{
			name: "invalid data format",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol(5),
				},
				TableConfig:   &dataflow.TableConfig{},
				StartPosition: &dataflow.StartPosition{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "check kafka data format",
		},
		{
			name: "invalid table config",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
				StartPosition: &dataflow.StartPosition{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "check table config",
		},
		{
			name: "invalid start position",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_TSO,
					Tso:  0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        true,
			errKeyword: "check start position",
		},
		{
			name: "valid kafka params",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
					Tso:  0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v8.1.0"}},
			}),
			err:        false,
			errKeyword: "",
		},
	}

	adaptor := newKafkaBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateCreateParams(tc.kafkaParams, tc.cluster)
			if tc.err {
				assert.Error(t, err)
				assert.Contains(t, err.Error(), tc.errKeyword)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_ValidateEditDownstreamParams(t *testing.T) {
	testcases := []struct {
		name           string
		kafkaParams    interface{}
		clusterVersion string
		err            bool
		errKeyword     string
	}{
		{
			name:           "kafka params nil",
			kafkaParams:    nil,
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "invalid kafka connection info",
		},
		{
			name:           "kafka params type error",
			kafkaParams:    &dataflow.MySQLInfo{},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "invalid kafka connection info",
		},
		{
			name:           "kafka params empty",
			kafkaParams:    &dataflow.KafkaInfo{},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "kafka_info",
		},
		{
			name: "invalid broker",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion: dataflow.KafkaVersion(5),
				},
				Authentication:       &dataflow.Authentication{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
			},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "check kafka broker",
		},
		{
			name: "invalid authentication",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType(5),
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
			},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "check kafka authentication",
		},
		{
			name: "invalid topic partition config",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType: dataflow.DispatchType(5),
				},
				DataFormat:  &dataflow.DataFormat{},
				TableConfig: &dataflow.TableConfig{},
			},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "check kafka topic partition config",
		},
		{
			name: "invalid data format",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol(5),
				},
				TableConfig: &dataflow.TableConfig{},
			},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "check kafka data format",
		},
		{
			name: "invalid table config",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
			},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "check table config",
		},
		{
			name: "empty start position",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
			},
			clusterVersion: "v8.1.0",
			err:            false,
			errKeyword:     "check start position",
		},
		{
			name: "no username kafka params",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
					Tso:  0,
				},
			},
			clusterVersion: "v8.1.0",
			err:            true,
			errKeyword:     "empty kafka auth username",
		},
		{
			name: "valid kafka params",
			kafkaParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					BrokerEndpoints: "abc:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "u",
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
					DefaultTopic:        "test",
					PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
					PartitionNum:        1,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
					Tso:  0,
				},
			},
			clusterVersion: "v8.1.0",
			err:            false,
			errKeyword:     "",
		},
	}

	adaptor := newKafkaBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateEditParams(tc.kafkaParams, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
				assert.Contains(t, err.Error(), tc.errKeyword)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_CheckClusterVersion(t *testing.T) {
	testcases := []struct {
		name           string
		kafkaInfo      *dataflow.KafkaInfo
		clusterVersion string
		err            bool
		errKeyword     string
	}{
		{
			name: "kafka auth not support",
			kafkaInfo: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				},
			},
			clusterVersion: "v6.0.0",
			err:            true,
			errKeyword:     string(kafkaAuthTypeSinkConstraintType),
		},
		{
			name: "event filter not support",
			kafkaInfo: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				},
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							Matcher: []string{"test.*"},
						},
					},
				},
			},
			clusterVersion: "v6.5.2",
			err:            true,
			errKeyword:     string(eventFilterConstraintType),
		},
		{
			name: "column partition dispatcher not support",
			kafkaInfo: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				},
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							Matcher: []string{"test.*"},
						},
					},
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
						{
							PartitionType: dataflow.PartitionDispatcher_COLUMNS,
						},
					},
				},
			},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeyword:     string(columnsDispatcherConstraintType),
		},
		{
			name: "debezium not support",
			kafkaInfo: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				},
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							Matcher: []string{"test.*"},
						},
					},
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
						{
							PartitionType: dataflow.PartitionDispatcher_COLUMNS,
						},
					},
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_DEBEZIUM,
				},
			},
			clusterVersion: "v7.5.0",
			err:            true,
			errKeyword:     string(debeziumFormatConstraintType),
		},
		{
			name: "column selectors not support",
			kafkaInfo: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				},
				ColumnSelectors: []*dataflow.ColumnSelector{
					{
						Matcher: []string{"a"},
						Columns: []string{"b"},
					},
				},
			},
			clusterVersion: "v6.5.2",
			err:            true,
			errKeyword:     string(columnSelectorsConstraintType),
		},
		{
			name: "all support",
			kafkaInfo: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				},
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							Matcher: []string{"test.*"},
						},
					},
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
						{
							PartitionType: dataflow.PartitionDispatcher_COLUMNS,
						},
					},
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_DEBEZIUM,
				},
			},
			clusterVersion: "v8.1.0",
			err:            false,
			errKeyword:     "",
		},
	}

	adaptor := &kafkaBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.checkClusterVersion(tc.kafkaInfo, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
				assert.Contains(t, err.Error(), tc.errKeyword)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_CheckPartitionDispatcherVersion(t *testing.T) {
	testcase := []struct {
		dispatchers    []*dataflow.PartitionDispatcherConfig
		clusterVersion string
		err            bool
	}{
		{
			dispatchers:    nil,
			clusterVersion: "v6.5.0",
			err:            false,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
			},
			clusterVersion: "v6.5.0",
			err:            false,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
				{
					PartitionType: dataflow.PartitionDispatcher_INDEX_VALUE,
				},
			},
			clusterVersion: "v6.5.0",
			err:            false,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
				{
					PartitionType: dataflow.PartitionDispatcher_INDEX_VALUE,
					IndexName:     "abc",
				},
			},
			clusterVersion: "v6.5.0",
			err:            true,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
				{
					PartitionType: dataflow.PartitionDispatcher_INDEX_VALUE,
					IndexName:     "abc",
				},
			},
			clusterVersion: "v7.5.0",
			err:            false,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
				{
					PartitionType: dataflow.PartitionDispatcher_COLUMNS,
				},
			},
			clusterVersion: "v6.5.0",
			err:            true,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
				{
					PartitionType: dataflow.PartitionDispatcher_COLUMNS,
				},
			},
			clusterVersion: "v7.5.0",
			err:            false,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_TABLE,
				},
				{
					PartitionType: dataflow.PartitionDispatcher_INDEX_VALUE,
					IndexName:     "abc",
				},
				{
					PartitionType: dataflow.PartitionDispatcher_COLUMNS,
				},
			},
			clusterVersion: "v7.5.0",
			err:            false,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_COLUMNS,
					Columns:       []string{"__$$abcd12345", "abcd", "update", "_"},
				},
			},
			clusterVersion: "v8.5.0",
			err:            false,
		},
		{
			dispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					PartitionType: dataflow.PartitionDispatcher_COLUMNS,
					Columns:       []string{"abc.efg"},
				},
			},
			clusterVersion: "v8.5.0",
			err:            true,
		},
	}

	adaptor := &kafkaBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcase {
		err := adaptor.checkPartitionDispatcherVersion(tc.dispatchers, tc.clusterVersion)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_CheckDataFormatVersion(t *testing.T) {
	testcase := []struct {
		dataFormat     *dataflow.DataFormat
		clusterVersion string
		err            bool
	}{
		{
			dataFormat:     nil,
			clusterVersion: "v6.0.0",
			err:            false,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_AVRO,
			},
			clusterVersion: "v6.0.0",
			err:            true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_AVRO,
			},
			clusterVersion: "v6.1.0",
			err:            false,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_DEBEZIUM_AVRO,
			},
			clusterVersion: "v6.0.0",
			err:            true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_DEBEZIUM_AVRO,
			},
			clusterVersion: "v6.1.0",
			err:            false,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_DEBEZIUM,
			},
			clusterVersion: "v6.5.0",
			err:            true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_DEBEZIUM,
			},
			clusterVersion: "v8.1.0",
			err:            false,
		},
	}

	adaptor := &kafkaBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcase {
		err := adaptor.checkDataFormatVersion(tc.dataFormat, tc.clusterVersion)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestKafkaBizChangefeedDownstreamAdaptor_AssembleCreateConnectorReq(t *testing.T) {
	testcases := []struct {
		downstreamParams interface{}
		assembledParams  *dataflow.KafkaInfo
		err              bool
	}{
		{
			downstreamParams: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "a.b.c:4000",
				},
			},
			assembledParams: nil,
			err:             true,
		},
		{
			downstreamParams: &dataflow.KafkaInfo{},
			assembledParams:  &dataflow.KafkaInfo{},
			err:              false,
		},
		{
			downstreamParams: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "abc",
				},
			},
			assembledParams: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "",
				},
			},
			err: false,
		},
		{
			downstreamParams: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "abc",
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_AVRO,
					SchemaRegistry: &dataflow.SchemaRegistry{
						EnableHttpAuth:          true,
						SchemaRegistryEndpoints: "a.b.c:4000",
						UserName:                "abc",
						Password:                "abc",
					},
				},
			},
			assembledParams: &dataflow.KafkaInfo{
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "",
				},
				DataFormat: &dataflow.DataFormat{
					Protocol: dataflow.Protocol_AVRO,
					SchemaRegistry: &dataflow.SchemaRegistry{
						EnableHttpAuth:          true,
						SchemaRegistryEndpoints: "a.b.c:4000",
						UserName:                "abc",
						Password:                "",
					},
				},
			},
			err: false,
		},
	}

	adaptor := newKafkaBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		paramsBytes, err := json.Marshal(tc.downstreamParams)
		require.NoError(t, err)
		bizChangefeed := &domain.BizChangefeed{
			DownstreamType:      entity.TiCDCDownStreamTypeKafka,
			DownstreamParameter: string(paramsBytes),
		}
		req, err := adaptor.AssembleGetResponse(bizChangefeed)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
			assert.Equal(t, dataflow.ChangefeedType_KAFKA, req.ChangefeedType)
			assert.Equal(t, tc.assembledParams, req.KafkaInfo)
		}
	}
}

func TestIsKafkaInfoEmpty(t *testing.T) {
	testcases := []struct {
		kafkaInfo *dataflow.KafkaInfo
		empty     bool
	}{
		{
			kafkaInfo: nil,
			empty:     true,
		},
		{
			kafkaInfo: &dataflow.KafkaInfo{
				Authentication:       &dataflow.Authentication{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
				StartPosition:        &dataflow.StartPosition{},
			},
			empty: true,
		},
		{
			kafkaInfo: &dataflow.KafkaInfo{
				Broker:               &dataflow.Broker{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
				StartPosition:        &dataflow.StartPosition{},
			},
			empty: true,
		},
		{
			kafkaInfo: &dataflow.KafkaInfo{
				Broker:         &dataflow.Broker{},
				Authentication: &dataflow.Authentication{},
				DataFormat:     &dataflow.DataFormat{},
				TableConfig:    &dataflow.TableConfig{},
				StartPosition:  &dataflow.StartPosition{},
			},
			empty: true,
		},
		{
			kafkaInfo: &dataflow.KafkaInfo{
				Broker:               &dataflow.Broker{},
				Authentication:       &dataflow.Authentication{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				TableConfig:          &dataflow.TableConfig{},
				StartPosition:        &dataflow.StartPosition{},
			},
			empty: true,
		},
		{
			kafkaInfo: &dataflow.KafkaInfo{
				Broker:               &dataflow.Broker{},
				Authentication:       &dataflow.Authentication{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				StartPosition:        &dataflow.StartPosition{},
			},
			empty: true,
		},
		{
			kafkaInfo: &dataflow.KafkaInfo{
				Broker:               &dataflow.Broker{},
				Authentication:       &dataflow.Authentication{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
			},
			empty: true,
		},
		{
			kafkaInfo: &dataflow.KafkaInfo{
				Broker:               &dataflow.Broker{},
				Authentication:       &dataflow.Authentication{},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{},
				DataFormat:           &dataflow.DataFormat{},
				TableConfig:          &dataflow.TableConfig{},
				StartPosition:        &dataflow.StartPosition{},
			},
			empty: false,
		},
	}

	for _, tc := range testcases {
		isEmpty := isKafkaInfoEmpty(tc.kafkaInfo)
		assert.Equal(t, tc.empty, isEmpty, "kafka info", tc.kafkaInfo)
	}
}

func TestCheckKafkaBroker(t *testing.T) {
	testcases := []struct {
		broker *dataflow.Broker
		isEdit bool
		err    bool
	}{
		{
			broker: nil,
			err:    true,
		},
		{
			broker: &dataflow.Broker{
				KafkaVersion: dataflow.KafkaVersion(5),
			},
			err: true,
		},
		{
			broker: &dataflow.Broker{
				KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_1XX,
				BrokerEndpoints: "",
			},
			err: true,
		},
		{
			broker: &dataflow.Broker{
				KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_1XX,
				BrokerEndpoints: "",
			},
			err: true,
		},
		{
			broker: &dataflow.Broker{
				KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_1XX,
				BrokerEndpoints: "abc",
			},
			err: true,
		},
		{
			broker: &dataflow.Broker{
				KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_1XX,
				BrokerEndpoints: "abc:9092,abc",
			},
			err: true,
		},
		{
			broker: &dataflow.Broker{
				KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_1XX,
				BrokerEndpoints: "abc:9092",
			},
			err: false,
		},
		{
			broker: &dataflow.Broker{
				KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_1XX,
				BrokerEndpoints: "abc:9092,abc:9093",
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkKafkaBroker(tc.broker, tc.isEdit)
		if tc.err {
			assert.Error(t, err, "broker", tc.broker)
		} else {
			assert.NoError(t, err, "broker", tc.broker)
		}
	}
}

func TestCheckKafkaAuthentication(t *testing.T) {
	testcases := []struct {
		authentication *dataflow.Authentication
		isEdit         bool
		err            bool
	}{
		{
			authentication: nil,
			err:            false,
		},
		{
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType(5),
			},
			err: true,
		},
		{
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_DISABLE,
			},
			err: false,
		},
		{
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				UserName: "abc",
				Password: "",
			},
			err: true,
		},
		{
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				UserName: "",
				Password: "abc",
			},
			err: true,
		},
		{
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				UserName: "abc",
				Password: "abc",
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkKafkaAuthentication(tc.authentication, tc.isEdit)
		if tc.err {
			assert.Error(t, err, "authentication", tc.authentication)
		} else {
			assert.NoError(t, err, "authentication", tc.authentication)
		}
	}
}

func TestCheckKafkaTopicPartitionConfig(t *testing.T) {
	testcases := []struct {
		tpCfg *dataflow.TopicPartitionConfig
		err   bool
	}{
		{
			tpCfg: nil,
			err:   true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType(1000),
			},
			err: true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				DefaultTopic: "",
			},
			err: true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				DefaultTopic:        "abc",
				PartitionDispatcher: dataflow.PartitionDispatcher(1000),
			},
			err: true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				DefaultTopic:        "abc",
				PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
				PartitionNum:        0,
			},
			err: true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType:         dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				DefaultTopic:         "abc",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{},
			},
			err: true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				DefaultTopic: "abc",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher(1000),
					},
				},
			},
			err: true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				DefaultTopic: "abc",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						Matcher:       []string{"*.*"},
						PartitionType: dataflow.PartitionDispatcher_TABLE,
					},
				},
				PartitionNum: 0,
			},
			err: true,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				DefaultTopic: "abc",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						Matcher:       []string{"*.*"},
						PartitionType: dataflow.PartitionDispatcher_TABLE,
					},
				},
				PartitionNum: 1,
			},
			err: false,
		},
		{
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType:        dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				DefaultTopic:        "",
				PartitionDispatcher: dataflow.PartitionDispatcher_TABLE,
				PartitionNum:        1,
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkKafkaTopicPartitionConfig(tc.tpCfg)
		if tc.err {
			assert.Error(t, err, "topic partition config", tc.tpCfg)
		} else {
			assert.NoError(t, err, "topic partition config", tc.tpCfg)
		}
	}
}

func TestCheckKafkaDataFormat(t *testing.T) {
	testcases := []struct {
		dataFormat *dataflow.DataFormat
		tpCfg      *dataflow.TopicPartitionConfig
		err        bool
	}{
		{
			dataFormat: nil,
			tpCfg:      nil,
			err:        false,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol(5),
			},
			tpCfg: nil,
			err:   true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_CANAL_JSON,
			},
			tpCfg: nil,
			err:   false,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol: dataflow.Protocol_AVRO,
			},
			tpCfg: nil,
			err:   true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol:                   dataflow.Protocol_AVRO,
				DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
				BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING,
				SchemaRegistry: &dataflow.SchemaRegistry{
					SchemaRegistryEndpoints: "http://test.com",
					EnableHttpAuth:          true,
					UserName:                "abc",
					Password:                "abc",
				},
			},
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: false,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol:                   dataflow.Protocol_DEBEZIUM_AVRO,
				DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
				BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING,
				SchemaRegistry: &dataflow.SchemaRegistry{
					SchemaRegistryEndpoints: "http://test.com",
				},
			},
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkKafkaDataFormat(tc.dataFormat, tc.tpCfg, false)
		if tc.err {
			assert.Error(t, err, "data format", tc.dataFormat, "topic partition config", tc.tpCfg)
		} else {
			assert.NoError(t, err, "data format", tc.dataFormat, "topic partition config", tc.tpCfg)
		}
	}
}

func TestCheckKafkaAvroDataFormat(t *testing.T) {
	testcases := []struct {
		dataFormat  *dataflow.DataFormat
		tpCfg       *dataflow.TopicPartitionConfig
		err         bool
		errContains string
	}{
		{
			dataFormat: nil,
			tpCfg:      nil,
			err:        true,
		},
		{
			dataFormat: nil,
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				DecimalHandlingMode: dataflow.DecimalHandlingMode(3),
			},
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
				BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode(3),
			},
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
				BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING,
			},
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: true,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol:                   dataflow.Protocol_DEBEZIUM_AVRO,
				DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
				BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING,
				SchemaRegistry: &dataflow.SchemaRegistry{
					SchemaRegistryEndpoints: "http://test.com",
					EnableHttpAuth:          true,
					UserName:                "abc",
					Password:                "abc",
				},
			},
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: false,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol:                   dataflow.Protocol_DEBEZIUM_AVRO,
				DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
				BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING,
				SchemaRegistry: &dataflow.SchemaRegistry{
					SchemaRegistryEndpoints: "http://test.com",
					EnableHttpAuth:          true,
					UserName:                "abc",
					Password:                "abc",
				},
			},
			tpCfg: &dataflow.TopicPartitionConfig{
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
			},
			err:         true,
			errContains: "debezium-avro data format",
		},
	}

	for _, tc := range testcases {
		err := checkKafkaAvroDataFormat(tc.dataFormat, tc.tpCfg, false)
		if tc.err {
			assert.Error(t, err, "data format", tc.dataFormat, "topic partition config", tc.tpCfg)
			if tc.errContains != "" {
				assert.Contains(t, err.Error(), tc.errContains)
			}
		} else {
			assert.NoError(t, err, "data format", tc.dataFormat, "topic partition config", tc.tpCfg)
		}
	}
}

func TestCheckKafkaSchemaRegistry(t *testing.T) {
	testcases := []struct {
		schemaRegistry *dataflow.SchemaRegistry
		isEdit         bool
		err            bool
	}{
		{
			schemaRegistry: nil,
			err:            true,
		},
		{
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "",
			},
			err: true,
		},
		{
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://test.com",
				EnableHttpAuth:          false,
			},
			err: false,
		},
		{
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://test.com",
				EnableHttpAuth:          true,
				UserName:                "abc",
				Password:                "",
			},
			err: true,
		},
		{
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://test.com",
				EnableHttpAuth:          true,
				UserName:                "abc",
				Password:                "",
			},
			isEdit: true,
			err:    false,
		},
		{
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://test.com",
				EnableHttpAuth:          true,
				UserName:                "",
				Password:                "abc",
			},
			err: true,
		},
		{
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://test.com",
				EnableHttpAuth:          true,
				UserName:                "abc",
				Password:                "abc",
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkKafkaSchemaRegistry(tc.schemaRegistry, tc.isEdit)
		if tc.err {
			assert.Error(t, err, "schema registry", tc.schemaRegistry)
		} else {
			assert.NoError(t, err, "schema registry", tc.schemaRegistry)
		}
	}
}

func TestCheckColumnSelectors(t *testing.T) {
	testcases := []struct {
		columnSelectors []*dataflow.ColumnSelector
		err             bool
	}{
		{
			columnSelectors: nil,
			err:             false,
		},
		{
			columnSelectors: []*dataflow.ColumnSelector{},
			err:             false,
		},
		{
			columnSelectors: []*dataflow.ColumnSelector{
				{
					Matcher: []string{"a+?>"},
					Columns: []string{"b"},
				},
			},
			err: true,
		},
		{
			columnSelectors: []*dataflow.ColumnSelector{
				{
					Matcher: []string{"a.*"},
					Columns: []string{"b+?>"},
				},
			},
			err: true,
		},
		{
			columnSelectors: []*dataflow.ColumnSelector{
				{
					Matcher: []string{"a.*"},
					Columns: []string{"b"},
				},
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkColumnSelectors(tc.columnSelectors)
		if tc.err {
			assert.Error(t, err, tc)
		} else {
			assert.NoError(t, err, tc)
		}
	}
}

func TestValidateKafkaTopic(t *testing.T) {
	testcases := []struct {
		topic string
		valid bool
	}{
		{
			topic: "123",
			valid: true,
		},
		{
			topic: "abc",
			valid: true,
		},
		{
			topic: "ABC",
			valid: true,
		}, {
			topic: "ABcd123",
			valid: true,
		},
		{
			topic: "abc-123",
			valid: true,
		},
		{
			topic: "abc.def",
			valid: true,
		},
		{
			topic: "abc_123",
			valid: true,
		},
		{
			topic: "AAAa_12a-ccc.ddd",
			valid: true,
		},
		{
			topic: "",
			valid: false,
		},
		{
			topic: "+abc",
			valid: false,
		},
		{
			topic: "abc*",
			valid: false,
		},
		{
			topic: "=*",
			valid: false,
		},
	}

	for _, tc := range testcases {
		isValid := validateKafkaTopic(tc.topic)
		assert.Equal(t, tc.valid, isValid, "topic", tc.topic)
	}
}

func TestCheckClusterVersion(t *testing.T) {
	version := "v6.1.0"
	gtVersion := "v7.1.0"
	ltVersion := "v6.0.1"

	constraintType := versionConstraintType("test")

	t.Run("check invalid version", func(t *testing.T) {
		constraint, err := semver.NewConstraint("> " + version)
		require.NoError(t, err)

		bizErr := checkClusterVersion(constraint, constraintType, "v1.2.3.4")
		require.Error(t, bizErr)
		assert.Equal(t, bizerr.ErrCodeUnknown, bizErr.RawErrCode())
	})

	t.Run("check eq", func(t *testing.T) {
		constraint, err := semver.NewConstraint("= " + version)
		require.NoError(t, err)

		bizErr := checkClusterVersion(constraint, constraintType, gtVersion)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())

		bizErr = checkClusterVersion(constraint, constraintType, ltVersion)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())

		bizErr = checkClusterVersion(constraint, constraintType, version)
		assert.NoError(t, bizErr)
	})

	t.Run("check gt", func(t *testing.T) {
		constraint, err := semver.NewConstraint("> " + version)
		require.NoError(t, err)

		bizErr := checkClusterVersion(constraint, constraintType, gtVersion)
		assert.NoError(t, bizErr)

		bizErr = checkClusterVersion(constraint, constraintType, ltVersion)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())

		bizErr = checkClusterVersion(constraint, constraintType, version)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())
	})

	t.Run("check gte", func(t *testing.T) {
		constraint, err := semver.NewConstraint(">= " + version)
		require.NoError(t, err)

		bizErr := checkClusterVersion(constraint, constraintType, gtVersion)
		assert.NoError(t, bizErr)

		bizErr = checkClusterVersion(constraint, constraintType, ltVersion)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())

		bizErr = checkClusterVersion(constraint, constraintType, version)
		assert.NoError(t, bizErr)
	})

	t.Run("check lt", func(t *testing.T) {
		constraint, err := semver.NewConstraint("< " + version)
		require.NoError(t, err)

		bizErr := checkClusterVersion(constraint, constraintType, gtVersion)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())

		bizErr = checkClusterVersion(constraint, constraintType, ltVersion)
		assert.NoError(t, bizErr)

		bizErr = checkClusterVersion(constraint, constraintType, version)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())
	})

	t.Run("check lte", func(t *testing.T) {
		constraint, err := semver.NewConstraint("<= " + version)
		require.NoError(t, err)

		bizErr := checkClusterVersion(constraint, constraintType, gtVersion)
		require.Error(t, bizErr)
		assert.Equal(t, config.ErrCodeClusterVersionNotSupport, bizErr.RawErrCode())

		bizErr = checkClusterVersion(constraint, constraintType, ltVersion)
		assert.NoError(t, bizErr)

		bizErr = checkClusterVersion(constraint, constraintType, version)
		assert.NoError(t, bizErr)
	})
}

func TestPulsarBizChangefeedDownstreamAdaptor_ValidateCreateParams(t *testing.T) {
	testcases := []struct {
		params  interface{}
		cluster *utils.Cluster
		err     bool
	}{
		{
			params: nil,
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.5.1"}},
			}),
			err: true,
		},
		{
			params: &dataflow.PulsarInfo{
				Endpoint: "abc",
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.5.1"}},
			}),
			err: true,
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol(1000),
				Endpoint:            "abc:9093",
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.5.1"}},
			}),
			err: true,
		},
		{
			params: &dataflow.PulsarInfo{
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
				Endpoint:            "abc:9093",
				CompressionType:     dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				TableConfig: &dataflow.TableConfig{
					Mode:       dataflow.TableMode_FORCE_SYNC,
					FilterRule: []string{"*.*"},
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: true,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
						Matcher:       []string{"*.*"},
					},
				},
				AdvancedSetting: &dataflow.PulsarAdvancedSetting{
					ConnectionTimeout:       5,
					OperationTimeout:        4,
					SendTimeout:             3,
					BatchingMaxMessages:     100,
					BatchingMaxPublishDelay: 1000,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.5.0"}},
			}),
			err: true,
		},
		{
			params: &dataflow.PulsarInfo{
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
				Endpoint:            "abc:9093",
				CompressionType:     dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				TableConfig: &dataflow.TableConfig{
					Mode:       dataflow.TableMode_FORCE_SYNC,
					FilterRule: []string{"*.*"},
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: true,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
						Matcher:       []string{"*.*"},
					},
				},
				AdvancedSetting: &dataflow.PulsarAdvancedSetting{
					ConnectionTimeout:       5,
					OperationTimeout:        4,
					SendTimeout:             3,
					BatchingMaxMessages:     100,
					BatchingMaxPublishDelay: 1000,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.5.1"}},
			}),
			err: false,
		},
	}
	adaptor := &pulsarBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		err := adaptor.ValidateCreateParams(tc.params, tc.cluster)
		if tc.err {
			assert.NotNil(t, err)
		} else {
			assert.Nil(t, err)
		}
	}
}

func TestPulsarBizChangefeedDownstreamAdaptor_validatePulsarSubParams(t *testing.T) {
	testcases := []struct {
		params      *dataflow.PulsarInfo
		errKeywords string
	}{
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol(1000),
			},
			errKeywords: "unsupported protocol",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication:      nil,
			},
			errKeywords: "check authentication",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				DataFormat: nil,
			},
			errKeywords: "check data format",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicDistribution: nil,
			},
			errKeywords: "check topic distribution",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{},
			},
			errKeywords: "check partition dispatchers",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						Matcher:       []string{"*.*"},
						PartitionType: dataflow.PartitionDispatcher_TABLE,
					},
				},
				TableConfig: nil,
			},
			errKeywords: "check table config",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						Matcher:       []string{"*.*"},
						PartitionType: dataflow.PartitionDispatcher_TABLE,
					},
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
				StartPosition: nil,
			},
			errKeywords: "check start position",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						Matcher:       []string{"*.*"},
						PartitionType: dataflow.PartitionDispatcher_TABLE,
					},
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
				StartPosition:   &dataflow.StartPosition{Mode: dataflow.StartMode_FROM_NOW},
				AdvancedSetting: nil,
			},
			errKeywords: "check advanced setting",
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
					Password: "password",
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						Matcher:       []string{"*.*"},
						PartitionType: dataflow.PartitionDispatcher_TABLE,
					},
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: []string{"*.*"},
				},
				StartPosition: &dataflow.StartPosition{Mode: dataflow.StartMode_FROM_NOW},
				AdvancedSetting: &dataflow.PulsarAdvancedSetting{
					ConnectionTimeout:       5,
					OperationTimeout:        4,
					SendTimeout:             3,
					BatchingMaxMessages:     1000,
					BatchingMaxPublishDelay: 100,
				},
			},
			errKeywords: "",
		},
	}
	adaptor := &pulsarBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		err := adaptor.validatePulsarSubParams(tc.params, false)
		if tc.errKeywords == "" {
			assert.NoError(t, err)
		} else {
			assert.Error(t, err)
			assert.Contains(t, err.Error(), tc.errKeywords)
		}
	}
}

func TestPulsarBizChangefeedDownstreamAdaptor_checkClusterVersion(t *testing.T) {
	testcases := []struct {
		version string
		err     bool
	}{
		{
			version: "v7.5.0",
			err:     true,
		},
		{
			version: "v7.5.1",
			err:     false,
		},
	}
	adaptor := &pulsarBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		err := adaptor.checkClusterVersion(tc.version)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestPulsarBizChangefeedDownstreamAdaptor_ValidateEditParams(t *testing.T) {
	testcases := []struct {
		params  interface{}
		version string
		err     bool
	}{
		{
			params:  nil,
			version: "v7.5.1",
			err:     true,
		},
		{
			params: &dataflow.PulsarInfo{
				DestinationProtocol: dataflow.PulsarDestinationProtocol(1000),
				Endpoint:            "abc:9093",
			},
			version: "v7.5.1",
			err:     true,
		},
		{
			params: &dataflow.PulsarInfo{
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
				Endpoint:            "abc:9093",
				CompressionType:     dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
				},
				TableConfig: &dataflow.TableConfig{
					Mode:       dataflow.TableMode_FORCE_SYNC,
					FilterRule: []string{"*.*"},
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: true,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
						Matcher:       []string{"*.*"},
					},
				},
				AdvancedSetting: &dataflow.PulsarAdvancedSetting{
					ConnectionTimeout:       5,
					OperationTimeout:        4,
					SendTimeout:             3,
					BatchingMaxMessages:     100,
					BatchingMaxPublishDelay: 1000,
				},
			},
			version: "v7.5.0",
			err:     true,
		},
		{
			params: &dataflow.PulsarInfo{
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				DestinationProtocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
				Endpoint:            "abc:9093",
				CompressionType:     dataflow.PulsarCompressionType_PULSAR_COMPRESSION_LZ4,
				Authentication: &dataflow.PulsarAuthentication{
					AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
					UserName: "user",
				},
				TableConfig: &dataflow.TableConfig{
					Mode:       dataflow.TableMode_FORCE_SYNC,
					FilterRule: []string{"*.*"},
				},
				DataFormat: &dataflow.PulsarDataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: true,
				},
				TopicDistribution: &dataflow.PulsarTopicDistribution{
					PulsarTenant:    "tenant",
					PulsarNamespace: "namespace",
					DefaultTopic:    "topic",
					DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				},
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
						Matcher:       []string{"*.*"},
					},
				},
				AdvancedSetting: &dataflow.PulsarAdvancedSetting{
					ConnectionTimeout:       5,
					OperationTimeout:        4,
					SendTimeout:             3,
					BatchingMaxMessages:     100,
					BatchingMaxPublishDelay: 1000,
				},
			},
			version: "v7.5.1",
			err:     false,
		},
	}
	adaptor := &pulsarBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		err := adaptor.ValidateEditParams(tc.params, tc.version)
		if tc.err {
			assert.NotNil(t, err)
		} else {
			assert.Nil(t, err)
		}
	}
}

func TestPulsarBizChangefeedDownstreamAdaptor_AssembleGetResponse(t *testing.T) {
	bizChangefeed := &domain.BizChangefeed{
		DownstreamType: entity.TiCDCDownStreamTypePulsar,
	}
	adaptor := &pulsarBizChangefeedDownstreamAdaptor{}
	t.Run("decode downstream params error", func(t *testing.T) {
		params := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "abc:3306",
			},
		}
		paramsBytes, err := json.Marshal(params)
		assert.NoError(t, err)
		bizChangefeed.DownstreamParameter = string(paramsBytes)
		_, err = adaptor.AssembleGetResponse(bizChangefeed)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "parse pulsar downstream parameter")
	})

	t.Run("assemble successfully", func(t *testing.T) {
		params := &dataflow.PulsarInfo{
			Authentication: &dataflow.PulsarAuthentication{
				UserName: "user",
				Password: "password",
				Token:    "token",
			},
		}
		paramsBytes, err := json.Marshal(params)
		assert.NoError(t, err)
		bizChangefeed.DownstreamParameter = string(paramsBytes)
		createReq, err := adaptor.AssembleGetResponse(bizChangefeed)
		assert.NoError(t, err)
		assert.Equal(t, dataflow.ChangefeedType_PULSAR, createReq.ChangefeedType)
		assert.Equal(t, "", createReq.PulsarInfo.Authentication.Password)
		assert.Equal(t, "", createReq.PulsarInfo.Authentication.Token)
	})
}

func TestCheckPulsarEndpoint(t *testing.T) {
	testcases := []struct {
		endpoint string
		valid    bool
	}{
		{
			endpoint: "",
			valid:    false,
		},
		{
			endpoint: "a.b.c",
			valid:    false,
		},
		{
			endpoint: "a.b.c:6650,a.b.c",
			valid:    false,
		},
		{
			endpoint: "a.b.c:6650",
			valid:    true,
		},
		{
			endpoint: "a.b.c:6650,a.b.c:6651",
			valid:    true,
		},
	}
	for _, tc := range testcases {
		result := checkPulsarEndpoint(tc.endpoint)
		assert.Equal(t, tc.valid, result)
	}
}

func TestCheckPulsarDestinationProtocol(t *testing.T) {
	testcases := []struct {
		protocol dataflow.PulsarDestinationProtocol
		err      bool
	}{
		{
			protocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR,
			err:      false,
		},
		{
			protocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_PULSAR_SSL,
			err:      false,
		},
		{
			protocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_HTTP,
			err:      true,
		},
		{
			protocol: dataflow.PulsarDestinationProtocol_PULSAR_PROTOCOL_HTTPS,
			err:      true,
		},
		{
			protocol: dataflow.PulsarDestinationProtocol(1000),
			err:      true,
		},
	}
	for _, tc := range testcases {
		err := checkPulsarDestinationProtocol(tc.protocol)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckPulsarAuthentication(t *testing.T) {
	testcases := []struct {
		auth   *dataflow.PulsarAuthentication
		isEdit bool
		err    bool
	}{
		{
			auth:   nil,
			isEdit: false,
			err:    true,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "",
				Password: "password",
			},
			isEdit: false,
			err:    true,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "user",
				Password: "",
			},
			isEdit: false,
			err:    true,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "user",
				Password: "",
			},
			isEdit: true,
			err:    false,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_BASIC,
				UserName: "user",
				Password: "password",
			},
			isEdit: false,
			err:    false,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
				Token:    "",
			},
			isEdit: false,
			err:    true,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
				Token:    "",
			},
			isEdit: true,
			err:    false,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType_PULSAR_AUTH_TYPE_TOKEN,
				Token:    "token",
			},
			isEdit: false,
			err:    false,
		},
		{
			auth: &dataflow.PulsarAuthentication{
				AuthType: dataflow.PulsarAuthType(1000),
			},
			isEdit: false,
			err:    true,
		},
	}
	for _, tc := range testcases {
		err := checkPulsarAuthentication(tc.auth, tc.isEdit)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckPulsarDataFormat(t *testing.T) {
	testcases := []struct {
		format *dataflow.PulsarDataFormat
		err    bool
	}{
		{
			format: nil,
			err:    true,
		},
		{
			format: &dataflow.PulsarDataFormat{
				Protocol: dataflow.Protocol_CANAL_JSON,
			},
			err: false,
		},
		{
			format: &dataflow.PulsarDataFormat{
				Protocol: dataflow.Protocol_AVRO,
			},
			err: true,
		},
		{
			format: &dataflow.PulsarDataFormat{
				Protocol: dataflow.Protocol(1000),
			},
			err: true,
		},
	}
	for _, tc := range testcases {
		err := checkPulsarDataFormat(tc.format)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckPulsarTopicDistribution(t *testing.T) {
	testcases := []struct {
		td  *dataflow.PulsarTopicDistribution
		err bool
	}{
		{
			td:  nil,
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant: "",
			},
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "",
			},
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "",
			},
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType(1000),
			},
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
			},
			err: false,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			},
			err: false,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				TopicPrefix:     "???",
			},
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				TopicPrefix:     "prefix_",
				Separator:       "???",
			},
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				TopicPrefix:     "prefix_",
				Separator:       "_sep_",
				TopicSuffix:     "???",
			},
			err: true,
		},
		{
			td: &dataflow.PulsarTopicDistribution{
				PulsarTenant:    "tenant",
				PulsarNamespace: "namespace",
				DefaultTopic:    "topic",
				DispatchType:    dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				TopicPrefix:     "prefix_",
				Separator:       "_sep_",
				TopicSuffix:     "_suffix",
			},
			err: false,
		},
	}
	for _, tc := range testcases {
		err := checkPulsarTopicDistribution(tc.td)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckPulsarAdvancedSetting(t *testing.T) {
	testcases := []struct {
		setting *dataflow.PulsarAdvancedSetting
		err     bool
	}{
		{
			setting: nil,
			err:     true,
		},
		{
			setting: &dataflow.PulsarAdvancedSetting{},
			err:     true,
		},
		{
			setting: &dataflow.PulsarAdvancedSetting{
				ConnectionTimeout: 5,
			},
			err: true,
		},
		{
			setting: &dataflow.PulsarAdvancedSetting{
				ConnectionTimeout: 5,
				OperationTimeout:  4,
			},
			err: true,
		},
		{
			setting: &dataflow.PulsarAdvancedSetting{
				ConnectionTimeout: 5,
				OperationTimeout:  4,
				SendTimeout:       3,
			},
			err: true,
		},
		{
			setting: &dataflow.PulsarAdvancedSetting{
				ConnectionTimeout:   5,
				OperationTimeout:    4,
				SendTimeout:         3,
				BatchingMaxMessages: 1000,
			},
			err: true,
		},
		{
			setting: &dataflow.PulsarAdvancedSetting{
				ConnectionTimeout:       5,
				OperationTimeout:        4,
				SendTimeout:             3,
				BatchingMaxMessages:     1000,
				BatchingMaxPublishDelay: 100,
			},
			err: false,
		},
	}
	for _, tc := range testcases {
		err := checkPulsarAdvancedSetting(tc.setting)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckPartitionDispatchers(t *testing.T) {
	testcases := []struct {
		pds []*dataflow.PartitionDispatcherConfig
		err bool
	}{
		{
			pds: nil,
			err: true,
		},
		{
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher: nil,
				},
			},
			err: true,
		},
		{
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher(1000),
				},
			},
			err: true,
		},
		{
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_INDEX_VALUE,
				},
			},
			err: false,
		},
		{
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_COLUMNS,
					Columns:       nil,
				},
			},
			err: true,
		},
		{
			pds: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_COLUMNS,
					Columns:       []string{"a"},
				},
			},
			err: false,
		},
	}
	for _, tc := range testcases {
		err := checkPartitionDispatchers(tc.pds)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestS3BizChangefeedDownstreamAdaptor_CheckClusterVersion(t *testing.T) {
	testcases := []struct {
		name           string
		params         *dataflow.S3Info
		clusterVersion string
		err            bool
	}{
		{
			name: "event filter version check failed",
			params: &dataflow.S3Info{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v6.5.2",
			err:            true,
		},
		{
			name: "cluster version check failed",
			params: &dataflow.S3Info{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "check successful",
			params: &dataflow.S3Info{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v7.1.0",
			err:            false,
		},
	}

	adaptor := &s3BizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.checkClusterVersion(tc.params, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestS3BizChangefeedDownstreamAdaptor_ValidateCreateDownstreamParams(t *testing.T) {
	testcases := []struct {
		name        string
		params      interface{}
		cluster     *utils.Cluster
		err         bool
		errKeywords string
	}{
		{
			name:   "downstream params nil",
			params: nil,
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "invalid s3 info",
		},
		{
			name:   "downstream params not s3 info type",
			params: &dataflow.KafkaInfo{},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "invalid s3 info",
		},
		{
			name: "invalid s3 endpoint",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri: "abc",
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}), err: true,
			errKeywords: "check endpoint",
		},
		{
			name: "cluster region and bucket region not match",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "check s3 region",
		},
		{
			name: "invalid data format",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CSV,
					Quote:    "abc",
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "check data format",
		},
		{
			name: "invalid table config",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "check table config",
		},
		{
			name: "invalid start position",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_TSO,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "check start position",
		},
		{
			name: "invalid storage flush config",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "20m",
					FileSize:      0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "check storage flush config",
		},
		{
			name: "invalid cluster version",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "",
					FileSize:      0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err:         true,
			errKeywords: "version constraint",
		},
		{
			name: "valid downstream params",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "",
					FileSize:      0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				Region: "us-west-2",
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err: false,
		},
	}

	adaptor := &s3BizChangefeedDownstreamAdaptor{
		session: mockAWSSession(testSetupGetBucketRegionServer("us-west-2", http.StatusOK, true)),
	}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateCreateParams(tc.params, tc.cluster)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestS3BizChangefeedDownstreamAdaptor_ValidateEditDownstreamParams(t *testing.T) {
	testcases := []struct {
		name           string
		params         interface{}
		clusterVersion string
		err            bool
		errKeywords    string
	}{
		{
			name:           "downstream params nil",
			params:         nil,
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "invalid s3 info",
		},
		{
			name:           "downstream params not s3 info type",
			params:         &dataflow.KafkaInfo{},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "invalid s3 info",
		},
		{
			name:           "nil s3 endpoint",
			params:         &dataflow.S3Info{},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "endpoint is empty",
		},
		{
			name: "access key empty",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri: "s3://bucket/dir",
				},
			},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "access key is empty",
		},
		{
			name: "invalid data format",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CSV,
					Quote:    "abc",
				},
			},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "check data format",
		},
		{
			name: "invalid table config",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
			},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "check table config",
		},
		{
			name: "invalid storage flush config",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "20m",
					FileSize:      0,
				},
			},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "check storage flush config",
		},
		{
			name: "invalid cluster version",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "1m",
					FileSize:      DefaultFileSize,
				},
			},
			clusterVersion: "v6.5.9",
			err:            true,
			errKeywords:    "version constraint",
		},
		{
			name: "valid downstream params",
			params: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "1m",
					FileSize:      DefaultFileSize,
				},
			},
			clusterVersion: "v7.1.0",
			err:            false,
		},
	}

	adaptor := &s3BizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateEditParams(tc.params, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestS3BizChangefeedDownstreamAdaptor_AssembleCreateConnectorReq(t *testing.T) {
	testcases := []struct {
		downstreamParams interface{}
		assembledParams  *dataflow.S3Info
		err              bool
	}{
		{
			downstreamParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "a.b.c:9092",
				},
			},
			assembledParams: nil,
			err:             true,
		},
		{
			downstreamParams: &dataflow.S3Info{},
			assembledParams:  &dataflow.S3Info{},
			err:              false,
		},
		{
			downstreamParams: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "sk",
				},
			},
			assembledParams: &dataflow.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri:     "s3://bucket/dir",
					AccessKey: "ak",
					SecretKey: "",
				},
			},
			err: false,
		},
	}

	adaptor := &s3BizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		paramsBytes, err := json.Marshal(tc.downstreamParams)
		require.NoError(t, err)
		bizChangefeed := &domain.BizChangefeed{
			DownstreamType:      entity.TiCDCDownStreamTypeS3,
			DownstreamParameter: string(paramsBytes),
		}
		req, err := adaptor.AssembleGetResponse(bizChangefeed)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
			assert.Equal(t, tc.assembledParams, req.S3Info)
			assert.Equal(t, dataflow.ChangefeedType_S3, req.ChangefeedType)
		}
	}
}

func TestCheckS3Region(t *testing.T) {
	clusterRegion := "us-west-2"
	normalSession := mockAWSSession(testSetupGetBucketRegionServer(clusterRegion, http.StatusOK, true))
	t.Run("s3 url invalid", func(t *testing.T) {
		s3URL := "ss://bucket/dir"
		err := checkS3Region(s3URL, clusterRegion, normalSession)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "is invalid")
	})

	s3URL := "s3://bucket/dir"
	t.Run("get region error", func(t *testing.T) {
		sess := mockAWSSession(testSetupGetBucketRegionServer(clusterRegion, http.StatusInternalServerError, false))
		err := checkS3Region(s3URL, clusterRegion, sess)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "get bucket region error")
	})

	t.Run("regions are not matched", func(t *testing.T) {
		sess := mockAWSSession(testSetupGetBucketRegionServer("us-east-1", http.StatusOK, true))
		err := checkS3Region(s3URL, clusterRegion, sess)
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "should be same as the cluster")
	})

	t.Run("regions are matched", func(t *testing.T) {
		err := checkS3Region(s3URL, clusterRegion, normalSession)
		assert.NoError(t, err)
	})
}

func TestCheckS3Endpoint(t *testing.T) {
	testcases := []struct {
		endpoint *dataflow.S3Endpoint
		isEdit   bool
		err      bool
	}{
		{
			endpoint: nil,
			isEdit:   false,
			err:      true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri: "abc",
			},
			isEdit: false,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri: "",
			},
			isEdit: false,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:    "s3://bucket/dir",
				AuthType: dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
			},
			isEdit: false,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/dir",
				AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
				AccessKey: "ak",
			},
			isEdit: false,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/dir",
				AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
				AccessKey: "ak",
				SecretKey: "sk",
			},
			isEdit: false,
			err:    false,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:    "s3://bucket/dir",
				AuthType: dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE,
			},
			isEdit: false,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:             "s3://bucket/dir",
				AuthType:          dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE,
				IamAssumedRoleArn: "role123",
			},
			isEdit: false,
			err:    false,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:    "s3://bucket/dir",
				AuthType: dataflow.S3AuthType(100),
			},
			isEdit: false,
			err:    true,
		},
		{
			endpoint: nil,
			isEdit:   true,
			err:      true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:    "s3://bucket/dir",
				AuthType: dataflow.S3AuthType(100),
			},
			isEdit: true,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:    "s3://bucket/dir",
				AuthType: dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
			},
			isEdit: true,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:     "",
				AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
				AccessKey: "ak",
			},
			isEdit: true,
			err:    false,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:    "",
				AuthType: dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE,
			},
			isEdit: true,
			err:    true,
		},
		{
			endpoint: &dataflow.S3Endpoint{
				S3Uri:             "",
				AuthType:          dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE,
				IamAssumedRoleArn: "role123",
			},
			isEdit: true,
			err:    false,
		},
		{
			// valid custom endpoint
			endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/dir",
				AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
				AccessKey: "ak",
				SecretKey: "sk",
				Endpoint:  "https://s3.oss-us-east-1.aliyuncs.com",
			},
			isEdit: false,
			err:    false,
		},
		{
			// invalid custom endpoint without scheme
			endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/dir",
				AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
				AccessKey: "ak",
				SecretKey: "sk",
				Endpoint:  "s3.oss-us-east-1.aliyuncs.com",
			},
			isEdit: false,
			err:    true,
		},
	}

	for _, tc := range testcases {
		err := checkS3Endpoint(tc.endpoint, tc.isEdit)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckS3DataFormat(t *testing.T) {
	testcases := []struct {
		dataFormat *dataflow.S3Dataformat
		err        bool
	}{
		{
			dataFormat: nil,
			err:        true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol: dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:    "ab",
			},
			err: true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol: dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:    "\r",
			},
			err: true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol: dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:    "\n",
			},
			err: true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol:  dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:     "'",
				Delimiter: "",
			},
			err: true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol:  dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:     "'",
				Delimiter: "a\r",
			},
			err: true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol:  dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:     "'",
				Delimiter: "a\n",
			},
			err: true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol:  dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:     "'",
				Delimiter: "a'b",
			},
			err: true,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol:  dataflow.S3Protocol_S3_PROTOCOL_CSV,
				Quote:     "'",
				Delimiter: ";",
			},
			err: false,
		},
		{
			dataFormat: &dataflow.S3Dataformat{
				Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkS3DataFormat(tc.dataFormat)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestCheckStorageFlushConfig(t *testing.T) {
	testcases := []struct {
		storageFlushConfig *dataflow.StorageFlushConfig
		err                bool
	}{
		{
			storageFlushConfig: nil,
			err:                true,
		},
		{
			storageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "abc",
			},
			err: true,
		},
		{
			storageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "1s",
			},
			err: true,
		},
		{
			storageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "20m",
			},
			err: true,
		},
		{
			storageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "5m",
				FileSize:      1048575,
			},
			err: true,
		},
		{
			storageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "5m",
				FileSize:      536870913,
			},
			err: true,
		},
		{
			storageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "5m",
				FileSize:      DefaultFileSize,
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkStorageFlushConfig(tc.storageFlushConfig)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestInitStorageFlushConfig(t *testing.T) {
	testcases := []struct {
		inputConfig    *dataflow.StorageFlushConfig
		expectedConfig *dataflow.StorageFlushConfig
	}{
		{
			inputConfig: nil,
			expectedConfig: &dataflow.StorageFlushConfig{
				FlushInterval: DefaultFlushInterval,
				FileSize:      DefaultFileSize,
			},
		},
		{
			inputConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "",
				FileSize:      0,
			},
			expectedConfig: &dataflow.StorageFlushConfig{
				FlushInterval: DefaultFlushInterval,
				FileSize:      DefaultFileSize,
			},
		},
		{
			inputConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "abc",
				FileSize:      0,
			},
			expectedConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "abc",
				FileSize:      DefaultFileSize,
			},
		},
		{
			inputConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "abc",
				FileSize:      123,
			},
			expectedConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "abc",
				FileSize:      123,
			},
		},
	}

	for _, tc := range testcases {
		s3Info := &dataflow.S3Info{
			StorageFlushConfig: tc.inputConfig,
		}
		storageFlushConfig := initStorageFlushConfig(s3Info.GetStorageFlushConfig())
		assert.Equal(t, tc.expectedConfig, storageFlushConfig)
	}
}

func TestGcsBizChangefeedDownstreamAdaptor_CheckClusterVersion(t *testing.T) {
	testcases := []struct {
		name           string
		params         *dataflow.GcsInfo
		clusterVersion string
		err            bool
	}{
		{
			name: "event filter version check failed",
			params: &dataflow.GcsInfo{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v6.5.2",
			err:            true,
		},
		{
			name: "cluster version check failed",
			params: &dataflow.GcsInfo{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v7.1.0",
			err:            true,
		},
		{
			name: "check successful",
			params: &dataflow.GcsInfo{
				TableConfig: &dataflow.TableConfig{
					EventFilterRule: []*dataflow.EventFilterRule{
						{
							IgnoreInsertValueExpr: "abc",
						},
					},
				},
			},
			clusterVersion: "v7.1.1",
			err:            false,
		},
	}

	adaptor := &gcsBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.checkClusterVersion(tc.params, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestGcsBizChangefeedDownstreamAdaptor_ValidateCreateDownstreamParams(t *testing.T) {
	testcases := []struct {
		name        string
		params      interface{}
		cluster     *utils.Cluster
		err         bool
		errKeywords string
	}{
		{
			name:   "downstream params nil",
			params: nil,
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err:         true,
			errKeywords: "invalid gcs info",
		},
		{
			name:   "downstream params not gcs info type",
			params: &dataflow.KafkaInfo{},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err:         true,
			errKeywords: "invalid gcs info",
		},
		{
			name: "invalid gcs endpoint",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "abc",
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err:         true,
			errKeywords: "check endpoint",
		},
		{
			name: "invalid data format",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CSV,
					Quote:    "abc",
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err:         true,
			errKeywords: "check data format",
		},
		{
			name: "invalid table config",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err:         true,
			errKeywords: "check table config",
		},
		{
			name: "invalid start position",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_TSO,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err:         true,
			errKeywords: "check start position",
		},
		{
			name: "invalid storage flush config",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "20m",
					FileSize:      0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err:         true,
			errKeywords: "check storage flush config",
		},
		{
			name: "invalid cluster version",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "",
					FileSize:      0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.0"}},
			}),
			err:         true,
			errKeywords: "version constraint",
		},
		{
			name: "valid downstream params",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "",
					FileSize:      0,
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v7.1.1"}},
			}),
			err: false,
		},
	}

	adaptor := &gcsBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateCreateParams(tc.params, tc.cluster)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestGcsBizChangefeedDownstreamAdaptor_ValidateEditDownstreamParams(t *testing.T) {
	testcases := []struct {
		name           string
		params         interface{}
		clusterVersion string
		err            bool
		errKeywords    string
	}{
		{
			name:           "downstream params nil",
			params:         nil,
			clusterVersion: "v7.1.1",
			err:            true,
			errKeywords:    "invalid gcs info",
		},
		{
			name:           "downstream params not gcs info type",
			params:         &dataflow.KafkaInfo{},
			clusterVersion: "v7.1.1",
			err:            true,
			errKeywords:    "invalid gcs info",
		},
		{
			name:           "nil gcs endpoint",
			params:         &dataflow.GcsInfo{},
			clusterVersion: "v7.1.1",
			err:            true,
			errKeywords:    "endpoint is empty",
		},
		{
			name: "invalid data format",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CSV,
					Quote:    "abc",
				},
			},
			clusterVersion: "v7.1.1",
			err:            true,
			errKeywords:    "check data format",
		},
		{
			name: "invalid table config",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule: genStrList(maxFilterRuleNum+1, 1),
				},
			},
			clusterVersion: "v7.1.1",
			err:            true,
			errKeywords:    "check table config",
		},
		{
			name: "invalid storage flush config",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "20m",
					FileSize:      0,
				},
			},
			clusterVersion: "v7.1.1",
			err:            true,
			errKeywords:    "check storage flush config",
		},
		{
			name: "invalid cluster version",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "1m",
					FileSize:      DefaultFileSize,
				},
			},
			clusterVersion: "v7.1.0",
			err:            true,
			errKeywords:    "version constraint",
		},
		{
			name: "valid downstream params",
			params: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
				DataFormat: &dataflow.S3Dataformat{
					Protocol: dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				},
				TableConfig: &dataflow.TableConfig{
					FilterRule:      []string{},
					EventFilterRule: []*dataflow.EventFilterRule{},
				},
				StartPosition: &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				},
				StorageFlushConfig: &dataflow.StorageFlushConfig{
					FlushInterval: "1m",
					FileSize:      DefaultFileSize,
				},
			},
			clusterVersion: "v7.1.1",
			err:            false,
		},
	}

	adaptor := &gcsBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateEditParams(tc.params, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestGcsBizChangefeedDownstreamAdaptor_AssembleCreateConnectorReq(t *testing.T) {
	testcases := []struct {
		downstreamParams interface{}
		assembledParams  *dataflow.GcsInfo
		err              bool
	}{
		{
			downstreamParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "a.b.c:9092",
				},
			},
			assembledParams: nil,
			err:             true,
		},
		{
			downstreamParams: &dataflow.GcsInfo{},
			assembledParams:  &dataflow.GcsInfo{},
			err:              false,
		},
		{
			downstreamParams: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
			},
			assembledParams: &dataflow.GcsInfo{
				Endpoint: &dataflow.GcsEndpoint{
					GcsUri: "gs://bucket/dir",
				},
			},
			err: false,
		},
	}

	adaptor := &gcsBizChangefeedDownstreamAdaptor{}
	for _, tc := range testcases {
		paramsBytes, err := json.Marshal(tc.downstreamParams)
		require.NoError(t, err)
		name := "changefeed-gcs"
		bizChangefeed := &domain.BizChangefeed{
			Name:                name,
			DownstreamType:      entity.TiCDCDownStreamTypeGCS,
			DownstreamParameter: string(paramsBytes),
		}
		req, err := adaptor.AssembleGetResponse(bizChangefeed)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
			assert.Equal(t, tc.assembledParams, req.GcsInfo)
			assert.Equal(t, dataflow.ChangefeedType_GCS, req.ChangefeedType)
			assert.Equal(t, name, req.Name)
		}
	}
}

func TestCheckGCSEndpoint(t *testing.T) {
	testcases := []struct {
		endpoint *dataflow.GcsEndpoint
		err      bool
	}{
		{
			endpoint: nil,
			err:      true,
		},
		{
			endpoint: &dataflow.GcsEndpoint{},
			err:      true,
		},
		{
			endpoint: &dataflow.GcsEndpoint{
				GcsUri: "s3://bucket/dir",
			},
			err: true,
		},
		{
			endpoint: &dataflow.GcsEndpoint{
				GcsUri: "gs://bucket/dir",
			},
			err: false,
		},
		{
			endpoint: &dataflow.GcsEndpoint{
				GcsUri: "gcs://bucket/dir",
			},
			err: false,
		},
	}

	for _, tc := range testcases {
		err := checkGCSEndpoint(tc.endpoint)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
		}
	}
}

func TestServerlessBizChangefeedDownstreamAdaptor_ValidateCreateDownstreamParams(t *testing.T) {
	testcases := []struct {
		name    string
		params  interface{}
		cluster *utils.Cluster
		err     bool
	}{
		{
			name:   "downstream params nil",
			params: nil,
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name:   "downstream params not serverless info type",
			params: &dataflow.KafkaInfo{},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "empty serverless cluster id",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 0,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:1000",
						Username: "u",
						Password: "p",
					},
					TableConfig: &dataflow.TableConfig{
						FilterRule:      []string{},
						EventFilterRule: []*dataflow.EventFilterRule{},
					},
					StartPosition: &dataflow.StartPosition{
						Mode: dataflow.StartMode_FROM_NOW,
					},
					PrivatelinkInfo: &dataflow.PrivatelinkInfo{
						DfPrivatelinkEndpointId: 111,
					},
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "invalid mysql info",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "invalid private link info",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:1000",
						Username: "u",
						Password: "p",
					},
					TableConfig: &dataflow.TableConfig{
						FilterRule:      []string{},
						EventFilterRule: []*dataflow.EventFilterRule{},
					},
					StartPosition: &dataflow.StartPosition{
						Mode: dataflow.StartMode_FROM_NOW,
					},
					PrivatelinkInfo: &dataflow.PrivatelinkInfo{},
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: true,
		},
		{
			name: "valid downstream params",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:1000",
						Username: "u",
						Password: "p",
					},
					TableConfig: &dataflow.TableConfig{
						FilterRule:      []string{},
						EventFilterRule: []*dataflow.EventFilterRule{},
					},
					StartPosition: &dataflow.StartPosition{
						Mode: dataflow.StartMode_FROM_NOW,
					},
					PrivatelinkInfo: &dataflow.PrivatelinkInfo{
						DfPrivatelinkEndpointId: 111,
					},
				},
			},
			cluster: utils.NewCluster(&mgmt.Cluster{
				ClusterInfo: &mgmt.Cluster_NextGenCluster{
					NextGenCluster: &v12.Cluster{Version: "v6.5.9"}},
			}),
			err: false,
		},
	}

	adaptor := newServerlessBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateCreateParams(tc.params, tc.cluster)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestServerlessBizChangefeedDownstreamAdaptor_ValidateEditDownstreamParams(t *testing.T) {
	testcases := []struct {
		name           string
		params         interface{}
		clusterVersion string
		err            bool
	}{
		{
			name:           "downstream params nil",
			params:         nil,
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name:           "downstream params not serverless info type",
			params:         &dataflow.KafkaInfo{},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "empty serverless cluster id",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 0,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:1000",
						Username: "u",
						Password: "p",
					},
					TableConfig: &dataflow.TableConfig{
						FilterRule:      []string{},
						EventFilterRule: []*dataflow.EventFilterRule{},
					},
					StartPosition: &dataflow.StartPosition{
						Mode: dataflow.StartMode_FROM_NOW,
					},
					PrivatelinkInfo: &dataflow.PrivatelinkInfo{
						DfPrivatelinkEndpointId: 111,
					},
				},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "invalid mysql info",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "invalid private link info",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:1000",
						Username: "u",
						Password: "p",
					},
					TableConfig: &dataflow.TableConfig{
						FilterRule:      []string{},
						EventFilterRule: []*dataflow.EventFilterRule{},
					},
					StartPosition: &dataflow.StartPosition{
						Mode: dataflow.StartMode_FROM_NOW,
					},
					PrivatelinkInfo: &dataflow.PrivatelinkInfo{},
				},
			},
			clusterVersion: "v6.5.9",
			err:            true,
		},
		{
			name: "valid downstream params",
			params: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:1000",
						Username: "u",
						Password: "p",
					},
					TableConfig: &dataflow.TableConfig{
						FilterRule:      []string{},
						EventFilterRule: []*dataflow.EventFilterRule{},
					},
					StartPosition: &dataflow.StartPosition{
						Mode: dataflow.StartMode_FROM_NOW,
					},
					PrivatelinkInfo: &dataflow.PrivatelinkInfo{
						DfPrivatelinkEndpointId: 111,
					},
				},
			},
			clusterVersion: "v6.5.9",
			err:            false,
		},
	}

	adaptor := newServerlessBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			err := adaptor.ValidateEditParams(tc.params, tc.clusterVersion)
			if tc.err {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
		})
	}
}

func TestServerlessBizChangefeedDownstreamAdaptor_AssembleCreateConnectorReq(t *testing.T) {
	testcases := []struct {
		downstreamParams interface{}
		assembledParams  *dataflow.ServerlessTidbInfo
		err              bool
	}{
		{
			downstreamParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "a.b.c:9092",
				},
			},
			assembledParams: nil,
			err:             true,
		},
		{
			downstreamParams: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:4000",
						Username: "u",
						Password: "p",
					},
				},
			},
			assembledParams: &dataflow.ServerlessTidbInfo{
				ClusterId: 111,
				MysqlInfo: &dataflow.MySQLInfo{
					Connection: &dataflow.Connection{
						Endpoint: "a.b.c:4000",
						Username: "u",
						Password: "",
					},
				},
			},
			err: false,
		},
	}

	adaptor := newServerlessBizChangefeedDownstreamAdaptor(nil)
	for _, tc := range testcases {
		paramsBytes, err := json.Marshal(tc.downstreamParams)
		require.NoError(t, err)
		bizChangefeed := &domain.BizChangefeed{
			DownstreamType:      entity.TiCDCDownStreamTypeServerless,
			DownstreamParameter: string(paramsBytes),
		}
		req, err := adaptor.AssembleGetResponse(bizChangefeed)
		if tc.err {
			assert.Error(t, err)
		} else {
			assert.NoError(t, err)
			assert.Equal(t, tc.assembledParams, req.ServerlessTidbInfo)
			assert.Equal(t, dataflow.ChangefeedType_SERVERLESS_TIDB, req.ChangefeedType)
		}
	}
}

func genEmptyEventFilters(length int) []*dataflow.EventFilterRule {
	rules := make([]*dataflow.EventFilterRule, length)
	for i := range rules {
		rules[i] = &dataflow.EventFilterRule{}
	}
	return rules
}

func genStrList(listLen, perStrLen int) []string {
	strs := make([]string, listLen)
	for i := range strs {
		strs[i] = genRandString(perStrLen)
	}
	return strs
}

func genRandString(n int) string {
	b := make([]rune, n)
	for i := range b {
		b[i] = letterRunes[rand.Intn(len(letterRunes))]
	}
	return string(b)
}

func testSetupGetBucketRegionServer(region string, statusCode int, incHeader bool) *httptest.Server {
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if incHeader {
			w.Header().Set("X-Amz-Bucket-Region", region)
		}
		w.WriteHeader(statusCode)
	}))
}

func mockAWSSession(server *httptest.Server) *session.Session {
	sess := aws_unit.Session.Copy()
	sess.Config.Region = aws.String("default-region")
	sess.Config.Endpoint = aws.String(server.URL)
	sess.Config.DisableSSL = aws.Bool(true)
	return sess
}