package impl

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

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/app/utils"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	encryptor_mock "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper/mock"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	regionalapi_mock "github.com/tidbcloud/dataflow-service/pkg/regional-api/mock"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

type kafkaDownstreamAdaptorTestSuite struct {
	suite.Suite
	kafkaAdaptor *kafkaDownstreamAdaptor

	mockPreCheckCli *regionalapi_mock.TiCDCPreCheckClient
	mockEncryptor   *encryptor_mock.Encryptor
}

func (s *kafkaDownstreamAdaptorTestSuite) SetupSuite() {
	s.mockPreCheckCli = regionalapi_mock.NewTiCDCPreCheckClient(s.T())
	s.mockEncryptor = encryptor_mock.NewEncryptor(s.T())

	s.kafkaAdaptor = &kafkaDownstreamAdaptor{
		cdcPreCheckCli: s.mockPreCheckCli,
		encryptor:      s.mockEncryptor,
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestParseKafkaParams() {
	testcases := []struct {
		name             string
		downstreamParams interface{}
		expectedError    bool
	}{
		{
			name:             "nil param",
			downstreamParams: nil,
			expectedError:    true,
		},
		{
			name:             "wrong type param",
			downstreamParams: &dataflow.MySQLInfo{},
			expectedError:    true,
		},
		{
			name:             "wrong type param2",
			downstreamParams: dataflow.KafkaInfo{},
			expectedError:    true,
		},
		{
			name:             "right type param",
			downstreamParams: &dataflow.KafkaInfo{},
			expectedError:    false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			_, err := s.kafkaAdaptor.parseKafkaParams(tc.downstreamParams)
			if tc.expectedError {
				s.Assert().Error(err)
			} else {
				s.Assert().NoError(err)
			}
		})
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestParseKafkaParamsStr() {
	testcases := []struct {
		name             string
		downstreamParams interface{}
		expectedError    bool
	}{
		{
			name: "wrong type param",
			downstreamParams: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "abc:3306",
				},
			},
			expectedError: true,
		},
		{
			name: "right type param",
			downstreamParams: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "abc:9092",
				},
			},
			expectedError: false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			paramsBytes, err := json.Marshal(tc.downstreamParams)
			s.Require().NoError(err)
			_, err = s.kafkaAdaptor.parseKafkaParamsStr(string(paramsBytes))
			if tc.expectedError {
				s.Assert().Error(err)
			} else {
				s.Assert().NoError(err)
			}
		})
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestTestConnection() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "us-west-2",
		InfraProviderType: "eks",
	}

	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.kafkaAdaptor.TestConnection(context.Background(), cluster, downstreamParams)
		s.Assert().Error(err)
	})

	mockDownstreamParams := &dataflow.KafkaInfo{
		Broker: &dataflow.Broker{
			BrokerEndpoints: "abc:9092",
		},
	}
	s.Run("test connection error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockPreCheckCli.
			On("TestNetworkConnection", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()
		err := s.kafkaAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.Assert().Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "us-west-2",
		TenantID:      1,
		CloudProvider: "aws",
	}
	expectedReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: "111",
		Type:      "kafka",
		KafkaParam: &ticdc_v1.TestNetworkConnectionRequest_KafkaParameter{
			BoostrapEndpoints: "abc:9092",
		},
	}
	call := s.mockPreCheckCli.
		On("TestNetworkConnection", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("test connection successfully", func() {
		err := s.kafkaAdaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.Assert().NoError(err)
	})
}

func (s *kafkaDownstreamAdaptorTestSuite) TestPreCheckChangefeedParameters() {
	cluster := &core.Cluster{
		Id:                111,
		ProjectId:         11,
		TenantId:          1,
		RegionName:        "us-west-2",
		InfraProviderType: "eks",
	}

	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.kafkaAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Assert().Error(err)
	})

	s.Run("parse start tso error", func() {
		downstreamParams := &dataflow.KafkaInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		err := s.kafkaAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Assert().Error(err)
	})

	downstreamParams := &dataflow.KafkaInfo{
		Broker: &dataflow.Broker{
			BrokerEndpoints: "abc:9092",
		},
		TopicPartitionConfig: &dataflow.TopicPartitionConfig{
			DefaultTopic:      "test",
			ReplicationFactor: 5,
			PartitionNum:      3,
		},
		DataFormat: &dataflow.DataFormat{
			Protocol: dataflow.Protocol_AVRO,
			SchemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://127.0.0.1:8080",
			},
		},
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  123123,
		},
	}
	s.Run("pre check error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockPreCheckCli.
			On("ValidateChangefeedParameters", mock.Anything, mock.Anything, mock.Anything).
			Return(mockErr)
		defer call.Unset()

		err := s.kafkaAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Assert().Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "us-west-2",
		TenantID:      1,
		CloudProvider: "aws",
	}
	expectedReq := &ticdc_v1.ValidateChangefeedParametersRequest{
		ClusterId: "111",
		Type:      "kafka",
		StartTso:  123123,
		KafkaParam: &ticdc_v1.ValidateChangefeedParametersRequest_KafkaParameter{
			Url:                     "abc:9092",
			Topic:                   "test",
			SchemaRegistryEndpoints: "http://127.0.0.1:8080",
			ReplicationFactor:       5,
			PartitionNum:            3,
		},
	}
	call := s.mockPreCheckCli.
		On("ValidateChangefeedParameters", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("pre check successfully", func() {
		err := s.kafkaAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Assert().NoError(err)
	})
}

