package backupv2_test

import (
	"context"
	"testing"
	"time"

	"github.com/AlekSi/pointer"
	dagentity "github.com/shiningrush/fastflow/pkg/entity"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/account/idl/pbgen/proto/account"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/config"
	cluster_job_mock "github.com/tidbcloud/dataflow-service/app/internal/cluster_job/mock"
	feature_gate_mock "github.com/tidbcloud/dataflow-service/app/internal/feature_gate/mock"
	"github.com/tidbcloud/dataflow-service/app/models/backup"
	"github.com/tidbcloud/dataflow-service/app/models/backup/impl"
	backup_model_mock "github.com/tidbcloud/dataflow-service/app/models/backup/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	workflow_mock "github.com/tidbcloud/dataflow-service/app/workflow/mock"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	internal_config "github.com/tidbcloud/dataflow-service/internal/config"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	dedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"gorm.io/gorm"
)

type snapshotBackupTestSuite struct {
	suite.Suite
	db *gorm.DB
	m  backup.SnapshotBackupModel

	mockBucketRepo         *repo_mock.BucketRepo
	mockBackupRepo         *repo_mock.BackupRepo
	mockRestoreRepo        *repo_mock.RestoreRepo
	mockBackupExportRepo   *repo_mock.BackupExportRepo
	mockClusterJob         *cluster_job_mock.Model
	mockCoreGrpcClient     *grpcli_mock.CoreGrpcClient
	mockMgmtGrpcClient     *grpcli_mock.MgmtSvcGrpcClient
	mockBackupScheduleRepo *repo_mock.BackupScheduleRepo
	mockWorkflow           *workflow_mock.Workflow
	mockImportCli          *grpcli_mock.ImportGrpcClient
	mockAccountCli         *grpcli_mock.AccountSvcGrpcClient

	mockBackupSettingM *backup_model_mock.BackupSettingModel
	mockFeatureGate    *feature_gate_mock.FeatureGate

	mockedUserID        uint64
	allMocks            []*mock.Call
	mockedCluster       *mgmt.Cluster
	mockedBackupSetting *entity.BackupSetting
}

func TestSnapshotBackupTestSuite(t *testing.T) {
	suite.Run(t, new(snapshotBackupTestSuite))
}

func (s *snapshotBackupTestSuite) SetupSuite() {
	s.mockBackupRepo = repo_mock.NewBackupRepo(s.T())
	s.mockRestoreRepo = repo_mock.NewRestoreRepo(s.T())
	s.mockBackupExportRepo = repo_mock.NewBackupExportRepo(s.T())
	s.mockBackupSettingM = backup_model_mock.NewBackupSettingModel(s.T())
	s.mockClusterJob = cluster_job_mock.NewModel(s.T())
	s.mockCoreGrpcClient = grpcli_mock.NewCoreGrpcClient(s.T())
	s.mockMgmtGrpcClient = grpcli_mock.NewMgmtSvcGrpcClient(s.T())
	s.mockBackupScheduleRepo = repo_mock.NewBackupScheduleRepo(s.T())
	s.mockWorkflow = workflow_mock.NewWorkflow(s.T())
	s.mockImportCli = grpcli_mock.NewImportGrpcClient(s.T())
	s.mockAccountCli = grpcli_mock.NewAccountSvcGrpcClient(s.T())
	s.mockFeatureGate = feature_gate_mock.NewFeatureGate(s.T())

	s.mockedUserID = 1
	s.mockedCluster = &mgmt.Cluster{
		ClusterId: 1, ProjectId: 1, OrganizationId: 1,
		DeployType: mgmt.DeployType_DEPLOY_TYPE_DEDICATED,
		ClusterInfo: &mgmt.Cluster_DedicatedCluster{
			DedicatedCluster: &dedicatedv1.Cluster{State: dedicatedv1.Cluster_AVAILABLE},
		},
	}
	s.mockedBackupSetting = &entity.BackupSetting{
		Status: entity.BackupSettingStatusNormal,
		Bucket: &entity.BucketSetting{
			Prefix:  pointer.ToString("prefix"),
			Buckets: []entity.BackupSettingBucket{{BucketId: 1, Master: true}},
		},
	}

	testDB, err := repo.GetFakeDB()
	require.NoError(s.T(), err)
	s.db = testDB
	trans := db.NewProvider(s.db)

	model := impl.NewSnapshotBackupModel(
		trans,
		s.mockClusterJob,
		s.mockCoreGrpcClient,
		s.mockMgmtGrpcClient,
		s.mockImportCli,
		s.mockAccountCli,
		s.mockWorkflow,
		s.mockBucketRepo,
		s.mockBackupRepo,
		s.mockBackupScheduleRepo,
		s.mockRestoreRepo,
		s.mockBackupExportRepo,
		s.mockBackupSettingM,
		s.mockFeatureGate,
	)
	s.m = model
}

