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/require"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_mock "github.com/tidbcloud/dataflow-service/app/models/ticdc/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	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 s3DownstreamAdaptorTestSuite struct {
	suite.Suite
	s3Adaptor *s3DownstreamAdaptor

	mockPreCheckCli     *regionalapi_mock.TiCDCPreCheckClient
	mockCDCClusterModel *ticdc_mock.ClusterModel
	mockEncryptor       *encryptor_mock.Encryptor
}

func (s *s3DownstreamAdaptorTestSuite) SetupSuite() {
	s.mockPreCheckCli = regionalapi_mock.NewTiCDCPreCheckClient(s.T())
	s.mockCDCClusterModel = ticdc_mock.NewClusterModel(s.T())
	s.mockEncryptor = encryptor_mock.NewEncryptor(s.T())

	s.s3Adaptor = &s3DownstreamAdaptor{
		cdcPreCheckCli:  s.mockPreCheckCli,
		cdcClusterModel: s.mockCDCClusterModel,
		encryptor:       s.mockEncryptor,
	}
}

func (s *s3DownstreamAdaptorTestSuite) TestParseS3Params() {
	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.S3Info{},
			expectedError:    true,
		},
		{
			name:             "right type param",
			downstreamParams: &dataflow.S3Info{},
			expectedError:    false,
		},
	}

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

func (s *s3DownstreamAdaptorTestSuite) TestParseS3ParamsStr() {
	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.S3Info{
				Endpoint: &dataflow.S3Endpoint{
					S3Uri: "s3://bucket/key",
				},
			},
			expectedError: false,
		},
	}

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

func (s *s3DownstreamAdaptorTestSuite) 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.s3Adaptor.TestConnection(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	mockDownstreamParams := &dataflow.S3Info{
		Endpoint: &dataflow.S3Endpoint{
			S3Uri: "s3://bucket/key",
		},
	}
	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.s3Adaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "us-west-2",
		TenantID:      1,
		CloudProvider: "aws",
	}
	expectedReq := &ticdc_v1.TestNetworkConnectionRequest{
		ClusterId: "111",
		Type:      "s3",
		S3Param: &ticdc_v1.TestNetworkConnectionRequest_S3Parameter{
			S3Uri: "s3://bucket/key",
		},
	}
	call := s.mockPreCheckCli.
		On("TestNetworkConnection", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("test connection successfully", func() {
		err := s.s3Adaptor.TestConnection(context.Background(), cluster, mockDownstreamParams)
		s.NoError(err)
	})
}

func (s *s3DownstreamAdaptorTestSuite) 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.s3Adaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

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

	downstreamParams := &dataflow.S3Info{
		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.s3Adaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Error(err)
	})

	expectedRegion := &regionalapi.Region{
		Name:          "us-west-2",
		TenantID:      1,
		CloudProvider: "aws",
	}
	expectedReq := &ticdc_v1.ValidateChangefeedParametersRequest{
		ClusterId: "111",
		Type:      "s3",
		StartTso:  123123,
	}
	call := s.mockPreCheckCli.
		On("ValidateChangefeedParameters", mock.Anything, expectedRegion, expectedReq).
		Return(nil)
	defer call.Unset()
	s.Run("pre check successfully", func() {
		err := s.s3Adaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.NoError(err)
	})
}

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

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

	downstreamParams := &dataflow.S3Info{
		Endpoint: &dataflow.S3Endpoint{
			S3Uri:     "s3://bucket/key",
			AccessKey: "abc",
			SecretKey: "",
		},
	}
	s.Run("password empty", func() {
		err := s.s3Adaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.NoError(err)
	})

	downstreamParams.Endpoint.SecretKey = "def"
	s.Run("encrypt secret key error", func() {
		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt secret key error")
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", mock.Anything).
			Return(encryptErr)
		defer call.Unset()

		err := s.s3Adaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Error(err)
		s.Contains(err.Error(), encryptErr.ErrMsg())
	})

	s.Run("encrypt successfully", func() {
		encryptedSecretKey := base64.StdEncoding.EncodeToString([]byte(downstreamParams.Endpoint.SecretKey))
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Endpoint.SecretKey).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedSecretKey
			})
		defer call.Unset()

		err := s.s3Adaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.NoError(err)
		s.Equal(encryptedSecretKey, downstreamParams.Endpoint.SecretKey)
	})
}

