package server

import (
	"context"
	"errors"
	"testing"

	v1alpha1br "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	operatorv1alpha1 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestValidateSharedPoolPauseReturnsAllFailuresInStableOrder(t *testing.T) {
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id: "100", Phase: apiv1.SharedPool_MODIFYING, DataplaneName: "dp-1",
	}}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Twice()

	dagExecutor := executormocks.NewRemoteExecutor(t)
	running := executor.NewExecTaskIdStatus()
	running["running"] = &executor.ExecTaskStatusEx{}
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanPauseSharedPool, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanUpdateComponent, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanUpgradeTiBRAutoBackup, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanApplyDMCluster, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateManualBackup, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)

	dataplaneRepo := repomocks.NewDataplaneRepo(t)
	dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(&entity.DataPlaneProvider{Name: "dp-1", GatewayUrl: "gateway.test"}, nil).Once()
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(newPauseValidationTiBRClient(t, true, true), nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		dataplaneRepo:    dataplaneRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor,
		bizCfg:           &config.BizConfig{},
	}

	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{
		"INVALID_LIFECYCLE_STATE",
		"UPGRADE_IN_PROGRESS",
		"BACKUP_IN_PROGRESS",
		"OPERATION_IN_PROGRESS",
	}, pauseValidationCodes(response))
	dagExecutor.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
	dagExecutor.AssertNotCalled(t, "Abort", mock.Anything, mock.Anything)
}

func TestPauseValidationCollectorSortsUnknownCodesLast(t *testing.T) {
	collector := newPauseValidationCollector(true)
	collector.add(newPauseValidationFailure("FUTURE_CODE", "future", nil))
	collector.add(newPauseValidationFailure("INVALID_LIFECYCLE_STATE", "invalid", nil))
	collector.add(newPauseValidationFailure("OPERATION_IN_PROGRESS", "mutation", nil))

	response := &entity.ValidateSharedPoolPauseResponse{Failures: collector.responseFailures()}
	require.Equal(t, []string{"INVALID_LIFECYCLE_STATE", "OPERATION_IN_PROGRESS", "FUTURE_CODE"}, pauseValidationCodes(response))
}

func TestValidateSharedPoolPausePreservesRunningBlockersWhenLaterPlanQueriesFail(t *testing.T) {
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id: "100", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "dp-1",
	}}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Twice()

	dagExecutor := executormocks.NewRemoteExecutor(t)
	running := executor.ExecTaskIdStatus{"running": &executor.ExecTaskStatusEx{}}
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanPauseSharedPool, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanResumeSharedPool, mock.Anything).Return(nil, errors.New("resume query unavailable")).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateSharedPool, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanDeleteSharedPool, mock.Anything).Return(nil, errors.New("delete query unavailable")).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)

	dataplaneRepo, dataplaneManager := pauseValidationDataplaneDependencies(t, newPauseValidationTiBRClient(t, true, true))
	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, dataplaneRepo: dataplaneRepo, dataplaneManager: dataplaneManager,
		dagExecutor: dagExecutor, bizCfg: &config.BizConfig{},
	}

	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{
		"DEPENDENCY_UNAVAILABLE",
		"OPERATION_IN_PROGRESS",
	}, pauseValidationCodes(response))
}

func TestValidateSharedPoolPauseAggregatesDependencyFailuresAndContinues(t *testing.T) {
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id: "100", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "dp-1",
	}}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Twice()

	dagExecutor := executormocks.NewRemoteExecutor(t)
	running := executor.NewExecTaskIdStatus()
	running["upgrade"] = &executor.ExecTaskStatusEx{}
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanUpdateComponent, mock.Anything).Return(nil, errors.New("executor unavailable")).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanUpgradeTiBRAutoBackup, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)

	dataplaneRepo := repomocks.NewDataplaneRepo(t)
	dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(nil, errors.New("dataplane repo unavailable")).Once()
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(newPauseValidationTiBRClient(t, false, false), nil).Once()

	s := &serverImpl{
		sharedPoolRepo:   sharedPoolRepo,
		dataplaneRepo:    dataplaneRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor,
		bizCfg:           &config.BizConfig{},
	}

	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{
		"DEPENDENCY_UNAVAILABLE",
		"UPGRADE_IN_PROGRESS",
	}, pauseValidationCodes(response))
	require.ErrorContains(t, errors.New(response.GetFailures()[0].GetMessage()), "failed to query conflicting operations")
	require.ErrorContains(t, errors.New(response.GetFailures()[0].GetMessage()), "load dataplane")
}