func (s *kafkaDownstreamAdaptorTestSuite) TestEncryptPasswordInPlace() {
	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.MySQLInfo{}
		err := s.kafkaAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().Error(err)
	})

	s.Run("password nil", func() {
		downstreamParams := &dataflow.KafkaInfo{}
		err := s.kafkaAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().NoError(err)
	})

	s.Run("password empty", func() {
		downstreamParams := &dataflow.KafkaInfo{
			Authentication: &dataflow.Authentication{
				Password: "",
			},
			DataFormat: &dataflow.DataFormat{
				SchemaRegistry: &dataflow.SchemaRegistry{
					Password: "",
				},
			},
		}
		err := s.kafkaAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().NoError(err)
	})

	s.Run("encrypt authentication password error", func() {
		downstreamParams := &dataflow.KafkaInfo{
			Authentication: &dataflow.Authentication{
				Password: "abc",
			},
		}
		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt password of authentication error")
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Authentication.Password).
			Return(encryptErr)
		defer call.Unset()

		err := s.kafkaAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().Error(err)
		s.Assert().Contains(err.Error(), encryptErr.ErrMsg())
		s.mockEncryptor.AssertExpectations(s.T())
	})

	s.Run("encrypt schema registry password error", func() {
		downstreamParams := &dataflow.KafkaInfo{
			Authentication: &dataflow.Authentication{
				Password: "abc",
			},
			DataFormat: &dataflow.DataFormat{
				SchemaRegistry: &dataflow.SchemaRegistry{
					Password: "def",
				},
			},
		}
		authCall := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Authentication.Password).
			Return(nil)
		defer authCall.Unset()

		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt password of schema registry error")
		registryCall := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.DataFormat.SchemaRegistry.Password).
			Return(encryptErr)
		defer registryCall.Unset()

		err := s.kafkaAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().Error(err)
		s.Assert().Contains(err.Error(), encryptErr.ErrMsg())
		s.mockEncryptor.AssertExpectations(s.T())
	})

	s.Run("encrypt successfully", func() {
		downstreamParams := &dataflow.KafkaInfo{
			Authentication: &dataflow.Authentication{
				Password: "abc",
			},
			DataFormat: &dataflow.DataFormat{
				SchemaRegistry: &dataflow.SchemaRegistry{
					Password: "def",
				},
			},
		}
		encryptedAuthPassword := base64.StdEncoding.EncodeToString([]byte(downstreamParams.Authentication.Password))
		authCall := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Authentication.Password).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedAuthPassword
			})
		defer authCall.Unset()

		encryptedRegistryPassword := base64.StdEncoding.EncodeToString(
			[]byte(downstreamParams.DataFormat.SchemaRegistry.Password))
		registryCall := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.DataFormat.SchemaRegistry.Password).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedRegistryPassword
			})
		defer registryCall.Unset()

		err := s.kafkaAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().NoError(err)
		s.Assert().Equal(encryptedAuthPassword, downstreamParams.Authentication.Password)
		s.Assert().Equal(encryptedRegistryPassword, downstreamParams.DataFormat.SchemaRegistry.Password)
		s.mockEncryptor.AssertExpectations(s.T())
	})
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleURIConfigData() {
	decryptedPassword := "decrypted-password"
	testcases := []struct {
		name              string
		kafkaInfo         *dataflow.KafkaInfo
		expectedUriConfig *domain.TiCDCChangefeedKafkaURIConfigData
		networkType       entity.ChangefeedDownstreamNetworkType
		err               bool
	}{
		{
			name: "invalid kafka version",
			kafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					KafkaVersion: dataflow.KafkaVersion(5),
				},
			},
			expectedUriConfig: nil,
			err:               true,
		},
		{
			name: "normal kafka info",
			kafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "abc:9092 ",
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					TlsEnable:       true,
					Compression:     dataflow.CompressionType_GZIP,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DefaultTopic:      "test",
					ReplicationFactor: 1,
					PartitionNum:      1,
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "def",
				},
			},
			expectedUriConfig: &domain.TiCDCChangefeedKafkaURIConfigData{
				BrokerEndpoints:     "abc:9092",
				EnableTLS:           true,
				DefaultTopic:        "test",
				Protocol:            utils.KafkaProtocolTypeCanalJson,
				ReplicationFactor:   1,
				PartitionNum:        1,
				KafkaVersion:        utils.KafkaVersion3xx,
				Compression:         pointer.ToString(utils.CompressionGzip),
				SASLMechanism:       pointer.ToString(utils.KafkaAuthTypePlain),
				SASLUser:            pointer.ToString("abc"),
				SASLPassword:        pointer.ToString(decryptedPassword),
				EnableTiDBExtension: pointer.ToBool(false),
				AutoCreateTopic:     pointer.ToBool(true),
			},
			networkType: entity.ChangefeedDownstreamNetworkTypeInternet,
			err:         false,
		},
		{
			name: "private link kafka info",
			kafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "abc:9092 ",
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					TlsEnable:       true,
					Compression:     dataflow.CompressionType_GZIP,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DefaultTopic:      "test",
					ReplicationFactor: 1,
					PartitionNum:      1,
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "def",
				},
			},
			expectedUriConfig: &domain.TiCDCChangefeedKafkaURIConfigData{
				BrokerEndpoints:     "abc:9092",
				EnableTLS:           true,
				DefaultTopic:        "test",
				Protocol:            utils.KafkaProtocolTypeCanalJson,
				ReplicationFactor:   1,
				PartitionNum:        1,
				KafkaVersion:        utils.KafkaVersion3xx,
				Compression:         pointer.ToString(utils.CompressionGzip),
				SASLMechanism:       pointer.ToString(utils.KafkaAuthTypePlain),
				SASLUser:            pointer.ToString("abc"),
				SASLPassword:        pointer.ToString(decryptedPassword),
				EnableTiDBExtension: pointer.ToBool(false),
				AutoCreateTopic:     pointer.ToBool(true),
				InsecureSkipVerify:  pointer.ToBool(true),
			},
			networkType: entity.DownstreamNetworkTypeManuallyPrivateLink,
			err:         false,
		},
		{
			name: "insecurity skip verify kafka info",
			kafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints:    "abc:9092",
					KafkaVersion:       dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					TlsEnable:          true,
					Compression:        dataflow.CompressionType_GZIP,
					InsecureSkipVerify: true,
				},
				DataFormat: &dataflow.DataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DefaultTopic:      "test",
					ReplicationFactor: 1,
					PartitionNum:      1,
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "def",
				},
			},
			expectedUriConfig: &domain.TiCDCChangefeedKafkaURIConfigData{
				BrokerEndpoints:     "abc:9092",
				EnableTLS:           true,
				DefaultTopic:        "test",
				Protocol:            utils.KafkaProtocolTypeCanalJson,
				ReplicationFactor:   1,
				PartitionNum:        1,
				KafkaVersion:        utils.KafkaVersion3xx,
				Compression:         pointer.ToString(utils.CompressionGzip),
				SASLMechanism:       pointer.ToString(utils.KafkaAuthTypePlain),
				SASLUser:            pointer.ToString("abc"),
				SASLPassword:        pointer.ToString(decryptedPassword),
				EnableTiDBExtension: pointer.ToBool(false),
				AutoCreateTopic:     pointer.ToBool(true),
				InsecureSkipVerify:  pointer.ToBool(true),
			},
			networkType: entity.ChangefeedDownstreamNetworkTypeInternet,
			err:         false,
		},
		{
			name: "kafka client id set",
			kafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "abc:9092",
					KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
					Compression:     dataflow.CompressionType_GZIP,
					KafkaClientId:   "my-client-id",
				},
				DataFormat: &dataflow.DataFormat{
					Protocol:            dataflow.Protocol_CANAL_JSON,
					EnableTidbExtension: false,
				},
				TopicPartitionConfig: &dataflow.TopicPartitionConfig{
					DefaultTopic:      "test",
					ReplicationFactor: 1,
					PartitionNum:      1,
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
					UserName: "abc",
					Password: "def",
				},
			},
			expectedUriConfig: &domain.TiCDCChangefeedKafkaURIConfigData{
				BrokerEndpoints:     "abc:9092",
				DefaultTopic:        "test",
				Protocol:            utils.KafkaProtocolTypeCanalJson,
				ReplicationFactor:   1,
				PartitionNum:        1,
				KafkaVersion:        utils.KafkaVersion3xx,
				Compression:         pointer.ToString(utils.CompressionGzip),
				SASLMechanism:       pointer.ToString(utils.KafkaAuthTypePlain),
				SASLUser:            pointer.ToString("abc"),
				SASLPassword:        pointer.ToString(decryptedPassword),
				EnableTiDBExtension: pointer.ToBool(false),
				AutoCreateTopic:     pointer.ToBool(true),
				KafkaClientID:       pointer.ToString("my-client-id"),
			},
			networkType: entity.ChangefeedDownstreamNetworkTypeInternet,
			err:         false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			if !tc.err {
				call := s.mockEncryptor.
					On("DecryptBase64", tc.kafkaInfo.Authentication.Password).
					Return(decryptedPassword, nil)
				defer call.Unset()
			}
			uriConfig, err := s.kafkaAdaptor.assembleURIConfigData(context.Background(), tc.kafkaInfo, tc.networkType)
			if tc.err {
				s.Assert().Error(err)
				return
			}
			s.Assert().Equal(tc.expectedUriConfig.BrokerEndpoints, uriConfig.BrokerEndpoints)
			s.Assert().Equal(tc.expectedUriConfig.EnableTLS, uriConfig.EnableTLS)
			s.Assert().Equal(tc.expectedUriConfig.Protocol, uriConfig.Protocol)
			s.Assert().Equal(tc.expectedUriConfig.DefaultTopic, uriConfig.DefaultTopic)
			s.Assert().Equal(tc.expectedUriConfig.ReplicationFactor, uriConfig.ReplicationFactor)
			s.Assert().Equal(tc.expectedUriConfig.PartitionNum, uriConfig.PartitionNum)
			s.Assert().Equal(tc.expectedUriConfig.KafkaVersion, uriConfig.KafkaVersion)
			if tc.expectedUriConfig.Compression != nil {
				s.Assert().NotNil(uriConfig.Compression)
				s.Assert().Equal(*tc.expectedUriConfig.Compression, *uriConfig.Compression)
			}
			if tc.expectedUriConfig.SASLMechanism != nil {
				s.Assert().NotNil(uriConfig.SASLMechanism)
				s.Assert().NotNil(uriConfig.SASLUser)
				s.Assert().NotNil(uriConfig.SASLPassword)
				s.Assert().Equal(*tc.expectedUriConfig.SASLMechanism, *uriConfig.SASLMechanism)
				s.Assert().Equal(*tc.expectedUriConfig.SASLUser, *uriConfig.SASLUser)
				s.Assert().Equal(*tc.expectedUriConfig.SASLPassword, *uriConfig.SASLPassword)
			}
			if tc.expectedUriConfig.EnableTiDBExtension != nil {
				s.Assert().NotNil(uriConfig.EnableTiDBExtension)
				s.Assert().Equal(*tc.expectedUriConfig.EnableTiDBExtension, *uriConfig.EnableTiDBExtension)
			}
			if tc.expectedUriConfig.AutoCreateTopic != nil {
				s.Assert().NotNil(uriConfig.AutoCreateTopic)
				s.Assert().Equal(*tc.expectedUriConfig.AutoCreateTopic, *uriConfig.AutoCreateTopic)
			}
			if tc.expectedUriConfig.InsecureSkipVerify != nil {
				s.Assert().NotNil(uriConfig.InsecureSkipVerify)
				s.Assert().Equal(*tc.expectedUriConfig.InsecureSkipVerify, *uriConfig.InsecureSkipVerify)
			}
			if tc.expectedUriConfig.KafkaClientID != nil {
				s.Assert().NotNil(uriConfig.KafkaClientID)
				s.Assert().Equal(*tc.expectedUriConfig.KafkaClientID, *uriConfig.KafkaClientID)
			} else {
				s.Assert().Nil(uriConfig.KafkaClientID)
			}
		})
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleURIDataFormat() {
	testcases := []struct {
		dataFormat                         *dataflow.DataFormat
		expectedDecimalHandingMode         string
		expectedBigintUnsignedHandlingMode string
		expectedEnableTiDBExtension        *bool
	}{
		{
			dataFormat: &dataflow.DataFormat{
				Protocol:            dataflow.Protocol_OPEN_PROTOCOL,
				EnableTidbExtension: true,
			},
			expectedDecimalHandingMode:         "",
			expectedBigintUnsignedHandlingMode: "",
			expectedEnableTiDBExtension:        nil,
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol:                   dataflow.Protocol_AVRO,
				DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
				BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING,
				EnableTidbExtension:        true,
			},
			expectedDecimalHandingMode:         utils.DecimalHandlingModeString,
			expectedBigintUnsignedHandlingMode: utils.BigintUnsignedHandlingModeString,
			expectedEnableTiDBExtension:        pointer.ToBool(true),
		},
		{
			dataFormat: &dataflow.DataFormat{
				Protocol:            dataflow.Protocol_CANAL_JSON,
				EnableTidbExtension: false,
			},
			expectedDecimalHandingMode:         "",
			expectedBigintUnsignedHandlingMode: "",
			expectedEnableTiDBExtension:        pointer.ToBool(false),
		},
	}

	for _, tc := range testcases {
		uriConfig := &domain.TiCDCChangefeedKafkaURIConfigData{}
		s.kafkaAdaptor.assembleURIDataFormat(uriConfig, tc.dataFormat)
		if tc.expectedDecimalHandingMode == "" {
			s.Assert().Nil(uriConfig.AvroDecimalHandlingMode)
		} else {
			s.Assert().Equal(tc.expectedDecimalHandingMode, *uriConfig.AvroDecimalHandlingMode)
		}
		if tc.expectedBigintUnsignedHandlingMode == "" {
			s.Assert().Nil(uriConfig.AvroBigintUnsignedHandlingMode)
		} else {
			s.Assert().Equal(tc.expectedBigintUnsignedHandlingMode, *uriConfig.AvroBigintUnsignedHandlingMode)
		}
		if tc.expectedEnableTiDBExtension == nil {
			s.Assert().Nil(uriConfig.EnableTiDBExtension)
		} else {
			s.Assert().Equal(*tc.expectedEnableTiDBExtension, *uriConfig.EnableTiDBExtension)
		}
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleURIAuthentication() {
	decryptedPassword := "decrypted-password"
	testcases := []struct {
		name             string
		authentication   *dataflow.Authentication
		decryptPassword  bool
		expectedAuthType string
	}{
		{
			name: "disable authentication",
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_DISABLE,
			},
			decryptPassword:  false,
			expectedAuthType: "",
		},
		{
			name: "sasl without password",
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				UserName: "abc",
			},
			decryptPassword:  false,
			expectedAuthType: utils.KafkaAuthTypePlain,
		},
		{
			name: "sasl with password",
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
				UserName: "abc",
				Password: "def",
			},
			decryptPassword:  true,
			expectedAuthType: utils.KafkaAuthTypePlain,
		},
		{
			name: "sha 256 with password",
			authentication: &dataflow.Authentication{
				AuthType: dataflow.KafkaAuthType_SASL_SCRAM_SHA_256,
				UserName: "abc",
				Password: "def",
			},
			decryptPassword:  true,
			expectedAuthType: utils.KafkaAuthTypeScramSha256,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			if tc.decryptPassword {
				call := s.mockEncryptor.
					On("DecryptBase64", tc.authentication.Password).
					Return(decryptedPassword, nil)
				defer call.Unset()
			}
			uriConfig := &domain.TiCDCChangefeedKafkaURIConfigData{}
			_ = s.kafkaAdaptor.assembleURIAuthentication(uriConfig, tc.authentication)
			if tc.expectedAuthType == "" {
				s.Assert().Nil(uriConfig.SASLMechanism)
				s.Assert().Nil(uriConfig.SASLUser)
				s.Assert().Nil(uriConfig.SASLPassword)
			} else {
				s.Assert().Equal(tc.expectedAuthType, *uriConfig.SASLMechanism)
				s.Assert().Equal(tc.authentication.UserName, *uriConfig.SASLUser)
				if tc.decryptPassword {
					s.Assert().Equal(decryptedPassword, *uriConfig.SASLPassword)
				} else {
					s.Assert().Equal("", *uriConfig.SASLPassword)
				}
			}
			if tc.decryptPassword {
				s.mockEncryptor.AssertExpectations(s.T())
			}
		})
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleChangefeedConfig() {
	rawPassword := "abc"
	decryptedPassword := "password"
	startTSO := uint64(123123)
	cluster := &core.Cluster{
		TenantId:  1,
		ProjectId: 11,
		Id:        111,
	}
	kafkaInfo := &dataflow.KafkaInfo{
		Broker: &dataflow.Broker{
			BrokerEndpoints: "abc:9092 ",
			KafkaVersion:    dataflow.KafkaVersion_KAFKA_VERSION_3XX,
			TlsEnable:       true,
			Compression:     dataflow.CompressionType_GZIP,
		},
		Authentication: &dataflow.Authentication{
			AuthType: dataflow.KafkaAuthType_SASL_PLAIN,
			UserName: "abc",
			Password: rawPassword,
		},
		TopicPartitionConfig: &dataflow.TopicPartitionConfig{
			DefaultTopic:      "test",
			ReplicationFactor: 1,
			PartitionNum:      1,
			DispatchType:      dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			TopicPrefix:       "prefix_",
			TopicSuffix:       "_suffix",
			PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_TS,
				},
			},
		},
		DataFormat: &dataflow.DataFormat{
			Protocol: dataflow.Protocol_AVRO,
			SchemaRegistry: &dataflow.SchemaRegistry{
				EnableHttpAuth:          true,
				SchemaRegistryEndpoints: "http://abc.com",
				UserName:                "abc",
				Password:                rawPassword,
			},
			EnableTidbExtension:        false,
			DecimalHandlingMode:        dataflow.DecimalHandlingMode_DECIMAL_HANDLING_MODE_STRING,
			BigintUnsignedHandlingMode: dataflow.BigintUnsignedHandlingMode_BIGINT_UNSIGNED_HANDLING_MODE_STRING,
		},
		TableConfig: &dataflow.TableConfig{
			FilterRule: []string{"*.*"},
			Mode:       dataflow.TableMode_FORCE_SYNC,
			EventFilterRule: []*dataflow.EventFilterRule{
				{
					Matcher: []string{"test.*"},
				},
			},
			CaseSensitive: true,
		},
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  startTSO,
		},
		ColumnSelectors: []*dataflow.ColumnSelector{
			{
				Matcher: []string{"a"},
				Columns: []string{"b"},
			},
		},
	}
	expectedChangefeedConfig := &domain.TiCDCChangefeedConfig{
		URIConfig: &domain.TiCDCChangefeedKafkaURIConfigData{
			BrokerEndpoints:                "abc:9092",
			EnableTLS:                      true,
			DefaultTopic:                   "test",
			Protocol:                       utils.KafkaProtocolTypeAvro,
			ReplicationFactor:              1,
			PartitionNum:                   1,
			KafkaVersion:                   utils.KafkaVersion3xx,
			Compression:                    pointer.ToString(utils.CompressionGzip),
			SASLMechanism:                  pointer.ToString(utils.KafkaAuthTypePlain),
			SASLUser:                       pointer.ToString("abc"),
			SASLPassword:                   pointer.ToString(decryptedPassword),
			EnableTiDBExtension:            pointer.ToBool(false),
			AvroDecimalHandlingMode:        pointer.ToString(utils.DecimalHandlingModeString),
			AvroBigintUnsignedHandlingMode: pointer.ToString(utils.BigintUnsignedHandlingModeString),
			AutoCreateTopic:                pointer.ToBool(true),
		},
		Config: &domain.TiCDCChangefeedConfigData{
			StartTs: pointer.ToUint64(startTSO),
			ReplicaConfig: &clusterproxy.ReplicaConfig{
				CaseSensitive:         pointer.ToBool(true),
				EnableOldValue:        pointer.ToBool(true),
				ForceReplicate:        pointer.ToBool(false),
				IgnoreIneligibleTable: pointer.ToBool(false),
				SQLMode:               "",
				Sink: &clusterproxy.SinkConfigV2{
					DispatchRules: []*clusterproxy.DispatchRuleV2{
						{
							Topic:     "prefix_{schema}_{table}_suffix",
							Partition: utils.PartitionTs,
							Matcher:   []string{"*.*"},
						},
					},
					Protocol:       utils.KafkaProtocolTypeAvro,
					SchemaRegistry: "http://abc:password@abc.com",
					ColumnSelectors: []*clusterproxy.ColumnSelector{
						{
							Matcher: []string{"a"},
							Columns: []string{"b"},
						},
					},
					KafkaSink: &clusterproxy.KafkaConfig{
						OutputRawChangeEvent: pointer.ToBool(false),
					},
				},
				Filter: &clusterproxy.FilterConfig{
					Rules: []string{"*.*"},
					EventFilters: []clusterproxy.EventFilterRule{
						{
							Matcher: []string{"test.*"},
						},
					},
				},
			},
		},
	}

	downstreamParamsBytes, err := json.Marshal(kafkaInfo)
	s.Require().NoError(err)
	call := s.mockEncryptor.
		On("DecryptBase64", rawPassword).
		Return(decryptedPassword, nil)
	defer call.Unset()
	changefeedConfig, err := s.kafkaAdaptor.AssembleChangefeedConfig(context.Background(), cluster,
		&entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
	s.Assert().NoError(err)
	s.Assert().Equal(expectedChangefeedConfig, changefeedConfig)
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleChangefeedConfigData() {
	rawPassword := "abc"
	decryptedPassword := "password"
	startTSO := uint64(123123)
	kafkaInfo := &dataflow.KafkaInfo{
		TopicPartitionConfig: &dataflow.TopicPartitionConfig{
			DefaultTopic: "test",
			DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
			TopicPrefix:  "prefix_",
			TopicSuffix:  "_suffix",
			PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
				{
					Matcher:       []string{"*.*"},
					PartitionType: dataflow.PartitionDispatcher_TS,
				},
			},
		},
		DataFormat: &dataflow.DataFormat{
			Protocol: dataflow.Protocol_AVRO,
			SchemaRegistry: &dataflow.SchemaRegistry{
				EnableHttpAuth:          true,
				SchemaRegistryEndpoints: "http://abc.com",
				UserName:                "abc",
				Password:                rawPassword,
			},
			OutputRawChangeEvent: true,
		},
		TableConfig: &dataflow.TableConfig{
			FilterRule: []string{"*.*"},
			Mode:       dataflow.TableMode_FORCE_SYNC,
			EventFilterRule: []*dataflow.EventFilterRule{
				{
					Matcher: []string{"test.*"},
				},
			},
		},
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  startTSO,
		},
		ColumnSelectors: []*dataflow.ColumnSelector{
			{
				Matcher: []string{"a"},
				Columns: []string{"b"},
			},
		},
	}
	expectedChangefeedConfig := &domain.TiCDCChangefeedConfigData{
		StartTs: pointer.ToUint64(startTSO),
		ReplicaConfig: &clusterproxy.ReplicaConfig{
			CaseSensitive:         pointer.ToBool(false),
			EnableOldValue:        pointer.ToBool(true),
			ForceReplicate:        pointer.ToBool(false),
			IgnoreIneligibleTable: pointer.ToBool(false),
			Sink: &clusterproxy.SinkConfigV2{
				DispatchRules: []*clusterproxy.DispatchRuleV2{
					{
						Topic:     "prefix_{schema}_{table}_suffix",
						Partition: utils.PartitionTs,
						Matcher:   []string{"*.*"},
					},
				},
				Protocol:       utils.KafkaProtocolTypeAvro,
				SchemaRegistry: "http://abc:password@abc.com",
				ColumnSelectors: []*clusterproxy.ColumnSelector{
					{
						Matcher: []string{"a"},
						Columns: []string{"b"},
					},
				},
				KafkaSink: &clusterproxy.KafkaConfig{
					OutputRawChangeEvent: pointer.ToBool(true),
				},
			},
			Filter: &clusterproxy.FilterConfig{
				Rules: []string{"*.*"},
				EventFilters: []clusterproxy.EventFilterRule{
					{
						Matcher: []string{"test.*"},
					},
				},
			},
		},
	}

	call := s.mockEncryptor.
		On("DecryptBase64", rawPassword).
		Return(decryptedPassword, nil)
	defer call.Unset()
	changefeedConfig, err := s.kafkaAdaptor.assembleChangefeedConfigData(context.Background(), kafkaInfo)
	s.Assert().NoError(err)
	s.Assert().Equal(expectedChangefeedConfig, changefeedConfig)
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleChangefeedSinkConfigDataFormat() {
	s.Run("avro", func() {
		dataFormat := &dataflow.DataFormat{
			Protocol: dataflow.Protocol_AVRO,
			SchemaRegistry: &dataflow.SchemaRegistry{
				EnableHttpAuth:          false,
				SchemaRegistryEndpoints: "http://abc.com",
			},
		}
		sinkConfig := &clusterproxy.SinkConfigV2{}
		err := s.kafkaAdaptor.assembleChangefeedSinkConfigDataFormat(dataFormat, sinkConfig)
		s.Assert().NoError(err)
		s.Assert().Equal(utils.KafkaProtocolTypeAvro, sinkConfig.Protocol)
		s.Assert().Equal(dataFormat.SchemaRegistry.SchemaRegistryEndpoints, sinkConfig.SchemaRegistry)
	})

	s.Run("debezium", func() {
		dataFormat := &dataflow.DataFormat{
			Protocol: dataflow.Protocol_DEBEZIUM,
			DebeziumConfig: &dataflow.DebeziumConfig{
				OutputOldValue: true,
			},
		}
		sinkConfig := &clusterproxy.SinkConfigV2{}
		err := s.kafkaAdaptor.assembleChangefeedSinkConfigDataFormat(dataFormat, sinkConfig)
		s.Assert().NoError(err)
		s.Assert().Equal(utils.KafkaProtocolDebezium, sinkConfig.Protocol)
		s.Assert().NotNil(sinkConfig.DebeziumConfig.OutputOldValue)
		s.Assert().Equal(dataFormat.DebeziumConfig.OutputOldValue, *sinkConfig.DebeziumConfig.OutputOldValue)
	})

	s.Run("output-raw-change-event", func() {
		dataFormat := &dataflow.DataFormat{
			Protocol:             dataflow.Protocol_CANAL_JSON,
			OutputRawChangeEvent: true,
		}
		sinkConfig := &clusterproxy.SinkConfigV2{}
		err := s.kafkaAdaptor.assembleChangefeedSinkConfigDataFormat(dataFormat, sinkConfig)
		s.Assert().NoError(err)
		s.Assert().Equal(utils.KafkaProtocolTypeCanalJson, sinkConfig.Protocol)
		s.Assert().Equal(dataFormat.OutputRawChangeEvent, *sinkConfig.KafkaSink.OutputRawChangeEvent)
	})
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleChangefeedSinkConfigAvro() {
	testcases := []struct {
		name            string
		schemaRegistry  *dataflow.SchemaRegistry
		decryptPassword bool
		err             bool
		expectedUri     string
	}{
		{
			name:            "schema registry nil",
			decryptPassword: false,
			err:             true,
			expectedUri:     "",
		},
		{
			name: "http auth not enabled",
			schemaRegistry: &dataflow.SchemaRegistry{
				EnableHttpAuth:          false,
				SchemaRegistryEndpoints: "http://abc.com",
			},
			decryptPassword: false,
			err:             false,
			expectedUri:     "http://abc.com",
		},
		{
			name: "http auth enabled but endpoint invalid",
			schemaRegistry: &dataflow.SchemaRegistry{
				EnableHttpAuth:          true,
				SchemaRegistryEndpoints: "abc",
			},
			decryptPassword: false,
			err:             true,
			expectedUri:     "",
		},
		{
			name: "http auth enabled with password",
			schemaRegistry: &dataflow.SchemaRegistry{
				EnableHttpAuth:          true,
				SchemaRegistryEndpoints: "http://abc.com",
				UserName:                "abc",
				Password:                "abc",
			},
			decryptPassword: true,
			err:             false,
			expectedUri:     "http://abc:password@abc.com",
		},
	}

	decryptedPassword := "password"
	for _, tc := range testcases {
		s.Run(tc.name, func() {
			if tc.decryptPassword {
				call := s.mockEncryptor.
					On("DecryptBase64", tc.schemaRegistry.Password).
					Return(decryptedPassword, nil)
				defer call.Unset()
			}
			sinkConfig := &clusterproxy.SinkConfigV2{}
			err := s.kafkaAdaptor.assembleChangefeedSinkConfigAvro(tc.schemaRegistry, sinkConfig)
			if tc.err {
				s.Assert().Error(err)
				return
			}
			s.Assert().NoError(err)
			s.Assert().Equal(tc.expectedUri, sinkConfig.SchemaRegistry)
			if tc.decryptPassword {
				s.mockEncryptor.AssertExpectations(s.T())
			}
		})
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleHttpAuthSchemaRegistry() {
	testcases := []struct {
		name            string
		schemaRegistry  *dataflow.SchemaRegistry
		decryptPassword bool
		err             bool
		expectedUri     string
	}{
		{
			name:            "schema registry nil",
			schemaRegistry:  nil,
			decryptPassword: false,
			err:             true,
			expectedUri:     "",
		},
		{
			name: "schema registry endpoint invalid",
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "a",
				UserName:                "",
				Password:                "",
			},
			decryptPassword: false,
			err:             true,
			expectedUri:     "",
		},
		{
			name: "password empty",
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://abc.com",
				UserName:                "abc",
				Password:                "",
				EnableHttpAuth:          true,
			},
			decryptPassword: false,
			err:             false,
			expectedUri:     "http://abc:@abc.com",
		},
		{
			name: "schema registry with password",
			schemaRegistry: &dataflow.SchemaRegistry{
				SchemaRegistryEndpoints: "http://abc.com",
				UserName:                "abc",
				Password:                "abc",
				EnableHttpAuth:          true,
			},
			decryptPassword: true,
			err:             false,
			expectedUri:     "http://abc:password@abc.com",
		},
	}

	decryptedPassword := "password"
	for _, tc := range testcases {
		s.Run(tc.name, func() {
			if tc.decryptPassword {
				call := s.mockEncryptor.
					On("DecryptBase64", tc.schemaRegistry.Password).
					Return(decryptedPassword, nil)
				defer call.Unset()
			}
			schemaRegistryUri, err := s.kafkaAdaptor.assembleSchemaRegistryURI(tc.schemaRegistry)
			if tc.err {
				s.Assert().Error(err)
				return
			}
			s.Assert().NoError(err)
			s.Assert().Equal(tc.expectedUri, schemaRegistryUri)
			if tc.decryptPassword {
				s.mockEncryptor.AssertExpectations(s.T())
			}
		})
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestAssembleChangefeedSinkConfigDebezium() {
	testcases := []struct {
		debeziumConfig         *dataflow.DebeziumConfig
		expectedOutputOldValue *bool
	}{
		{
			debeziumConfig:         nil,
			expectedOutputOldValue: pointer.ToBool(true),
		},
		{
			debeziumConfig: &dataflow.DebeziumConfig{
				OutputOldValue: true,
			},
			expectedOutputOldValue: pointer.ToBool(true),
		},
		{
			debeziumConfig: &dataflow.DebeziumConfig{
				OutputOldValue: false,
			},
			expectedOutputOldValue: pointer.ToBool(false),
		},
	}

	for _, tc := range testcases {
		sinkConfig := &clusterproxy.SinkConfigV2{}
		s.kafkaAdaptor.assembleChangefeedSinkConfigDebezium(tc.debeziumConfig, sinkConfig)
		if tc.expectedOutputOldValue == nil {
			s.Assert().Nil(sinkConfig.DebeziumConfig.OutputOldValue, "debezium config", tc.debeziumConfig)
		} else {
			s.Assert().Equal(
				*tc.expectedOutputOldValue,
				*sinkConfig.DebeziumConfig.OutputOldValue,
				"debezium config", tc.debeziumConfig)
		}
	}
}

func (s *kafkaDownstreamAdaptorTestSuite) TestCompleteEditParams() {
	currentBrokerEndpoint := "abc:9092"
	currentPassword := "current password"
	currentParams := &dataflow.KafkaInfo{
		Broker: &dataflow.Broker{
			BrokerEndpoints: currentBrokerEndpoint,
		},
		Authentication: &dataflow.Authentication{
			Password: currentPassword,
		},
	}
	currentParamsBytes, err := json.Marshal(currentParams)
	s.Require().NoError(err)
	currentParamsStr := string(currentParamsBytes)

	s.Run("edit params wrong type", func() {
		editParams := &dataflow.MySQLInfo{}
		err := s.kafkaAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Assert().Error(err)
		s.Assert().Contains(err.Error(), "downstreamParams is not")
	})

	s.Run("current params wrong type", func() {
		editParams := &dataflow.KafkaInfo{}
		invalidCurrentParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "abc:3306",
			},
		}
		invalidCurrentParamsBytes, err := json.Marshal(invalidCurrentParams)
		s.Require().NoError(err)
		err = s.kafkaAdaptor.CompleteEditParams(context.Background(), editParams, string(invalidCurrentParamsBytes))
		s.Assert().Error(err)
		s.Assert().Contains(err.Error(), "unmarshal kafka params error")
	})

	s.Run("edit broker endpoint", func() {
		editParams := &dataflow.KafkaInfo{
			Broker: &dataflow.Broker{
				BrokerEndpoints: "def:9093",
			},
			Authentication: &dataflow.Authentication{
				Password: "",
			},
		}
		err = s.kafkaAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Assert().NoError(err)
		s.Assert().Equal(currentBrokerEndpoint, editParams.Broker.BrokerEndpoints)
	})

	s.Run("all passwords empty", func() {
		editParams := &dataflow.KafkaInfo{
			Broker: &dataflow.Broker{
				BrokerEndpoints: currentBrokerEndpoint,
			},
			Authentication: &dataflow.Authentication{
				Password: "",
			},
		}
		err = s.kafkaAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Assert().NoError(err)
		s.Assert().Equal(currentPassword, editParams.Authentication.Password)
	})

	encryptedPassword := "encrypt password"
	s.Run("encrypt password of authentication", func() {
		editParams := &dataflow.KafkaInfo{
			Broker: &dataflow.Broker{
				BrokerEndpoints: currentBrokerEndpoint,
			},
			Authentication: &dataflow.Authentication{
				Password: "abc",
			},
		}
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &editParams.Authentication.Password).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedPassword
			})
		defer call.Unset()

		err = s.kafkaAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Assert().NoError(err)
		s.Assert().Equal(encryptedPassword, editParams.Authentication.Password)
	})

	s.Run("encrypt password of schema registry", func() {
		editParams := &dataflow.KafkaInfo{
			Broker: &dataflow.Broker{
				BrokerEndpoints: currentBrokerEndpoint,
			},
			Authentication: &dataflow.Authentication{
				Password: "",
			},
			DataFormat: &dataflow.DataFormat{
				SchemaRegistry: &dataflow.SchemaRegistry{
					Password: "abc",
				},
			},
		}
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &editParams.DataFormat.SchemaRegistry.Password).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedPassword
			})
		defer call.Unset()

		err = s.kafkaAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Assert().NoError(err)
		s.Assert().Equal(encryptedPassword, editParams.DataFormat.SchemaRegistry.Password)
	})
}