func (s *s3DownstreamAdaptorTestSuite) TestAssembleURIConfig() {
	mockCluster := &core.Cluster{}
	s.Run("decrypt secret key error", func() {
		s3Params := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
				SecretKey: "def",
			},
		}
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "decrypt secret key error")
		call := s.mockEncryptor.
			On("DecryptBase64", mock.Anything).
			Return("", mockErr)
		defer call.Unset()
		_, err := s.s3Adaptor.assembleURIConfig(context.Background(), s3Params, mockCluster)
		s.Error(err)
		s.Contains(err.Error(), "decrypt secret key")
	})

	s.Run("get external id error", func() {
		s3Params := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				AuthType:          dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE,
				IamAssumedRoleArn: "role123",
			},
		}
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockCDCClusterModel.
			On("GetRole", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.s3Adaptor.assembleURIConfig(context.Background(), s3Params, mockCluster)
		s.Error(err)
		s.Contains(err.Error(), "get external id")
	})

	s.Run("normal uri config with ak sk", func() {
		s3Params := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/key",
				AuthType:  dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK,
				AccessKey: "abc",
				SecretKey: "def",
			},
			StorageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "5m",
				FileSize:      1000,
			},
			DataFormat: &dataflow.S3Dataformat{
				Protocol:            dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				EnableTidbExtension: true,
			},
		}
		decryptedSecretKey := "decrypted secret key"
		call := s.mockEncryptor.
			On("DecryptBase64", s3Params.Endpoint.SecretKey).
			Return(decryptedSecretKey, nil)
		defer call.Unset()

		expectedURIConfig := &domain.TiCDCChangefeedS3URIConfigData{
			S3URI:                "s3://bucket/key",
			AccessKey:            pointer.ToString("abc"),
			SecretAccessKey:      pointer.ToString(decryptedSecretKey),
			ForcePathStyle:       pointer.ToBool(true),
			FlushInterval:        "5m",
			FileSize:             1000,
			TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
			EnableTiDBExtension:  pointer.ToBool(true),
		}

		uriConfig, err := s.s3Adaptor.assembleURIConfig(context.Background(), s3Params, mockCluster)
		s.NoError(err)
		s.Equal(expectedURIConfig, uriConfig)
	})

	s.Run("normal uri config with iam role", func() {
		s3Params := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:             "s3://bucket/key",
				AuthType:          dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE,
				IamAssumedRoleArn: "role123",
			},
			StorageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "5m",
				FileSize:      1000,
			},
			DataFormat: &dataflow.S3Dataformat{
				Protocol:            dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
				EnableTidbExtension: true,
			},
		}
		mockRoleResp := &domain.GetTiCDCClusterRoleResp{
			AWSExternalID: "123",
		}
		call := s.mockCDCClusterModel.
			On("GetRole", mock.Anything, mock.Anything).
			Return(mockRoleResp, nil)
		defer call.Unset()

		expectedURIConfig := &domain.TiCDCChangefeedS3URIConfigData{
			S3URI:                "s3://bucket/key",
			RoleARN:              pointer.ToString("role123"),
			ExternalID:           pointer.ToString("123"),
			ForcePathStyle:       pointer.ToBool(true),
			FlushInterval:        "5m",
			FileSize:             1000,
			TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
			EnableTiDBExtension:  pointer.ToBool(true),
		}

		uriConfig, err := s.s3Adaptor.assembleURIConfig(context.Background(), s3Params, mockCluster)
		s.NoError(err)
		s.Equal(expectedURIConfig, uriConfig)
	})
}

