package impl

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

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/dataflow-service-ng/app/config"
	"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"
	encryptor_mock "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper/mock"
	regional_mock "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/mock/gomock"
)

type mysqlDownstreamAdaptorTestSuite struct {
	suite.Suite
	mysqlAdaptor *mysqlDownstreamAdaptor

	cdcPreCheckCli *regional_mock.TiCDCPreCheckClient
	mockEncryptor  *encryptor_mock.Encryptor
}

func (s *mysqlDownstreamAdaptorTestSuite) SetupSuite() {
	s.cdcPreCheckCli = regional_mock.NewTiCDCPreCheckClient(s.T())
	s.mockEncryptor = encryptor_mock.NewEncryptor(s.T())

	s.mysqlAdaptor = &mysqlDownstreamAdaptor{
		cdcPreCheckCli: s.cdcPreCheckCli,
		encryptor:      s.mockEncryptor,
	}
}

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

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

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

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

func (s *mysqlDownstreamAdaptorTestSuite) TestParseTestMySQLConnectionError() {
	accessDeniedErrMsg := "err: Access denied for user"
	otherErrMsg := "err: no such host"
	testcases := []struct {
		name             string
		err              error
		expectedKeywords string
	}{
		{
			name:             "nil error",
			err:              nil,
			expectedKeywords: "",
		},
		{
			name:             "access denied error",
			err:              errors.New(accessDeniedErrMsg),
			expectedKeywords: config.ErrMsgConnectToMySQL_AccessDenied,
		},
		{
			name:             "other error",
			err:              errors.New(otherErrMsg),
			expectedKeywords: otherErrMsg,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			err := s.mysqlAdaptor.parseTestMySQLConnectionError(tc.err)
			if tc.expectedKeywords == "" {
				s.Assert().NoError(err)
			} else {
				s.Assert().Error(err)
				s.Assert().Contains(err.Error(), tc.expectedKeywords)
			}
		})
	}
}

func (s *mysqlDownstreamAdaptorTestSuite) TestTestConnection() {
	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		OrganizationId: 1,
	})

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

	s.Run("test connection access denied error", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "example.com:3306",
				Username: "test_user",
				Password: "test_password",
			},
		}

		connErr := bizerr.New(config.ErrCodeCDCFailedToConnectErr, config.ErrMsgConnectToMySQL_AccessDenied)
		call := s.cdcPreCheckCli.On("TestNetworkConnection", mock.Anything, mock.Anything, mock.Anything).
			Return(connErr)
		defer call.Unset()

		err := s.mysqlAdaptor.TestConnection(context.Background(), cluster, downstreamParams)
		s.Assert().Error(err)
		s.Assert().Contains(err.Error(), config.ErrMsgConnectToMySQL_AccessDenied)
		s.Assert().Contains(err.Error(), fmt.Sprintf(config.ErrMsgCDCFailedToConnectErr,
			DownstreamDisplayNameMapping[dataflow.ChangefeedType_MYSQL], ""))
		s.cdcPreCheckCli.AssertExpectations(s.T())
	})

	s.Run("test connection successfully", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Endpoint: "valid.host:3306",
				Username: "valid_user",
				Password: "valid_password",
			},
		}

		call := s.cdcPreCheckCli.On("TestNetworkConnection", mock.Anything, mock.Anything, mock.Anything).
			Return(nil)
		defer call.Unset()

		err := s.mysqlAdaptor.TestConnection(context.Background(), cluster, downstreamParams)
		s.Assert().NoError(err)
		s.cdcPreCheckCli.AssertExpectations(s.T())
	})
}