func TestBuildTopicDispatchers(t *testing.T) {
	tests := []struct {
		name          string
		info          *dataflow.TopicPartitionConfig
		expectedTopic string
	}{
		{
			name: "One Topic Dispatch Type",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "",
		},
		{
			name: "Dispatch By Table with Default Separator",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:  "prefix_",
				TopicSuffix:  "_suffix",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "prefix_{schema}_{table}_suffix",
		},
		{
			name: "Dispatch By Table with Custom Separator",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:  "prefix_",
				TopicSuffix:  "_suffix",
				Separator:    ".",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "prefix_{schema}.{table}_suffix",
		},
		{
			name: "Dispatch By Table with empty prefix",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:  "",
				TopicSuffix:  "_suffix",
				Separator:    ".",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "{schema}.{table}_suffix",
		},
		{
			name: "Dispatch By Table with empty suffix",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:  "prefix_",
				TopicSuffix:  "",
				Separator:    ".",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "prefix_{schema}.{table}",
		},
		{
			name: "Dispatch By Table with empty middle",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:  "prefix_",
				TopicSuffix:  "_suffix",
				Separator:    "",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "prefix_{schema}_{table}_suffix",
		},
		{
			name: "Dispatch By Table with all empty",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_TABLE,
				TopicPrefix:  "",
				TopicSuffix:  "",
				Separator:    "",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "{schema}_{table}",
		},
		{
			name: "Dispatch By Database",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE,
				TopicPrefix:  "prefix_",
				TopicSuffix:  "_suffix",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "prefix_{schema}_suffix",
		},
		{
			name: "Dispatch By Database with empty prefix",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE,
				TopicPrefix:  "",
				TopicSuffix:  "_suffix",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "{schema}_suffix",
		},
		{
			name: "Dispatch By Database with empty suffix",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE,
				TopicPrefix:  "prefix_",
				TopicSuffix:  "",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "prefix_{schema}",
		},
		{
			name: "Dispatch By Database with all empty",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_BY_DATABASE,
				TopicPrefix:  "",
				TopicSuffix:  "",
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
					},
				},
			},
			expectedTopic: "{schema}",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dispatchers := buildTopicDispatchers(tt.info)
			assert.Equal(t, 1, len(dispatchers))
			assert.Equal(t, tt.expectedTopic, dispatchers[0].Topic)
		})
	}
}