func (s *s3DownstreamAdaptorTestSuite) TestAssembleChangefeedConfig() {
	cluster := &core.Cluster{}
	bizCf := &entity.BizChangefeed{}
	s.Run("parse s3 params error", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "abc:3306",
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		_, err = s.s3Adaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "unmarshal s3 params")
	})

	s.Run("assemble uri config error", func() {
		downstreamParams := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				SecretKey: "def",
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "decrypt secret key error")
		call := s.mockEncryptor.
			On("DecryptBase64", mock.Anything).
			Return("", mockErr)
		defer call.Unset()

		_, err = s.s3Adaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), mockErr.ErrMsg())
	})

	decryptedSecretKey := "decrypted secret key"
	call := s.mockEncryptor.
		On("DecryptBase64", mock.Anything).
		Return(decryptedSecretKey, nil)
	defer call.Unset()

	s.Run("assemble changefeed config error", func() {
		downstreamParams := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/key",
				AccessKey: "abc",
				SecretKey: "def",
			},
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		_, err = s.s3Adaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.Error(err)
		s.Contains(err.Error(), "invalid start tso")
	})

	s.Run("assemble changefeed config successfully", func() {
		downstreamParams := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/key",
				AccessKey: "abc",
				SecretKey: "def",
			},
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  123123123,
			},
			DataFormat: &dataflow.S3Dataformat{
				Terminator:    ";",
				DateSeparator: "day",
				Protocol:      dataflow.S3Protocol_S3_PROTOCOL_CSV,
			},
			StorageFlushConfig: &dataflow.StorageFlushConfig{
				FlushInterval: "5m",
				FileSize:      1000,
			},
			TableConfig: &dataflow.TableConfig{
				FilterRule: []string{"*.*"},
				EventFilterRule: []*dataflow.EventFilterRule{
					{
						Matcher:     []string{"test.*"},
						IgnoreEvent: []string{"delete"},
					},
				},
				CaseSensitive: true,
				Mode:          dataflow.TableMode_FORCE_SYNC,
			},
		}
		expectedConfig := &domain.TiCDCChangefeedConfig{
			URIConfig: &domain.TiCDCChangefeedS3URIConfigData{
				S3URI:                "s3://bucket/key",
				AccessKey:            pointer.ToString("abc"),
				SecretAccessKey:      pointer.ToString(decryptedSecretKey),
				ForcePathStyle:       pointer.ToBool(true),
				FlushInterval:        "5m",
				FileSize:             1000,
				TransactionAtomicity: pointer.ToString(defaultTransactionAtomicity),
			},
			Config: &domain.TiCDCChangefeedConfigData{
				StartTs: pointer.ToUint64(123123123),
				ReplicaConfig: &clusterproxy.ReplicaConfig{
					CaseSensitive:         pointer.ToBool(true),
					ForceReplicate:        pointer.ToBool(true),
					IgnoreIneligibleTable: pointer.ToBool(false),
					Sink: &clusterproxy.SinkConfigV2{
						Terminator:    ";",
						DateSeparator: "day",
						Protocol:      "csv",
						CSVConfig: &clusterproxy.CSVConfig{
							Delimiter:            ",",
							Quote:                "\"",
							NullString:           "\\N",
							IncludeCommitTs:      pointer.ToBool(false),
							BinaryEncodingMethod: BinaryEncodingMethodBase64,
							OutputFieldHeader:    pointer.ToBool(false),
						},
						CloudStorageConfig: &clusterproxy.CloudStorageConfig{
							OutputRawChangeEvent: pointer.ToBool(false),
						},
					},
					Filter: &clusterproxy.FilterConfig{
						Rules: []string{"*.*"},
						EventFilters: []clusterproxy.EventFilterRule{
							{
								Matcher:     []string{"test.*"},
								IgnoreEvent: []string{"delete"},
							},
						},
					},
				},
			},
		}
		downstreamParamsBytes, err := json.Marshal(downstreamParams)
		s.NoError(err)
		bizCf.DownstreamParams = string(downstreamParamsBytes)
		changefeedConfig, err := s.s3Adaptor.AssembleChangefeedConfig(context.Background(), cluster, bizCf)
		s.NoError(err)
		s.Equal(expectedConfig, changefeedConfig)
	})
}