func TestValidateSharedPoolPauseMapsFlushWALPrerequisiteCodes(t *testing.T) {
	tests := []struct {
		name      string
		disabled  bool
		setup     func(*testing.T, *repomocks.SharedPoolRepo) (*repomocks.DataplaneRepo, *dataplanemocks.Manager)
		wantCode  string
		poolLoads int
	}{
		{name: "FlushWAL disabled", disabled: true, wantCode: "DEPENDENCY_UNAVAILABLE", poolLoads: 1},
		{
			name: "dataplane not found", wantCode: "DEPENDENCY_UNAVAILABLE", poolLoads: 2,
			setup: func(t *testing.T, _ *repomocks.SharedPoolRepo) (*repomocks.DataplaneRepo, *dataplanemocks.Manager) {
				t.Helper()
				repo := repomocks.NewDataplaneRepo(t)
				repo.On("LoadByName", mock.Anything, "dp-1").Return(nil, status.Error(codes.NotFound, "missing")).Once()
				return repo, nil
			},
		},
		{
			name: "gateway URL missing", wantCode: "DEPENDENCY_UNAVAILABLE", poolLoads: 2,
			setup: func(t *testing.T, _ *repomocks.SharedPoolRepo) (*repomocks.DataplaneRepo, *dataplanemocks.Manager) {
				t.Helper()
				repo := repomocks.NewDataplaneRepo(t)
				repo.On("LoadByName", mock.Anything, "dp-1").Return(&entity.DataPlaneProvider{Name: "dp-1"}, nil).Once()
				return repo, nil
			},
		},
		{
			name: "TiBR not found", wantCode: "DEPENDENCY_UNAVAILABLE", poolLoads: 2,
			setup: func(t *testing.T, _ *repomocks.SharedPoolRepo) (*repomocks.DataplaneRepo, *dataplanemocks.Manager) {
				t.Helper()
				return pauseValidationDataplaneDependencies(t, newPauseValidationTiBRClient(t, false, false))
			},
		},
		{
			name: "TiBR not ready", wantCode: "DEPENDENCY_UNAVAILABLE", poolLoads: 2,
			setup: func(t *testing.T, _ *repomocks.SharedPoolRepo) (*repomocks.DataplaneRepo, *dataplanemocks.Manager) {
				t.Helper()
				return pauseValidationDataplaneDependencies(t, newPauseValidationTiBRClient(t, true, false))
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id: "100", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "dp-1",
			}}
			sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
			sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Times(tt.poolLoads)
			var dataplaneRepo *repomocks.DataplaneRepo
			var dataplaneManager *dataplanemocks.Manager
			if tt.setup != nil {
				dataplaneRepo, dataplaneManager = tt.setup(t, sharedPoolRepo)
			}
			if dataplaneManager == nil {
				dataplaneManager = dataplanemocks.NewManager(t)
				dataplaneManager.On("GetClient", "dp-1").Return(newPauseValidationTiBRClient(t, false, false), nil).Once()
			}
			dagExecutor := executormocks.NewRemoteExecutor(t)
			dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
			s := &serverImpl{
				sharedPoolRepo: sharedPoolRepo, dataplaneRepo: dataplaneRepo, dataplaneManager: dataplaneManager,
				dagExecutor: dagExecutor, bizCfg: &config.BizConfig{DefaultTiBRConfig: config.TiBRConfig{DisableFlushWAL: tt.disabled}},
			}

			response, err := s.ValidateSharedPoolPause(context.Background(), "100")
			require.NoError(t, err)
			require.Equal(t, []string{tt.wantCode}, pauseValidationCodes(response))
		})
	}
}

