package config

import (
	"context"
	"fmt"
	"runtime"
	"slices"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	apprent "github.com/tidbcloud/devops/entity/approval"
	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	maintent "github.com/tidbcloud/devops/entity/maintenance"
	tntent "github.com/tidbcloud/devops/entity/tenant"
	cccli "github.com/tidbcloud/devops/pkg/infra/clients/configcentercli"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	dfcli "github.com/tidbcloud/devops/pkg/infra/clients/dataflowcli"
	"github.com/tidbcloud/devops/pkg/infra/log"
	"github.com/tidbcloud/devops/pkg/infra/utils/pointer"
	"github.com/tidbcloud/devops/pkg/infra/utils/test"
	apprsvcapi "github.com/tidbcloud/devops/pkg/service/approval/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/change/config/applier"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/repo"
	cmxapi "github.com/tidbcloud/devops/pkg/service/clustermaintenance/api"
	tntsvc "github.com/tidbcloud/devops/pkg/service/tenant"

	gk "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/testing/protocmp"
	"google.golang.org/protobuf/types/known/structpb"
)

func TestRepo(t *testing.T) {
	RegisterFailHandler(gk.Fail)
	gk.RunSpecs(t, "ConfigChange Actor Suite", test.ReporterConfig())
}

var _ = gk.Describe("[ConfigChange Actor]", func() {
	var fw = test.Framework()
	var ctx context.Context
	var comps Components
	var ori *ccent.ConfigurationChange

	gk.BeforeEach(func() {
		ctx = log.NewContext(context.Background(), fw.Logger)
		comps = Components{
			Repo: repo.NewMockCloudConfigRepo(fw.Ctrl),
			ConfigCenter: &cccli.Client{
				ConfigAPI: cccli.NewMockConfigCenterServiceClient(fw.Ctrl),
			},
			CoreSvc: &coresvccli.CoreSvcCli{
				Core: coresvccli.NewMockCoreServiceClient(fw.Ctrl),
				Mgmt: &coresvccli.MgmtService{
					Cluster:             coresvccli.NewMockClusterServiceClient(fw.Ctrl),
					ClusterCustomStatus: coresvccli.NewMockClusterCustomStatusServiceClient(fw.Ctrl),
				},
			},
			Dataflow: &dfcli.DataflowCli{
				Ops: dfcli.NewMockOpsInternalServiceClient(fw.Ctrl),
			},
			ApprovalService: apprsvcapi.NewMockApprovalServiceAPI(fw.Ctrl),
			MaintenanceSvc:  cmxapi.NewMockServiceAPI(fw.Ctrl),
			TenantService:   tntsvc.NewMockTenantServiceAPI(fw.Ctrl),
			ConfigService:   api.NewMockCloudConfigServiceAPI(fw.Ctrl),
		}
		ori = &ccent.ConfigurationChange{}
	})

	gk.Context("[Preview]", func() {
		var param api.PreviewConfigurationChangeParam
		var clusterVer string

		gk.BeforeEach(func() {
			param = api.PreviewConfigurationChangeParam{
				CreateConfigurationChangeBaseParam: api.CreateConfigurationChangeBaseParam{
					Files: []*api.ChangeFileParam{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tidb.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: "warn",
									},
								},
								{
									Name: "tidb.config.2",
									ParameterValue: ccent.ParameterValue{
										Value: 1,
									},
								},
							},
						},
						{
							Filename:   "tikv",
							ConfigType: "tikv",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tikv.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: 111.1,
									},
								},
								{
									Name: "tikv.config.2",
									ParameterValue: ccent.ParameterValue{
										Value: "custom",
									},
								},
							},
						},
					},
					TargetCluster: &ccent.ClusterConfigurationLocator{
						ClusterType: ccent.ClusterTypeDedicatedTiDB,
						ClusterID:   "1234",
					},
				},
			}
			clusterVer = "v1.0.0"
			comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
				GetClusterMeta(gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccproto.ClusterMeta{Version: clusterVer}, nil)
		})

		gk.It("unsupported phases", func() {
			for _, phase := range ccent.ConfigurationChangePhaseValues {
				ori.Phase = phase
				actor := NewConfigurationChangeActor(ctx, comps, ori)

				result, err := actor.Preview(param)
				Expect(result).To(BeNil())
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("unsupported action when phase is"))
			}
		})

		gk.It("return 'HotReload' when all parameters are hotreloadable", func() {
			actor := NewConfigurationChangeActor(ctx, comps, nil)
			comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
				GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccent.Configuration{
					Files: map[string]*ccent.FileConfiguration{
						"tidb": {
							Filename:   "tidb",
							ConfigType: "tidb",
							Parameters: map[string]*ccent.UnifiedParameter{
								"tidb.config.2": {
									UsedParameter: ccent.UsedParameter{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
							},
						},
						"tikv": {
							Filename:   "tikv",
							ConfigType: "tikv",
							Parameters: map[string]*ccent.UnifiedParameter{},
						},
					},
				}, nil)
			comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
				GetTemplate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccent.Template{}, nil)
			comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
				ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccproto.ListConfigurationDictItemsResponse{
					ConfigurationDictItems: []*ccproto.ConfigurationDictItem{ // all modified paramters are hotreloadable
						{
							ConfigurationType: "tidb",
							Version:           clusterVer,
							ParamKey:          "tidb.config.1",
							HotReloadable:     pointer.Pointer(true),
						},
						{
							ConfigurationType: "tikv",
							Version:           clusterVer,
							ParamKey:          "tikv.config.1",
							HotReloadable:     pointer.Pointer(true),
						},
						{
							ConfigurationType: "tikv",
							Version:           clusterVer,
							ParamKey:          "tikv.config.2",
							HotReloadable:     pointer.Pointer(true),
						},
					},
				}, nil)

			result, err := actor.Preview(param)
			Expect(err).To(BeNil())
			Expect(result.ApplyPolicies).To(Equal([]ccent.ParameterApplyPolicy{
				ccent.ParameterApplyPolicyHotReload,
				ccent.ParameterApplyPolicyRestart,
				ccent.ParameterApplyPolicyEnsureUpdateToTC,
			}))
			Expect(result.Reason).NotTo(BeEmpty())
		})

		gk.It("return 'Restart' when any parameter is not hotreloadable", func() {
			actor := NewConfigurationChangeActor(ctx, comps, nil)
			comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
				GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccent.Configuration{
					Files: map[string]*ccent.FileConfiguration{
						"tidb": {
							Filename:   "tidb",
							ConfigType: "tidb",
							Parameters: map[string]*ccent.UnifiedParameter{
								"tidb.config.2": {
									UsedParameter: ccent.UsedParameter{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
							},
						},
						"tikv": {
							Filename:   "tikv",
							ConfigType: "tikv",
							Parameters: map[string]*ccent.UnifiedParameter{},
						},
					},
				}, nil)
			comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
				GetTemplate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccent.Template{}, nil)
			comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
				ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccproto.ListConfigurationDictItemsResponse{
					ConfigurationDictItems: []*ccproto.ConfigurationDictItem{
						{
							ConfigurationType: "tidb",
							Version:           clusterVer,
							ParamKey:          "tidb.config.1",
							HotReloadable:     pointer.Pointer(true),
						},
						{
							ConfigurationType: "tikv",
							Version:           clusterVer,
							ParamKey:          "tikv.config.1",
							HotReloadable:     pointer.Pointer(false), // not hotreloadable
						},
						{
							ConfigurationType: "tikv",
							Version:           clusterVer,
							ParamKey:          "tikv.config.2",
							HotReloadable:     pointer.Pointer(true),
						},
					},
				}, nil)

			result, err := actor.Preview(param)
			Expect(err).To(BeNil())
			Expect(result.ApplyPolicies).To(Equal([]ccent.ParameterApplyPolicy{
				ccent.ParameterApplyPolicyRestart,
				ccent.ParameterApplyPolicyEnsureUpdateToTC,
			}))
			Expect(result.Reason).NotTo(BeEmpty())
		})

		gk.It("return 'Restart' when any parameter is missing in dict", func() {
			actor := NewConfigurationChangeActor(ctx, comps, nil)
			comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
				GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccent.Configuration{
					Files: map[string]*ccent.FileConfiguration{
						"tidb": {
							Filename:   "tidb",
							ConfigType: "tidb",
							Parameters: map[string]*ccent.UnifiedParameter{
								"tidb.config.2": {
									UsedParameter: ccent.UsedParameter{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
							},
						},
						"tikv": {
							Filename:   "tikv",
							ConfigType: "tikv",
							Parameters: map[string]*ccent.UnifiedParameter{},
						},
					},
				}, nil)
			comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
				GetTemplate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccent.Template{}, nil)
			comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
				ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
				Return(&ccproto.ListConfigurationDictItemsResponse{
					ConfigurationDictItems: []*ccproto.ConfigurationDictItem{ // miss tikv.config.1
						{
							ConfigurationType: "tidb",
							Version:           clusterVer,
							ParamKey:          "tidb.config.1",
							HotReloadable:     pointer.Pointer(true),
						},
						{
							ConfigurationType: "tikv",
							Version:           clusterVer,
							ParamKey:          "tikv.config.2",
							HotReloadable:     pointer.Pointer(true),
						},
					},
				}, nil)

			result, err := actor.Preview(param)
			Expect(err).To(BeNil())
			Expect(result.ApplyPolicies).To(Equal([]ccent.ParameterApplyPolicy{
				ccent.ParameterApplyPolicyRestart,
				ccent.ParameterApplyPolicyEnsureUpdateToTC,
			}))
			Expect(result.Reason).NotTo(BeEmpty())
		})

	})

	gk.Context("[Create]", func() {
		var actor ConfigurationChangeActor
		var param api.CreateConfigurationChangeParam

		gk.BeforeEach(func() {
			actor = NewConfigurationChangeActor(ctx, comps, nil)
			param = api.CreateConfigurationChangeParam{}
		})

		gk.It("unsupported phases", func() {
			for _, phase := range ccent.ConfigurationChangePhaseValues {
				ori.Phase = phase
				actor := NewConfigurationChangeActor(ctx, comps, ori)

				err := actor.Create(param)
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("unsupported action when phase is"))
			}
		})

		gk.Context("[TenantInit]", func() {
			var cur *ccent.Configuration

			gk.BeforeEach(func() {
				param.TargetTenantInit = &ccent.TenantInitConfigurationLocator{
					ClusterType: ccent.ClusterTypeDedicatedTiDB,
					TenantID:    1111,
				}
				// add mock for GetDefaultVersion to avoid unexpected call
				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					GetDefaultVersion(gomock.Any(), gomock.Any()).AnyTimes().
					Return(&ccproto.DefaultVersion{Version: "v7.0.0"}, nil)
				cur = &ccent.Configuration{
					Version: "v7.0.0",
					Files: map[string]*ccent.FileConfiguration{
						"tidb": {
							Filename:   "tidb",
							ConfigType: "tidb",
							Parameters: map[string]*ccent.UnifiedParameter{
								"tidb.config.2": {
									UsedParameter: ccent.UsedParameter{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
							},
						},
					},
				}

				comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
					GetTenantInitConfiguration(gomock.Any(), gomock.Any()).AnyTimes().
					Return(cur, nil)
			})

			gk.It("return error when a change is in progress", func() {
				comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
					ListConfigurationChanges(gomock.Any(), gomock.Any()).AnyTimes().
					Return([]*ccent.ConfigurationChange{
						{
							ID: 100,
						},
					}, nil)

				err := actor.Create(param)
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("some tenant-init configuration changes are already in progress"))
			})

			gk.Context("create a change", func() {

				gk.BeforeEach(func() {
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						ListConfigurationChanges(gomock.Any(), gomock.Any()).AnyTimes().
						Return(nil, nil)
				})

				gk.It("with file change", func() {
					param.CreateConfigurationChangeBaseParam.Files = []*api.ChangeFileParam{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tidb.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: "warn",
									},
								},
								{
									Name: "tidb.config.2",
									ParameterValue: ccent.ParameterValue{
										Value: 1,
									},
								},
							},
						},
						{
							Filename:   "tikv",
							ConfigType: "tikv",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tikv.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: 111.1,
									},
								},
								{
									Name: "tikv.config.2",
									ParameterValue: ccent.ParameterValue{
										Value: "custom",
									},
								},
							},
						},
					}
					exp := ccent.ConfigurationChangeScope{
						Files: []*ccent.ConfigurationFileChange{
							{
								Filename:   "tidb",
								ConfigType: "tidb",
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: "warn",
											},
										},
										Source: ccent.ParameterCustom,
									},
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tidb.config.1",
										From:       nil,
										To:         "warn",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
							{
								Filename:   "tikv",
								ConfigType: "tikv",
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tikv.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: 111.1,
											},
										},
										Source: ccent.ParameterCustom,
									},
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tikv.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: "custom",
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tikv.config.1",
										From:       nil,
										To:         111.1,
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
									{
										Name:       "tikv.config.2",
										From:       nil,
										To:         "custom",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
						},
						Components: nil, // not change components
						Version:    nil, // not change version
					}

					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{
							ConfigurationDictItems: []*ccproto.ConfigurationDictItem{
								{
									ConfigurationType: "tidb",
									ParamKey:          "tidb.config.1",
									HotReloadable:     pointer.Pointer(true),
								},
								{
									ConfigurationType: "tikv",
									ParamKey:          "tikv.config.2",
									HotReloadable:     pointer.Pointer(true),
								},
							},
						}, nil)
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).AnyTimes().
						DoAndReturn(func(ctx context.Context, change *ccent.ConfigurationChange) (*ccent.ConfigurationChange, error) {
							SortChange(change)
							Expect(test.IsEqual(exp, change.Scope)).To(BeNil())
							return nil, nil
						})
					err := actor.Create(param)
					Expect(err).To(BeNil())
				})

				gk.It("with version change", func() {
					param.CreateConfigurationChangeBaseParam.Version = &api.ChangeVersionParam{
						ToVersion: "v8.0.0",
					}
					exp := ccent.ConfigurationChangeScope{
						Version: &ccent.ConfigurationVersionChange{
							From: "v7.0.0",
							To:   "v8.0.0",
						},
					}

					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						ListAvailableComponentVersions(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return([]string{"v8.0.0"}, nil)
					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{
							ConfigurationDictItems: []*ccproto.ConfigurationDictItem{
								{
									ConfigurationType: "tidb",
									ParamKey:          "tidb.config.1",
									HotReloadable:     pointer.Pointer(true),
								},
								{
									ConfigurationType: "tikv",
									ParamKey:          "tikv.config.2",
									HotReloadable:     pointer.Pointer(true),
								},
							},
						}, nil)
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).AnyTimes().
						DoAndReturn(
							func(ctx context.Context, change *ccent.ConfigurationChange) (*ccent.ConfigurationChange, error) {
								SortChange(change)
								Expect(test.IsEqual(exp, change.Scope)).To(BeNil())
								return nil, nil
							},
						)
					err := actor.Create(param)
					Expect(err).To(BeNil())
				})

				gk.It("with execution window", func() {
					param.CreateConfigurationChangeBaseParam.Files = []*api.ChangeFileParam{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tidb.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: "warn",
									},
								},
							},
						},
					}
					param.ExecutionWindow = &cmxapi.CreateOpsExecutionWindowParam{
						StartAt: 100,
						EndAt:   200,
					}

					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{}, nil)
					comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
						CreateOpsExecutionWindow(gomock.Any(), *param.ExecutionWindow).
						Return(&maintent.OpsExecutionWindow{ID: 123, StartAt: 100, EndAt: 200}, nil)
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).
						DoAndReturn(func(ctx context.Context, change *ccent.ConfigurationChange) (*ccent.ConfigurationChange, error) {
							Expect(change.ExecutionWindowID).To(Equal(pointer.Pointer(uint64(123))))
							return change, nil
						})

					err := actor.Create(param)
					Expect(err).To(BeNil())
					Expect(actor.GetContext().Change.ExecutionWindowID).To(Equal(pointer.Pointer(uint64(123))))
				})

				gk.It("returns error when execution window creation failed", func() {
					param.CreateConfigurationChangeBaseParam.Files = []*api.ChangeFileParam{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tidb.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: "warn",
									},
								},
							},
						},
					}
					param.ExecutionWindow = &cmxapi.CreateOpsExecutionWindowParam{
						StartAt: 100,
						EndAt:   200,
					}

					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{}, nil)
					comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
						CreateOpsExecutionWindow(gomock.Any(), *param.ExecutionWindow).
						Return(nil, fmt.Errorf("create execution window failed"))
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).
						Times(0)

					err := actor.Create(param)
					Expect(err).To(HaveOccurred())
					Expect(err.Error()).To(ContainSubstring("create execution window failed"))
				})

			})

		})

		gk.Context("[Cluster]", func() {
			var loc *ccent.ClusterConfigurationLocator
			var cur *ccent.Configuration

			gk.BeforeEach(func() {
				loc = &ccent.ClusterConfigurationLocator{
					ClusterType: ccent.ClusterTypeDedicatedTiDB,
					ClusterID:   "1234",
				}
				cur = &ccent.Configuration{
					TargetCluster: loc,
					Components: map[string]*ccent.Component{
						"tidb": {
							Name:              "tidb",
							EffectiveImageTag: "v7.0.0",
							OverwriteImageTag: "v7.0.0",
						},
						"tikv": {
							Name:              "tikv",
							EffectiveImageTag: "v7.0.0",
							OverwriteImageTag: "v7.0.0",
						},
					},
					Files: map[string]*ccent.FileConfiguration{
						"tidb": {
							Filename:          "tidb",
							ConfigType:        "tidb",
							BelongToComponent: "tidb",
							Parameters: map[string]*ccent.UnifiedParameter{
								"tidb.config.2": {
									UsedParameter: ccent.UsedParameter{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
									Custom: &ccent.ParameterValue{
										Value: 1,
									},
								},
							},
						},
						"tikv": {
							Filename:          "tikv",
							ConfigType:        "tikv",
							BelongToComponent: "tikv",
							Parameters:        map[string]*ccent.UnifiedParameter{},
						},
					},
				}
				param.TargetCluster = loc
			})

			gk.It("return error when a change is in progress", func() {
				comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
					GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(cur, nil)
				comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
					ListConfigurationChanges(gomock.Any(), gomock.Any()).AnyTimes().
					Return([]*ccent.ConfigurationChange{
						{
							ID: 100,
						},
					}, nil)

				err := actor.Create(param)
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("some cluster configuration changes are already in progress"))
			})

			gk.Context("create a change", func() {

				gk.BeforeEach(func() {
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						ListConfigurationChanges(gomock.Any(), gomock.Any()).AnyTimes().
						Return(nil, nil)
				})

				gk.It("with file change", func() {
					param.CreateConfigurationChangeBaseParam.Files = []*api.ChangeFileParam{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tidb.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: "warn",
									},
								},
								{
									Name: "tidb.config.2",
									ParameterValue: ccent.ParameterValue{
										Value: 1,
									},
								},
							},
						},
						{
							Filename:   "tikv",
							ConfigType: "tikv",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tikv.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: 111.1,
									},
								},
								{
									Name: "tikv.config.2",
									ParameterValue: ccent.ParameterValue{
										Value: "custom",
									},
								},
							},
						},
					}
					exp := ccent.ConfigurationChangeScope{
						Files: []*ccent.ConfigurationFileChange{
							{
								Filename:   "tidb",
								ConfigType: "tidb",
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: "warn",
											},
										},
										Source: ccent.ParameterCustom,
									},
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tidb.config.1",
										From:       nil,
										To:         "warn",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
							{
								Filename:   "tikv",
								ConfigType: "tikv",
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tikv.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: 111.1,
											},
										},
										Source: ccent.ParameterCustom,
									},
									{
										NamedParameter: ccent.NamedParameter{
											Name: "tikv.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: "custom",
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tikv.config.1",
										From:       nil,
										To:         111.1,
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
									{
										Name:       "tikv.config.2",
										From:       nil,
										To:         "custom",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
						},
						Components: nil, // not change components
						Version:    nil, // not change version
					}

					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(cur, nil)
					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetTemplate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccent.Template{}, nil)
					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{
							ConfigurationDictItems: []*ccproto.ConfigurationDictItem{ // miss tikv.config.1
								{
									ConfigurationType: "tidb",
									ParamKey:          "tidb.config.1",
									HotReloadable:     pointer.Pointer(true),
								},
								{
									ConfigurationType: "tikv",
									ParamKey:          "tikv.config.2",
									HotReloadable:     pointer.Pointer(true),
								},
							},
						}, nil)
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).AnyTimes().
						DoAndReturn(
							func(ctx context.Context, change *ccent.ConfigurationChange) (*ccent.ConfigurationChange, error) {
								SortChange(change)
								Expect(test.IsEqual(exp, change.Scope)).To(BeNil())
								return nil, nil
							},
						)

					err := actor.Create(param)
					Expect(err).To(BeNil())
				})

				gk.It("with version change", func() {
					param.CreateConfigurationChangeBaseParam.Version = &api.ChangeVersionParam{
						ToVersion: "v8.0.0",
					}
					defTemp := &ccent.Template{
						ID:   101,
						Name: "default",
						Parameters: map[string]*ccent.NamedParameter{
							"tidb.config.1": {
								Name: "tidb.config.1",
								ParameterValue: ccent.ParameterValue{
									Value: "warn",
								},
							},
							"tidb.config.2": {
								Name: "tidb.config.2",
								ParameterValue: ccent.ParameterValue{
									Value: 100,
								},
							},
						},
					}
					exp := ccent.ConfigurationChangeScope{
						Components: []*ccent.ConfigurationComponentChange{
							{
								Component: "tidb",
								ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
									From: "v7.0.0",
									To:   "v8.0.0",
								},
								ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
									From: "v7.0.0",
									To:   "",
								},
							},
							{
								Component: "tikv",
								ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
									From: "v7.0.0",
									To:   "v8.0.0",
								},
								ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
									From: "v7.0.0",
									To:   "",
								},
							},
						},
						Files: []*ccent.ConfigurationFileChange{
							{
								Filename:   "tidb",
								ConfigType: "tidb",
								TemplateID: 101,
								ModifiedTemplate: &ccent.ChangedTemplate{
									From: 0,
									To:   101,
								},
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{ // new parameter from template
											Name: "tidb.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: "warn",
											},
										},
										Source: ccent.ParameterTemplate,
									},
									{
										NamedParameter: ccent.NamedParameter{ // existing parameter from custom
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 1,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tidb.config.1",
										From:       nil,
										To:         "warn",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterTemplate,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
							{
								Filename:   "tikv",
								ConfigType: "tikv",
								TemplateID: 101,
								ModifiedTemplate: &ccent.ChangedTemplate{
									From: 0,
									To:   101,
								},
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{ // new parameter from template
											Name: "tidb.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: "warn",
											},
										},
										Source: ccent.ParameterTemplate,
									},
									{
										NamedParameter: ccent.NamedParameter{ // existing parameter from template
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 100,
											},
										},
										Source: ccent.ParameterTemplate,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tidb.config.1",
										From:       nil,
										To:         "warn",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterTemplate,
										RiskLevel:  ccent.RiskLevelHigh,
									},
									{
										Name:       "tidb.config.2",
										From:       nil,
										To:         100,
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterTemplate,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
						},
						Version: &ccent.ConfigurationVersionChange{
							To: "v8.0.0",
						},
					}

					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						ListAvailableComponentVersions(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return([]string{"v8.0.0"}, nil)
					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(cur, nil)
					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetTemplateByDefaultCase(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(defTemp, nil)
					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetTemplate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(defTemp, nil)
					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						GetComponentImage(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ComponentImage{
							Tag: "v8.0.0",
						}, nil)
					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{}, nil)
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).AnyTimes().
						DoAndReturn(
							func(ctx context.Context, change *ccent.ConfigurationChange) (*ccent.ConfigurationChange, error) {
								SortChange(change)
								Expect(test.IsEqual(exp, change.Scope)).To(BeNil())
								return nil, nil
							},
						)
					comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
						ListClusters(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&mgmtv1.ListClustersResponse{
							Clusters: []*mgmtv1.Cluster{
								{},
							},
						}, nil)

					err := actor.Create(param)
					Expect(err).To(BeNil())
				})

				gk.It("with version/image/file change", func() {
					param.CreateConfigurationChangeBaseParam.Version = &api.ChangeVersionParam{
						ToVersion: "v8.0.0",
					}
					param.Components = []*api.ChangeComponentParam{
						{
							Component: "tidb",
							OverwriteImageTag: &api.ChangeImageTagParam{
								ToImageTag: "v6.0.0",
							},
						},
					}
					param.Files = []*api.ChangeFileParam{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							CustomParameters: []*ccent.NamedParameter{
								{
									Name: "tidb.config.1",
									ParameterValue: ccent.ParameterValue{
										Value: "info",
									},
								},
								{
									Name: "tidb.config.2",
									ParameterValue: ccent.ParameterValue{
										Value: 10010,
									},
								},
							},
						},
					}
					defTemp := &ccent.Template{
						ID:   101,
						Name: "default",
						Parameters: map[string]*ccent.NamedParameter{
							"tidb.config.1": {
								Name: "tidb.config.1",
								ParameterValue: ccent.ParameterValue{
									Value: "warn",
								},
							},
							"tidb.config.2": {
								Name: "tidb.config.2",
								ParameterValue: ccent.ParameterValue{
									Value: 100,
								},
							},
						},
					}
					exp := ccent.ConfigurationChangeScope{
						Components: []*ccent.ConfigurationComponentChange{
							{
								Component: "tidb",
								ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as overwrite_image_tag
									From: "v7.0.0",
									To:   "v6.0.0",
								},
								ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be changed
									From: "v7.0.0",
									To:   "v6.0.0",
								},
							},
							{
								Component: "tikv",
								ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as overwrite_image_tag
									From: "v7.0.0",
									To:   "v8.0.0",
								},
								ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be changed
									From: "v7.0.0",
									To:   "",
								},
							},
						},
						Files: []*ccent.ConfigurationFileChange{
							{
								Filename:   "tidb",
								ConfigType: "tidb",
								TemplateID: 101,
								ModifiedTemplate: &ccent.ChangedTemplate{
									From: 0,
									To:   101,
								},
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{ // new parameter from custom
											Name: "tidb.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: "info",
											},
										},
										Source: ccent.ParameterCustom,
									},
									{
										NamedParameter: ccent.NamedParameter{ // existing parameter from custom
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 10010,
											},
										},
										Source: ccent.ParameterCustom,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tidb.config.1",
										From:       nil,
										To:         "info",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
									{
										Name:       "tidb.config.2",
										From:       1,
										To:         10010,
										FromSource: ccent.ParameterCustom,
										ToSource:   ccent.ParameterCustom,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
							{
								Filename:   "tikv",
								ConfigType: "tikv",
								TemplateID: 101,
								ModifiedTemplate: &ccent.ChangedTemplate{
									From: 0,
									To:   101,
								},
								AllParameters: []ccent.UsedParameter{
									{
										NamedParameter: ccent.NamedParameter{ // new parameter from template
											Name: "tidb.config.1",
											ParameterValue: ccent.ParameterValue{
												Value: "warn",
											},
										},
										Source: ccent.ParameterTemplate,
									},
									{
										NamedParameter: ccent.NamedParameter{ // existing parameter from template
											Name: "tidb.config.2",
											ParameterValue: ccent.ParameterValue{
												Value: 100,
											},
										},
										Source: ccent.ParameterTemplate,
									},
								},
								ModifiedParameters: []ccent.ChangedParameter{
									{
										Name:       "tidb.config.1",
										From:       nil,
										To:         "warn",
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterTemplate,
										RiskLevel:  ccent.RiskLevelHigh,
									},
									{
										Name:       "tidb.config.2",
										From:       nil,
										To:         100,
										FromSource: ccent.ParameterKernelDefault,
										ToSource:   ccent.ParameterTemplate,
										RiskLevel:  ccent.RiskLevelHigh,
									},
								},
							},
						},
						Version: &ccent.ConfigurationVersionChange{
							To: "v8.0.0",
						},
					}

					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						ListAvailableComponentVersions(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return([]string{"v8.0.0"}, nil)
					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(cur, nil)
					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetTemplateByDefaultCase(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(defTemp, nil)
					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetTemplate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(defTemp, nil)
					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{}, nil)
					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						GetComponentImage(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ComponentImage{
							Tag: "v8.0.0",
						}, nil)
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).AnyTimes().
						DoAndReturn(
							func(ctx context.Context, change *ccent.ConfigurationChange) (*ccent.ConfigurationChange, error) {
								SortChange(change)
								Expect(test.IsEqual(exp, change.Scope)).To(BeNil())
								return nil, nil
							},
						)
					comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
						ListClusters(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&mgmtv1.ListClustersResponse{
							Clusters: []*mgmtv1.Cluster{
								{},
							},
						}, nil)

					err := actor.Create(param)
					Expect(err).To(BeNil())
				})

				gk.It("with image tag change", func() {
					param.CreateConfigurationChangeBaseParam.Components = []*api.ChangeComponentParam{
						{
							Component: "tidb",
							OverwriteImageTag: &api.ChangeImageTagParam{
								ToImageTag: "v6.0.0",
							},
						},
					}
					exp := ccent.ConfigurationChangeScope{
						Components: []*ccent.ConfigurationComponentChange{
							{
								Component: "tidb",
								ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
									From: "v7.0.0",
									To:   "v6.0.0",
								},
								ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
									From: "v7.0.0",
									To:   "v6.0.0",
								},
							},
						},
					}

					comps.ConfigService.(*api.MockCloudConfigServiceAPI).EXPECT().
						GetClusterConfiguration(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
						Return(cur, nil)
					comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
						ListConfigurationDictItems(gomock.Any(), gomock.Any()).AnyTimes().
						Return(&ccproto.ListConfigurationDictItemsResponse{}, nil)
					comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
						CreateConfigurationChange(gomock.Any(), gomock.Any()).AnyTimes().
						DoAndReturn(
							func(ctx context.Context, change *ccent.ConfigurationChange) (*ccent.ConfigurationChange, error) {
								SortChange(change)
								Expect(test.IsEqual(exp, change.Scope)).To(BeNil())
								return nil, nil
							},
						)

					err := actor.Create(param)
					Expect(err).To(BeNil())
				})

			})

		})

	})

	gk.Context("[SendApproval]", func() {
		var tenant tntent.Tenant

		gk.BeforeEach(func() {
			ori.Phase = ccent.ConfigurationChangePhaseCreated
			tenant = tntent.Tenant{
				Id:               1001,
				Name:             "tenant-1001",
				SalesForceRegion: tntent.TenantRegionAPAC,
				CountryRegion:    tntent.TenantRegionAPAC,
			}
			comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
				PatchConfigurationChange(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(nil).AnyTimes()
		})

		gk.It("unsupported phases", func() {
			for _, phase := range ccent.ConfigurationChangePhaseValues {
				ori.Phase = phase
				actor := NewConfigurationChangeActor(ctx, comps, ori)

				err := actor.SendApproval()
				switch phase {
				case ccent.ConfigurationChangePhaseCreated:
				default:
					Expect(err).To(HaveOccurred())
					Expect(err.Error()).To(ContainSubstring("unsupported action when phase is"))
				}
			}
		})

		gk.Context("For Cluster", func() {
			gk.BeforeEach(func() {
				ori.TargetType = ccent.ConfigurationChangeTargetTypeCluster
				ori.ChangeType = "dedicated"
				ori.TargetCluster = &ccent.ClusterConfigurationLocator{
					ClusterType: ccent.ClusterTypeDedicatedTiDB,
					ClusterID:   "1001",
				}
				comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
					ListClusters(gomock.Any(), gomock.Any()).
					Return(&mgmtv1.ListClustersResponse{Clusters: []*mgmtv1.Cluster{{
						ClusterName:    "cluster-1001",
						OrganizationId: tenant.Id,
					}}}, nil).AnyTimes()
				comps.TenantService.(*tntsvc.MockTenantServiceAPI).EXPECT().
					GetTenant(gomock.Any(), gomock.Any()).
					Return(&tenant, nil).AnyTimes()
			})

			gk.It("send approval normally when in maintenance window", func() {
				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					IsInClusterOpsMaintenanceWindow(gomock.Any(), "1001", nil).
					Return(true, &maintent.ClusterOpsMaintenanceWindow{
						Approvers: []string{"mw-1@pingcap.com"},
					}, nil).Times(1)
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, mutates ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(param.FromTemplate).To(Equal(ApprovalTemplateModifyConfig))
						Expect(param.Name).To(Equal(ApprovalNameCluster))
						Expect(param.Parameters[ApprovalParamClusterID]).To(Equal("1001"))
						Expect(param.Overview[ApprovalParamClusterName]).To(Equal("cluster-1001"))
						Expect(mutates).To(BeEmpty())

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})

			gk.It("checks execution window times by execution window id", func() {
				ori.ExecutionWindowID = pointer.Pointer(uint64(123))

				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					GetOpsExecutionWindow(gomock.Any(), uint64(123)).
					Return(&maintent.OpsExecutionWindow{
						StartAt: 100,
						EndAt:   200,
					}, nil).Times(1)

				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					IsInClusterOpsMaintenanceWindow(gomock.Any(), "1001", []time.Time{time.Unix(100, 0), time.Unix(200, 0)}).
					Return(true, &maintent.ClusterOpsMaintenanceWindow{
						Approvers: []string{"mw-1@pingcap.com"},
					}, nil).Times(1)
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, mutates ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(mutates).To(BeEmpty())

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})

			gk.It("append maintenance approvers when not in maintenance window", func() {
				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					IsInClusterOpsMaintenanceWindow(gomock.Any(), "1001", nil).
					Return(false, &maintent.ClusterOpsMaintenanceWindow{
						Approvers: []string{"mw-1@pingcap.com", "mw-2@pingcap.com"},
					}, nil).Times(1)
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, mutates ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(mutates).To(HaveLen(1))

						approvalParam := &apprsvcapi.CreateApprovalParam{
							Nodes: []*apprent.ApproverNode{{
								Method: apprent.ApproveMethodAny,
								Approver: &apprent.Approver{
									Emails: []string{"template@pingcap.com"},
								},
							}},
						}
						mutates[0](approvalParam)

						Expect(approvalParam.Nodes).To(HaveLen(2))
						Expect(approvalParam.Nodes[1].Method).To(Equal(apprent.ApproveMethodAny))
						Expect(approvalParam.Nodes[1].Approver.Emails).To(Equal([]string{"mw-1@pingcap.com", "mw-2@pingcap.com"}))

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})

			gk.It("append maintenance approvers when execution window from id is not fully in maintenance window", func() {
				ori.ExecutionWindowID = pointer.Pointer(uint64(123))

				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					GetOpsExecutionWindow(gomock.Any(), uint64(123)).
					Return(&maintent.OpsExecutionWindow{
						StartAt: 100,
						EndAt:   200,
					}, nil).Times(1)

				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					IsInClusterOpsMaintenanceWindow(gomock.Any(), "1001", []time.Time{time.Unix(100, 0), time.Unix(200, 0)}).
					Return(false, &maintent.ClusterOpsMaintenanceWindow{
						Approvers: []string{"mw-1@pingcap.com", "mw-2@pingcap.com"},
					}, nil).Times(1)
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, mutates ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(mutates).To(HaveLen(1))

						approvalParam := &apprsvcapi.CreateApprovalParam{}
						mutates[0](approvalParam)

						Expect(approvalParam.Nodes).To(HaveLen(1))
						Expect(approvalParam.Nodes[0].Method).To(Equal(apprent.ApproveMethodAny))
						Expect(approvalParam.Nodes[0].Approver.Emails).To(Equal([]string{"mw-1@pingcap.com", "mw-2@pingcap.com"}))

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})

			gk.It("skip maintenance node when not in maintenance window but approvers are empty", func() {
				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					IsInClusterOpsMaintenanceWindow(gomock.Any(), "1001", nil).
					Return(false, &maintent.ClusterOpsMaintenanceWindow{
						Approvers: []string{},
					}, nil).Times(1)
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, mutates ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(mutates).To(BeEmpty())

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})

			gk.It("return error when maintenance window check fails", func() {
				comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
					IsInClusterOpsMaintenanceWindow(gomock.Any(), "1001", nil).
					Return(false, nil, fmt.Errorf("maintenance check failed")).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("build maintenance window approval mutate failed"))
			})
		})

		gk.Context("For TenantInit", func() {
			gk.BeforeEach(func() {
				ori.TargetType = ccent.ConfigurationChangeTargetTypeTenantInit
				ori.TargetTenantInit = &ccent.TenantInitConfigurationLocator{
					ClusterType: ccent.ClusterTypeDedicatedTiDB,
					TenantID:    1001,
				}
			})

			gk.It("send approval normally", func() {
				comps.TenantService.(*tntsvc.MockTenantServiceAPI).EXPECT().
					GetTenant(gomock.Any(), gomock.Any()).
					Return(&tenant, nil).AnyTimes()
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, _ ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(param.FromTemplate).To(Equal(ApprovalTemplateModifyConfig))
						Expect(param.Name).To(Equal(ApprovalNameTenantInit))

						Expect(param.Parameters[ApprovalParamTenantID]).To(Equal(strconv.FormatUint(tenant.Id, 10)))
						Expect(param.Parameters[ApprovalParamTenantName]).To(Equal(tenant.Name))
						Expect(param.Parameters[ApprovalParamTenantRegion]).To(Equal(string(tenant.GetTenantRegion())))

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})
		})

		gk.Context("For Template", func() {
			gk.BeforeEach(func() {
				ori.TargetType = ccent.ConfigurationChangeTargetTypeTemplate
				ori.TargetTemplate = &ccent.ConfigurationTemplateLocator{
					ClusterType: ccent.ClusterTypeDedicatedTiDB,
				}
			})

			gk.It("send approval normally", func() {
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, _ ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(param.FromTemplate).To(Equal(ApprovalTemplateModifyTemplate))
						Expect(param.Name).To(Equal(ApprovalNameTemplate))

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})

			gk.It("skips applicant notification when operator entity is api key", func() {
				ori.OperatorEntity = "api_key:key-a"
				comps.ApprovalService.(*apprsvcapi.MockApprovalServiceAPI).EXPECT().
					CreateApprovalFromTemplate(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, param apprsvcapi.CreateApprovalFromTemplateParam, _ ...func(*apprsvcapi.CreateApprovalParam)) (*uint64, error) {
						Expect(param.FromTemplate).To(Equal(ApprovalTemplateModifyTemplate))
						Expect(param.Name).To(Equal(ApprovalNameTemplate))
						Expect(param.Configuration.SkipApplicantNotification).To(BeTrue())

						return pointer.Pointer(uint64(1)), nil
					}).Times(1)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				err := actor.SendApproval()
				Expect(err).To(BeNil())
			})
		})
	})

	gk.Context("[Apply]", func() {
		var ori *ccent.ConfigurationChange

		gk.BeforeEach(func() {
			ori = &ccent.ConfigurationChange{
				ID:     1,
				Reason: "test",
				Phase:  ccent.ConfigurationChangePhaseWaitForApply,
			}

			comps.Repo.(*repo.MockCloudConfigRepo).EXPECT().
				PatchConfigurationChange(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
				Return(nil)
		})

		gk.It("failed if target type is unknown", func() {
			ori.TargetType = ""

			actor := NewConfigurationChangeActor(ctx, comps, ori)
			param := api.ApplyConfigurationChangeParam{}
			err := actor.Apply(param)
			Expect(err).To(HaveOccurred())
			Expect(err.Error()).To(ContainSubstring("target type is unknown"))

		})

		gk.It("failed if current time is outside execution window", func() {
			now := time.Now()
			ori.ExecutionWindowID = pointer.Pointer(uint64(123))
			ori.TargetType = ccent.ConfigurationChangeTargetTypeCluster
			ori.TargetCluster = &ccent.ClusterConfigurationLocator{
				ClusterType: ccent.ClusterTypeDedicatedTiDB,
				ClusterID:   "1001",
			}

			comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
				GetOpsExecutionWindow(gomock.Any(), uint64(123)).
				Return(&maintent.OpsExecutionWindow{
					ID:      123,
					StartAt: now.Add(time.Hour).Unix(),
					EndAt:   now.Add(2 * time.Hour).Unix(),
				}, nil).Times(1)

			actor := NewConfigurationChangeActor(ctx, comps, ori)
			param := api.ApplyConfigurationChangeParam{}
			err := actor.Apply(param)
			Expect(err).To(HaveOccurred())
			Expect(err.Error()).To(ContainSubstring("current time is outside execute window"))
		})

		gk.It("continues applying if current time is inside execution window", func() {
			now := time.Now()
			ori.ExecutionWindowID = pointer.Pointer(uint64(123))
			ori.TargetType = ""

			comps.MaintenanceSvc.(*cmxapi.MockServiceAPI).EXPECT().
				GetOpsExecutionWindow(gomock.Any(), uint64(123)).
				Return(&maintent.OpsExecutionWindow{
					ID:      123,
					StartAt: now.Add(-time.Hour).Unix(),
					EndAt:   now.Add(time.Hour).Unix(),
				}, nil).Times(1)

			actor := NewConfigurationChangeActor(ctx, comps, ori)
			param := api.ApplyConfigurationChangeParam{}
			err := actor.Apply(param)
			Expect(err).To(HaveOccurred())
			Expect(err.Error()).To(ContainSubstring("target type is unknown"))
		})

		gk.Context("[TenantInit]", func() {
			gk.BeforeEach(func() {
				ori.TargetType = ccent.ConfigurationChangeTargetTypeTenantInit
				ori.TargetTenantInit = &ccent.TenantInitConfigurationLocator{
					ClusterType: ccent.ClusterTypeDedicatedTiDB,
					TenantID:    1001,
				}
				// add this so GetDefaultVersion won't be unexpected during Apply tests
				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					GetDefaultVersion(gomock.Any(), gomock.Any()).AnyTimes().
					Return(&ccproto.DefaultVersion{Version: "v7.0.0"}, nil)
			})

			gk.It("with file change", func() {
				ori.Scope = ccent.ConfigurationChangeScope{
					Files: []*ccent.ConfigurationFileChange{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							TemplateID: 101,
							AllParameters: []ccent.UsedParameter{
								{
									NamedParameter: ccent.NamedParameter{ // new parameter from custom
										Name: "tidb.config.1",
										ParameterValue: ccent.ParameterValue{
											Value: "info",
										},
									},
									Source: ccent.ParameterCustom,
								},
								{
									NamedParameter: ccent.NamedParameter{ // existing parameter from custom
										Name: "tidb.config.2",
										ParameterValue: ccent.ParameterValue{
											Value: 10010,
										},
									},
									Source: ccent.ParameterCustom,
								},
							},
							ModifiedParameters: []ccent.ChangedParameter{
								{
									Name:       "tidb.config.1",
									From:       nil,
									To:         "info",
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
							},
						},
					},
				}
				exp := &applier.ConfigCenterTenantInitApplierRequest{
					ParamRequest: &ccproto.BatchUpdateTenantInitialCustomParamsRequest{
						Requests: []*ccproto.UpdateTenantInitialCustomParamsRequest{
							{
								ClusterType:       ccproto.ClusterType_DEDICATED_TIDB,
								ConfigurationType: ccproto.DEDICATED_TIDB_CONFIG_TIDB,
								TenantId:          1001,
								InitialParams: map[string]*ccproto.ParamItem{
									"tidb.config.1": {
										Value: MustNewPbValue("info"),
									},
									"tidb.config.2": {
										Value: MustNewPbValue(10010),
									},
								},
							},
						},
					},
				}

				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					BatchUpdateTenantInitialCustomParams(gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *ccproto.BatchUpdateTenantInitialCustomParamsRequest, arg2 ...interface{}) (*ccproto.BatchUpdateTenantInitialCustomParamsResponse, error) {
							Expect(test.IsEqual(exp.ParamRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					UpdateDefaultVersion(gomock.Any(), gomock.Any()).AnyTimes().
					Return(nil, fmt.Errorf("should not be called"))

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(BeNil())
			})

			gk.It("with version change", func() {
				ori.Scope = ccent.ConfigurationChangeScope{
					Version: &ccent.ConfigurationVersionChange{
						From: "v7.0.0",
						To:   "v6.0.0",
					},
				}
				exp := &applier.ConfigCenterTenantInitApplierRequest{
					DefaultVersionRequest: &ccproto.UpdateDefaultVersionRequest{
						ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
						Component:   ccproto.DEDICATED_TIDB_COMPONENT_TIDB,
						TenantId:    pointer.Pointer(uint64(1001)),
						Version:     "v6.0.0",
					},
				}

				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					BatchUpdateTenantInitialCustomParams(gomock.Any(), gomock.Any()).AnyTimes().
					Return(nil, fmt.Errorf("should not be called"))
				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					UpdateDefaultVersion(gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *ccproto.UpdateDefaultVersionRequest, arg2 ...interface{}) (*ccproto.DefaultVersion, error) {
							Expect(test.IsEqual(exp.DefaultVersionRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(BeNil())
			})

		})

		gk.Context("[Cluster]", func() {
			gk.BeforeEach(func() {
				ori.TargetType = ccent.ConfigurationChangeTargetTypeCluster
				ori.TargetCluster = &ccent.ClusterConfigurationLocator{
					ClusterType: ccent.ClusterTypeDedicatedTiDB,
					ClusterID:   "1234",
				}
			})

			gk.It("failed if a file change is unaccepted", func() {
				ori.ApplyPolicy = ccent.ParameterApplyPolicyRestart
				ori.Scope = ccent.ConfigurationChangeScope{
					Files: []*ccent.ConfigurationFileChange{
						{
							Filename:   "unknown-file",
							ConfigType: "unknown-file",
							AllParameters: []ccent.UsedParameter{
								{
									NamedParameter: ccent.NamedParameter{
										Name: "config.1",
										ParameterValue: ccent.ParameterValue{
											Value: "warn",
										},
									},
									Source: ccent.ParameterCustom,
								},
							},
							ModifiedParameters: []ccent.ChangedParameter{
								{
									Name:       "tidb.config.1",
									From:       nil,
									To:         "warn",
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
							},
							PreviousText: "previous_text",
						},
					},
				}

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("no applier can handle the file change"))
			})

			gk.It("with restart file change via dataflow and mgmt", func() {
				ori.ApplyPolicy = ccent.ParameterApplyPolicyRestart
				ori.Scope = ccent.ConfigurationChangeScope{
					Files: []*ccent.ConfigurationFileChange{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							AllParameters: []ccent.UsedParameter{
								{
									NamedParameter: ccent.NamedParameter{
										Name: "tidb.config.1",
										ParameterValue: ccent.ParameterValue{
											Value: "warn",
										},
									},
									Source: ccent.ParameterCustom,
								},
								{
									NamedParameter: ccent.NamedParameter{
										Name: "tidb.config.2",
										ParameterValue: ccent.ParameterValue{
											Value: 1,
										},
									},
									Source: ccent.ParameterCustom,
								},
							},
							ModifiedParameters: []ccent.ChangedParameter{
								{
									Name:       "tidb.config.1",
									From:       nil,
									To:         "warn",
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
							},
							PreviousText: "previous_text",
						},
						{
							Filename:   "ticdc",
							ConfigType: "ticdc",
							AllParameters: []ccent.UsedParameter{
								{
									NamedParameter: ccent.NamedParameter{
										Name: "ticdc.config.1",
										ParameterValue: ccent.ParameterValue{
											Value: 111.1,
										},
									},
									Source: ccent.ParameterCustom,
								},
								{
									NamedParameter: ccent.NamedParameter{
										Name: "ticdc.config.2",
										ParameterValue: ccent.ParameterValue{
											Value: "custom",
										},
									},
									Source: ccent.ParameterCustom,
								},
							},
							ModifiedParameters: []ccent.ChangedParameter{
								{
									Name:       "ticdc.config.1",
									From:       nil,
									To:         111.1,
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
								{
									Name:       "ticdc.config.2",
									From:       nil,
									To:         "custom",
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
							},
							PreviousText: "previous_text",
						},
					},
					Components: nil, // not change components
					Version:    nil, // not change version
				}
				expMgmt := &applier.MgmtServiceApplierRequest{
					ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
					ClusterId:   "1234",
					CCRequest: &ccproto.UpdateClusterMetaRequest{
						ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
						ClusterId:   "1234",
						ConfigurationFileRequests: []*ccproto.UpdateClusterConfigurationFileRequest{
							{
								ClusterType:           ccproto.ClusterType_DEDICATED_TIDB,
								ClusterId:             "1234",
								ConfigurationFilename: "tidb",
								ConfigurationType:     pointer.Pointer("tidb"),
								CustomParams: map[string]*ccproto.ParamItem{
									"tidb.config.1": {
										Value: MustNewPbValue("warn"),
									},
									"tidb.config.2": {
										Value: MustNewPbValue(1),
									},
								},
								AllowMissing: true,
								PreviousText: pointer.Pointer("previous_text"),
							},
						},
					},
					MgmtRequest: &mgmtv1.ApplyClusterRequest{
						ClusterId: 1234,
						StatusId:  101,
						ClusterApplyMask: &dedicatedv1.ClusterApplyMask{
							ComponentApplyMasks: []*dedicatedv1.ClusterApplyMask_ComponentApplyMask{
								{
									ComponentType: dedicatedv1.ComponentType_TIDB,
									Configuration: true,
								},
							},
							Strategy: dedicatedv1.ClusterApplyMask_ROLLING_RESTART,
						},
					},
				}
				expDf := &applier.DataflowApplierRequest{
					ClusterType: dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
					ClusterId:   1234,
					Requests: []*dataflow.ApplyConfigChangesReq{
						{
							ClusterType:       dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
							ClusterId:         1234,
							ConfigurationFile: dataflow.ConfigFileType_TICDC_FILE,
							CustomParams: map[string]*dataflow.ParamItem{
								"ticdc.config.1": {
									Value: MustNewPbValue(111.1),
								},
								"ticdc.config.2": {
									Value: MustNewPbValue("custom"),
								},
							},
						},
					},
				}

				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					UpdateClusterMeta(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *ccproto.UpdateClusterMetaRequest, arg2 ...interface{}) (*ccproto.UpdateClusterMetaResponse, error) {
							Expect(test.IsEqual(expMgmt.CCRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					SetClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(&mgmtv1.ClusterCustomStatus{StatusId: 101}, nil)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					DeleteClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(nil, fmt.Errorf("should not be called"))
				comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
					ApplyCluster(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *mgmtv1.ApplyClusterRequest, arg2 ...interface{}) (*mgmtv1.Cluster, error) {
							Expect(test.IsEqual(expMgmt.MgmtRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.Dataflow.Ops.(*dfcli.MockOpsInternalServiceClient).EXPECT().
					ApplyConfigChanges(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(func(ctx context.Context, req *dataflow.ApplyConfigChangesReq, arg2 ...interface{}) (*dataflow.ApplyConfigChangesResp, error) {
						for _, expReq := range expDf.Requests {
							if expReq.ConfigurationFile == req.ConfigurationFile {
								Expect(test.IsEqual(expReq, req, protocmp.Transform())).To(BeNil())
								return nil, nil
							}
						}

						return nil, fmt.Errorf("not found")
					})

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(BeNil())
			})

			gk.It("with hotreload file change via mgmt", func() {
				ori.ChangeType = "fulltidb"
				ori.ApplyPolicy = ccent.ParameterApplyPolicyHotReload
				ori.Scope = ccent.ConfigurationChangeScope{
					Files: []*ccent.ConfigurationFileChange{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							AllParameters: []ccent.UsedParameter{
								{
									NamedParameter: ccent.NamedParameter{
										Name: "tidb.config.1",
										ParameterValue: ccent.ParameterValue{
											Value: "warn",
										},
									},
									Source: ccent.ParameterCustom,
								},
							},
							ModifiedParameters: []ccent.ChangedParameter{
								{
									Name:       "tidb.config.1",
									From:       nil,
									To:         "warn",
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
							},
							PreviousText: "previous_text",
						},
					},
				}
				expMgmt := &applier.MgmtServiceApplierRequest{
					ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
					ClusterId:   "1234",
					CCRequest: &ccproto.UpdateClusterMetaRequest{
						ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
						ClusterId:   "1234",
						ConfigurationFileRequests: []*ccproto.UpdateClusterConfigurationFileRequest{
							{
								ClusterType:           ccproto.ClusterType_DEDICATED_TIDB,
								ClusterId:             "1234",
								ConfigurationFilename: "tidb",
								ConfigurationType:     pointer.Pointer("tidb"),
								CustomParams: map[string]*ccproto.ParamItem{
									"tidb.config.1": {
										Value: MustNewPbValue("warn"),
									},
								},
								AllowMissing: true,
								PreviousText: pointer.Pointer("previous_text"),
							},
						},
					},
					MgmtRequest: &mgmtv1.ApplyClusterRequest{
						ClusterId: 1234,
						StatusId:  101,
						ClusterApplyMask: &dedicatedv1.ClusterApplyMask{
							ComponentApplyMasks: []*dedicatedv1.ClusterApplyMask_ComponentApplyMask{
								{
									ComponentType: dedicatedv1.ComponentType_TIDB,
									Configuration: true,
								},
							},
							Strategy: dedicatedv1.ClusterApplyMask_HOT_RELOAD,
						},
					},
				}

				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					UpdateClusterMeta(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *ccproto.UpdateClusterMetaRequest, arg2 ...interface{}) (*ccproto.UpdateClusterMetaResponse, error) {
							Expect(test.IsEqual(expMgmt.CCRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					SetClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(&mgmtv1.ClusterCustomStatus{StatusId: 101}, nil)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					DeleteClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(nil, fmt.Errorf("should not be called"))
				comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
					ApplyCluster(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *mgmtv1.ApplyClusterRequest, arg2 ...interface{}) (*mgmtv1.Cluster, error) {
							Expect(test.IsEqual(expMgmt.MgmtRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(BeNil())
			})

			gk.It("with version change", func() {
				ori.ChangeType = "tidb"
				ori.Scope = ccent.ConfigurationChangeScope{
					Version: &ccent.ConfigurationVersionChange{
						To: "v8.0.0",
					},
					Components: []*ccent.ConfigurationComponentChange{
						{
							Component: "tidb",
							ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
								From: "v7.0.0",
								To:   "v8.0.0",
							},
							ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
								From: "v7.0.0",
								To:   "",
							},
						},
						{
							Component: "ticdc",
							ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
								From: "v7.0.0",
								To:   "v8.0.0",
							},
							ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
								From: "v7.0.0",
								To:   "",
							},
						},
					},
				}
				expMgmt := &applier.MgmtServiceApplierRequest{
					ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
					ClusterId:   "1234",
					CCRequest: &ccproto.UpdateClusterMetaRequest{
						ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
						ClusterId:   "1234",
						ComponentRequests: []*ccproto.UpdateClusterComponentRequest{
							{
								ClusterType:       ccproto.ClusterType_DEDICATED_TIDB,
								ClusterId:         "1234",
								ComponentName:     "tidb",
								Version:           pointer.Pointer("v8.0.0"),
								OverwriteImageTag: pointer.Pointer(""),
							},
						},
					},
					MgmtRequest: &mgmtv1.ApplyClusterRequest{
						ClusterId: 1234,
						StatusId:  101,
						ClusterApplyMask: &dedicatedv1.ClusterApplyMask{
							Version: true,
							ComponentApplyMasks: []*dedicatedv1.ClusterApplyMask_ComponentApplyMask{
								{
									ComponentType: dedicatedv1.ComponentType_TIDB,
									Image:         true,
								},
							},
						},
					},
				}
				expDf := &applier.DataflowApplierRequest{
					ClusterType: dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
					ClusterId:   1234,
					Requests: []*dataflow.ApplyConfigChangesReq{
						{
							ClusterType:       dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
							ClusterId:         1234,
							ConfigurationFile: dataflow.ConfigFileType_TICDC_FILE,
							Version:           pointer.Pointer("v8.0.0"),
							CustomImageTag:    pointer.Pointer(""),
						},
					},
				}

				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					UpdateClusterMeta(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *ccproto.UpdateClusterMetaRequest, arg2 ...interface{}) (*ccproto.UpdateClusterMetaResponse, error) {
							pc, _, _, _ := runtime.Caller(6)
							callerName := runtime.FuncForPC(pc).Name()
							if strings.Contains(callerName, "MgmtServiceApplier") {
								Expect(test.IsEqual(expMgmt.CCRequest, req, protocmp.Transform())).To(BeNil())
							}
							return nil, nil
						},
					)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					SetClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(&mgmtv1.ClusterCustomStatus{StatusId: 101}, nil)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					DeleteClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(nil, fmt.Errorf("should not be called"))
				comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
					ApplyCluster(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *mgmtv1.ApplyClusterRequest, arg2 ...interface{}) (*mgmtv1.Cluster, error) {
							Expect(test.IsEqual(expMgmt.MgmtRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.Dataflow.Ops.(*dfcli.MockOpsInternalServiceClient).EXPECT().
					ApplyConfigChanges(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(func(ctx context.Context, req *dataflow.ApplyConfigChangesReq, arg2 ...interface{}) (*dataflow.ApplyConfigChangesResp, error) {
						for _, expReq := range expDf.Requests {
							if expReq.ConfigurationFile == req.ConfigurationFile {
								Expect(test.IsEqual(expReq, req, protocmp.Transform())).To(BeNil())
								return nil, nil
							}
						}

						return nil, fmt.Errorf("not found")
					})

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(BeNil())
			})

			gk.It("failed when a component change is unknown", func() {
				ori.ChangeType = "fulltidb"
				ori.Scope = ccent.ConfigurationChangeScope{

					Components: []*ccent.ConfigurationComponentChange{
						{
							Component: "comp-unknown",
							ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
								From: "v7.0.0",
								To:   "vxx",
							},
							ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
								From: "v7.0.0",
								To:   "vxx",
							},
						},
					},
				}

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(HaveOccurred())
				Expect(err.Error()).To(ContainSubstring("no applier can handle the component change"))
			})

			gk.It("with component change", func() {
				ori.ChangeType = "fulltidb"
				ori.Scope = ccent.ConfigurationChangeScope{

					Components: []*ccent.ConfigurationComponentChange{
						{
							Component: "tidb",
							ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
								From: "v7.0.0",
								To:   "vxx",
							},
							ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
								From: "v7.0.0",
								To:   "vxx",
							},
						},
						{
							Component: "ticdc",
							ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
								From: "v7.0.0",
								To:   "vxx",
							},
							ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
								From: "v7.0.0",
								To:   "vxx",
							},
						},
					},
				}
				expMgmt := &applier.MgmtServiceApplierRequest{
					ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
					ClusterId:   "1234",
					CCRequest: &ccproto.UpdateClusterMetaRequest{
						ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
						ClusterId:   "1234",
						ComponentRequests: []*ccproto.UpdateClusterComponentRequest{
							{
								ClusterType:       ccproto.ClusterType_DEDICATED_TIDB,
								ClusterId:         "1234",
								ComponentName:     "tidb",
								OverwriteImageTag: pointer.Pointer("vxx"),
							},
						},
					},
					MgmtRequest: &mgmtv1.ApplyClusterRequest{
						ClusterId: 1234,
						StatusId:  101,
						ClusterApplyMask: &dedicatedv1.ClusterApplyMask{
							ComponentApplyMasks: []*dedicatedv1.ClusterApplyMask_ComponentApplyMask{
								{
									ComponentType: dedicatedv1.ComponentType_TIDB,
									Image:         true,
								},
							},
						},
					},
				}
				expDf := &applier.DataflowApplierRequest{
					ClusterType: dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
					ClusterId:   1234,
					Requests: []*dataflow.ApplyConfigChangesReq{
						{
							ClusterType:       dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
							ClusterId:         1234,
							ConfigurationFile: dataflow.ConfigFileType_TICDC_FILE,
							CustomImageTag:    pointer.Pointer("vxx"),
						},
					},
				}

				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					UpdateClusterMeta(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *ccproto.UpdateClusterMetaRequest, arg2 ...interface{}) (*ccproto.UpdateClusterMetaResponse, error) {
							Expect(test.IsEqual(expMgmt.CCRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					SetClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(&mgmtv1.ClusterCustomStatus{StatusId: 101}, nil)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					DeleteClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(nil, fmt.Errorf("should not be called"))
				comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
					ApplyCluster(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *mgmtv1.ApplyClusterRequest, arg2 ...interface{}) (*mgmtv1.Cluster, error) {
							Expect(test.IsEqual(expMgmt.MgmtRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.Dataflow.Ops.(*dfcli.MockOpsInternalServiceClient).EXPECT().
					ApplyConfigChanges(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(func(ctx context.Context, req *dataflow.ApplyConfigChangesReq, arg2 ...interface{}) (*dataflow.ApplyConfigChangesResp, error) {
						for _, expReq := range expDf.Requests {
							if expReq.ConfigurationFile == req.ConfigurationFile {
								Expect(test.IsEqual(expReq, req, protocmp.Transform())).To(BeNil())
								return nil, nil
							}
						}

						return nil, fmt.Errorf("not found")
					})

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(BeNil())
			})

			gk.It("with file/version/component change via dataflow and mgmt", func() {
				ori.ChangeType = "tidb"
				ori.Scope = ccent.ConfigurationChangeScope{
					Version: &ccent.ConfigurationVersionChange{
						To: "v8.0.0",
					},
					Components: []*ccent.ConfigurationComponentChange{
						{
							Component: "tidb",
							ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
								From: "v7.0.0",
								To:   "v8.0.0",
							},
							ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
								From: "v7.0.0",
								To:   "vxx",
							},
						},
						{
							Component: "ticdc",
							ModifiedImageTag: &ccent.ChangedImageTag{ // image_tag should be same as version
								From: "v7.0.0",
								To:   "v8.0.0",
							},
							ModifiedOverwriteImageTag: &ccent.ChangedImageTag{ // overwrite_image_tag should be reverted to ""
								From: "v7.0.0",
								To:   "",
							},
						},
					},
					Files: []*ccent.ConfigurationFileChange{
						{
							Filename:   "tidb",
							ConfigType: "tidb",
							AllParameters: []ccent.UsedParameter{
								{
									NamedParameter: ccent.NamedParameter{
										Name: "tidb.config.1",
										ParameterValue: ccent.ParameterValue{
											Value: "warn",
										},
									},
									Source: ccent.ParameterCustom,
								},
								{
									NamedParameter: ccent.NamedParameter{
										Name: "tidb.config.2",
										ParameterValue: ccent.ParameterValue{
											Value: 1,
										},
									},
									Source: ccent.ParameterCustom,
								},
							},
							ModifiedParameters: []ccent.ChangedParameter{
								{
									Name:       "tidb.config.1",
									From:       nil,
									To:         "warn",
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
							},
							PreviousText: "previous_text",
						},
						{
							Filename:   "ticdc",
							ConfigType: "ticdc",
							AllParameters: []ccent.UsedParameter{
								{
									NamedParameter: ccent.NamedParameter{
										Name: "ticdc.config.1",
										ParameterValue: ccent.ParameterValue{
											Value: 111.1,
										},
									},
									Source: ccent.ParameterCustom,
								},
								{
									NamedParameter: ccent.NamedParameter{
										Name: "ticdc.config.2",
										ParameterValue: ccent.ParameterValue{
											Value: "custom",
										},
									},
									Source: ccent.ParameterCustom,
								},
							},
							ModifiedParameters: []ccent.ChangedParameter{
								{
									Name:       "ticdc.config.1",
									From:       nil,
									To:         111.1,
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
								{
									Name:       "ticdc.config.2",
									From:       nil,
									To:         "custom",
									FromSource: ccent.ParameterKernelDefault,
									ToSource:   ccent.ParameterCustom,
									RiskLevel:  ccent.RiskLevelHigh,
								},
							},
							PreviousText: "previous_text",
						},
					},
				}
				expMgmt := &applier.MgmtServiceApplierRequest{
					ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
					ClusterId:   "1234",
					CCRequest: &ccproto.UpdateClusterMetaRequest{
						ClusterType: ccproto.ClusterType_DEDICATED_TIDB,
						ClusterId:   "1234",
						ComponentRequests: []*ccproto.UpdateClusterComponentRequest{
							{
								ClusterType:       ccproto.ClusterType_DEDICATED_TIDB,
								ClusterId:         "1234",
								ComponentName:     "tidb",
								Version:           pointer.Pointer("v8.0.0"),
								OverwriteImageTag: pointer.Pointer("vxx"),
							},
						},
						ConfigurationFileRequests: []*ccproto.UpdateClusterConfigurationFileRequest{
							{
								ClusterType:           ccproto.ClusterType_DEDICATED_TIDB,
								ClusterId:             "1234",
								ConfigurationFilename: "tidb",
								ConfigurationType:     pointer.Pointer("tidb"),
								CustomParams: map[string]*ccproto.ParamItem{
									"tidb.config.1": {
										Value: MustNewPbValue("warn"),
									},
									"tidb.config.2": {
										Value: MustNewPbValue(1),
									},
								},
								AllowMissing: true,
								PreviousText: pointer.Pointer("previous_text"),
							},
						},
					},
					MgmtRequest: &mgmtv1.ApplyClusterRequest{
						ClusterId: 1234,
						StatusId:  101,
						ClusterApplyMask: &dedicatedv1.ClusterApplyMask{
							Version: true,
							ComponentApplyMasks: []*dedicatedv1.ClusterApplyMask_ComponentApplyMask{
								{
									ComponentType: dedicatedv1.ComponentType_TIDB,
									Image:         true,
									Configuration: true,
								},
							},
						},
					},
				}
				expDf := &applier.DataflowApplierRequest{
					ClusterType: dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
					ClusterId:   1234,
					Requests: []*dataflow.ApplyConfigChangesReq{
						{
							ClusterType:       dataflow.ConfigClusterType_TIDB_DEDICATED_CLUSTER,
							ClusterId:         1234,
							ConfigurationFile: dataflow.ConfigFileType_TICDC_FILE,
							Version:           pointer.Pointer("v8.0.0"),
							CustomImageTag:    pointer.Pointer(""),
							CustomParams: map[string]*dataflow.ParamItem{
								"ticdc.config.1": {
									Value: MustNewPbValue(111.1),
								},
								"ticdc.config.2": {
									Value: MustNewPbValue("custom"),
								},
							},
						},
					},
				}

				comps.ConfigCenter.ConfigAPI.(*cccli.MockConfigCenterServiceClient).EXPECT().
					UpdateClusterMeta(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *ccproto.UpdateClusterMetaRequest, arg2 ...interface{}) (*ccproto.UpdateClusterMetaResponse, error) {
							pc, _, _, _ := runtime.Caller(6)
							callerName := runtime.FuncForPC(pc).Name()
							if strings.Contains(callerName, "MgmtServiceApplier") {
								Expect(test.IsEqual(expMgmt.CCRequest, req, protocmp.Transform())).To(BeNil())
							}
							return nil, nil
						},
					)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					SetClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(&mgmtv1.ClusterCustomStatus{StatusId: 101}, nil)
				comps.CoreSvc.Mgmt.ClusterCustomStatus.(*coresvccli.MockClusterCustomStatusServiceClient).EXPECT().
					DeleteClusterCustomStatus(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					Return(nil, fmt.Errorf("should not be called"))
				comps.CoreSvc.Mgmt.Cluster.(*coresvccli.MockClusterServiceClient).EXPECT().
					ApplyCluster(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(
						func(ctx context.Context, req *mgmtv1.ApplyClusterRequest, arg2 ...interface{}) (*mgmtv1.Cluster, error) {
							Expect(test.IsEqual(expMgmt.MgmtRequest, req, protocmp.Transform())).To(BeNil())
							return nil, nil
						},
					)
				comps.Dataflow.Ops.(*dfcli.MockOpsInternalServiceClient).EXPECT().
					ApplyConfigChanges(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes().
					DoAndReturn(func(ctx context.Context, req *dataflow.ApplyConfigChangesReq, arg2 ...interface{}) (*dataflow.ApplyConfigChangesResp, error) {
						for _, expReq := range expDf.Requests {
							if expReq.ConfigurationFile == req.ConfigurationFile {
								Expect(test.IsEqual(expReq, req, protocmp.Transform())).To(BeNil())
								return nil, nil
							}
						}

						return nil, fmt.Errorf("not found")
					})

				actor := NewConfigurationChangeActor(ctx, comps, ori)
				param := api.ApplyConfigurationChangeParam{}
				err := actor.Apply(param)
				Expect(err).To(BeNil())
			})

		})
	})
})

func SortChange(change *ccent.ConfigurationChange) {
	slices.SortFunc(change.Scope.Files, func(a, b *ccent.ConfigurationFileChange) int {
		if a.Filename > b.Filename {
			return 1
		} else if a.Filename < b.Filename {
			return -1
		}

		if a.ConfigType > b.ConfigType {
			return 1
		} else if a.ConfigType < b.ConfigType {
			return -1
		}

		return 0
	})

	for _, filgChg := range change.Scope.Files {
		slices.SortFunc(filgChg.AllParameters, func(a, b ccent.UsedParameter) int {
			if a.Name > b.Name {
				return 1
			} else if a.Name < b.Name {
				return -1
			} else {
				return 0
			}
		})
		slices.SortFunc(filgChg.ModifiedParameters, func(a, b ccent.ChangedParameter) int {
			if a.Name > b.Name {
				return 1
			} else if a.Name < b.Name {
				return -1
			} else {
				return 0
			}
		})
	}
	slices.SortFunc(change.Scope.Components, func(a, b *ccent.ConfigurationComponentChange) int {
		if a.Component > b.Component {
			return 1
		} else if a.Component < b.Component {
			return -1
		} else {
			return 0
		}
	})
}

func MustNewPbValue(val any) *structpb.Value {
	v, err := structpb.NewValue(val)
	if err != nil {
		panic(err)
	}
	return v

}