func (s *s3DownstreamAdaptorTestSuite) TestCompleteEditParams() {
	invalidCurrentParamsBytes, _ := json.Marshal(&dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: "abc:3306",
		},
	})
	currentParamsBytes, _ := json.Marshal(&dataflow.S3Info{
		Endpoint: &dataflow.S3Endpoint{
			S3Uri:     "s3://bucket/key",
			AccessKey: "abc",
			SecretKey: "def",
		},
	})
	s.Run("invalid edit params", func() {
		editParams := &dataflow.MySQLInfo{}
		err := s.s3Adaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), "downstreamParams is not *dataflow.S3Info type")
	})

	s.Run("invalid current params", func() {
		editParams := &dataflow.S3Info{}
		err := s.s3Adaptor.CompleteEditParams(context.Background(), editParams, string(invalidCurrentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), "unmarshal s3 params error")
	})

	s.Run("endpoint is changed and assume role auth type", func() {
		editParams := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:    "s3://bucket-2/key",
				AuthType: dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE,
			},
		}
		err := s.s3Adaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.NoError(err)
		s.Equal("s3://bucket/key", editParams.Endpoint.S3Uri)
		s.Empty(editParams.Endpoint.SecretKey)
	})

	s.Run("endpoint is changed and secret key is empty", func() {
		editParams := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri: "s3://bucket-2/key",
			},
		}
		err := s.s3Adaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.NoError(err)
		s.Equal("s3://bucket/key", editParams.Endpoint.S3Uri)
		s.Equal("def", editParams.Endpoint.SecretKey)
	})

	s.Run("encrypt secret key error", func() {
		editParams := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/key",
				AccessKey: "abc",
				SecretKey: "ghi",
			},
		}
		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt secret key error")
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", mock.Anything).
			Return(encryptErr)
		defer call.Unset()
		err := s.s3Adaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.Error(err)
		s.Contains(err.Error(), encryptErr.ErrMsg())
	})

	s.Run("encrypt secret key successfully", func() {
		editParams := &dataflow.S3Info{
			Endpoint: &dataflow.S3Endpoint{
				S3Uri:     "s3://bucket/key",
				AccessKey: "abc",
				SecretKey: "ghi",
			},
		}
		encryptedSecretKey := base64.StdEncoding.EncodeToString([]byte(editParams.Endpoint.SecretKey))
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &editParams.Endpoint.SecretKey).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedSecretKey
			})
		defer call.Unset()
		err := s.s3Adaptor.CompleteEditParams(context.Background(), editParams, string(currentParamsBytes))
		s.NoError(err)
		s.Equal(encryptedSecretKey, editParams.Endpoint.SecretKey)
	})
}

func (s *s3DownstreamAdaptorTestSuite) TestGetExternalID() {
	mockCluster := &core.Cluster{
		TenantId:          1,
		ProjectId:         11,
		Id:                111,
		RegionName:        "us-west-2",
		CloudProviderName: "aws",
	}
	s.Run("get cdc cluster role error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockCDCClusterModel.
			On("GetRole", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		_, err := s.s3Adaptor.getExternalID(context.Background(), mockCluster)
		s.Error(err)
		s.Contains(err.Error(), "get ticdc cluster role")
	})

	s.Run("get external id successfully", func() {
		expectedReq := &domain.GetTiCDCClusterRoleReq{
			TenantID:      1,
			ProjectID:     11,
			TiDBClusterID: 111,
			Region:        "us-west-2",
			CloudProvider: domain.CloudProviderTypeAWS,
		}
		mockResp := &domain.GetTiCDCClusterRoleResp{
			CloudIAMIdentity: "role123",
			CloudAccountID:   "123123",
			AWSExternalID:    "123",
		}
		call := s.mockCDCClusterModel.
			On("GetRole", mock.Anything, expectedReq).
			Return(mockResp, nil)
		defer call.Unset()
		externalID, err := s.s3Adaptor.getExternalID(context.Background(), mockCluster)
		s.NoError(err)
		s.Equal("123", externalID)
	})
}