func TestBuildPartionDispatchers(t *testing.T) {
	tests := []struct {
		name              string
		info              *dataflow.TopicPartitionConfig
		expectedPartition string
	}{
		{
			name: "ts partition type",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TS,
						Matcher:       []string{"*.*"},
						IndexName:     "index_name",
						Columns:       []string{"a", "b"},
					},
				},
			},
			expectedPartition: "ts",
		},
		{
			name: "index value partition type",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_INDEX_VALUE,
						Matcher:       []string{"*.*"},
						IndexName:     "index_name",
						Columns:       []string{"a", "b"},
					},
				},
			},
			expectedPartition: "index-value",
		},
		{
			name: "columns partition type",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_COLUMNS,
						Matcher:       []string{"*.*"},
						IndexName:     "index_name",
						Columns:       []string{"a", "b"},
					},
				},
			},
			expectedPartition: "columns",
		},
		{
			name: "table partition type",
			info: &dataflow.TopicPartitionConfig{
				DefaultTopic: "abc",
				DispatchType: dataflow.DispatchType_DISPATCH_TYPE_ONE_TOPIC,
				PartitionDispatchers: []*dataflow.PartitionDispatcherConfig{
					{
						PartitionType: dataflow.PartitionDispatcher_TABLE,
						Matcher:       []string{"*.*"},
						IndexName:     "index_name",
						Columns:       []string{"a", "b"},
					},
				},
			},
			expectedPartition: "table",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dispatchers := buildTopicDispatchers(tt.info)
			assert.Equal(t, 1, len(dispatchers))
			assert.Equal(t, tt.expectedPartition, dispatchers[0].Partition)
		})
	}
}