func (s *mysqlDownstreamAdaptorTestSuite) TestPreCheckChangefeedParameters() {
	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		OrganizationId: 1,
	})

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

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

	s.Run("no need to check start tso", func() {
		downstreamParams := &dataflow.MySQLInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_NOW,
			},
		}
		err := s.mysqlAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Assert().NoError(err)
	})

	var mockTSO uint64 = 123123
	s.Run("check failed", func() {
		downstreamParams := &dataflow.MySQLInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  mockTSO,
			},
		}
		call := s.cdcPreCheckCli.
			On("ValidateChangefeedParameters",
				mock.Anything, mock.Anything, mock.Anything).
			Return(bizerr.New(config.ErrCodeCDCInvalidParam, "failed to validate start tso"))
		defer call.Unset()

		err := s.mysqlAdaptor.PreCheckChangefeedParameters(context.Background(), cluster, downstreamParams)
		s.Assert().Error(err)
		s.cdcPreCheckCli.AssertExpectations(s.T())
	})

	s.Run("check successful", func() {
		downstreamParams := &dataflow.MySQLInfo{
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  mockTSO,
			},
		}
		ctrl := gomock.NewController(s.T())
		defer ctrl.Finish()
		call := s.cdcPreCheckCli.
			On("ValidateChangefeedParameters",
				mock.Anything, mock.Anything, mock.Anything).
			Return(nil)
		defer call.Unset()

		err := s.mysqlAdaptor.PreCheckChangefeedParameters(context.Background(),
			cluster, downstreamParams)
		s.Assert().NoError(err)
		s.cdcPreCheckCli.AssertExpectations(s.T())
	})
}

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

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

	s.Run("password empty", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Password: "",
			},
		}
		err := s.mysqlAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().NoError(err)
	})

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

		err := s.mysqlAdaptor.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.MySQLInfo{
			Connection: &dataflow.Connection{
				Password: "abc",
			},
		}
		encryptedPassword := base64.StdEncoding.EncodeToString([]byte(downstreamParams.Connection.Password))
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &downstreamParams.Connection.Password).
			Return(nil).
			Run(func(args mock.Arguments) {
				password := args.Get(0).(*string)
				*password = encryptedPassword
			})
		defer call.Unset()

		err := s.mysqlAdaptor.EncryptPasswordInPlace(context.Background(), downstreamParams)
		s.Assert().NoError(err)
		s.Assert().Equal(encryptedPassword, downstreamParams.Connection.Password)
		s.mockEncryptor.AssertExpectations(s.T())
	})
}

func (s *mysqlDownstreamAdaptorTestSuite) TestAssembleURIConfigData() {
	testcases := []struct {
		name             string
		username         string
		password         string
		endpoint         string
		mockDecryptError *bool
		expectedError    bool
	}{
		{
			name:             "all empty",
			username:         "",
			password:         "",
			endpoint:         "",
			mockDecryptError: nil,
			expectedError:    false,
		},
		{
			name:             "username is not empty",
			username:         "user1",
			password:         "",
			endpoint:         "",
			mockDecryptError: nil,
			expectedError:    false,
		},
		{
			name:             "password is not empty",
			username:         "user1",
			password:         "password1",
			endpoint:         "",
			mockDecryptError: pointer.ToBool(false),
			expectedError:    false,
		},
		{
			name:             "decrypt password error",
			username:         "user1",
			password:         "password1",
			endpoint:         "",
			mockDecryptError: pointer.ToBool(true),
			expectedError:    true,
		},
		{
			name:             "endpoint is not empty",
			username:         "user1",
			password:         "password1",
			endpoint:         "endpoint1",
			mockDecryptError: pointer.ToBool(false),
			expectedError:    false,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			decryptedPassword := "decrypted password1"
			if tc.mockDecryptError != nil {
				if *tc.mockDecryptError {
					decryptErr := bizerr.New(bizerr.ErrCodeUnknown, "decrypt error")
					call := s.mockEncryptor.
						On("DecryptBase64", tc.password).
						Return("", decryptErr)
					defer call.Unset()
				} else {
					call := s.mockEncryptor.
						On("DecryptBase64", tc.password).
						Return(decryptedPassword, nil)
					defer call.Unset()
				}
			}

			mysqlParams := &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Username: tc.username,
					Password: tc.password,
					Endpoint: tc.endpoint,
				},
			}
			uriConfig, err := s.mysqlAdaptor.assembleURIConfigData(context.Background(), mysqlParams)
			if tc.expectedError {
				s.Assert().Error(err)
				return
			}
			s.Assert().NoError(err)
			s.Assert().Equal(defaultTransactionAtomicity, *uriConfig.TransactionAtomicity)
			if tc.username == "" {
				s.Assert().Nil(uriConfig.User)
			} else {
				s.Assert().Equal(tc.username, *uriConfig.User)
			}
			if tc.password == "" {
				s.Assert().Nil(uriConfig.Password)
			} else {
				s.Assert().Equal(decryptedPassword, *uriConfig.Password)
			}
			if tc.endpoint == "" {
				s.Assert().Nil(uriConfig.Endpoint)
			} else {
				s.Assert().Equal(tc.endpoint, *uriConfig.Endpoint)
			}
		})
	}
}