func TestAssembleCSVConfig(t *testing.T) {
	t.Run("empty data format", func(t *testing.T) {
		dataFormat := &dataflow.S3Dataformat{}
		expectedCSVConfig := &clusterproxy.CSVConfig{
			Delimiter:            ",",
			Quote:                "\"",
			NullString:           "\\N",
			IncludeCommitTs:      pointer.ToBool(false),
			BinaryEncodingMethod: BinaryEncodingMethodBase64,
			OutputFieldHeader:    pointer.ToBool(false),
		}
		csvConfig := assembleCSVConfig(dataFormat)
		assert.Equal(t, expectedCSVConfig, csvConfig)
	})

	t.Run("normal data format", func(t *testing.T) {
		dataFormat := &dataflow.S3Dataformat{
			Delimiter:            ";",
			Quote:                "'",
			Null:                 "null",
			IncludeCommitTs:      true,
			BinaryEncodingMethod: dataflow.BinaryEncodingMethod_HEX,
			OutputFieldHeader:    true,
		}
		expectedCSVConfig := &clusterproxy.CSVConfig{
			Delimiter:            ";",
			Quote:                "'",
			NullString:           "null",
			IncludeCommitTs:      pointer.ToBool(true),
			BinaryEncodingMethod: BinaryEncodingMethodHex,
			OutputFieldHeader:    pointer.ToBool(true),
		}
		csvConfig := assembleCSVConfig(dataFormat)
		assert.Equal(t, expectedCSVConfig, csvConfig)
	})
}