func TestValidateSharedPoolPauseMapsOperatorStateInconsistency(t *testing.T) {
	sharedPool := &entity.SharedPool{
		SharedPool:            apiv1.SharedPool{Id: "100", Phase: apiv1.SharedPool_PAUSING, DataplaneName: "dp-1"},
		PauseResumeState:      entity.PauseResumeStatePausing,
		SuspendComputeDesired: false,
	}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Once()
	operationRepo := repomocks.NewPauseResumeOperationRepo(t)
	operationRepo.On("LoadLatestBySharedPoolID", mock.Anything, "100").Return(&entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "failed-pause", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_FAILED,
	}, nil).Once()
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(newPauseValidationSuspendedClient(t, false), nil).Once()

	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, pauseResumeOperationRepo: operationRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor, bizCfg: &config.BizConfig{},
	}
	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{"INVALID_LIFECYCLE_STATE"}, pauseValidationCodes(response))
}

func TestValidateSharedPoolPauseChecksOperatorForStablePool(t *testing.T) {
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id: "100", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "dp-1",
	}}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Once()
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(newPauseValidationSuspendedClient(t, false), nil).Once()
	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, dataplaneManager: dataplaneManager,
		dagExecutor: dagExecutor, bizCfg: &config.BizConfig{},
	}

	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{"INVALID_LIFECYCLE_STATE"}, pauseValidationCodes(response))
}

func TestValidateSharedPoolPauseDetectsPendingOperatorSuspension(t *testing.T) {
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id: "100", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "dp-1",
	}}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Once()
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(newPauseValidationPendingSuspensionClient(t), nil).Once()
	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, dataplaneManager: dataplaneManager,
		dagExecutor: dagExecutor, bizCfg: &config.BizConfig{},
	}

	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{"INVALID_LIFECYCLE_STATE"}, pauseValidationCodes(response))
}

func TestValidateSharedPoolPauseChecksFlushWALWhileOperatorSuspensionIsPending(t *testing.T) {
	sharedPool := &entity.SharedPool{
		SharedPool:            apiv1.SharedPool{Id: "100", Phase: apiv1.SharedPool_PAUSING, DataplaneName: "dp-1"},
		PauseResumeState:      entity.PauseResumeStatePausing,
		SuspendComputeDesired: true,
	}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Once()
	operationRepo := repomocks.NewPauseResumeOperationRepo(t)
	operationRepo.On("LoadLatestBySharedPoolID", mock.Anything, "100").Return(&entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "failed-pause", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_FAILED,
	}, nil).Once()
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(newPauseValidationPendingSuspensionClient(t), nil).Once()
	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, pauseResumeOperationRepo: operationRepo,
		dataplaneManager: dataplaneManager, dagExecutor: dagExecutor,
		bizCfg: &config.BizConfig{DefaultTiBRConfig: config.TiBRConfig{DisableFlushWAL: true}},
	}

	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{"DEPENDENCY_UNAVAILABLE"}, pauseValidationCodes(response))
}

func TestValidateSharedPoolPauseContinuesIndependentFlushWALCheckWhenOperatorUnavailable(t *testing.T) {
	sharedPool := &entity.SharedPool{
		SharedPool:       apiv1.SharedPool{Id: "100", Phase: apiv1.SharedPool_PAUSING, DataplaneName: "dp-1"},
		PauseResumeState: entity.PauseResumeStatePausing,
	}
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(sharedPool, nil).Once()
	operationRepo := repomocks.NewPauseResumeOperationRepo(t)
	operationRepo.On("LoadLatestBySharedPoolID", mock.Anything, "100").Return(&entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "failed-pause", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_FAILED,
	}, nil).Once()
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(nil, errors.New("operator unavailable")).Once()
	s := &serverImpl{
		sharedPoolRepo: sharedPoolRepo, pauseResumeOperationRepo: operationRepo,
		dataplaneManager: dataplaneManager,
		dagExecutor:      dagExecutor,
		bizCfg:           &config.BizConfig{DefaultTiBRConfig: config.TiBRConfig{DisableFlushWAL: true}},
	}

	response, err := s.ValidateSharedPoolPause(context.Background(), "100")
	require.NoError(t, err)
	require.Equal(t, []string{"DEPENDENCY_UNAVAILABLE"}, pauseValidationCodes(response))
}