func (s *mysqlDownstreamAdaptorTestSuite) TestAssembleChangefeedConfigData() {
	testcases := []struct {
		name              string
		tableMode         dataflow.TableMode
		filterRules       []string
		eventFilterRules  []*dataflow.EventFilterRule
		parseStarTSOError bool
	}{
		{
			name:              "parse start tso error",
			parseStarTSOError: true,
		},
		{
			name:             "force sink table mode",
			tableMode:        dataflow.TableMode_FORCE_SYNC,
			filterRules:      []string{},
			eventFilterRules: []*dataflow.EventFilterRule{},
		},
		{
			name:             "ignore not support table mode",
			tableMode:        dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			filterRules:      []string{},
			eventFilterRules: []*dataflow.EventFilterRule{},
		},
		{
			name:             "filter rules are nil",
			tableMode:        dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			filterRules:      nil,
			eventFilterRules: []*dataflow.EventFilterRule{},
		},
		{
			name:             "event filter rules are nil",
			tableMode:        dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			filterRules:      []string{},
			eventFilterRules: nil,
		},
		{
			name:             "filter rules are not empty",
			tableMode:        dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			filterRules:      []string{"a", "b"},
			eventFilterRules: []*dataflow.EventFilterRule{},
		},
		{
			name:        "event filter rules are not empty",
			tableMode:   dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			filterRules: []string{"a", "b"},
			eventFilterRules: []*dataflow.EventFilterRule{
				{
					IgnoreInsertValueExpr: "a",
				},
				{
					IgnoreInsertValueExpr: "b",
				},
			},
		},
	}

	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		ProjectId:      11,
		OrganizationId: 1,
	})

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			mysqlParams := &dataflow.MySQLInfo{
				TableConfig: &dataflow.TableConfig{
					Mode:            tc.tableMode,
					FilterRule:      tc.filterRules,
					EventFilterRule: tc.eventFilterRules,
				},
			}
			if tc.parseStarTSOError {
				mysqlParams.StartPosition = &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_UTC,
					Utc:  time.Now().Format(time.RFC1123),
				}
			} else {
				mysqlParams.StartPosition = &dataflow.StartPosition{
					Mode: dataflow.StartMode_FROM_NOW,
				}
			}

			changefeedConfig, err := s.mysqlAdaptor.assembleChangefeedConfigData(
				context.Background(), cluster, mysqlParams)
			if tc.parseStarTSOError {
				s.Assert().Error(err)
				return
			}
			s.Assert().NoError(err)
			s.Assert().Equal(uint64(0), *changefeedConfig.StartTs)
			s.Assert().Nil(changefeedConfig.TargetTs)
			replicaConfig := changefeedConfig.ReplicaConfig
			s.Assert().Equal(true, *replicaConfig.EnableOldValue)
			s.Assert().Equal(tc.tableMode == dataflow.TableMode_FORCE_SYNC, *replicaConfig.ForceReplicate)
			s.Assert().Equal(
				tc.tableMode == dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE, *replicaConfig.IgnoreIneligibleTable)
			filterConfig := replicaConfig.Filter
			s.Assert().Equal(len(tc.filterRules), len(filterConfig.Rules))
			s.Assert().Equal(len(tc.eventFilterRules), len(filterConfig.EventFilters))
		})
	}
}