func TestAssembleChangefeedConfigData(t *testing.T) {
	t.Run("parse start tso error", func(t *testing.T) {
		startPosition := &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_UTC,
			Utc:  time.Now().Format(time.RFC1123),
		}
		dataFormat := &dataflow.S3Dataformat{
			Terminator:    ";",
			DateSeparator: "day",
			Protocol:      dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
		}
		tableConfig := &dataflow.TableConfig{
			FilterRule: []string{"*.*"},
			EventFilterRule: []*dataflow.EventFilterRule{
				{
					Matcher:     []string{"test.*"},
					IgnoreEvent: []string{"delete"},
				},
			},
		}
		_, err := assembleCloudStorageChangefeedConfigData(tableConfig, startPosition, dataFormat)
		assert.Error(t, err)
	})

	t.Run("canal-json protocol", func(t *testing.T) {
		startPosition := &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  123123123,
		}
		dataFormat := &dataflow.S3Dataformat{
			Terminator:    ";",
			DateSeparator: "day",
			Protocol:      dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
		}
		tableConfig := &dataflow.TableConfig{
			FilterRule: []string{"*.*"},
			EventFilterRule: []*dataflow.EventFilterRule{
				{
					Matcher:     []string{"test.*"},
					IgnoreEvent: []string{"delete"},
				},
			},
			Mode: dataflow.TableMode_FORCE_SYNC,
		}
		expectedConfig := &domain.TiCDCChangefeedConfigData{
			StartTs: pointer.ToUint64(123123123),
			ReplicaConfig: &clusterproxy.ReplicaConfig{
				CaseSensitive:         pointer.ToBool(false),
				ForceReplicate:        pointer.ToBool(true),
				IgnoreIneligibleTable: pointer.ToBool(false),
				Sink: &clusterproxy.SinkConfigV2{
					Terminator:    ";",
					DateSeparator: "day",
					Protocol:      "canal-json",
					CloudStorageConfig: &clusterproxy.CloudStorageConfig{
						OutputRawChangeEvent: pointer.ToBool(false),
					},
				},
				Filter: &clusterproxy.FilterConfig{
					Rules: []string{"*.*"},
					EventFilters: []clusterproxy.EventFilterRule{
						{
							Matcher:     []string{"test.*"},
							IgnoreEvent: []string{"delete"},
						},
					},
				},
			},
		}
		config, err := assembleCloudStorageChangefeedConfigData(tableConfig, startPosition, dataFormat)
		assert.NoError(t, err)
		assert.Equal(t, expectedConfig, config)
	})

	t.Run("csv protocol", func(t *testing.T) {
		startPosition := &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  123123123,
		}
		dataFormat := &dataflow.S3Dataformat{
			Terminator:           ";",
			DateSeparator:        "day",
			Protocol:             dataflow.S3Protocol_S3_PROTOCOL_CSV,
			OutputRawChangeEvent: true,
		}
		tableConfig := &dataflow.TableConfig{
			FilterRule: []string{"*.*"},
			EventFilterRule: []*dataflow.EventFilterRule{
				{
					Matcher:     []string{"test.*"},
					IgnoreEvent: []string{"delete"},
				},
			},
			Mode: dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
		}
		expectedConfig := &domain.TiCDCChangefeedConfigData{
			StartTs: pointer.ToUint64(123123123),
			ReplicaConfig: &clusterproxy.ReplicaConfig{
				CaseSensitive:         pointer.ToBool(false),
				ForceReplicate:        pointer.ToBool(false),
				IgnoreIneligibleTable: pointer.ToBool(true),
				Sink: &clusterproxy.SinkConfigV2{
					Terminator:    ";",
					DateSeparator: "day",
					Protocol:      "csv",
					CSVConfig: &clusterproxy.CSVConfig{
						Delimiter:            ",",
						Quote:                "\"",
						NullString:           "\\N",
						IncludeCommitTs:      pointer.ToBool(false),
						BinaryEncodingMethod: BinaryEncodingMethodBase64,
						OutputFieldHeader:    pointer.ToBool(false),
					},
					CloudStorageConfig: &clusterproxy.CloudStorageConfig{
						OutputRawChangeEvent: pointer.ToBool(true),
					},
				},
				Filter: &clusterproxy.FilterConfig{
					Rules: []string{"*.*"},
					EventFilters: []clusterproxy.EventFilterRule{
						{
							Matcher:     []string{"test.*"},
							IgnoreEvent: []string{"delete"},
						},
					},
				},
			},
		}
		config, err := assembleCloudStorageChangefeedConfigData(tableConfig, startPosition, dataFormat)
		assert.NoError(t, err)
		assert.Equal(t, expectedConfig, config)
	})
}

func TestNormalizeCloudStorageDateSeparator(t *testing.T) {
	tests := map[string]struct {
		input       string
		expected    string
		expectedErr bool
	}{
		"blank defaults to none": {expected: DateSeparatorNone},
		"none":                   {input: "none", expected: DateSeparatorNone},
		"mixed case and whitespace year": {
			input:    "  YeAr ",
			expected: DateSeparatorYear,
		},
		"mixed case month": {input: "MONTH", expected: DateSeparatorMonth},
		"day":              {input: "day", expected: DateSeparatorDay},
		"invalid value":    {input: "week", expectedErr: true},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			dataFormat := &dataflow.S3Dataformat{DateSeparator: tc.input}
			config, err := assembleCloudStorageChangefeedConfigData(
				&dataflow.TableConfig{},
				&dataflow.StartPosition{Mode: dataflow.StartMode_FROM_TSO, Tso: 1},
				dataFormat,
			)
			if tc.expectedErr {
				require.Error(t, err)
				require.Equal(t, bizerr.ErrCodeParams, err.ErrCode())
				return
			}
			require.NoError(t, err)
			require.Equal(t, tc.expected, dataFormat.GetDateSeparator())
			require.Equal(t, tc.expected, config.ReplicaConfig.Sink.DateSeparator)
		})
	}
}

func TestS3DownstreamAdaptorTestSuite(t *testing.T) {
	suite.Run(t, new(s3DownstreamAdaptorTestSuite))
}