func (s *snapshotBackupTestSuite) SetupTest() {
	// mock all to make suer async create backup success
	s.allMocks = s.mockAll()
}

func (s *snapshotBackupTestSuite) TearDownTest() {
	for _, m := range s.allMocks {
		m.Unset()
	}
}

func (s *snapshotBackupTestSuite) mockAll() []*mock.Call {
	dagInstance := &dagentity.DagInstance{BaseInfo: dagentity.BaseInfo{ID: "fake"}}
	backupSettingResp := &domain.QueryBackupSettingRespV2{BackupSetting: s.mockedBackupSetting}
	masterBucketResp := &domain.QueryMasterBucketRespV2{BucketID: pointer.ToUint64(1), BucketURI: pointer.ToString("url")}

	mocks := []*mock.Call{
		s.mockBackupRepo.On("Save", mock.Anything, mock.Anything).Return(nil),
		s.mockBackupScheduleRepo.On("Create", mock.Anything, mock.Anything).Return(nil),
		s.mockBackupScheduleRepo.On("SoftDelete", mock.Anything, mock.Anything).Return(nil),
		s.mockBackupRepo.On("Count", mock.Anything, mock.Anything).Return(int64(0), nil),

		s.mockBackupScheduleRepo.On("Count", mock.Anything, mock.Anything).Return(int64(0), nil),
		s.mockBackupScheduleRepo.On("Load", mock.Anything, mock.Anything).Return(&entity.BackupSchedule{}, nil),

		s.mockCoreGrpcClient.On("CreateClusterSnapshot", mock.Anything, mock.Anything).Return(uint64(1), nil),
		s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(s.mockedCluster, nil),

		s.mockClusterJob.On("RegisterBackupClusterJob", mock.Anything, mock.Anything, mock.Anything).Return(nil),
		s.mockWorkflow.On("RunDagWithTags", mock.Anything, mock.Anything, mock.Anything).Return(dagInstance, nil),
		s.mockImportCli.On("List", mock.Anything, mock.Anything, mock.Anything).Return(&dataflow.ListImportResp{}, nil),

		s.mockBackupSettingM.On("QueryBackupSetting", mock.Anything, mock.Anything).Return(backupSettingResp, nil),
		s.mockBackupSettingM.On("QueryMasterBucket", mock.Anything, mock.Anything).Return(masterBucketResp, nil),

		s.mockFeatureGate.On("IsFeatureEnabled", mock.Anything, mock.Anything).Return(false),
	}
	return mocks
}

func (s *snapshotBackupTestSuite) makeMockOptional() {
	for _, m := range s.allMocks {
		m.Maybe()
	}
}

func (s *snapshotBackupTestSuite) TestAsyncCreateSuccess() {
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		Name: "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}
	// test auto backup
	req.BackupType = dataflow.BackupType_AUTO
	req.AutoBackupScheduleID = pointer.ToUint64(1)
	_, err := s.m.AsyncCreate(ctx, req)
	s.NoError(err)

	// test manual backup
	req.BackupType = dataflow.BackupType_MANUAL
	req.AutoBackupScheduleID = nil
	_, err = s.m.AsyncCreate(ctx, req)
	s.NoError(err)
}