func (s *mysqlDownstreamAdaptorTestSuite) TestAssembleChangefeedConfig() {
	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      111,
		OrganizationId: 1,
	})
	s.Run("downstream param error", func() {
		downstreamParams := &dataflow.KafkaInfo{
			Broker: &dataflow.Broker{
				BrokerEndpoints: "aa",
			},
		}
		downstreamParamsBytes, jsonErr := json.Marshal(downstreamParams)
		s.Assert().NoError(jsonErr)
		_, err := s.mysqlAdaptor.AssembleChangefeedConfig(context.Background(), cluster,
			&entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
		s.Assert().Error(err)
	})

	s.Run("assemble uri config data error", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Password: "abc",
			},
		}
		decryptErr := bizerr.New(bizerr.ErrCodeUnknown, "decrypt error")
		call := s.mockEncryptor.
			On("DecryptBase64", downstreamParams.Connection.Password).
			Return("", decryptErr)
		defer call.Unset()

		downstreamParamsBytes, jsonErr := json.Marshal(downstreamParams)
		s.Assert().NoError(jsonErr)
		_, err := s.mysqlAdaptor.AssembleChangefeedConfig(context.Background(), cluster,
			&entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
		s.Assert().Error(err)
		s.Assert().Contains(err.Error(), decryptErr.ErrMsg())
	})

	s.Run("assemble changefeed config data error", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Password: "",
			},
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  time.Now().Format(time.RFC1123),
			},
		}
		downstreamParamsBytes, jsonErr := json.Marshal(downstreamParams)
		s.Assert().NoError(jsonErr)
		_, err := s.mysqlAdaptor.AssembleChangefeedConfig(context.Background(), cluster,
			&entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
		s.Assert().Error(err)
	})

	s.Run("assemble changefeed config successfully", func() {
		downstreamParams := &dataflow.MySQLInfo{
			Connection: &dataflow.Connection{
				Password: "",
			},
			StartPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_NOW,
			},
			TableConfig: &dataflow.TableConfig{
				FilterRule:      []string{},
				EventFilterRule: []*dataflow.EventFilterRule{},
				Mode:            dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			},
		}
		downstreamParamsBytes, jsonErr := json.Marshal(downstreamParams)
		s.Assert().NoError(jsonErr)
		changefeedConfig, err := s.mysqlAdaptor.AssembleChangefeedConfig(context.Background(), cluster,
			&entity.BizChangefeed{DownstreamParams: string(downstreamParamsBytes)})
		s.Assert().NoError(err)
		s.Assert().NotNil(changefeedConfig.URIConfig)
		s.Assert().NotNil(changefeedConfig.Config)
	})
}

func (s *mysqlDownstreamAdaptorTestSuite) TestManageEditParams() {
	invalidParams := &dataflow.KafkaInfo{
		Broker: &dataflow.Broker{
			BrokerEndpoints: "aa",
		},
	}
	endpoint := "mysql-endpoint"
	currentParams := &dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: endpoint,
			Username: "a",
			Password: "password",
		},
	}
	editParams := &dataflow.MySQLInfo{
		Connection: &dataflow.Connection{
			Endpoint: endpoint,
			Username: "b",
		},
	}
	currentParamsBytes, err := json.Marshal(currentParams)
	s.NoError(err)
	currentParamsStr := string(currentParamsBytes)

	s.Run("parse edit params error", func() {
		err := s.mysqlAdaptor.CompleteEditParams(context.Background(), invalidParams, currentParamsStr)
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
	})

	s.Run("parse current params error", func() {
		invalidParamsBytes, jsonErr := json.Marshal(invalidParams)
		s.NoError(jsonErr)
		err := s.mysqlAdaptor.CompleteEditParams(context.Background(), editParams, string(invalidParamsBytes))
		s.Error(err)
		s.Equal(bizerr.ErrCodeUnknown, err.RawErrCode())
	})

	s.Run("endpoints are not equal", func() {
		editParams.Connection.Endpoint = "changed"
		err := s.mysqlAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.NoError(err)
		s.Equal(currentParams.Connection.Endpoint, editParams.Connection.Endpoint)
	})

	editParams.Connection.Password = ""
	s.Run("password is empty", func() {
		err := s.mysqlAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.NoError(err)
		s.Equal(currentParams.Connection.Password, editParams.Connection.Password)
	})

	editParams.Connection.Password = "changed"
	s.Run("password is not empty, encrypt error", func() {
		encryptErr := bizerr.New(bizerr.ErrCodeUnknown, "encrypt error")
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &editParams.Connection.Password).
			Return(encryptErr)
		defer call.Unset()

		err := s.mysqlAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.Error(err)
		s.Equal(encryptErr.RawErrCode(), err.RawErrCode())
		s.Contains(err.Error(), "encrypt password")
		s.mockEncryptor.AssertExpectations(s.T())
	})

	s.Run("password is not empty, encrypt successfully", func() {
		call := s.mockEncryptor.
			On("EncryptBase64Inplace", &editParams.Connection.Password).
			Return(nil)
		defer call.Unset()

		err := s.mysqlAdaptor.CompleteEditParams(context.Background(), editParams, currentParamsStr)
		s.NoError(err)
		s.mockEncryptor.AssertExpectations(s.T())
	})
}

func TestMySQLDownstreamAdaptorTestSuite(t *testing.T) {
	suite.Run(t, new(mysqlDownstreamAdaptorTestSuite))
}