func TestConvertColumnSelectors(t *testing.T) {
	testcases := []struct {
		columnSelectors []*dataflow.ColumnSelector
		expectedResult  []*clusterproxy.ColumnSelector
	}{
		{
			columnSelectors: nil,
			expectedResult:  []*clusterproxy.ColumnSelector{},
		},
		{
			columnSelectors: []*dataflow.ColumnSelector{},
			expectedResult:  []*clusterproxy.ColumnSelector{},
		},
		{
			columnSelectors: []*dataflow.ColumnSelector{
				{
					Matcher: []string{"a"},
					Columns: []string{"b"},
				},
			},
			expectedResult: []*clusterproxy.ColumnSelector{
				{
					Matcher: []string{"a"},
					Columns: []string{"b"},
				},
			},
		},
		{
			columnSelectors: []*dataflow.ColumnSelector{
				{
					Matcher: []string{"a"},
					Columns: []string{"b"},
				},
				{
					Matcher: []string{"c", "d"},
					Columns: []string{"e", "f"},
				},
			},
			expectedResult: []*clusterproxy.ColumnSelector{
				{
					Matcher: []string{"a"},
					Columns: []string{"b"},
				},
				{
					Matcher: []string{"c", "d"},
					Columns: []string{"e", "f"},
				},
			},
		},
	}

	for _, tc := range testcases {
		result := convertColumnSelectors(tc.columnSelectors)
		assert.Equal(t, tc.expectedResult, result)
	}
}

func TestKafkaDownstreamAdaptorTestSuite(t *testing.T) {
	suite.Run(t, new(kafkaDownstreamAdaptorTestSuite))
}