func pauseValidationCodes(response *entity.ValidateSharedPoolPauseResponse) []string {
	result := make([]string, 0, len(response.GetFailures()))
	for _, failure := range response.GetFailures() {
		result = append(result, failure.GetCode())
	}
	return result
}

func pauseValidationDataplaneDependencies(t *testing.T, remoteClient client.Client) (*repomocks.DataplaneRepo, *dataplanemocks.Manager) {
	t.Helper()
	repo := repomocks.NewDataplaneRepo(t)
	repo.On("LoadByName", mock.Anything, "dp-1").Return(&entity.DataPlaneProvider{Name: "dp-1", GatewayUrl: "gateway.test"}, nil).Once()
	manager := dataplanemocks.NewManager(t)
	manager.On("GetClient", "dp-1").Return(remoteClient, nil).Once()
	return repo, manager
}

func newPauseValidationTiBRClient(t *testing.T, includeTiBR, ready bool) client.Client {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, v1alpha1br.AddToScheme(scheme))
	require.NoError(t, operatorv1alpha1.AddToScheme(scheme))
	namespace := entity.GetDataPlaneNamespace("100")
	cluster := &operatorv1alpha1.Cluster{
		ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: namespace, Generation: 2},
		Status:     operatorv1alpha1.ClusterStatus{ObservedGeneration: 2},
	}
	builder := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster)
	if includeTiBR {
		conditions := []metav1.Condition(nil)
		if ready {
			conditions = []metav1.Condition{
				{Type: "Ready", Status: metav1.ConditionTrue},
				{Type: "Synced", Status: metav1.ConditionTrue},
			}
		}
		tibr := &v1alpha1br.TiBR{
			ObjectMeta: metav1.ObjectMeta{Name: tioperator.DefaultTiBRName, Namespace: namespace},
			Status:     v1alpha1br.TiBRStatus{CommonStatus: operatorv1alpha1.CommonStatus{Conditions: conditions}},
		}
		builder = builder.WithObjects(tibr)
	}
	return builder.Build()
}

func newPauseValidationSuspendedClient(t *testing.T, includeTiBR bool) client.Client {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, v1alpha1br.AddToScheme(scheme))
	require.NoError(t, operatorv1alpha1.AddToScheme(scheme))
	namespace := entity.GetDataPlaneNamespace("100")
	cluster := &operatorv1alpha1.Cluster{
		ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: namespace, Generation: 2},
		Spec:       operatorv1alpha1.ClusterSpec{SuspendAction: &operatorv1alpha1.SuspendAction{SuspendCompute: true}},
		Status: operatorv1alpha1.ClusterStatus{ObservedGeneration: 2, Conditions: []metav1.Condition{
			{Type: operatorv1alpha1.ClusterCondSuspended, Status: metav1.ConditionTrue, ObservedGeneration: 2},
		}},
	}
	builder := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster)
	if includeTiBR {
		builder = builder.WithObjects(&v1alpha1br.TiBR{
			ObjectMeta: metav1.ObjectMeta{Name: tioperator.DefaultTiBRName, Namespace: namespace},
			Status: v1alpha1br.TiBRStatus{CommonStatus: operatorv1alpha1.CommonStatus{Conditions: []metav1.Condition{
				{Type: "Ready", Status: metav1.ConditionTrue},
				{Type: "Synced", Status: metav1.ConditionTrue},
			}}},
		})
	}
	return builder.Build()
}

func newPauseValidationPendingSuspensionClient(t *testing.T) client.Client {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, operatorv1alpha1.AddToScheme(scheme))
	namespace := entity.GetDataPlaneNamespace("100")
	cluster := &operatorv1alpha1.Cluster{
		ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: namespace, Generation: 2},
		Spec:       operatorv1alpha1.ClusterSpec{SuspendAction: &operatorv1alpha1.SuspendAction{SuspendCompute: true}},
		Status:     operatorv1alpha1.ClusterStatus{ObservedGeneration: 1},
	}
	return fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build()
}