func (s *snapshotBackupTestSuite) TestAsyncCreate_ValidateReq() {
	s.makeMockOptional()
	ctx := context.Background()
	req := &domain.AsyncCreateBackupReq{
		ClusterID: s.mockedCluster.ClusterId, ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	// backup name is empty
	_, err := s.m.AsyncCreate(ctx, req)
	s.Equal(bizerr.ErrCodeParams, err.ErrCode())
	s.ErrorContains(err, "backup name is empty")

	// backup type is invalid
	req.Name = "test"
	req.BackupType = dataflow.BackupType_ALL
	_, err = s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster id")

	// auto backup schedule id is empty
	req.BackupType = dataflow.BackupType_AUTO
	_, err = s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster id")
}

func (s *snapshotBackupTestSuite) TestAsyncCreatePreCheck_BackupSetting() {
	s.makeMockOptional()
	// unset backup setting mocks
	for len(s.mockBackupSettingM.ExpectedCalls) > 0 {
		m := s.mockBackupSettingM.ExpectedCalls[0]
		m.Unset()
		// remove this mock from allMocks
		for idx, m2 := range s.allMocks {
			if m2 == m {
				s.allMocks = append(s.allMocks[:idx], s.allMocks[idx+1:]...)
				break
			}
		}
	}

	ctx := context.Background()
	mockErr := bizerr_wrapper.NewInternalErr("db error")
	req := &domain.AsyncCreateBackupReq{
		BackupType: dataflow.BackupType_MANUAL,
		Name:       "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	// error on get backup setting
	m := s.mockBackupSettingM.On("QueryBackupSetting", mock.Anything, mock.Anything).Return(nil, mockErr)
	_, err := s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster")
	m.Unset()

	// error on check backup setting status
	setting := *s.mockedBackupSetting // do a copy
	setting.Status = entity.BackupSettingStatusDeleted
	backupSettingResp := &domain.QueryBackupSettingRespV2{BackupSetting: &setting}
	m = s.mockBackupSettingM.On("QueryBackupSetting", mock.Anything, mock.Anything).Return(backupSettingResp, nil)
	_, err = s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster")
	m.Unset()
}

func (s *snapshotBackupTestSuite) TestAsyncCreatePreCheck_DuplicateName() {
	s.makeMockOptional()
	// unset backup schedule mocks
	for len(s.mockBackupScheduleRepo.ExpectedCalls) > 0 {
		m := s.mockBackupScheduleRepo.ExpectedCalls[0]
		m.Unset()
		// remove this mock from allMocks
		for idx, m2 := range s.allMocks {
			if m2 == m {
				s.allMocks = append(s.allMocks[:idx], s.allMocks[idx+1:]...)
				break
			}
		}
	}

	ctx := context.Background()
	mockErr := bizerr_wrapper.NewInternalErr("db error")
	req := &domain.AsyncCreateBackupReq{
		BackupType: dataflow.BackupType_MANUAL,
		Name:       "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	cnt := int64(1)
	m := s.mockBackupScheduleRepo.On("Count", mock.Anything, mock.Anything).Return(cnt, mockErr)
	_, err := s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster")
	m.Unset()

	// duplicate name
	m = s.mockBackupScheduleRepo.On("Count", mock.Anything, mock.Anything).Return(cnt, nil)
	_, err = s.m.AsyncCreate(ctx, req)
	s.Equal(config.ErrCodeAsyncCreateBackupPreCheckDuplicateName, err.ErrCode())
	m.Unset()
}

func (s *snapshotBackupTestSuite) TestAsyncCreatePreCheck_RunningBackups() {
	s.makeMockOptional()
	// unset backup mocks
	for len(s.mockBackupRepo.ExpectedCalls) > 0 {
		m := s.mockBackupRepo.ExpectedCalls[0]
		m.Unset()
		// remove this mock from allMocks
		for idx, m2 := range s.allMocks {
			if m2 == m {
				s.allMocks = append(s.allMocks[:idx], s.allMocks[idx+1:]...)
				break
			}
		}
	}

	ctx := context.Background()
	mockErr := bizerr_wrapper.NewInternalErr("db error")
	req := &domain.AsyncCreateBackupReq{
		BackupType: dataflow.BackupType_MANUAL,
		Name:       "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	cnt := int64(1)
	m := s.mockBackupRepo.On("Count", mock.Anything, mock.Anything).Return(cnt, mockErr)
	_, err := s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster")
	m.Unset()

	// duplicate name
	m = s.mockBackupRepo.On("Count", mock.Anything, mock.Anything).Return(cnt, nil)
	_, err = s.m.AsyncCreate(ctx, req)
	s.Equal(config.ErrCodeAsyncCreateBackupPreCheckRunningBackups, err.ErrCode())
	m.Unset()
}

func (s *snapshotBackupTestSuite) TestAsyncCreatePreCheck_Cluster() {
	s.makeMockOptional()
	// unset cluster mocks
	for len(s.mockMgmtGrpcClient.ExpectedCalls) > 0 {
		m := s.mockMgmtGrpcClient.ExpectedCalls[0]
		m.Unset()
		// remove this mock from allMocks
		for idx, m2 := range s.allMocks {
			if m2 == m {
				s.allMocks = append(s.allMocks[:idx], s.allMocks[idx+1:]...)
				break
			}
		}
	}

	ctx := context.Background()
	mockErr := bizerr_wrapper.NewInternalErr("db error")
	req := &domain.AsyncCreateBackupReq{
		BackupType: dataflow.BackupType_MANUAL,
		Name:       "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	cluster := s.mockedCluster
	// cluster not found
	m := s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(cluster, mockErr)
	_, err := s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster")
	m.Unset()

	// cluster is paused
	cluster.GetDedicatedCluster().State = dedicatedv1.Cluster_PAUSED
	m = s.mockMgmtGrpcClient.On("GetAnyDedicatedClusterBasicByID", mock.Anything, mock.Anything).Return(cluster, nil)
	_, err = s.m.AsyncCreate(ctx, req)
	s.Equal(config.ErrCodeAsyncCreateBackupPreCheckClusterNotNormal, err.ErrCode())
	m.Unset()

	// restore cluster state for subsequent tests
	cluster.GetDedicatedCluster().State = dedicatedv1.Cluster_AVAILABLE
}

func (s *snapshotBackupTestSuite) TestAsyncCreatePreCheck_RunningImports() {
	s.makeMockOptional()
	// unset import mocks
	for len(s.mockImportCli.ExpectedCalls) > 0 {
		m := s.mockImportCli.ExpectedCalls[0]
		m.Unset()
		// remove this mock from allMocks
		for idx, m2 := range s.allMocks {
			if m2 == m {
				s.allMocks = append(s.allMocks[:idx], s.allMocks[idx+1:]...)
				break
			}
		}
	}

	importResp := &dataflow.ListImportResp{}

	ctx := context.Background()
	mockErr := bizerr_wrapper.NewInternalErr("error")
	req := &domain.AsyncCreateBackupReq{
		BackupType: dataflow.BackupType_MANUAL,
		Name:       "test", ClusterID: s.mockedCluster.ClusterId,
		ProjectID: s.mockedCluster.ProjectId, UserID: s.mockedUserID,
	}

	// rpc err
	m := s.mockImportCli.On("List", mock.Anything, mock.Anything, mock.Anything).Return(importResp, mockErr)
	_, err := s.m.AsyncCreate(ctx, req)
	s.ErrorContains(err, "Temporarily internal error for cluster")
	m.Unset()

	// running imports
	importResp.Imports = []*dataflow.ImportGetResp{{ClusterId: s.mockedCluster.ClusterId}}
	m = s.mockImportCli.On("List", mock.Anything, mock.Anything, mock.Anything).Return(importResp, nil)
	_, err = s.m.AsyncCreate(ctx, req)
	s.Equal(config.ErrCodeAsyncCreateBackupPreCheckRunningImports, err.ErrCode())
	m.Unset()
}

//nolint:funlen
func (s *snapshotBackupTestSuite) TestAsyncDeleteSuccess() {
	ctx := context.Background()

	// disable global backup repo mock
	for len(s.mockBackupRepo.ExpectedCalls) > 0 {
		m := s.mockBackupRepo.ExpectedCalls[0]
		m.Unset()
		// remove this mock from allMocks
		for idx, m2 := range s.allMocks {
			if m2 == m {
				s.allMocks = append(s.allMocks[:idx], s.allMocks[idx+1:]...)
				break
			}
		}
	}

	// disable global backup schedule repo mock
	for len(s.mockBackupScheduleRepo.ExpectedCalls) > 0 {
		m := s.mockBackupScheduleRepo.ExpectedCalls[0]
		m.Unset()
		// remove this mock from allMocks
		for idx, m2 := range s.allMocks {
			if m2 == m {
				s.allMocks = append(s.allMocks[:idx], s.allMocks[idx+1:]...)
				break
			}
		}
	}

	mockedBackupID := entity.ID(1)
	mockedBackupSchedule := &entity.BackupSchedule{Name: internal_config.FinalBackupName, Type: entity.BackupTypeManual}
	mockedBackup := &entity.Backup{Status: entity.BackupStatusSuccess}
	mockedBackup.BackupSchedule = mockedBackupSchedule

	req := &domain.AsyncDeleteBackupReq{BackupID: mockedBackupID.Uint64()}

	m := s.mockBackupRepo.On("Load", mock.Anything, mock.Anything).Return(mockedBackup, nil)
	defer m.Unset()

	// can not delete running final backup
	{
		mockedBackup.DeletedAt = nil
		mockedBackup.Status = entity.BackupStatusRunning
		_, err := s.m.AsyncDelete(ctx, req)
		s.Equal(err.ErrCode(),
			bizerr_wrapper.NewInvalidInputError(config.ErrMsgCanNotDeleteFinalBackup, mockedBackup.ClusterID).ErrCode())
	}

	// can not delete when a restore is running
	{
		mockedBackup.Status = entity.BackupStatusSuccess

		m1 := s.mockBackupExportRepo.On("CountByFilter", mock.Anything, mock.Anything).Return(int64(0), nil)
		m2 := s.mockRestoreRepo.On("Count", mock.Anything, mock.Anything).Return(int64(1), nil)
		_, err := s.m.AsyncDelete(ctx, req)
		s.Equal(err.ErrCode(),
			bizerr_wrapper.NewInvalidInputError(
				config.ErrMsgAsyncDeleteBackupPreCheckRunningRestore, mockedBackup.ClusterID).ErrCode())

		m1.Unset()
		m2.Unset()
	}

	// can not delete when a export is running
	{
		mockedBackup.Status = entity.BackupStatusSuccess

		m1 := s.mockBackupExportRepo.On("CountByFilter", mock.Anything, mock.Anything).Return(int64(1), nil)
		_, err := s.m.AsyncDelete(ctx, req)
		s.Equal(err.ErrCode(),
			bizerr_wrapper.NewInvalidInputError(
				config.ErrMsgAsyncDeleteBackupPreCheckRunningExport, mockedBackup.ClusterID).ErrCode())

		m1.Unset()
	}

	// success
	{
		m1 := s.mockBackupRepo.On("Save", mock.Anything, mock.Anything).Return(nil)
		m2 := s.mockBackupScheduleRepo.On("SoftDelete", mock.Anything, mock.Anything).
			Run(func(args mock.Arguments) {
				mockedBackupSchedule.DeletedAt = pointer.ToTime(time.Now())
			}).
			Return(nil)
		m3 := s.mockBackupExportRepo.On("CountByFilter", mock.Anything, mock.Anything).Return(int64(0), nil)
		m4 := s.mockRestoreRepo.On("Count", mock.Anything, mock.Anything).Return(int64(0), nil)

		mockedBackup.Status = entity.BackupStatusSuccess
		resp, err := s.m.AsyncDelete(ctx, req)
		s.NoError(err)

		// backup deletedAt should be set
		s.NotNil(mockedBackup.DeletedAt)

		// manual backup's backup schedule should be deleted
		s.mockBackupScheduleRepo.AssertCalled(s.T(), "SoftDelete", mock.Anything, mock.Anything)
		s.NotNil(mockedBackupSchedule.DeletedAt)

		// workflow should be called
		s.mockWorkflow.AssertCalled(s.T(), "RunDagWithTags", mock.Anything, mock.Anything, mock.Anything)
		s.NotEmpty(resp.WorkflowID)

		m1.Unset()
		m2.Unset()
		m3.Unset()
		m4.Unset()
	}

	// delete again success
	{
		mockedBackup.DeletedAt = pointer.ToTime(time.Now())
		resp, err := s.m.AsyncDelete(ctx, req)
		s.NoError(err)
		// no new workflow should be created
		s.Empty(resp.WorkflowID)
	}
}

//nolint:funlen
func (s *snapshotBackupTestSuite) TestListRecycleBinClusters() {
	s.makeMockOptional()
	testcases := []struct {
		name                string
		projectID           uint64
		tenantID            uint64
		requestAccount      bool
		mockProject         *account.GetProjectByIDResponse
		preCheckPass        bool
		mockClusterIDs      []uint64
		mockTotal           int64
		mockListClusterResp *core.ListClustersResp
		expectedClusterNum  int
	}{
		{
			name:      "project id is empty",
			projectID: 0,
			tenantID:  1,
		},
		{
			name:      "tenant id is empty",
			projectID: 1,
			tenantID:  0,
		},
		{
			name:           "tenant id and project id are not match",
			projectID:      1,
			tenantID:       1,
			requestAccount: true,
			mockProject: &account.GetProjectByIDResponse{
				Project: &account.Project{
					Id:       1,
					TenantId: 2,
				},
			},
		},
		{
			name:           "cluster id list is empty",
			projectID:      1,
			tenantID:       1,
			requestAccount: true,
			mockProject: &account.GetProjectByIDResponse{
				Project: &account.Project{
					Id:       1,
					TenantId: 1,
				},
			},
			preCheckPass:        true,
			mockClusterIDs:      []uint64{},
			mockTotal:           0,
			mockListClusterResp: &core.ListClustersResp{},
			expectedClusterNum:  0,
		},
		{
			name:           "contains not deleted cluster",
			projectID:      1,
			tenantID:       1,
			requestAccount: true,
			mockProject: &account.GetProjectByIDResponse{
				Project: &account.Project{
					Id:       1,
					TenantId: 1,
				},
			},
			preCheckPass:   true,
			mockClusterIDs: []uint64{1, 2, 3},
			mockTotal:      0,
			mockListClusterResp: &core.ListClustersResp{
				OrderedClusterIds: []uint64{1, 2},
				Clusters: map[uint64]*core.Cluster{
					1: {
						Id:        1,
						DeletedAt: pointer.ToInt64(time.Now().Add(-2 * time.Hour).Unix()),
					},
					2: {
						Id:        2,
						DeletedAt: pointer.ToInt64(time.Now().Add(-1 * time.Hour).Unix()),
					},
				},
			},
			expectedClusterNum: 2,
		},
		{
			name:           "all clusters are deleted",
			projectID:      1,
			tenantID:       1,
			requestAccount: true,
			mockProject: &account.GetProjectByIDResponse{
				Project: &account.Project{
					Id:       1,
					TenantId: 1,
				},
			},
			preCheckPass:   true,
			mockClusterIDs: []uint64{1, 2, 3},
			mockTotal:      0,
			mockListClusterResp: &core.ListClustersResp{
				OrderedClusterIds: []uint64{1, 2, 3},
				Clusters: map[uint64]*core.Cluster{
					1: {
						Id:        1,
						DeletedAt: pointer.ToInt64(time.Now().Add(-2 * time.Hour).Unix()),
					},
					2: {
						Id:        2,
						DeletedAt: pointer.ToInt64(time.Now().Add(-1 * time.Hour).Unix()),
					},
					3: {
						Id:        3,
						DeletedAt: pointer.ToInt64(time.Now().Unix()),
					},
				},
			},
			expectedClusterNum: 3,
		},
	}

	for _, tc := range testcases {
		s.Run(tc.name, func() {
			ctx := context.Background()
			req := &domain.ListRecycleBinClustersReq{
				TenantID:  tc.tenantID,
				ProjectID: tc.projectID,
				Pager: &domain.Pager{
					Page:     1,
					PageSize: 10,
				},
			}

			if tc.requestAccount {
				call := s.mockAccountCli.On("GetProjectByID", mock.Anything, &account.GetProjectByIDRequest{
					Id: tc.projectID,
				}).Return(tc.mockProject, nil)
				defer call.Unset()
			}
			if tc.preCheckPass {
				call1 := s.mockBackupRepo.On("ListClustersWithBackup", mock.Anything, tc.projectID).
					Return(tc.mockClusterIDs, nil)
				defer call1.Unset()
				if len(tc.mockClusterIDs) > 0 {
					call2 := s.mockCoreGrpcClient.On("CountClusters", mock.Anything, mock.Anything).
						Return(tc.mockTotal, nil)
					defer call2.Unset()
					call3 := s.mockCoreGrpcClient.On("ListClusters", mock.Anything, mock.Anything).
						Return(tc.mockListClusterResp, nil)
					defer call3.Unset()
				}
			}

			resp, err := s.m.ListRecycleBinClusters(ctx, req)
			if !tc.preCheckPass {
				s.Error(err)
			} else {
				s.Equal(tc.expectedClusterNum, len(resp.Items))
			}
			s.mockBackupRepo.AssertExpectations(s.T())
			s.mockAccountCli.AssertExpectations(s.T())
			s.mockCoreGrpcClient.AssertExpectations(s.T())
		})
	}
}